《Deep Agents 实战笔记》第 12 篇 | 基于 deepagents (Python) 官方文档 原文：https://docs.langchain.com/oss/python/deepagents/rag 、https://docs.langchain.com/oss/python/deepagents/rubric

上一篇的 Deep Research 代理解决的是"信息从哪来"。这一篇换个更工程化的问题：答案凭什么可信。官方 RAG 教程给出的答案不是换个更强的模型，而是把 Deep Agents 的组件箱打开——检索工具、文件系统 Backend、子代理、技能、评分标准（rubric）——重新组装一遍 RAG。

一、Baseline：没有检索的"RAG"会怎样

教程的起点很犀利：先证明"不检索有多离谱"。同一个问题——"如何从子代理流式传输中间工具结果？"——交给一个没有任何检索工具、只挂了个"文档助手"提示词的 Deep Agent：

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)

结果不出所料：回答流于泛泛，遗漏关键指引，甚至混入过时信息——模型只靠训练数据作答，永远不会以你的事实来源为依据。这就是所有 RAG 系统存在的理由。

二、传统 RAG 与 Agentic RAG：差的不只是"多调几次"

传统 RAG 是一条流水线：嵌入 → 检索一次 → 拼上下文 → 生成，检索质量好坏全看那一次相似度搜索的运气。官方教程实现的 Agentic RAG 模式叫检索、卸载与委派（retrieve, offload, delegate），整条链路变成了一个会自我修正的循环：

flowchart TB
    subgraph BASE["Baseline RAG：单程流水线"]
        Q1["用户问题"] --> R1["相似度检索一次"]
        R1 --> G1["拼进上下文直接生成"]
        G1 --> A1["答案（对错听天由命）"]
    end
    subgraph AG["Agentic RAG：检索、卸载与委派"]
        Q2["用户问题"] --> P2["规划聚焦的搜索查询"]
        P2 --> S2["search_documentation 检索"]
        S2 --> O2["分块卸载到文件系统<br/>只返回文件路径"]
        O2 --> T2["task 委派 chunk-analyst<br/>子代理并行读文件"]
        T2 --> Y2["综合子代理摘要成答案"]
        Y2 --> V2{"校验：证据够吗？"}
        V2 -->|"不足，换查询再搜"| P2
        V2 -->|"充分"| A2["带文档链接的最终答案"]
    end
    style A1 fill:#fde8e8,color:#8b1a1a,stroke:#e05252
    style A2 fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a
    style V2 fill:#fdf3ff,color:#504b5f,stroke:#7e65ae

正在渲染图表…

三个结构性差异值得记住：

  1. 检索结果不进上下文，进文件系统——主代理只拿到文件路径，这是卸载（offload）思想在 RAG 场景的直接应用；
  2. 分析外包给子代理——每个分块一个 chunk-analyst，独立上下文、可并行，主代理的窗口始终干净；
  3. 生成之后有校验回路——证据不足就换查询再搜，而不是硬着头皮编一个。

三、索引：标准四步，没什么花活

索引阶段就是经典 RAG 的四步曲：加载、切分、向量化、存储。教程对 docs.langchain.com 的 14 个页面建索引：

import requests
from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

def load_langchain_docs(doc_paths: list[str] | None = None) -> list[Document]:
    """Fetch LangChain documentation pages as Documents."""
    paths = doc_paths or DOC_PATHS
    docs: list[Document] = []
    for path in paths:
        url = f"{DOCS_BASE}/{path}.md"
        try:
            response = requests.get(url, timeout=20)
            response.raise_for_status()
        except requests.RequestException:
            continue
        source = f"{DOCS_BASE}/{path}"
        docs.append(
            Document(page_content=response.text, metadata={"source": source})
        )
    return docs

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
all_splits = text_splitter.split_documents(docs)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_store = InMemoryVectorStore(embedding=embeddings)
vector_store.add_documents(documents=all_splits)

实测数据：14 个页面共约 59 万字符，切成 782 个分块。教程特意提醒：即便某些模型窗口塞得下整个语料库，模型也很难在超长输入里精准找信息，token 效率也不划算——"窗口大"不等于"不用检索"。另外生产环境记得把 vector store 持久化，并按计划在文档更新时刷新索引。

四、核心巧思：检索工具返回的不是内容，是文件路径

整个教程最漂亮的一笔在 search_documentation 工具里——它把检索和卸载焊在了一起：

from langchain_core.utils.uuid import uuid7
from deepagents.backends import StateBackend
from langchain.tools import tool

backend = StateBackend()

