RAG 与 Agent 应用实战
大模型应用落地的两大工程范式:RAG(让模型回答它没学过的知识)和 Agent(让模型调用工具干活)。这篇从本地部署开始,把两条链路的实战代码走通。
前置阅读:AI 模型学习与对比(理论篇),本文专注动手。
一、先把模型跑起来:Ollama 本地部署
Ollama 是本地跑开源模型最省事的方案(2026 年仍在活跃迭代),一行命令拉模型:
# 安装后拉取模型(按显存选择)
ollama pull qwen3:8b # 主力:中文能力强,日常任务够用
ollama pull gemma3:4b # 轻量:4GB 显存甚至纯 CPU 都能跑
ollama pull deepseek-v4-flash # 需要联网能力时备选
ollama run qwen3:8b "你好" # 终端直接对话测试代码里调用(兼容 OpenAI SDK 格式):
from openai import OpenAI
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
resp = client.chat.completions.create(
model="qwen3:8b",
messages=[{"role": "user", "content": "用一句话解释什么是RAG"}],
)
print(resp.choices[0].message.content)选型经验:8B 级模型是本地开发的甜点位——RAG 场景下它的「组织语言」能力完全够用(事实由检索提供),而显存需求(8-10GB)消费级显卡可以承受。
二、RAG:五步流水线
RAG 解决的问题:模型的知识冻结在训练截止日,且没见过你的私有数据。思路是先检索、后生成:
用户问题 → ①嵌入向量化 → ②向量检索Top-K → ③重排序
↓
⑤生成答案 ← ④拼装Prompt(问题+检索片段)2.1 文档切片与向量化
import chromadb
from openai import OpenAI
client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
db = chromadb.PersistentClient(path="./rag_store")
collection = db.get_or_create_collection("docs")
def chunk_text(text, size=300, overlap=50):
"""按句子边界滑动窗口切片——切在句子中间,检索质量明显下降"""
chunks, start = [], 0
while start < len(text):
end = min(start + size, len(text))
# 回退到最后一个句号,保证语义完整
cut = text.rfind("。", start, end)
if cut > start + size // 2:
end = cut + 1
chunks.append(text[start:end])
start = end - overlap
return chunks
def build_knowledge_base(docs: dict[str, str]):
"""docs: {文档名: 文本内容}"""
for name, text in docs.items():
chunks = chunk_text(text)
embeddings = [
client.embeddings.create(model="quentinz/bge-large-zh-v1.5",
input=c).data[0].embedding
for c in chunks
]
collection.add(
ids=[f"{name}-{i}" for i in range(len(chunks))],
documents=chunks,
embeddings=embeddings,
)工程要点:
- 嵌入模型选中文专用(bge 系列效果好且 Ollama 可直接跑),别用英文模型硬翻
- 表格和代码块整块保留,不要切——切断的表格检索回来是乱码
- 切片存元数据(来源文档、章节标题),答案里要引用,没有溯源的 RAG 没人敢用
2.2 检索 + 重排 + 生成
def retrieve(query: str, top_k: int = 8) -> list[str]:
q_emb = client.embeddings.create(
model="quentinz/bge-large-zh-v1.5", input=query
).data[0].embedding
res = collection.query(query_embeddings=[q_emb], n_results=top_k)
return res["documents"][0]
def answer(query: str) -> str:
chunks = retrieve(query)
context = "\n\n".join(f"[片段{i+1}] {c}" for i, c in enumerate(chunks[:4]))
prompt = f"""仅根据以下资料回答问题。资料里没有的就说"根据现有资料无法回答",不要编造。
{context}
问题:{query}"""
resp = client.chat.completions.create(
model="qwen3:8b",
messages=[{"role": "user", "content": prompt}],
temperature=0.1, # 事实型问答调低随机性
)
return resp.choices[0].message.content2.3 进阶:GraphRAG 适用场景
向量检索的短板是「跨文档的全局性问题」(比如"这套系统的整体风险点有哪些")。图增强检索(GraphRAG)把实体关系抽成知识图谱,顺着关系边检索——Neo4j 的 GraphRAG 包已提供生产级支持。经验法则:FAQ 型问题用向量 RAG,关系型/汇总型问题才上 GraphRAG,后者构建成本高不少。
三、Agent:让模型用工具
Agent = 模型 + 工具调用 + 循环决策。模型看到用户请求 → 决定调哪个工具 → 拿到结果 → 决定继续调还是回答。
3.1 Function Calling 最小实现
import json
tools = [{
"type": "function",
"function": {
"name": "query_order",
"description": "查询订单状态",
"parameters": {
"type": "object",
"properties": {
"order_id": {"type": "string", "description": "订单号"}
},
"required": ["order_id"],
},
},
}]
def query_order(order_id: str) -> str:
# 真实场景里是查数据库/调内部API
return json.dumps({"order_id": order_id, "status": "已发货", "eta": "明天"})
def agent_chat(user_input: str) -> str:
messages = [{"role": "user", "content": user_input}]
while True:
resp = client.chat.completions.create(
model="qwen3:8b", messages=messages, tools=tools
)
msg = resp.choices[0].message
if not msg.tool_calls:
return msg.content # 没有工具调用 → 最终回答
messages.append(msg)
for call in msg.tool_calls:
result = query_order(**json.loads(call.function.arguments))
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": result,
})
# 循环继续:模型看到工具结果后决定下一步注意 while True 循环——Agent 的本质就是「模型在循环里自主决策」。生产代码必须加最大循环次数限制,防止模型陷入工具调用死循环烧 token。
3.2 MCP:把工具做成标准件
自己维护工具清单只能算 demo。MCP(Model Context Protocol)把「工具暴露」标准化了——写一次 MCP Server,任何支持 MCP 的客户端(Claude、ChatGPT、Cursor、VS Code、Ollama 等)都能直接用你的工具:
# MCP Server 示例(Python SDK)
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("order-tools")
@mcp.tool()
def query_order(order_id: str) -> str:
"""查询订单状态与物流信息"""
return {"order_id": order_id, "status": "已发货", "eta": "明天"}
mcp.run() # 客户端自动发现这个工具,无需任何接入代码到 2026 年,MCP 的月 SDK 下载量已过亿,工具生态数以万计——新项目接工具,直接按 MCP 标准做,别再造私有协议。
3.3 框架选型(2026 现状)
- LangGraph(2025 年 10 月 1.0):状态机式智能体编排,适合复杂多步流程,生产采用最多
- LlamaIndex:检索侧最强(RAG 数据管道成熟),编排已拆分为独立 Workflows 包
- 不用框架:简单的「模型+两三个工具」场景,裸写 Function Calling 循环更清晰——框架的抽象成本只有在流程复杂时才回本
我的实践结论:RAG 用 LlamaIndex 的数据管道,Agent 编排用 LangGraph,简单场景两者都不用。
四、评估:没有 eval 的 AI 应用都是玄学
上线前必须回答「改了 prompt,效果变好还是变坏了」——靠人肉试是不行的:
# 最小可用的回归评测集
eval_cases = [
{"q": "退货流程是什么", "must_contain": ["7天", "运费"], "must_not_contain": ["抱歉,我不知道"]},
{"q": "你们老板是谁", "must_contain": ["无法回答"]}, # 知识库外的问题必须拒答
]
def run_eval():
passed = 0
for case in eval_cases:
ans = answer(case["q"])
ok = all(k in ans for k in case.get("must_contain", [])) and \
not any(k in ans for k in case.get("must_not_contain", []))
passed += ok
print(f"{'✓' if ok else '✗'} {case['q']}")
print(f"通过率: {passed}/{len(eval_cases)}")维护一个几十条 badcase 的评测集,每次改动跑一遍——这是 AI 应用工程化的分水岭。Ragas 之类的框架(月下载 40 万+)提供检索质量、忠实度等现成指标,需要更细的度量时再上。
五、总结:落地优先级
- 先 RAG 后 Agent:RAG 链路简单、收益直接,是大多数业务的第一步
- 本地小模型做开发,云端大模型做兜底——开发迭代快且零成本
- 工具一律 MCP 化,生态红利还在放大期
- 评测集从第一天就建,不然每次改 prompt 都是赌博
更多实战记录见 AI 客服系统 项目页。
参考来源
- Ollama 官方文档与 Release(ollama.com,2026 年 8 月 v0.33.2)
- MCP 官方文档(modelcontextprotocol.io)
- Neo4j GraphRAG 文档(neo4j.com)
- LangChain / LlamaIndex 官方博客(2026 现状)
- Ragas 评估框架文档(docs.ragas.io)