给大模型接文档问答（RAG）人人会做：切块、向量化、检索、拼进提示词。但 demo 级 RAG 和生产级文档助手的差距，恰恰藏在那些”看着能跑”的细节里：检索到的长文本把主上下文撑爆、多文档证据互相打架、答案没有引用无从核验、模型一本正经地编 API。LangChain 的 Deep Agents 给出一套生产级 RAG 原语：自定义检索工具、filesystem backend 落盘、subagent 并行精读、skills 封装流程、grading rubric 校验 grounded。本文以官方”文档问答智能体”为蓝本，从零实现 检索 → 落盘 → 子代理精读 → 汇总引用 的完整模式，并讲透其余三种 RAG 模式与 rubric 校验。
一、背景：为什么朴素 RAG 不够用
先做一个对照实验。拿一个问题——”子代理的中间工具结果怎么流式输出？”——直接扔给没有任何检索工具的 Deep Agent，它只能凭训练记忆回答：大概率给出一个看似合理、实则过期或压根不存在的 API，还不会告诉你它不确定。这就是无检索问答的常态：流畅、自信、不可验。
朴素 RAG（检索 top-k 切块直塞提示词）修了一半问题，又引入新问题。第一，上下文污染：十个长切块塞进主上下文，真正关键的一段被淹没，模型注意力涣散，还烧 token。第二，证据不可追溯：答案里不写引用，用户无法核验，错了都不知道找谁。第三，长文档无能：单个文件超长时，要么截断丢信息，要么分片后丢失跨片逻辑。第四，无 grounded 校验：检索和生成之间没有质检员，模型夹带训练记忆”自由发挥”也没人拦。
Deep Agents 的解法是把 RAG 做成”一支队伍”而不是”一次检索”：主智能体只负责检索和调度，把证据写到 filesystem backend 里落盘，多个子代理（chunk-analyst）并行精读各自的文件，主智能体最后只收子代理的摘要做综合。这样主上下文永远干净，证据天然带文件级引用，长文档可以分页精读，还能叠加 rubric 打分员做 grounded 质检。
二、原理：Deep Agents 的 RAG 原语与四种模式
先认识五个原语。
自定义检索工具（retrieval tools）：你自己实现的函数，比如 search_documentation(query)，内部做向量检索返回匹配切块。智能体按需调用它，而不是你手写检索拼接逻辑。工具的 docstring 就是它的”使用说明书”，写清楚”何时调、参数是什么、返回什么格式”，智能体调用质量翻倍。
Filesystem backend：Deep Agents 的虚拟文件系统。检索到的切块不直接回填对话，而是写成 /retrieved/chunk_*.md 文件。子代理去读文件而不是读主上下文——这就是”offload（卸载）”，整个模式的名字”retrieve, offload, and delegate”由此而来。
Subagents（子代理）：有独立上下文窗口的下属智能体。你定义 chunk-analyst 子代理：职责是”读一个切块文件，回答它与用户问题的相关度，给出摘要和引用”。主智能体可以并行派 3 个分析员各读一块，互不干扰。
Skills：把”怎么检索、查哪个索引、引用格式是什么”写成可复用的技能说明，智能体按技能行事。适合检索流程固定、要复用到多个智能体的场景。
Grading rubrics（评分细则，需 deepagents>=0.6.5，beta）：RubricMiddleware 配置的质检员子代理，按你写的细则（如”每条事实必须有引用”）给初稿打分，不通过就打回重写，直到通过或达到迭代上限。
四种 RAG 模式的关系：
| 模式 | 流程 | 适合场景 |
|---|---|---|
| Skills-guided retrieval | 加载技能 → 按技能检索 → 综合 | 检索流程固定、语料稳定 |
| Rubric-checked grounding | 检索起草 → 质检员按细则打分 → 循环改写 | 答案必须严格 grounded，如法务医疗 |
| Todo-driven investigation | 规划待办 → 逐项检索 → 综合 | 复杂问题要拆多路调查 |
| Retrieve, offload, delegate | 检索 → 落盘 → 子代理并行精读 → 汇总引用 | 文档大、证据多、上下文贵 |
本文完整实现第四种（最通用、 mechanical 最多的一种），并在实战后讲清如何叠加 rubric 做质检。
三、环境准备
需要 Python 3.10+，以及一个 LLM 服务商 Key。官方示例用 google_genai:gemini-3.6-flash，你也可以换 OpenAI/Anthropic/国产 OpenAI 兼容 endpoint。安装：
pip install -U deepagents langchain langchain-community faiss-cpu beautifulsoup4
export GOOGLE_API_KEY="你的key" # 或 OPENAI_API_KEY / ANTHROPIC_API_KEY
准备一个文档语料做实验：可以是 LangChain 官方文档的子集，也可以是你自己的产品文档（几十个 HTML/Markdown 文件即可）。下面实战先用 3 个本地 markdown 文件演示全链路，你换成真实语料只需改索引构建一步。
四、分步实战
贯穿全程的问题：
How do I stream intermediate tool results from a subagent?（子代理的中间工具结果怎么流式输出？）
步骤 1：对照组——无检索的回答有多不可靠
from deepagents import create_deep_agent
from langchain.messages import HumanMessage
EXAMPLE_QUERY = "How do I stream intermediate tool results from a subagent?"
baseline_agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
tools=[],
system_prompt=(
"You are a helpful LangChain documentation assistant. "
"Answer questions about LangChain APIs and patterns."
),
)
result = baseline_agent.invoke(
{"messages": [HumanMessage(content=EXAMPLE_QUERY)]}
)
print(result["messages"][-1].text)
跑一遍，记下它的回答：大概率泛泛而谈，提不到 subagent streaming 前端文档，也没有任何引用。这就是后面一切工作的”before”照片。
步骤 2：建索引——文档切块 + 向量化
生产里用向量库，本教程用最小可跑的 TF-IDF/embedding 二选一。先给能跑通的版本（换 embedding 只需替换检索函数内部）：
import os, glob
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
DOC_DIR = "docs_corpus" # 放你的 .md 文档
paths = sorted(glob.glob(os.path.join(DOC_DIR, "*.md")))
docs = [open(p, encoding="utf-8").read() for p in paths]
def chunk(text, size=800, overlap=120):
parts, start = [], 0
while start < len(text):
parts.append(text[start:start + size])
start += size - overlap
return parts
chunks, meta = [], []
for p, d in zip(paths, docs):
for i, c in enumerate(chunk(d)):
chunks.append(c)
meta.append({"source": os.path.basename(p), "chunk_id": i})
vectorizer = TfidfVectorizer(max_features=20000).fit(chunks)
print(f"索引就绪：{len(paths)} 个文件，{len(chunks)} 个切块")
生产替换点：把 TF-IDF 换成 text-embedding-3-small 之类 embedding + FAISS/Chroma，下面的工具函数签名不变。
步骤 3：自定义检索工具——智能体的”手”
from langchain.tools import tool
@tool
def search_documentation(query: str, top_k: int = 6) -> str:
"""Search product documentation for passages relevant to the query.
Args:
query: 用户问题的关键词化改写，英文文档用英文关键词。
top_k: 返回切块数，默认 6。
Returns:
形如 [chunk_id] source | 内容摘要 的候选列表，含 chunk_id 供后续精读。
"""
qv = vectorizer.transform([query])
dv = vectorizer.transform(chunks)
sims = cosine_similarity(qv, dv)[0]
top = sims.argsort()[::-1][:top_k]
lines = []
for rank, idx in enumerate(top):
m = meta[idx]
lines.append(
f"[{idx}] {m['source']}#chunk{m['chunk_id']} "
f"(score={sims[idx]:.3f})\n{chunks[idx][:500]}"
)
return "\n\n---\n\n".join(lines)
注意 docstring 写得越具体，智能体检索词质量越高。”关键词化改写”这句尤其重要，否则它会把整句中文问题原样传进来，英文语料的检索分数会很难看。
步骤 4：检索 → 落盘——主上下文的”泄压阀”
核心规矩写进主智能体的 workflow 指令：检索结果不许在对话里展开，只许写文件。
RAG_WORKFLOW_INSTRUCTIONS = """\
你是一个文档问答 orchestrator。工作流铁律：
1. 先调用 search_documentation 检索，必要时换关键词检索 2-3 次。
2. 把每个有价值的切块全文写入 /retrieved/ 下的独立文件
（如 /retrieved/chunk_12.md），文件头注明 source 和 chunk_id。
不要把切块全文留在对话里——你的上下文只保留文件路径清单。
3. 然后并行委派 chunk-analyst 子代理精读每个文件。
4. 收到所有分析结果后综合作答，每条事实标注 [source#chunkN] 引用。
5. 证据不足时直接说"文档中未覆盖"，绝不编造 API。
"""
落盘这一步是整套模式的灵魂：主上下文只保留”路径清单 + 摘要”，切块全文躺在 backend 文件里。十个切块也好、五十个也好，主智能体的注意力永远只放在调度上。
步骤 5：子代理精读——并行分析员
定义 chunk-analyst 子代理：一次只读一个文件，输出结构化分析。
CHUNK_ANALYST_INSTRUCTIONS = """\
你是切块分析员。输入：用户问题 + /retrieved/ 下的一个文件路径。
输出（固定格式）：
- relevance: 0-5 分，这个切块与问题的相关度
- summary: 3-5 句中文摘要，只写文件中有的内容
- quotes: 最多 3 条原文短引用（英文原文照录）
- citation: 文件头注明的 source#chunk 编号
无关切块 relevance 打 0-1 分并一句话说明，不许硬答。
"""
SUBAGENT_DELEGATION_INSTRUCTIONS = """\
委派规则：
- 一次最多并行 {max_concurrent_analysts} 个 chunk-analyst。
- 每个子代理只给：用户问题 + 一个文件路径，不给其他切块。
- 子代理返回后，丢弃 relevance <= 1 的切块，用其余的综合。
- 合并同类事实，引用去重，优先采用带代码的文档原文。
"""
max_concurrent_analysts = 3
INSTRUCTIONS = (
RAG_WORKFLOW_INSTRUCTIONS
+ "\n\n" + "=" * 80 + "\n\n"
+ SUBAGENT_DELEGATION_INSTRUCTIONS.format(
max_concurrent_analysts=max_concurrent_analysts)
)
chunk_analyst_subagent = {
"name": "chunk-analyst",
"description": (
"Analyze one retrieved documentation chunk file. "
"Pass the user question and a single file path under /retrieved/."
),
"system_prompt": CHUNK_ANALYST_INSTRUCTIONS,
}
子代理的 description 是主智能体决定”何时派它”的依据，写清输入输出契约，别写空话。
步骤 6：组装 Agent 并运行
from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend # backend 挂载
from langchain.chat_models import init_chat_model
backend = FilesystemBackend(root_dir="./rag_workspace")
model = init_chat_model(model="google_genai:gemini-3.6-flash")
agent = create_deep_agent(
model=model,
tools=[search_documentation],
backend=backend,
system_prompt=INSTRUCTIONS,
subagents=[chunk_analyst_subagent],
)
if __name__ == "__main__":
result = agent.invoke(
{"messages": [HumanMessage(content=EXAMPLE_QUERY)]}
)
for msg in result.get("messages", []):
if msg.text:
print(msg.text)
对照步骤 1 的 before 照片看 after：答案应该出现具体 API、步骤化指导和 [xxx.md#chunkN] 引用。去 ./rag_workspace/retrieved/ 翻落盘文件，确认”主上下文干净、证据在文件里”的架构真的发生了。
步骤 7：大文档分页 + 代码解释器（进阶）
单个文档超长时，子代理不要一次读全文件：backend 自带搜索工具，让分析员用”关键词搜索 → 定位行 → 分页读”的方式精读，相当于给子代理配了”书内 grep”。需要输出表格、时间线、可视化时，再挂代码解释器（code interpreter）：子代理从源数据里抽数字、跑计算、画图，而不是让主智能体背数字。
步骤 8：叠加 Rubric 质检——grounded 的守门员
对引用有硬要求的场景（如法务、医疗、对外客服），加一道 RubricMiddleware 质检：
from deepagents.middleware import RubricMiddleware
rubric = RubricMiddleware(
rubric="每条事实性陈述必须带 [source#chunk] 引用；引用必须真实存在于 /retrieved/ 文件中；证据不足必须明示，不得编造。",
max_iterations=2,
)
流程变为：检索落盘 → 子代理精读 → 起草 → 质检员打分 → 不通过打回重写（最多 2 轮）。这就是第二种模式 Rubric-checked grounding 与第四种模式的叠加——原语是拼装的，不是互斥的。同理，复杂问题可以叠加 opt-in 的任务规划（todo 列表拆多路调查），即第三种模式。
五、常见坑
坑 1：检索工具 docstring 敷衍。 docstring 是智能体唯一的工具说明书。”搜索文档”四个字 vs “何时调、传什么关键词、返回格式是什么”，检索质量差两档。每个工具都值得写 5 行以上 docstring。
坑 2：切块全文回填主上下文。 一旦允许切块进对话，落盘架构名存实亡。workflow 指令里把”全文只许写文件”写成铁律，并抽查 /retrieved/ 是否真的被写了。
坑 3：top_k 一设了之。 top_k 太小漏证据，太大子代理排队慢。建议 5~8 起步，按”子代理丢弃率”调：丢弃率高说明捞得太滥，减小 top_k 或加严检索词。
坑 4：子代理一次喂多个文件。 并行精读的前提是一代理一文件。多文件塞给一个子代理，它的上下文立刻变脏，还不如不用子代理。
坑 5：引用不验存在性。 模型会编引用编号（[doc.md#chunk99] 查无此块）。Rubric 质检或后处理脚本必须核对引用真实存在，这是 grounded 的最后一米。
坑 6：中文问题直搜英文语料。 跨语言检索分数天然吃亏。工具 docstring 要求”关键词化改写”，或在工具内部先做一次查询改写/翻译。
坑 7：无评估就上线。 RAG 的回归只能靠评测集守：攒 30~50 个”问题 + golden 引用 + 期望要点”的评测对，用 LangSmith 的 RAG 评估教程方法定期跑，改动提示词、切块、模型任一环节都重跑。
附：生产替换清单（TF-IDF → 向量库）
教程用 TF-IDF 是为了零依赖跑通，上生产按这个清单替换，工具函数签名保持不变：
- Embedding 模型：换成
text-embedding-3-small或开源bge-m3（中英双语语料优先后者），切块长度 500~800 字、 overlap 10%~15%。 - 向量库：FAISS（单机够用）、Chroma（开发友好）、Milvus/Elasticsearch（大规模）。存
source#chunk元数据，引用链不断。 - 混合检索：向量召回 + BM25 召回取并集，再用 reranker（如 bge-reranker）精排取 top-k， grounded 率通常涨一大截。
- 评测集：攒 30~50 个”问题 + golden 引用 + 期望要点”三元组，换 embedding、改切块、升级模型任一动作后重跑，用 LangSmith 的 RAG 评估方法做前后对比。
常见问答
问：子代理和直接多检索一次有什么区别？ 区别在上下文隔离。串行多检索把全部切块堆进主上下文，证据越多主智能体越晕；子代理各读一块，干扰为零，证据越多反而越稳。切块超过 5 个，子代理模式几乎总是更优。
问：引用格式怎么定？ 正文教程用 source#chunk 文件级引用，简单可验。段落级要求就加行号或段落编号进落盘文件头。关键是引用必须机器可验：写个后处理脚本逐条核对编号存在，不存在就打回，这才是 grounded 的闭环。
六、总结
这套模式浓缩成四步：索引建好 → 工具包好（docstring 写透）→ 切块落盘（主上下文只留路径）→ 子代理并行精读、主代理汇总引用。它的本质是上下文工程：把有限的注意力花在调度与综合上，把脏活（读长文本）隔离到子代理的独立上下文里。按需再拼装 skills（固化流程）、rubric（grounded 质检）、todo 规划（复杂调查），以及 LangSmith 评估与部署，你的文档问答就从”能跑的 demo”升级成了”敢上线的助理”：每句话都有出处，每处引用都可验。 点击阅读原文
参考资料：https://docs.langchain.com/oss/python/deepagents/rag 点击阅读原文
相关文章