@tool(parse_docstring=True)
def search_documentation(query: str) -> str:
    """Search LangChain documentation and save matching chunks to the agent filesystem.

    Args:
        query: Natural language search query.

    Returns:
        File paths where retrieved chunks were saved under /retrieved/.
    """
    retrieved_docs = vector_store.similarity_search(query, k=4)
    batch_id = uuid7().hex[:8]
    uploads: list[tuple[str, bytes]] = []
    saved_paths: list[str] = []

    for index, doc in enumerate(retrieved_docs, start=1):
        path = f"/retrieved/{batch_id}/chunk_{index}.md"
        content = (
            f"# Source: {doc.metadata.get('source', 'unknown')}\n\n"
            f"{doc.page_content}"
        )
        uploads.append((path, content.encode("utf-8")))
        saved_paths.append(path)

    backend.upload_files(uploads)
    return (
        f"Saved {len(saved_paths)} documentation chunks:\n"
        + "\n".join(saved_paths)
    )

三个细节：工具内部调用 backend.upload_files() 把分块写入 Backend，必须把同一个 backend 实例传给 create_deep_agent，否则内置的 read_file、grep 读不到这些路径；每个分块头部写入 # Source: 来源 URL，为后续引用做准备；返回值只有路径清单，比如 /retrieved/a1b2c3d4/chunk_1.md——主代理的上下文里从头到尾没出现过分块正文。

五、提示词分工与子代理并行分析

和第 11 篇的 Deep Research 一样，这个项目也是"主代理指令 + 委派指令 + 子代理指令"三段式。主代理的五步工作流：

1. Plan：把复杂问题拆成聚焦的搜索查询
2. Search：调用 search_documentation，拿到文件路径
3. Analyze：用 task() 把每个分块文件委派给 chunk-analyst，
   多个分块就在一次响应里发起多个并行 task 调用
4. Synthesize：合并子代理摘要，附上文档来源链接
5. Verify：如果摘要没能完整回答问题，用更精准的查询再搜

委派指令里有一条铁律值得抄进你的项目："Do not paste full chunk contents into your own messages. Let subagents read files."——不要把分块全文粘进自己的消息，让子代理自己去读文件。这正是第 7 篇讲的上下文工程纪律在实战里的落点。

子代理 chunk-analyst 的指令则极其克制：读分配到的文件，提取有助于回答问题的事实，返回 300 字以内的摘要，附上来源 URL。注意它没有检索工具，只用内置文件系统工具——职责单一，上下文天然干净：

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,
}

agent = create_deep_agent(
    model=model,
    tools=[search_documentation],
    backend=backend,
    system_prompt=INSTRUCTIONS,
    subagents=[chunk_analyst_subagent],
)

运行时代码就是标准的 invoke。跑一次示例查询，代理会：检索 → 收到文件路径 → 发起一个或多个 task()（每个负责一个分块）→ 综合出带文档链接的答案。用 LangSmith 可以清楚看到搜索调用、文件写入、子代理委派的完整链路。

六、Rubric：给答案装一个"评审判官"

检索回来了、答案生成了，但怎么保证答案确实以检索到的材料为依据（grounded），而不是模型又偷偷靠记忆发挥？官方的答案是 RubricMiddleware——把 LLM-as-a-judge 模式搬进运行时（需要 deepagents>=0.6.5，目前处于 beta 阶段，API 可能变化）。

用法分两步。第一步，创建代理时挂上评中间件：

from deepagents import RubricMiddleware, create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_deep_agent(
    model="google_genai:gemini-3.6-flash",
    middleware=[
        RubricMiddleware(
            model="anthropic:claude-haiku-4-5",
            max_iterations=3,
        ),
    ],
    checkpointer=InMemorySaver(),
)

第二步，调用时在输入状态里传 rubric 字符串。不传就不会触发评分循环：

from langchain.messages import HumanMessage

config = {"configurable": {"thread_id": "my-rubric-thread"}}
result = agent.invoke(
    {
        "messages": [HumanMessage("Write a haiku about spring.")],
        "rubric": (
            "- The poem has three lines\n"
            "- Lines follow a 5-7-5 syllable pattern\n"
            "- The theme is spring"
        ),
    },
    config=config,
)

工作循环长这样：

