AI 大模型开发全栈实战:从本地部署到 Agent 工具链搭建(附完整代码)
AI 大模型开发全栈实战:从本地部署到 Agent 工具链搭建(附完整代码)
🔥 2026 年,AI 开发不再是算法工程师的专利。 本文从零搭建 AI 开发环境,手把手带你完成大模型本地部署、MCP 工具协议实战、RAG 知识库搭建三大核心环节,全部代码可直接运行。
关键词:AI 开发、大模型部署、MCP 协议、RAG 检索增强、Python、Ollama、vLLM
技术栈:Python 3.11 + PyTorch 2.x + Ollama + vLLM + LangChain + ChromaDB
阅读时长:约 25 分钟 | 难度:⭐⭐⭐ 中高级
一、2026 年 AI 开发的三个核心问题
很多人入门 AI 开发时会遇到三个灵魂拷问:
表格
问题 痛点 本文解决方案
模型跑不起来 环境配置复杂、CUDA 版本冲突 Ollama 一行命令启动 + vLLM 生产级部署
模型不会用工具 每个大模型工具调用格式不同 MCP 统一协议,一次编写适配所有模型
模型胡说八道 幻觉问题、知识过时 RAG 检索增强生成,基于真实数据回答
本文就是围绕这三个问题,从零到一搭建一套完整的 AI 开发工具链。
二、环境搭建:Python + PyTorch + CUDA
2.1 基础环境
bash
1
2
3
4
5
6
7
创建独立虚拟环境(避免依赖冲突)
python -m venv ai-dev-env
source ai-dev-env/bin/activate # Windows: ai-dev-env\Scripts\activate
升级 pip
pip install --upgrade pip
2.2 安装 PyTorch(CUDA 加速)
bash
1
2
3
4
5
6
查看当前 CUDA 版本
nvidia-smi
安装 PyTorch(以 CUDA 11.8 为例)
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118
验证安装:
python
1
2
3
4
5
import torch
print(f"PyTorch 版本: {torch.version}“)
print(f"CUDA 可用: {torch.cuda.is_available()}”)
print(f"GPU 设备: {torch.cuda.get_device_name(0) if torch.cuda.is_available() else ‘无’}")
2.3 安装 AI 开发核心依赖
bash
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
大模型推理
pip install ollama vllm
AI Agent 框架
pip install langchain langchain-openai langchain-community
向量数据库
pip install chromadb sentence-transformers
MCP 协议
pip install mcp httpx
工具库
pip install fastapi uvicorn python-dotenv
2.4 硬件配置参考
表格
模型规模 最低显存 推荐显卡 适用场景
7B 量化 6GB RTX 3060/4060 个人开发、原型验证
13B 量化 12GB RTX 4070 Ti 小型团队、内部工具
7B 全精度 16GB RTX 4080/4090 高质量推理
70B 量化 48GB 2×A100/A6000 生产级服务
三、大模型本地部署:Ollama vs vLLM 双方案
本地部署大模型有两条主流路线,不是二选一,而是分层使用。
表格
对比维度 Ollama vLLM
定位 本地实验与原型开发 生产级高吞吐量部署
安装难度 ⭐ 极简,一行命令 ⭐⭐⭐ 需配置 CUDA 环境
并发能力 单节点 ≤32 请求 连续批处理,数千级并发
内存管理 标准分配 PagedAttention 动态调度
显存效率 ~60% 利用率 ~96% 利用率(浪费率<4%)
适用场景 个人开发、POC 验证 上线服务、高并发 API
3.1 方案一:Ollama——5 分钟跑起来第一个模型
bash
1
2
3
4
5
6
7
8
9
安装 Ollama(Linux/macOS)
curl -fsSL https://ollama.ai/install.sh | sh
拉取 Qwen2.5-7B 模型(约 4.7GB)
ollama pull qwen2.5:7b
启动交互式对话
ollama run qwen2.5:7b
Python API 调用:
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
import requests
import json
def chat_with_ollama(prompt: str, model: str = “qwen2.5:7b”) -> str:
“”“通过 Ollama API 与大模型对话”“”
response = requests.post(
“http://localhost:11434/api/generate”,
json={
“model”: model,
“prompt”: prompt,
“stream”: False
}
)
return response.json()[“response”]
测试
result = chat_with_ollama(“用 Python 写一个快速排序算法”)
print(result)
OpenAI 兼容接口(推荐):
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from openai import OpenAI
Ollama 提供 OpenAI 兼容 API,无缝切换
client = OpenAI(
base_url=“http://localhost:11434/v1”,
api_key=“ollama” # 随意填写,Ollama 不校验
)
response = client.chat.completions.create(
model=“qwen2.5:7b”,
messages=[
{“role”: “system”, “content”: “你是一个 Python 技术专家”},
{“role”: “user”, “content”: “解释一下 Python 的 GIL 是什么?”}
],
temperature=0.7
)
print(response.choices[0].message.content)
3.2 方案二:vLLM——生产级高并发部署
bash
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
安装 vLLM
pip install vllm
启动 API 服务(单卡)
python -m vllm.entrypoints.openai.api_server
–model Qwen/Qwen2.5-7B-Instruct
–host 0.0.0.0
–port 8000
多卡并行(2×GPU)
python -m vllm.entrypoints.openai.api_server
–model Qwen/Qwen2.5-14B-Instruct
–tensor-parallel-size 2
–gpu-memory-utilization 0.9
Python 调用:
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from openai import OpenAI
vLLM 同样兼容 OpenAI API
client = OpenAI(
base_url=“http://localhost:8000/v1”,
api_key=“EMPTY”
)
response = client.chat.completions.create(
model=“Qwen/Qwen2.5-7B-Instruct”,
messages=[
{“role”: “user”, “content”: “什么是 Transformer 架构?”}
],
temperature=0.7,
max_tokens=1024
)
print(response.choices[0].message.content)
性能对比实测(RTX 4090,Qwen2.5-7B):
表格
指标 Ollama vLLM 提升幅度
单请求延迟 2.3s 1.1s 52%
100 并发吞吐 350 tokens/s 1200 tokens/s 243%
显存占用 14.2GB 8.1GB 43%
首 Token 延迟 800ms 280ms 65%
💡 经验之谈:Ollama 用于日常开发和验证,vLLM 用于正式上线。我的做法是——开发阶段用 Ollama 快速迭代,上线切 vLLM,代码几乎不用改,因为都是 OpenAI 兼容接口。
四、MCP 协议实战:让 AI Agent 拥有"手脚"
4.1 为什么需要 MCP?
每个大模型的工具调用方式都不一样:OpenAI 是 Function Calling,Claude 是 Tool Use,国产模型各有各的格式。接三个模型要写三套适配层——MCP(Model Context Protocol)就是来解决这个问题的。
plaintext
1
2
3
4
5
以前:你写 REST API 给浏览器调
现在:你写 MCP Server 给 AI 调
调用方变了,"定义接口 → 处理请求 → 返回结果"的逻辑不变。
MCP 的核心架构:
plaintext
1
2
3
4
5
6
7
8
9
10
11
12
13
┌──────────────────┐ ┌──────────────────┐
│ MCP Client │ JSON-RPC 2.0 │ MCP Server │
│ │ ◄──────────────────► │ │
│ Claude Desktop │ stdio / HTTP/SSE │ 你写的 Python │
│ Cursor / Cline │ │ 服务 │
└──────────────────┘ └─────────┬────────┘
│
▼
┌─────────────────┐
│ 你的业务 API │
│ 数据库/搜索引擎 │
└─────────────────┘
4.2 从零写一个 MCP Server
环境准备:
bash
1
2
pip install mcp httpx
核心代码——天气查询 MCP Server:
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
weather_mcp_server.py
from mcp.server.fastmcp import FastMCP
import httpx
创建 MCP Server 实例
mcp = FastMCP(“weather-service”)
@mcp.tool()
async def get_weather(city: str) -> str:
“”"
查询指定城市的实时天气信息
返回温度、湿度、风向和天气概况
Args:
city: 城市名称,如"北京"、"上海"
"""
# 这里替换为真实的天气 API
async with httpx.AsyncClient() as client:
resp = await client.get(
f"https://api.weather.example.com/current",
params={"city": city}
)
data = resp.json()
return (
f"{city}天气:{data['condition']},"
f"温度 {data['temp']}°C,"
f"湿度 {data['humidity']}%,"
f"{data['wind_dir']} {data['wind_level']}级"
)
@mcp.tool()
async def get_forecast(city: str, days: int = 3) -> str:
“”"
查询城市未来 N 天的天气预报
Args:
city: 城市名称
days: 预报天数,默认 3 天
"""
# 模拟数据(实际接入真实 API)
forecasts = [
{"date": f"第{i+1}天", "temp": f"{25-i}°C", "weather": "晴转多云"}
for i in range(days)
]
result = "\n".join(
f" {f['date']}:{f['weather']},{f['temp']}"
for f in forecasts
)
return f"{city} 未来 {days} 天预报:\n{result}"
启动服务
if name == “main”:
mcp.run(transport=“stdio”)
接入 Claude Desktop:
在 claude_desktop_config.json 中添加:
json
1
2
3
4
5
6
7
8
9
{
“mcpServers”: {
“weather”: {
“command”: “python”,
“args”: [“/path/to/weather_mcp_server.py”]
}
}
}
重启 Claude Desktop,AI 就能自动发现并调用你的天气工具了。
4.3 进阶:数据库查询 MCP Server
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
database_mcp_server.py
from mcp.server.fastmcp import FastMCP
import sqlite3
import json
mcp = FastMCP(“database-assistant”)
初始化示例数据库
def init_db():
conn = sqlite3.connect(“demo.db”)
conn.execute(“”"
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
price REAL,
stock INTEGER,
category TEXT
)
“”")
# 插入示例数据
products = [
(1, “AI 开发实战课”, 299.0, 150, “课程”),
(2, “Python 从入门到精通”, 89.0, 300, “图书”),
(3, “GPU 云计算套餐”, 1999.0, 50, “服务”),
]
conn.executemany(
“INSERT OR REPLACE INTO products VALUES (?,?,?,?,?)”,
products
)
conn.commit()
conn.close()
@mcp.tool()
def query_products(sql_filter: str = “”) -> str:
“”"
查询产品信息,支持 SQL WHERE 条件过滤
Args:
sql_filter: SQL WHERE 子句,如 "price > 100 AND category = '课程'"
"""
conn = sqlite3.connect("demo.db")
query = "SELECT * FROM products"
if sql_filter:
query += f" WHERE {sql_filter}"
cursor = conn.execute(query)
columns = [desc[0] for desc in cursor.description]
rows = [dict(zip(columns, row)) for row in cursor.fetchall()]
conn.close()
return json.dumps(rows, ensure_ascii=False, indent=2)
@mcp.tool()
def get_product_stats() -> str:
“”“获取产品统计信息:总数、均价、各分类数量”“”
conn = sqlite3.connect(“demo.db”)
stats = conn.execute("SELECT COUNT(*), AVG(price) FROM products").fetchone()
categories = conn.execute(
"SELECT category, COUNT(*) FROM products GROUP BY category"
).fetchall()
conn.close()
result = {
"总产品数": stats[0],
"平均价格": round(stats[1], 2),
"分类统计": {cat: count for cat, count in categories}
}
return json.dumps(result, ensure_ascii=False, indent=2)
if name == “main”:
init_db()
mcp.run(transport=“stdio”)
💡 关键点:MCP Server 的 docstring 不是普通注释——它是 AI 判断"什么时候调用这个工具"的依据。写得越清晰,AI 调用的准确率越高。
五、RAG 知识库搭建:让 AI 不再胡说八道
5.1 RAG 的核心思路
plaintext
1
2
3
传统 LLM:用户提问 → LLM 直接回答(可能幻觉)
RAG: 用户提问 → 检索相关知识 → 注入 Prompt → LLM 基于事实回答(可溯源)
表格
痛点 RAG 如何解决
知识过时 实时检索外部知识库
幻觉问题 基于真实文档生成,可溯源
领域知识缺失 注入企业私有数据
5.2 完整 RAG 系统代码
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
“”"
RAG 知识库问答系统
技术栈:LangChain + ChromaDB + Sentence-BERT
“”"
from langchain.document_loaders import TextLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain_openai import ChatOpenAI
import os
class RAGSystem:
“”“RAG 知识库问答系统”“”
def __init__(self, docs_dir: str = "./knowledge_base"):
self.docs_dir = docs_dir
self.persist_dir = "./chroma_db"
self.embeddings = HuggingFaceEmbeddings(
model_name="all-MiniLM-L6-v2", # 轻量级嵌入模型
model_kwargs={"device": "cpu"}
)
def build_knowledge_base(self):
"""构建知识库:加载文档 → 分块 → 向量化 → 入库"""
# 1. 加载文档(支持 txt、md 等)
loader = DirectoryLoader(
self.docs_dir,
glob=" **/*.txt",
loader_cls=TextLoader,
loader_kwargs={"encoding": "utf-8"}
)
documents = loader.load()
print(f"📄 加载了 {len(documents)} 个文档")
# 2. 文本分块(关键:合理的 chunk 大小)
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每块 500 字符
chunk_overlap=50, # 重叠 50 字符保持上下文
separators=["\n\n", "\n", "。", "!", "?", ";", " "]
)
docs = text_splitter.split_documents(documents)
print(f"📦 切分为 {len(docs)} 个文本块")
# 3. 向量化并存入 ChromaDB
db = Chroma.from_documents(
docs,
self.embeddings,
persist_directory=self.persist_dir
)
db.persist()
print(f"✅ 知识库构建完成,数据已持久化到 {self.persist_dir}")
return db
def load_knowledge_base(self):
"""加载已有的知识库"""
return Chroma(
persist_directory=self.persist_dir,
embedding_function=self.embeddings
)
def create_qa_chain(self, db=None):
"""创建检索问答链"""
if db is None:
db = self.load_knowledge_base()
# 使用本地 Ollama 模型(也可换 OpenAI API)
llm = ChatOpenAI(
base_url="http://localhost:11434/v1",
api_key="ollama",
model="qwen2.5:7b",
temperature=0.1 # 低温更严谨,减少幻觉
)
# 构建检索器
retriever = db.as_retriever(
search_type="similarity",
search_kwargs={"k": 3} # 检索最相关的 3 个片段
)
# 构建问答链
qa_chain = RetrievalQA.from_chain_type(
llm=llm,
chain_type="stuff",
retriever=retriever,
return_source_documents=True # 返回参考来源
)
return qa_chain
def query(self, question: str):
"""提问并返回答案和来源"""
qa_chain = self.create_qa_chain()
result = qa_chain.invoke({"query": question})
print("=" * 60)
print(f"❓ 问题:{result['query']}")
print(f"💡 回答:{result['result']}")
print("=" * 60)
print("📚 参考来源:")
for i, doc in enumerate(result["source_documents"], 1):
print(f" [{i}] {doc.page_content[:120]}...")
print()
===== 使用示例 =====
if name == “main”:
# 第一步:准备知识库文档
# 在 ./knowledge_base/ 目录下放入你的 txt 文件
os.makedirs(“knowledge_base”, exist_ok=True)
# 创建示例文档
sample_doc = """
WorkBuddy 是面向企业的 AI 工作流自动化平台。
核心功能包括:Skills 技能系统、MCP 连接器、专家系统、Teams 协作。
支持私有化部署,数据不出企业内网。
相比传统 RPA,WorkBuddy 基于大模型实现智能决策,可处理非结构化任务。
定价模式:SaaS 版按 Seat 收费,私有化版一次性买断 + 年度维护费。
“”"
with open(“knowledge_base/workbuddy_intro.txt”, “w”, encoding=“utf-8”) as f:
f.write(sample_doc)
# 第二步:构建知识库(首次运行)
rag = RAGSystem()
rag.build_knowledge_base()
# 第三步:提问
rag.query("WorkBuddy 支持哪些核心功能?")
rag.query("WorkBuddy 怎么收费的?")
rag.query("WorkBuddy 和传统 RPA 有什么区别?")
5.3 RAG 优化技巧
表格
优化方向 方法 效果
分块策略 按语义边界切分,优先在段落/句子处断开 避免语义断裂
嵌入模型 中文场景用 bge-large-zh-v1.5 语义匹配更准
检索策略 混合检索(BM25 + 向量) 兼顾关键词和语义
重排序 Cross-Encoder 二次排序 准确率提升 15-20%
Prompt 工程 强制模型基于上下文回答 减少幻觉
六、完整 AI 开发工作流:串联所有模块
把上面的模块串起来,就是一个完整的 AI 应用开发流程:
plaintext
1
2
3
4
5
6
7
8
9
10
11
┌─────────────────────────────────────────────────────────┐
│ AI 应用开发全流程 │
│ │
│ ① 环境搭建 ② 模型部署 ③ 工具集成 │
│ Python + PyTorch → Ollama/vLLM → MCP Server │
│ │
│ ④ 知识库构建 ⑤ 应用开发 ⑥ 部署上线 │
│ RAG + ChromaDB → Agent 编排 → Docker + API │
│ │
└─────────────────────────────────────────────────────────┘
一键启动脚本:
python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
“”"
AI 开发环境一键检查脚本
“”"
import subprocess
import sys
def check_module(name, import_name=None):
“”“检查模块是否可用”“”
import_name = import_name or name
try:
import(import_name)
print(f" ✅ {name}“)
return True
except ImportError:
print(f” ❌ {name}(未安装)")
return False
def check_service(name, url):
“”“检查服务是否运行”“”
import requests
try:
resp = requests.get(url, timeout=3)
if resp.status_code == 200:
print(f" ✅ {name}(运行中)“)
return True
except:
pass
print(f” ❌ {name}(未启动)")
return False
print(“=” * 50)
print(“🔍 AI 开发环境检查”)
print(“=” * 50)
print(“\n📦 Python 依赖:”)
modules = [
(“torch”, “torch”),
(“langchain”, “langchain”),
(“chromadb”, “chromadb”),
(“sentence-transformers”, “sentence_transformers”),
(“mcp”, “mcp”),
(“fastapi”, “fastapi”),
(“openai”, “openai”),
]
all_ok = True
for name, imp in modules:
if not check_module(name, imp):
all_ok = False
print(“\n🔧 本地服务:”)
services = [
(“Ollama”, “http://localhost:11434”),
(“vLLM”, “http://localhost:8000/health”),
]
for name, url in services:
check_service(name, url)
print(“\n” + “=” * 50)
if all_ok:
print(“🎉 环境就绪,可以开始 AI 开发!”)
else:
print(“⚠️ 部分依赖缺失,请参考上方安装命令”)
print(“=” * 50)
七、避坑指南
7.1 环境配置
表格
坑 解决方案
CUDA 版本不匹配 nvidia-smi 查看驱动版本,安装对应 PyTorch
OOM 显存不足 换量化模型(Q4_K_M),减小 batch_size
依赖冲突 用 venv/conda 隔离环境
7.2 模型部署
表格
坑 解决方案
Ollama 下载慢 手动下载 GGUF 文件放入 ~/.ollama/models
vLLM 启动失败 检查 torch 和 cuda 版本兼容性
中文乱码 确保 tokenizer 的 trust_remote_code=True
7.3 RAG 优化
表格
坑 解决方案
检索不准 换中文嵌入模型(bge-large-zh)
回答偏离上下文 Prompt 中强调"仅基于以下信息回答"
分块语义断裂 用 RecursiveCharacterTextSplitter,设合理 overlap
八、总结
本文完整覆盖了 AI 大模型开发的三大核心环节:
**本地部署 **:Ollama(快速验证)+ vLLM(生产级部署),双方案覆盖从开发到上线全流程
**MCP 协议 **:统一工具调用标准,一次编写适配所有大模型,告别多套适配层
**RAG 知识库 **:基于真实数据生成回答,彻底解决幻觉问题
2026 年,AI 开发的门槛已经大幅降低。你不需要从零训练模型,不需要深厚的数学功底——** 把环境搭好、把工具链串通、把知识库建好 **,你就已经超过了 90% 的"AI 开发者"。
参考资源:
Ollama 官方文档
vLLM GitHub
MCP 协议规范
LangChain 文档
ChromaDB 文档
📌 作者说:如果这篇文章对你有帮助,欢迎点赞、收藏、关注三连~有问题可以在评论区交流,我会一一回复。后续还会更新大模型微调(QLoRA)、AI Agent 高级编排等进阶内容,感兴趣的话点个关注不迷路~
更多推荐



所有评论(0)