flowchart LR
    Start["调用时携带 rubric"] --> Agent["Deep agent 产出答案"]
    Agent --> Grader{"评分子代理判定"}
    Grader -->|"satisfied"| Done["结束：全部标准通过"]
    Grader -->|"failed 或 grader_error"| Done
    Grader -->|"needs_revision"| Cap{"迭代次数小于<br/>max_iterations？"}
    Cap -->|"是"| Inject["逐条标准反馈注入对话<br/>代理再次运行"]
    Cap -->|"否"| MaxOut["max_iterations_reached 终止"]
    Inject --> Agent
    style Grader fill:#fdf3ff,color:#504b5f,stroke:#7e65ae
    style Done fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a
    style MaxOut fill:#fde8e8,color:#8b1a1a,stroke:#e05252

正在渲染图表…

判定有五种：satisfied（全部通过）、needs_revision（至少一条未过，反馈注入后重跑）、max_iterations_reached（到上限仍未过）、failed（评分标准本身有问题）、grader_error（评分子代理自身异常）。几个实践要点：

  • 评分模型用小模型：RubricMiddleware 的 model 参数接受 "provider:model-id" 字符串，官方示例用比工作模型更小更便宜的 claude-haiku-4-5——评分是判断题不是创作题，没必要用大模型烧钱；
  • 给评分器配工具：tools 参数可以让评分器在判定前收集证据。官方的代码生成示例给评分器挂了个 run_test_suite 工具，直接跑测试套件验证代码，而不是抽象地"推理"代码对不对——这一步把评分从主观变成客观；
  • 观察迭代进度：用 on_evaluation 回调（每次评分迭代后收到 RubricEvaluation，含逐条标准的 criteria 和未通过项的 gap 反馈），或 stream_events(version="v3") 配合 CustomTransformer 从 stream.custom 读取 rubric_evaluation_start / rubric_evaluation_end 事件；
  • 跨调用延续：配上 checkpointer 并复用同一个 thread_id，同一份 rubric 会持续作用于后续调用，直到传入新 rubric。

在 RAG 场景里，rubric 的典型条目就是"答案中的每个论断都能对应到检索到的源材料""引用的文档路径真实存在于检索结果中"——评分子代理逐条核对，逼着工作模型有据可查。

七、技能引导检索：把检索经验写成文件

教程还列了几种 RAG 模式，其中技能引导的检索（skills-guided retrieval）值得一提：用户提问后，代理加载一个相关技能，技能里写清楚"搜哪个索引、查询怎么构造、引用用什么格式"，代理照着指引调检索工具再综合答案。

本质上，这是第 8 篇讲的技能系统在 RAG 场景的应用：把"怎么搜"的方法论从提示词里搬进 SKILL.md 文件，启动时只占一行摘要，检索时才展开全文。同一个代理可以按问题类型加载不同的检索技能——技术文档怎么搜、营销资料怎么搜、代码仓库怎么搜，各自一份。教程还提到其余两种模式：基于评分标准（rubric）的依据校验（上文已实现）、待办清单驱动的调查（启用任务规划后由 write_todos 列出要查的页面或查询）。这些模式共享同一套基础组件，按语料规模、延迟要求和"答案必须贴合源数据的程度"自由组合。

八、实战作业

用你自己领域的文档库跑通整套流程：把教程里的 DOC_PATHS 换成你的文档地址（内部 wiki、产品手册、团队规范都行），索引 → 检索 → 卸载 → 子代理分析 → 综合，最后加一条 rubric："每个论断必须引用真实存在的文档 URL"。对比开篇那个裸模型 baseline 的回答，差距会非常直观。

本篇踩坑点

  • 间接提示词注入：这是官方专门用警告框强调的风险——检索到的文档里可能含有看似指令的文本，与你的系统提示词共享上下文窗口，模型可能遵从文档里的"指令"。没有任何提示词策略能完全防住；教程的做法是让模型把检索内容当纯数据处理、分块头部加 # Source: 区分元数据，并在输出呈现给用户前人工验证引用。生产环境务必重视；
  • backend 实例不一致：search_documentation 里写入的 backend 和传给 create_deep_agent 的 backend 必须是同一个实例，否则子代理读不到文件，报的错却很像"路径不存在"；
  • InMemoryVectorStore 不持久：教程用它图省事，进程一退索引就没了，生产要换持久化 vector store 并规划索引刷新；
  • rubric 是 beta 功能：deepagents>=0.6.5 才有 RubricMiddleware，且 API 未来可能变化，锁定版本再上生产；
  • 评分标准写得太主观：rubric 条目要写成可逐条判定的清单，"写得专业一点"这种条目评分子代理没法判，只会带来 needs_revision 空转，直到撞上 max_iterations。

下一篇预告：实战做完了，第 13 篇是系列收官——走向生产：速率限制与重试、模型回退、LangGraph Platform 部署与自托管、LangSmith 可观测性与评估，把 demo 补成能扛流量的系统。

参考资料：