1. 本章目标 #
第 39 章把 create_deep_agent 拆开看了一遍,结论是:它的能力 = 9 个内置工具 + 1.1 万字符的工具说明书。那一章全在「看结构」,这一章开始「跑起来」。
本章要做的事很具体:从零装依赖开始,跑出一个能自己检索资料、自己写报告文件、必要时还会派子智能体的研究助手。
学完你应能:
- 装好
deepagents,并按provider:model格式配好模型 - 用官方的 eval 成绩表判断「我这个模型能不能撑住 deep agent」
- 三种联网搜索方案(供应商内置 / Tavily / 自建工具)各自怎么接、什么时候用哪个
- 读懂一次 deep agent 运行的完整消息轨迹,指出它在哪一步决定写文件
- 从
state["files"]里正确取出文件内容(这里有个结构上的坑) - 让它派子智能体,并说清这么做的代价(实测贵 11 倍)
- 用
checkpointer让写出的文件跨轮次保留 - 用流式输出实时观察它每一步在干什么
前置依赖: 第 39 章(本章直接接着它讲)、第 2 章(消息轨迹怎么读)、第 8 章(@tool 怎么写)、第 11 章(checkpointer 与 thread_id)。
参考文档:
建议阅读顺序: §2 是环境准备,已经装好的可以只看 §2.2(选模型那张表很值得一看)。§3 三行代码先跑通。§5 是本章主线,那个研究助手的完整代码在这里。§6 讲子智能体,里面的成本对比很重要,别跳。§7~§8 是两个必备的实用技巧。
本章验证环境:deepagents 0.7.12、langchain 1.3.18、langgraph 1.2.11、langchain-deepseek 1.1.0,模型为 deepseek-v4-flash。本章的实验会真实调用模型,全部跑完约花几毛钱(§6 那个派子智能体的实验最贵,单次约 19 万 tokens)。
2. 环境准备 #
2.1. 装依赖 #
deepagents 是独立的库,不随 langchain 一起装:
pip install -U deepagents
# 或者用 uv(本教程用的是这个)
uv add deepagents它会顺带把 langchain 和 langchain-core 升到自己要求的版本。本章环境里 langchain 从 1.3.14 被升到了 1.3.18,属于正常现象。
装完确认一下版本,因为 0.7 改过默认行为(第 39 章 §6 实测过:write_todos 不再默认开启),版本对不上的话本章的输出会和你看到的不一样:
# deepagents 是独立顶层包,不是 langchain 的子模块
import deepagents
import langchain
# 打印两个关键版本号
print("deepagents:", deepagents.__version__)
print("langchain :", langchain.__version__)运行输出:
deepagents: 0.7.12
langchain : 1.3.182.2. 选一个撑得住的模型 #
Deep Agents 对模型有一个硬要求和一个软要求。
硬要求:必须支持工具调用(tool calling)。 这没得商量——第 39 章 §3.3 已经看到,它的全部能力都是以工具的形式给出去的,模型不会调工具就什么都做不了。
软要求:模型得扛得住大工具面。 第 39 章实测过,一个 deep agent 的工具说明书有 11058 字符、9 个内置工具。工具一多,弱模型就容易挑错工具——想写文件却调了 ls,或者该派子智能体时自己硬扛。
模型用 provider:model 格式指定,冒号前面是 LangChain 的集成名,后面原样传给供应商:
# 几种常见写法,冒号后面的部分要按各家的模型目录来写
"deepseek:deepseek-v4-flash" # 本教程默认
"openai:gpt-5.5" # OpenAI
"anthropic:claude-sonnet-4-6" # Anthropic
"google_genai:gemini-3.6-flash" # Google
"ollama:north-mini-code-1.0" # 本地 Ollama
"baseten:zai-org/GLM-5.2" # 注意模型名里带斜杠也照写冒号后面的模型标识必须和供应商的写法完全一致。 有的供应商用简单名字(gpt-5.5),有的用带命名空间的路径(zai-org/GLM-5.2),写错了会在初始化时报错而不是运行时——这算是好事,至少发现得早。
那怎么知道自己的模型行不行?官方在 Models 页公布了 Deep Agents eval suite 的成绩单,这是选模型时最实用的一张表。摘录几行:
| 模型 | 总分 | 文件操作 | 检索 | 工具使用 | 对话 | 摘要 |
|---|---|---|---|---|---|---|
openrouter:z-ai/glm-5.1 |
89% | 92% | 100% | 89% | 33% | 80% |
google_genai:gemini-3.6-flash |
82% | 100% | 100% | 90% | 38% | 80% |
openrouter:deepseek/deepseek-v4-flash |
81% | 100% | 80% | 90% | 33% | 80% |
openai:gpt-5.5 |
80% | 92% | 100% | 84% | 52% | 80% |
anthropic:claude-opus-4-7 |
80% | 100% | 100% | 82% | 48% | 100% |
openai:gpt-5.4 |
18% | 100% | 100% | 18% | 38% | 100% |
注意最后一行:gpt-5.4 的文件操作和检索都是 100%,但工具使用只有 18%,总分被拖到 18%。这说明「模型强不强」和「能不能当 deep agent 用」是两件事——它单项能力都在,就是不擅长在一大堆工具里挑对的那个。这正是前面说的「软要求」。
本教程用 deepseek-v4-flash 的理由就在这张表里:它总分 81%,其中文件操作 100%、工具使用 90%——而这两项恰好是本章要观察的核心行为(写文件、派子智能体)。表里那一行是 OpenRouter 转发的同款模型,我们直接用 DeepSeek 官方接口,provider 写 deepseek。
官方另外给了一份「建议模型」清单(通过了 eval 的那些):
| 供应商 | 建议模型 |
|---|---|
gemini-3.1-pro-preview、gemini-3.6-flash |
|
| OpenAI | gpt-5.5、gpt-5.4 |
| Anthropic | claude-opus-4-8、claude-opus-4-7、claude-opus-4-6 |
| 开源权重 | GLM-5.2、Kimi-K2.7 Code、MiniMax-M3 |
官方对这份清单有一句很克制的说明,值得原样记住:
这些模型在 Deep Agents eval suite 上表现良好,该测试考察的是基本智能体操作。通过这些测试是必要条件,但不足以保证在更长、更复杂的任务上表现好。
翻译成实践建议:这张表能帮你排除不合格的模型,但不能替你选出最合适的。 真实任务上还得自己评测——那就是第 29~38 章 LangSmith 那套东西的用处。
2.3. 配 API Key #
Key 放 .env,用 python-dotenv 加载,不要写进代码或提交进仓库(第 1 章 §7.3 的规矩,这里照旧)。
本章只需要一个模型的 Key:
# 本教程默认用 DeepSeek
DEEPSEEK_API_KEY="sk-xxx"
# 强烈建议一并打开 LangSmith,理由见下
LANGSMITH_TRACING=true
LANGSMITH_API_KEY="lsv2_pt_xxx"
LANGSMITH_PROJECT="deepagents"这一章开始,LangSmith 从「建议」变成「强烈建议」。 原因看一眼 §6 的实验就明白了:那次运行有 12 次模型调用、3 个子智能体、19 万 tokens,光靠 print 已经很难看清发生了什么。官方 quickstart 也把「设置 LangSmith tracing」单列成了一个步骤。第 29~38 章讲过怎么读 Trace,这里正好用上。
3. 三行代码跑通第一个 deep agent #
先跑一个最小的,确认环境没问题。这一步的目的不是做出有用的东西,而是确认「能跑」:
# 从 .env 读取 API Key
from dotenv import load_dotenv
# Deep Agents 的唯一入口函数
from deepagents import create_deep_agent
# override=True 让 .env 覆盖系统同名环境变量
load_dotenv(override=True)
# 最小配置:只给一个模型,什么工具都不给
agent = create_deep_agent(model="deepseek:deepseek-v4-flash")
# 跑一次,输入格式和 create_agent 完全一样
result = agent.invoke({"messages": [{"role": "user", "content": "你好,你能做什么?"}]})
# 取最后一条消息就是回答
print(result["messages"][-1].content)运行输出(模型每次措辞不同,你跑出来的不会和这里一字一样,但报出来的能力清单是一致的):
你好!我是一个可以帮你处理文件和代码任务的 AI 助手。我能做很多事情,比如:
**文件操作**
- 浏览、查找工作区中的文件(`ls`、`glob`)
- 查看文件内容(`read_file`)
- 创建、编辑、删除文件(`write_file`、`edit_file`、`delete`)
- 在文件中搜索指定内容(`grep`)
**代码与工程任务**
- 阅读代码、理解项目结构和逻辑
- 编写或修改代码、修复 bug
**复杂任务**
- 我可以派出子代理(subagent)并行处理一些独立的大型搜索或分析任务,然后帮你汇总结果
不过要注意,我没有联网浏览网页、执行系统命令或运行程序的能力——我只能在你提供的工作区里操作文件。它把工具名一个不差地报了出来:ls、glob、read_file、write_file、edit_file、delete、grep,还提到能派子智能体。而这些工具你一个都没传——它们就是第 39 章 §3.3 打印出来的那 9 个内置工具。
最后那句免责声明尤其值得注意:「我没有执行系统命令或运行程序的能力」。 这话说得完全准确,而且正好交叉印证了第 39 章 §3.3 的那个发现:execute 工具虽然被注册进了 ToolNode(共 10 个),但因为默认的 StateBackend 不支持执行,它没有被暴露给模型(模型只看到 9 个)。模型自己都知道它没有这个能力——第 39 章是从代码层面数出来的,这里是模型从它自己的工具面「读」出来的,两边对上了。
做个对照更能看清差别。把同一个问题问给不给工具的 create_agent:
from dotenv import load_dotenv
# 注意这次换成 langchain 的 create_agent(第 9 章那个)
from langchain.agents import create_agent
load_dotenv(override=True)
plain = create_agent(model="deepseek:deepseek-v4-flash")
r = plain.invoke({"messages": [{"role": "user", "content": "你好,你能做什么?"}]})
print(r["messages"][-1].content)运行输出:
你好!我能做很多事情,比如:
- **回答问题**:知识百科、学习辅导、生活常识等。
- **写作协助**:写文章、邮件、文案、诗歌、故事等。
- **翻译**:中英文及其他语言互译。
- **编程帮助**:代码调试、算法讲解、项目思路等。
- **数据分析**:整理信息、总结要点、做表格建议等。
- **创意灵感**:起名字、做策划、头脑风暴。
- **日常聊天**:陪你聊天、解答疑惑、提供建议。同一个模型、同一个问题、同样都没传工具,一个说「我能读写文件、派子智能体」,另一个说「我能陪你聊天」。 差别全部来自 create_agent 换成了 create_deep_agent。
这两段自我介绍并排放在一起,就是 harness 和 framework 的差别最直观的样子:framework 给你的是一个空壳子,工具得你自己装;harness 给你的是一台开箱就有手脚的机器。
如果这一步报错,先按这两条排查: ①
ModuleNotFoundError: No module named 'deepagents'→ 没装或装到了别的环境;②模型报「不支持工具调用」→ 换一个支持 tool calling 的模型(§2.2)。
4. 联网搜索工具怎么接 #
跑通之后,第一件要加的事就是让它能查资料。官方 quickstart 用的例子是「研究助手」,搜索工具有三种接法。
4.1. 方案一:供应商内置搜索(最省事) #
Google、OpenAI、Anthropic 都提供服务端执行的联网搜索,不用装包、不用额外申请 Key——搜索在供应商那边跑完,直接把结果给模型。
接法是传一个字典给 tools=,而不是传函数:
from deepagents import create_deep_agent
# 三家的写法各不相同,按各自文档来
# Google:一个空字典就够了
internet_search = {"google_search": {}}
# OpenAI:用 type 指定
# internet_search = {"type": "web_search"}
# Anthropic:要带版本号和名字
# internet_search = {"type": "web_search_20260209", "name": "web_search"}
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
# 字典和普通函数工具可以混在一个列表里
tools=[internet_search],
)用法上要注意的是:这类工具和你的模型是绑定的。 上面那个 {"google_search": {}} 只有配 Google 的模型时才能用,换成 DeepSeek 就不行了。所以它省事,但也把你锁在一家供应商上。
4.2. 方案二:Tavily(任意供应商都能用) #
Tavily 是一个专门给 AI 应用做的搜索 API,好处是和模型解耦——不管你用哪家模型都能接。代价是要多装一个包、多申请一个 Key:
pip install tavily-python# .env 里加一行
TAVILY_API_KEY="tvly-xxx"接法就是普通的自定义工具(第 8 章那套写法):
import os
# Literal 用来把参数限定在几个固定取值里,模型能看到这个约束
from typing import Literal
from dotenv import load_dotenv
from tavily import TavilyClient
from deepagents import create_deep_agent
load_dotenv(override=True)
# 客户端只需建一次,复用它避免每次请求都重建连接
tavily_client = TavilyClient(api_key=os.environ["TAVILY_API_KEY"])
def internet_search(
query: str,
# 返回几条结果,给默认值让模型可以不传
max_results: int = 5,
# 用 Literal 限定主题,模型只能从这三个里选
topic: Literal["general", "news", "finance"] = "general",
# 是否要网页原文,原文很占上下文,默认关掉
include_raw_content: bool = False,
):
"""Run a web search"""
# 直接把参数透传给 Tavily
return tavily_client.search(
query,
max_results=max_results,
include_raw_content=include_raw_content,
topic=topic,
)
agent = create_deep_agent(
# 换成任何支持工具调用的模型都行
model="deepseek:deepseek-v4-flash",
# 注意:普通函数不加 @tool 也能传,Deep Agents 会自动包装
tools=[internet_search],
)有两个细节值得说:
一、include_raw_content 默认给 False 是有讲究的。 网页原文动辄几万字,一开就会把上下文撑爆。这正是 deep agent 该用文件系统的场景:真要原文,让它先 write_file 存下来,需要哪段再 read_file 读回,而不是整篇堆在对话里(第 46 章会专门讲这个)。
二、函数不加 @tool 也能传。 上面这个 internet_search 是个普通函数,create_deep_agent 会自动包装成工具——docstring 变成给模型的说明,类型标注变成参数 schema。加 @tool 当然也行,效果一样。
4.3. 本章的选择:自建一个资料库工具 #
本章的实验不用上面两个方案,而是自己写一个「企业内部资料库」工具。三个理由:
| 理由 | 说明 |
|---|---|
| 零外部依赖 | 不需要 Tavily Key,也不绑定某家模型,你照抄就能跑 |
| 结果可复现 | 联网搜索每次结果都不一样,实验没法对照;本地资料是固定的 |
| 更像真实业务 | 企业里更常见的是「查内部知识库」,而不是「查公网」——这正好接第 12~15 章那条 RAG 主线 |
第三条值得多说一句:你在第 12~15 章搭的那个知识库检索器,包装成工具就能直接接给 deep agent。 本章为了让代码自包含,用一个字典模拟检索结果,但接口形状和真实检索器是一样的——换成 retriever.invoke(query) 就是生产代码。
5. 主线:一个会自己写报告的研究助手 #
现在把东西拼起来,做本章的主要产出。任务设定成企业里很常见的一种:给三个技术主题,逐个查资料,最后汇总成一份报告文件。
5.1. 完整代码 #
这段代码可以直接跑,不需要任何额外 Key(只要配好模型的 Key):
# 计时用,看清一次完整运行要多久
import time
from dotenv import load_dotenv
from langchain.tools import tool
from deepagents import create_deep_agent
load_dotenv(override=True)
MODEL = "deepseek:deepseek-v4-flash"
# 埋点列表:记录业务工具的真实调用情况(第 19 章 §3.1 那个技巧)
TRACE = []
# 模拟企业内部资料库。每条资料故意写得长一些,用来观察它怎么处理大段内容
DOCS = {
"向量库": """【内部资料 · 向量库选型评审记录 v3】
Chroma:单机嵌入式,pip 装完即用,适合 10 万条以内的原型与内部工具。
支持元数据过滤与持久化目录,但没有分布式部署方案,副本与分片需自行解决。
实测 5 万条 768 维向量,检索 P95 约 18ms,内存占用约 1.2GB。
FAISS:Meta 开源的向量检索库,IndexFlatL2 精确检索,IVF/HNSW 近似检索。
性能最强但只是「库」不是「服务」,没有元数据过滤、没有持久化协议,要自己包一层。
实测同样数据集 HNSW 索引 P95 约 3ms,但构建索引耗时 4 分钟。
pgvector:PostgreSQL 扩展,最大优势是和业务库同源,事务、备份、权限体系全部复用。
适合已经在用 Postgres 的团队,检索性能弱于 FAISS 但运维成本最低。
实测 5 万条 P95 约 45ms,开启 ivfflat 索引后降到 12ms。
结论:原型用 Chroma,已有 Postgres 的业务系统用 pgvector,超大规模且能投人力的用 FAISS。""",
"切分策略": """【内部资料 · 文档切分策略沉淀 v2】
固定长度切分:按字符数硬切,实现最简单,但会把句子和表格切断,语义完整性差。
适合日志、流水这类本身没有强结构的文本。推荐 chunk_size 512、overlap 50。
递归字符切分:按段落→句子→词逐级回退,是大多数场景的默认选择。
实测在中文技术文档上,比固定长度切分的召回率高约 12%。
按结构切分:Markdown 按标题层级切、代码按函数切、HTML 按标签切。
最适合有明确结构的文档,能保住「这段属于哪一节」的信息,但要为每种格式写规则。
语义切分:用嵌入模型算相邻句子的相似度,在语义跳变处下刀。
效果最好但成本最高,每次切分都要跑一遍嵌入,10 万字文档约花 3 元。
踩过的坑:表格一定要整块保留,切断后模型会把表头和数据对错行,这类错误很难在评测里发现。""",
"重排": """【内部资料 · 重排(rerank)上线复盘】
为什么要重排:向量检索用的是双塔模型,问题和文档分别编码,快但精度有限。
交叉编码器把问题和文档拼在一起过一遍模型,精度高但慢,所以只能用来重排少量候选。
标准做法:先用向量或混合检索召回 50 条,再用重排模型精排出前 5 条喂给大模型。
实测效果:在内部 800 条评测集上,Top-5 命中率从 71% 提升到 89%。
代价:单次查询延迟增加约 300ms,重排 50 条的 API 费用约 0.001 元。
选型:阿里 DashScope 的 gte-rerank、Cohere Rerank、以及自部署 bge-reranker-base。
自部署适合数据不能出内网的场景,需要一张显卡,吞吐约 200 条/秒。
踩过的坑:候选数给太少(比如只召回 10 条)时重排几乎没有收益,因为好文档根本没进候选。""",
}
# docstring 会原样作为工具说明发给模型,所以要把「可用主题」写进去
@tool
def search_docs(topic: str) -> str:
"""按主题检索企业内部资料库。可用主题:向量库、切分策略、重排。"""
# 埋点:只有工具真被执行才会记录
TRACE.append(f"search_docs({topic})")
# 用「包含」匹配,容忍模型传「向量库选型」这类稍微多几个字的主题
for k, v in DOCS.items():
if k in topic:
return v
# 没匹配上时,把可用主题回给模型,让它自己纠正(第 8 章的错误回灌)
return f"没有找到「{topic}」相关资料。可用主题:{', '.join(DOCS)}"
agent = create_deep_agent(
model=MODEL,
# 只传一个业务工具,另外 9 个内置工具由 harness 提供
tools=[search_docs],
# 系统提示词完全属于你,不会被内置内容污染(第 39 章 §3.4 实测过)
system_prompt="你是企业技术调研助手。资料要逐个主题检索,最后交付一份结构清晰的中文报告。",
)
# 任务里明确要求「保存为 report.md」,这是触发文件工具的关键
TASK = ("调研我们知识库项目的三个技术主题:向量库、切分策略、重排。"
"每个主题单独检索资料,然后写一份汇总报告保存为 report.md。")
t0 = time.perf_counter()
result = agent.invoke({"messages": [{"role": "user", "content": TASK}]})
print(f"耗时 {time.perf_counter() - t0:.1f}s")
# 业务工具被调了几次、查了哪些主题
print("业务工具调用:", TRACE)运行输出:
耗时 13.8s
业务工具调用: ['search_docs(切分策略)', 'search_docs(向量库)', 'search_docs(重排)']三个主题都查了。但光看这两行看不出 deep agent 做了什么——真正有意思的东西在消息轨迹里。
5.2. 读消息轨迹:它在哪一步决定写文件 #
第 2 章讲过「先看完整 messages,再看最后一句回答」。这个习惯在 deep agent 上更重要,因为它的步数明显更多。
接着上面的代码打印轨迹:
# 接 §5.1 的 result 继续
print("=== 完整消息轨迹 ===")
for i, m in enumerate(result["messages"]):
kind = type(m).__name__
# AIMessage 上的 tool_calls 表示这一步模型决定调工具
tcs = getattr(m, "tool_calls", None) or []
if tcs:
# 打印工具名和它传了哪些参数名(参数值可能很长,只看键)
names = [f"{c['name']}({list(c['args'])[:3]})" for c in tcs]
print(f"{i:2d} {kind:12s} -> 调用 {names}")
else:
# 其余消息打印字数和开头,避免刷屏
txt = m.content if isinstance(m.content, str) else str(m.content)
print(f"{i:2d} {kind:12s} {len(txt):5d}字 | {txt[:60]}")运行输出(内容做了截断):
=== 完整消息轨迹 ===
0 HumanMessage 63字 | 调研我们知识库项目的三个技术主题:向量库、切分策略、重排。...
1 AIMessage -> 调用 ["search_docs(['topic'])", "search_docs(['topic'])", "search_docs(['topic'])"]
2 ToolMessage 472字 | 【内部资料 · 向量库选型评审记录 v3】Chroma:单机嵌入式...
3 ToolMessage 364字 | 【内部资料 · 文档切分策略沉淀 v2】固定长度切分:按字符数硬切...
4 ToolMessage 382字 | 【内部资料 · 重排(rerank)上线复盘】为什么要重排:向量检索...
5 AIMessage -> 调用 ["ls(['path'])"]
6 ToolMessage 14字 | No files found
7 AIMessage -> 调用 ["write_file(['file_path', 'content'])"]
8 ToolMessage 23字 | Updated file /report.md
9 AIMessage 700字 | 已完成全部三个主题的调研并生成了汇总报告 `report.md`。...这九步值得逐段读,因为它把 deep agent 的工作方式完整暴露了:
| 步骤 | 发生了什么 | 说明 |
|---|---|---|
| 1 | 一条消息里并行调了 3 次 search_docs |
三个主题彼此独立,模型一次全发出去(第 8 章的并行工具调用) |
| 2~4 | 三条资料返回 | 这是你的业务工具在干活 |
| 5~6 | 调 ls,返回 No files found |
它先看了一眼文件系统——这一步没人教它做 |
| 7~8 | 调 write_file,返回 Updated file /report.md |
报告真的落盘了 |
| 9 | 组织一段回复 | 告诉用户「报告已生成」 |
第 5 步最有意思:它在写文件之前,先调 ls 探了一下。 这个行为不是我们要求的,系统提示词里一个字都没提文件系统。它来自第 39 章 §3.4 那 11058 字符的工具说明书——write_file 和 read_file 的说明里写着诸如「改文件前一定要先读」这样的纪律,模型顺着这些纪律形成了「先看看有什么,再动手」的习惯。
这就是 harness 的价值具象化: 用 create_agent 想做到这一步,你得自己写文件工具、自己写一段「操作文件前先确认状态」的提示词、自己调试模型什么时候该写文件。这里全是白送的。
再看一眼成本:
# 接 §5.1 的 result 继续
inp = out = n = 0
for m in result["messages"]:
# 只有 AIMessage 带 usage_metadata,它是唯一可靠的用量来源
u = getattr(m, "usage_metadata", None)
if u:
n += 1
inp += u.get("input_tokens", 0)
out += u.get("output_tokens", 0)
print(f"模型调用 {n} 次,输入 {inp} tokens,输出 {out} tokens")运行输出:
模型调用 4 次,输入 16403 tokens,输出 2173 tokens4 次模型调用、1.6 万输入 tokens、13.8 秒,产出一份 2448 字符的报告。记住这组数字,§6.3 要拿它做对比。
5.3. 取文件内容:files 不是「路径 → 字符串」 #
报告写在哪了?在 state["files"] 里。但这里有个结构上的坑,很容易第一次就踩:
# 接 §5.1 的 result 继续
files = result.get("files") or {}
print("文件列表:", list(files))
# 取第一个文件看看它到底是什么结构
path = list(files)[0]
val = files[path]
print("值的 Python 类型:", type(val).__name__)
# 如果是字典,把键打出来
if isinstance(val, dict):
print("字典的键:", list(val))运行输出:
文件列表: ['/report.md']
值的 Python 类型: dict
字典的键: ['content', 'encoding', 'created_at', 'modified_at']files[path] 不是字符串,而是一个带元数据的字典。 正文在 content 键里,另外还有编码和两个时间戳。
所以直接 print(files["/report.md"]) 会得到这么一坨:
{'content': '# 知识库项目技术调研报告\n\n> 主题:向量库选型、文档切分策略...', 'encoding': 'utf-8', 'created_at': ..., 'modified_at': ...}正确的取法是写一个兼容两种情况的小函数——因为不同后端的存储结构可能不同(第 42 章会讲四种后端),兼容一下更稳:
# 接 §5.1 的 result 继续
def read_vfs(files: dict, path: str) -> str:
"""从 state["files"] 里取出某个文件的正文。"""
val = files.get(path)
# 有的后端直接存字符串
if isinstance(val, str):
return val
# StateBackend 存的是带元数据的字典,正文在 content 键里
if isinstance(val, dict):
return val.get("content", "")
# 其他情况兜底成空串,避免抛异常
return ""
files = result.get("files") or {}
body = read_vfs(files, "/report.md")
print(f"/report.md 共 {len(body)} 字符\n")
# 打印前 400 字看看质量
print(body[:400])运行输出:
/report.md 共 2448 字符
# 知识库项目技术调研报告
> 主题:向量库选型、文档切分策略、检索重排(Rerank)
> 依据:企业内部资料库检索结果(向量库选型评审记录 v3、文档切分策略沉淀 v2、重排上线复盘)
---
## 一、向量库
### 1.1 可选方案对比
| 方案 | 类型 | 核心能力 | 实测表现(5 万条 768 维) | 适用场景 |
| --- | --- | --- | --- | --- |
| **Chroma** | 单机嵌入式(pip 即装即用) | 支持元数据过滤与持久化目录;无分布式方案 | 检索 P95 约 **18ms**,内存约 **1.2GB** | 10 万条以内的原型与内部工具 |
| **FAISS**(Meta) | 开源向量检索库(非服务) | IndexFlatL2 精确检索;IVF/HNSW 近似检索 | HNSW 检索 P95 约 **3ms**;构建索引约 4 分钟 | 超大规模场景 |
| **pgvector** | PostgreSQL 扩展 | 与业务库同源,事务、备份、权限全部复用 | P95 约 **45ms**;开 ivfflat 后 **12ms** | 已在用 Postgres 的团队 |它把三段散文资料整理成了带表格的结构化报告,实测数字一个没丢。 这份产出是真实文件,不是聊天记录里的一段文本——下一轮对话可以让它继续修改(§7 会演示)。
一个容易混淆的地方:这个「文件」不在你的硬盘上。 默认后端是
StateBackend,文件存在 LangGraph 的状态里,进程结束就没了。想真正落到磁盘要换FilesystemBackend(第 42 章)。所以现在别去项目目录里找report.md——找不到是正常的。
6. 让它派子智能体 #
report.md 已经拿到了,但整个过程一个子智能体都没派。第 39 章 §3.3 明明看到 task 工具就在工具面里。
这一节回答两个问题:它为什么不派?怎么让它派?以及——派了之后代价多大?
6.1. 默认它不会派 #
回看 §5.2 那份轨迹:查资料、ls、write_file,全程没有 task。
这是对的行为,不是 bug。 原因在 task 工具的说明书里(第 39 章 §3.4 打印过全文),其中有这么两句:
Launch an ephemeral subagent to handle a complex, multi-step task.
...
- When only general-purpose is available, use it for any complex, context-heavy task关键词是 complex(复杂)、multi-step(多步)、context-heavy(上下文密集)。而 §5 那个任务对模型来说太简单了——三次检索拿到的资料总共才 1200 字,它一口气就能读完写完,没有任何理由再去派子智能体。
派子智能体的意义是隔离上下文:子任务的中间过程留在子智能体自己的上下文里,只把最终结论带回来。任务本来就装得下时,隔离没有收益,只有开销。
所以「模型不用某个内置工具」多半不是问题。 第 39 章 §9 那张坑表里列过:
write_todos不被用是因为默认没开,execute不被用是因为后端不支持——而task不被用,往往只是因为任务还不够复杂。
6.2. 引导它派 #
要让它派,得让任务「看起来值得派」。两处一起改:
一、在系统提示词里给出明确的委派策略(告诉它什么时候该派、派了要做什么); 二、把任务描述成「彼此独立的多个子课题」(这正好命中说明书里那条「独立任务可以并行扇出」)。
import time
from dotenv import load_dotenv
from langchain.tools import tool
from deepagents import create_deep_agent
load_dotenv(override=True)
MODEL = "deepseek:deepseek-v4-flash"
TRACE = []
# 这次资料写短一些,避免实验成本过高(子智能体会反复读写文件)
DOCS = {
"向量库": "【向量库评审】Chroma 单机嵌入式,10 万条以内够用,P95 18ms。FAISS 性能最强 P95 3ms 但只是库不是服务。pgvector 和业务库同源,运维最省,P95 45ms。",
"切分策略": "【切分策略】固定长度最简单但切断句子。递归字符切分是默认选择,中文技术文档召回率高 12%。按结构切分能保住层级信息。语义切分效果最好但 10 万字约 3 元。",
"重排": "【重排复盘】召回 50 条再精排 5 条,Top-5 命中率 71%→89%,延迟 +300ms。选型有 gte-rerank、Cohere、自部署 bge-reranker。候选太少时重排没收益。",
}
@tool
def search_docs(topic: str) -> str:
"""按主题检索企业内部资料库。可用主题:向量库、切分策略、重排。"""
TRACE.append(f"search_docs({topic})")
for k, v in DOCS.items():
if k in topic:
return v
return f"没有找到「{topic}」。可用主题:{', '.join(DOCS)}"
agent = create_deep_agent(
model=MODEL,
tools=[search_docs],
# 关键改动:把委派策略写进系统提示词
system_prompt=(
"你是企业技术调研助手。\n"
"遇到多个彼此独立的子课题时,用 task 工具为每个子课题派一个 general-purpose "
"子智能体去独立调研,并要求它把结果写成单独的文件;等子智能体全部返回后,"
"你再汇总成总报告。"
),
)
# 关键改动:任务里强调「彼此独立」,并明确每个子课题要产出独立文件
TASK = ("调研三个彼此独立的主题:向量库、切分策略、重排。"
"请为每个主题各派一个子智能体独立调研,各自把调研结果写成 "
"向量库.md、切分策略.md、重排.md,最后你把三份汇总成 report.md。")
t0 = time.perf_counter()
result = agent.invoke({"messages": [{"role": "user", "content": TASK}]})
print(f"耗时 {time.perf_counter() - t0:.1f}s")
# 注意这个数字:子智能体也会调业务工具,所以次数会超过 3
print("业务工具调用次数:", len(TRACE))
print(TRACE)
# 打印主智能体的轨迹,重点看 task 调用
print("\n=== 主智能体的消息轨迹 ===")
for i, m in enumerate(result["messages"]):
kind = type(m).__name__
tcs = getattr(m, "tool_calls", None) or []
if tcs:
for c in tcs:
# task 调用看 subagent_type,文件操作看 file_path
arg = c["args"].get("subagent_type") or c["args"].get("file_path") or ""
print(f"{i:2d} {kind:12s} -> {c['name']} {arg}")
else:
txt = m.content if isinstance(m.content, str) else str(m.content)
print(f"{i:2d} {kind:12s} {len(txt):5d}字 | {txt[:60]}")运行输出(这次很长,做了删减,保留关键步骤):
耗时 153.4s
业务工具调用次数: 9
['search_docs(重排)', 'search_docs(向量库)', 'search_docs(切分策略)',
'search_docs(向量库)', 'search_docs(切分策略)', 'search_docs(切分策略)',
'search_docs(重排)', 'search_docs(向量库)', 'search_docs(重排)']
=== 主智能体的消息轨迹 ===
0 HumanMessage 94字 | 调研三个彼此独立的主题:向量库、切分策略、重排。...
1 AIMessage -> ls
1 AIMessage -> glob
2 ToolMessage 14字 | No files found
3 ToolMessage 14字 | No files found
4 AIMessage -> write_file /probe_path_check.txt
5 ToolMessage 34字 | Updated file /probe_path_check.txt
6 AIMessage -> delete /probe_path_check.txt
7 ToolMessage 29字 | Deleted /probe_path_check.txt
8 AIMessage -> task general-purpose
8 AIMessage -> task general-purpose
8 AIMessage -> task general-purpose
9 ToolMessage 687字 | 调研报告已完成并真实写入 `/向量库.md`(约 195 行、约 5,800 字...
10 ToolMessage 549字 | 已完成调研并真实写入文件 **`/切分策略.md`**(共 250 行,7,000+ 字符)...
11 ToolMessage 474字 | 已完成。调研结果已真实写入 `/重排.md`(共 314 行,9000+ 中文字符)...
12 AIMessage -> ls
13 ToolMessage 33字 | ['/切分策略.md', '/向量库.md', '/重排.md']
14 AIMessage -> read_file /向量库.md
14 AIMessage -> read_file /切分策略.md
14 AIMessage -> read_file /重排.md
15 ToolMessage 5371字 | 1 # 向量库技术调研报告(企业 RAG / 向量检索)...
16 ToolMessage 3839字 | 1 # RAG 切分策略(Chunking)技术调研报告...
17 ToolMessage 3324字 | 1 # RAG 中的「重排」(Rerank) 技术调研报告...
18 AIMessage -> read_file /向量库.md
18 AIMessage -> read_file /切分策略.md
18 AIMessage -> read_file /重排.md
19 ToolMessage 4615字 | 101 --- 102 103 ## 5. 向量库选型考量(维度细化)...
20 ToolMessage 4387字 | 101 | 过小(<100 token) | 向量聚焦、召回精确...
21 ToolMessage 4953字 | 101 关键定位: 102 103 - 重排位于**召回/融合之后、LLM 生成之前**...
22 AIMessage -> read_file /切分策略.md
22 AIMessage -> read_file /重排.md
...
27 AIMessage -> write_file /report.md
28 ToolMessage 23字 | Updated file /report.md
29 AIMessage -> ls
30 ToolMessage 47字 | ['/report.md', '/切分策略.md', '/向量库.md', '/重排.md']
31 AIMessage 713字 | 全部完成。三个子智能体已分别完成独立调研并写出各自文件...这份轨迹信息量很大,挑五处最值得看的:
一、第 8 步:一条消息里三个 task 调用。 这就是并行扇出——三个子智能体同时开工,而不是排队。这正是 task 说明书里那句「独立任务用一条消息里的多个工具调用同时派出去」的效果。第 18 章我们用 LangGraph 手写并行时,要自己处理状态合并;这里模型自己就做了。
二、第 9~11 步:子智能体只回了一段几百字的摘要。 三个子智能体各自写出了 5800~9000 字的报告,但返回给主智能体的只是 474~687 字的交付说明。这就是第 39 章反复提到的「单次交接、只回一份最终报告」——上下文隔离在这里变得可见了:子智能体读了多少资料、改了几版,主智能体一概不知,只拿到结论。
三、第 14~26 步:主智能体在分页读文件。 注意那几条 ToolMessage 的开头——1 #...、101 ---、201 ...、301 ...。它在按 100 行一页地翻。这印证了第 39 章 §3.4 里 read_file 说明书那句「默认从头读最多 100 行,用 offset/limit 分页处理大文件」。说明书里的纪律,模型真的在遵守。
四、第 4~7 步:它先做了一次「探路」。 写了个 /probe_path_check.txt 又马上 delete 掉——在试探这个文件系统的路径规则能不能用。这类自发行为没人教,也说明了为什么 delete 工具会被内置。
五、业务工具被调了 9 次,不是 3 次。 因为三个子智能体各自都在调 search_docs,而且有的调了不止一次。子智能体是独立的智能体,不共享主智能体已经查到的资料——它们从零开始。这是上下文隔离的另一面:干净,但重复劳动。
6.3. 代价:贵了 11 倍 #
把两次运行的账单摆在一起,这一节的重点就出来了:
# 分别对 §5 和 §6 的 result 跑这个统计
def bill(result):
inp = out = n = 0
for m in result["messages"]:
u = getattr(m, "usage_metadata", None)
if u:
n += 1
inp += u.get("input_tokens", 0)
out += u.get("output_tokens", 0)
return n, inp, out
n, inp, out = bill(result)
print(f"主智能体模型调用 {n} 次,输入 {inp},输出 {out}")运行输出:
主智能体模型调用 12 次,输入 190972,输出 10650对照 §5 那次:
| 指标 | §5 不派子智能体 | §6 派了 3 个子智能体 | 倍数 |
|---|---|---|---|
| 耗时 | 13.8s | 153.4s | 11.1 倍 |
| 主智能体模型调用 | 4 次 | 12 次 | 3 倍 |
| 输入 tokens | 16403 | 190972 | 11.6 倍 |
| 输出 tokens | 2173 | 10650 | 4.9 倍 |
| 业务工具调用 | 3 次 | 9 次 | 3 倍 |
| 产出 | 1 份报告(2448 字符) | 4 份文件(合计约 3.9 万字符) | — |
而且这 19 万 tokens 只是主智能体的账。 三个子智能体各自的模型调用不在 result["messages"] 里(它们的过程被隔离了),真实总成本比这更高。要看全貌得靠 LangSmith Trace——这就是 §2.3 说「LangSmith 从建议变成强烈建议」的原因:光靠打印,你连账单都算不全。
那什么时候该派?判断标准和第 39 章 §8 那张表是一致的:
| 情况 | 该不该派 | 理由 |
|---|---|---|
| 子任务的中间过程很占上下文(读几十个文件、翻很多网页) | 该派 | 隔离掉过程,只留结论 |
| 子任务彼此独立,可以同时做 | 该派 | 并行能省墙上时间 |
| 每个子任务都要产出独立的大篇幅内容 | 该派 | §6 这个实验就是典型 |
| 任务一次就能装进上下文 | 别派 | 白付 11 倍成本(§5 那种) |
| 子任务之间要来回讨论 | 别派 | 子智能体无状态,只能单次交接 |
最后一行是个硬限制,值得单独强调:子智能体不能来回对话。 task 说明书里写得很清楚——「每次调用默认都是无状态的:它只看到你给的提示词,然后返回一份最终报告」。所以派它的时候,要把全部细节写进提示词,并明确说清要它返回什么。指望「先派出去、之后再补充要求」是行不通的。
本节结论:子智能体是「拿成本换上下文空间」的工具,不是「让结果更好」的工具。 §6 这次的四份文件确实比 §5 那一份丰富得多,但如果任务本来就装得下,这 11 倍成本换来的只是更啰嗦的输出。
7. 让文件跨轮次保留 #
§5 写出的 report.md 有个问题:下一轮对话它就找不到了。
这不是 bug,而是默认后端 StateBackend 的定义——官方原话是「线程作用域的文件系统,存在 langgraph 状态里;文件在同一个线程内跨轮次保留(通过你的 checkpointer)」。关键在括号里那句:没配 checkpointer,就没有「同一个线程」这回事。
这一节用一组 A/B 对照把它跑出来。
7.1. A 组:不配 checkpointer #
from dotenv import load_dotenv
from langchain.tools import tool
from deepagents import create_deep_agent
load_dotenv(override=True)
MODEL = "deepseek:deepseek-v4-flash"
@tool
def search_docs(topic: str) -> str:
"""按主题检索企业内部资料库。可用主题:向量库、切分策略、重排。"""
data = {
"向量库": "【向量库评审】Chroma 单机嵌入式,10 万条以内够用,P95 18ms。FAISS 性能最强 P95 3ms 但只是库不是服务。pgvector 和业务库同源,运维最省,P95 45ms。",
"切分策略": "【切分策略】固定长度最简单但切断句子。递归字符切分是默认选择,中文技术文档召回率高 12%。语义切分效果最好但 10 万字约 3 元。",
"重排": "【重排复盘】召回 50 条再精排 5 条,Top-5 命中率 71%→89%,延迟 +300ms。候选太少时重排没收益。",
}
for k, v in data.items():
if k in topic:
return v
return f"没有找到「{topic}」。可用主题:{', '.join(data)}"
# A 组:不传 checkpointer
plain = create_deep_agent(model=MODEL, tools=[search_docs])
# 第一轮:让它写文件
r1 = plain.invoke({"messages": [{"role": "user", "content": "检索向量库资料,写成 vec.md 保存。"}]})
print("第一轮写出的文件:", list(r1.get("files") or {}))
# 第二轮:另起一次 invoke,让它读回刚写的文件
r2 = plain.invoke({"messages": [{"role": "user", "content": "读一下 vec.md,告诉我里面提到了几种向量库。"}]})
print("第二轮回答:", r2["messages"][-1].content[:120])
print("第二轮的 files:", list(r2.get("files") or {}))运行输出:
第一轮写出的文件: ['/vec.md']
第二轮回答: 我在文件系统里没有找到 `vec.md` 文件,于是通过企业内部资料库检索了「向量库」主题的相关文档,内容如下:...
第二轮的 files: []第一轮明明写出了 /vec.md,第二轮的 files 是空的。 模型老实说「没找到这个文件」,然后自己重新检索了一遍资料来回答——行为上没毛病,但那份文件白写了。
原因是每次 invoke 都是一次全新的运行,状态不会自动延续。files 存在状态里,状态没了,文件也就没了。
7.2. B 组:配上 checkpointer 和 thread_id #
改动只有两处:建 Agent 时传 checkpointer,调用时传 thread_id。这两样都是第 11 章学过的东西,用法一字不差:
# 复用 §7.1 的 search_docs 和 MODEL
# InMemorySaver 是内存版 checkpointer,进程结束就没了,适合本地实验
from langgraph.checkpoint.memory import InMemorySaver
from deepagents import create_deep_agent
# 改动一:建 Agent 时传 checkpointer
saver = InMemorySaver()
agent = create_deep_agent(model=MODEL, tools=[search_docs], checkpointer=saver)
# 改动二:两轮用同一个 thread_id,它们才算「同一个会话」
cfg = {"configurable": {"thread_id": "demo-1"}}
# 第一轮:写文件
r1 = agent.invoke(
{"messages": [{"role": "user", "content": "检索向量库资料,写成 vec.md 保存。"}]}, cfg
)
print("第一轮写出的文件:", list(r1.get("files") or {}))
# 第二轮:读回文件。注意要带上同一个 cfg
r2 = agent.invoke(
{"messages": [{"role": "user",
"content": "读一下 vec.md,告诉我里面提到了几种向量库,只回答数字和名字。"}]},
cfg,
)
# 打印第二轮用到的工具,看它有没有真的去读文件
print("第二轮用到的工具:")
for m in r2["messages"]:
for c in (getattr(m, "tool_calls", None) or []):
print(" ", c["name"], c["args"].get("file_path", ""))
print("第二轮回答:", r2["messages"][-1].content[:120])运行输出:
第一轮写出的文件: ['/vec.md']
第二轮用到的工具:
search_docs
search_docs
search_docs
ls
ls
glob
write_file /vec.md
read_file /vec.md
read_file /vec.md
第二轮回答: 3 种:Chroma、FAISS、pgvector。read_file /vec.md 出现了,答案也对:3 种,Chroma、FAISS、pgvector。 文件确实跨轮次留下来了。
不过这份输出有个地方值得诚实地讲一下:它在读之前又多做了一堆动作——重新 search_docs 三次、ls 两次、glob 一次,还把 vec.md 重写了一遍。这些都不是必需的。
原因是恢复历史后模型的判断带有随机性:它可能先怀疑文件不存在,于是又查又写了一遍,然后才去读。这类「多余但无害」的动作在 deep agent 里很常见,因为它的工具面大、自主性高。要收紧的话有两个办法:
| 办法 | 做法 | 效果 |
|---|---|---|
| 提示词里明确 | 「文件已存在,直接 read_file 读,不要重新检索」 |
立竿见影,但属于概率性约束(第 19 章 §3.3) |
| 收窄工具面 | 只给 read_file / ls,不给 write_file |
结构性约束,第 41 章讲 FilesystemMiddleware(tools=[...]) |
第二种更可靠,理由第 19 章已经论证过:工具面里没有的东西,模型再怎么想也调不出来。
实验用
InMemorySaver就够,但它进程一结束就没了。 要真正持久化(重启还在、多进程共享),换langgraph-checkpoint-postgres那套——第 11 章配过,用法完全一样。
8. 流式:实时看它在干什么 #
§6 那次运行花了 153 秒。如果什么都不显示,用户会以为程序卡死了。而且从开发角度看,你也需要知道它当前卡在哪一步。
Deep Agents 的流式和 LangGraph 一样(毕竟它就是一张图),第 27 章讲过 stream_mode。这里用 updates 模式——它每完成一个节点就吐一次,正好用来看「进行到哪一步了」:
# 复用 §7.1 的 search_docs 和 MODEL
from deepagents import create_deep_agent
agent = create_deep_agent(model=MODEL, tools=[search_docs])
# stream_mode="updates" 表示每个节点执行完就产出一次增量
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "检索重排资料,写成 rerank.md。"}]},
stream_mode="updates",
):
# chunk 的结构是 {节点名: {状态字段: 值}}
for node, payload in chunk.items():
# 这一步产生的新消息
msgs = (payload or {}).get("messages") or []
for m in msgs:
tcs = getattr(m, "tool_calls", None) or []
if tcs:
# 模型决定调工具
for c in tcs:
print(f"[{node}] 决定调用 {c['name']}")
elif type(m).__name__ == "ToolMessage":
# 工具返回了结果,只报长度避免刷屏
body = m.content if isinstance(m.content, str) else str(m.content)
print(f"[{node}] {m.name} 返回 {len(body)} 字符")
elif getattr(m, "content", ""):
# 模型给出了最终回答
body = m.content if isinstance(m.content, str) else str(m.content)
print(f"[{node}] 输出回答 {len(body)} 字符")运行输出:
[model] 决定调用 search_docs
[model] 决定调用 ls
[tools] search_docs 返回 60 字符
[tools] ls 返回 14 字符
[model] 决定调用 search_docs
[model] 决定调用 search_docs
[model] 决定调用 glob
[tools] search_docs 返回 97 字符
[tools] search_docs 返回 67 字符
[tools] glob 返回 14 字符
[model] 决定调用 write_file
[tools] write_file 返回 23 字符
[model] 输出回答 399 字符这份输出把第 39 章 §3.2 那张图跑活了: 节点名只有 model 和 tools 两个,来回交替——就是那个「模型 ↔ 工具」循环。所有内置能力(文件系统、子智能体)都以工具的形式出现在 tools 节点里,没有任何额外节点。第 39 章用静态结构证明过一次,这里是动态验证。
还能看出两个细节:
一、第一轮它同时发了 search_docs 和 ls。 一边查资料一边看文件系统,并行的。
二、write_file 之前它调了 glob。 又一次「先确认再动手」,和 §5.2 观察到的行为一致。
如果要做前端界面,官方另有一套更完整的事件流 API,还提供 stream.subagents 给每个子智能体独立的流式句柄——§6 那三个并行的子智能体,用它就能分别显示各自进度。那是第 49 章的内容。
9. 配模型参数与运行时换模型 #
最后补两个实用配置。
9.1. 要调模型参数时,传实例而不是字符串 #
provider:model 字符串写起来快,但没法配 temperature、思考长度这类参数。要配就用 init_chat_model 建好实例再传进去:
from dotenv import load_dotenv
# init_chat_model 是第 3 章那个统一的模型构造函数
from langchain.chat_models import init_chat_model
from deepagents import create_deep_agent
load_dotenv(override=True)
# 先按需要把模型配好
model = init_chat_model(
model="deepseek:deepseek-v4-flash",
# 调研类任务希望结果稳定,把随机性降到最低
temperature=0,
# 超时和重试这类生产参数也在这里配
timeout=60,
max_retries=2,
)
# 再把实例传给 create_deep_agent,代替字符串
agent = create_deep_agent(model=model, tools=[])
print("模型已配置:", type(model).__name__)运行输出:
模型已配置: ChatDeepSeek可配的参数因供应商而异(Google 有 thinking_level,OpenAI 有 reasoning_effort),具体看各家的集成文档。
这里有个坑要提前说:官方文档提到 ProviderProfile(按供应商或按模型登记默认初始化参数)只在你传 provider:model 字符串时生效,传配置好的实例时不生效——因为模型已经建好了,没有它插手的机会。
9.2. 让用户在运行时选模型 #
如果你的产品要让用户自己挑模型(比如界面上一个下拉框),不需要为每个模型各建一个 Agent。官方的做法是用 wrap_model_call 中间件在每次调用时替换模型:
from dataclasses import dataclass
from typing import Callable
from dotenv import load_dotenv
from langchain.agents.middleware import ModelRequest, ModelResponse, wrap_model_call
from langchain.chat_models import init_chat_model
from deepagents import create_deep_agent
load_dotenv(override=True)
# 运行时上下文的结构:这里只放一个模型名
@dataclass
class Context:
model: str
# wrap_model_call 是「包装型」钩子,不会往图里插节点(第 39 章 §3.2)
@wrap_model_call
def configurable_model(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
# 从运行时上下文里取出用户选的模型名
model_name = request.runtime.context.model
# 按名字建模型
model = init_chat_model(model_name)
# override 换掉这次请求用的模型,再交给下一层处理
return handler(request.override(model=model))
agent = create_deep_agent(
# 这里给的是兜底默认值
model="deepseek:deepseek-v4-flash",
middleware=[configurable_model],
# 必须声明 context_schema,否则运行时拿不到 context
context_schema=Context,
)
# 调用时通过 context 把用户的选择传进去
result = agent.invoke(
{"messages": [{"role": "user", "content": "你好"}]},
context=Context(model="deepseek:deepseek-v4-flash"),
)
print(result["messages"][-1].content[:60])这段代码把第 10 章的 middleware、第 9 章的 context 和 Deep Agents 拼在了一起——又一次印证第 39 章那句「前面 38 章一点没白学」。
10. 实用约定与坑 #
| 约定 | 说明 |
|---|---|
先确认 deepagents 版本 |
0.7 改过默认行为(§2.1) |
| 选模型先查官方 eval 表 | 看「工具使用」那一列,不只看总分(§2.2) |
从 files 取正文要走 content 键 |
它是带元数据的字典,不是字符串(§5.3) |
想让文件跨轮次留下来,必须配 checkpointer |
还要每轮传同一个 thread_id(§7) |
| 任务里明确写「保存为 xxx.md」 | 不明说的话它可能只把内容贴在回答里(§5.1) |
| 派子智能体前先算算值不值 | 实测贵 11 倍(§6.3) |
| 派子智能体时把细节一次写全 | 它无状态,只能单次交接,不能追加要求(§6.3) |
| 长任务一定要用流式 | 否则用户以为卡死了(§8) |
| 这一章开始打开 LangSmith | 子智能体的账单在 messages 里看不到(§6.3) |
一、静默失效(不报错,但结果不对)
| 现象 | 原因 | 处理 |
|---|---|---|
| 第二轮读不到上一轮写的文件 | 没配 checkpointer,或两轮 thread_id 不同 |
见 §7.2 |
打印文件内容得到一坨 {'content': ...} |
files 的值是带元数据的字典 |
取 val["content"](§5.3) |
| 在项目目录里找不到写出的文件 | 默认 StateBackend 存在状态里,不落磁盘 |
换 FilesystemBackend(第 42 章) |
| 它从不派子智能体 | 任务对模型来说不够复杂,没必要派 | 引导或改任务描述(§6.2) |
| 它答完了但没写文件 | 任务里没明确要求产出文件 | 明说「保存为 xxx.md」 |
| 子智能体重复查了同样的资料 | 它们上下文隔离,不共享主智能体的已有结果 | 正常现象(§6.2),或把资料先写成文件让它们读 |
| 换了模型后行为大变 | 大工具面对模型能力敏感 | 查 eval 表的「工具使用」列(§2.2) |
二、会明确报错
| 现象 | 原因 | 处理 |
|---|---|---|
ModuleNotFoundError: No module named 'deepagents' |
独立库,不随 langchain 装 |
pip install -U deepagents |
| 模型报「不支持工具调用」 | Deep Agents 强依赖 tool calling | 换模型(§2.2) |
| 供应商内置搜索工具报错 | 它和模型绑定,换了供应商就不能用 | 改用 Tavily 或自建工具(§4) |
KeyError: 'TAVILY_API_KEY' |
用了 Tavily 方案但没配 Key | 配 Key,或改用 §4.3 的自建工具 |
传实例后 ProviderProfile 不生效 |
它只对 provider:model 字符串生效 |
把参数直接写进 init_chat_model(§9.1) |
三、概念性误解
| 误解 | 纠正 |
|---|---|
「写出的 report.md 在我硬盘上」 |
默认存在 LangGraph 状态里(§5.3) |
| 「派子智能体结果一定更好」 | 它换的是上下文空间,不是质量(§6.3) |
| 「可以先派出去、之后再补要求」 | 子智能体无状态,只能单次交接(§6.3) |
| 「eval 总分高就一定适合我的任务」 | 官方明说通过 eval 是必要非充分条件(§2.2) |
| 「deep agent 步数多,说明图很复杂」 | 流式输出里只有 model / tools 两个节点(§8) |
11. 练习 #
基础(跑通即可)
- 换掉资料库:把 §5 的
DOCS换成你自己业务里的三段资料(或者接第 15 章那个真实检索器),跑一遍,看它写出的报告结构对不对。 - 观察「先看再动手」:把 §5.2 的轨迹打印跑三次,统计它有几次在
write_file之前先调了ls或glob。这个行为是概率性的还是每次都有? - 修文件而不是重写:在 §7.2 的 B 组基础上加第三轮——「在 vec.md 末尾补一节『选型建议』」。看它用的是
write_file(整体覆盖)还是edit_file(局部替换),并解释哪种更好。
进阶(要动手设计)
- 成本对照:分别跑 §5 和 §6 的两个版本各三次,取平均值,把 §6.3 那张表用你自己的数据填一遍。如果倍数和本章差很多,想想是任务规模不同还是模型不同造成的。
- 给子智能体分工:§6 用的是内置的
general-purpose。改成三个专职子智能体(用subagents=参数,写法见第 39 章 §6),各自只给它需要的工具。对比一下:分工之后总 token 是变多还是变少? - 收紧工具面:§7.2 里模型多做了一堆动作。用提示词约束一次、再想办法只给它读类工具,对比两种做法的效果——这道题做完,第 19 章 §3 那个「概率性约束 vs 结构性约束」的结论你会记得更牢。
- 接上真实搜索:申请一个 Tavily Key,把 §4.2 那段接进 §5 的助手里,让它调研一个公网话题。注意观察
include_raw_content=True时上下文涨得多快,以及它会不会自己把原文卸载到文件里。 - (扩展)配上持久化:把 §7.2 的
InMemorySaver换成 Postgres 版(第 11 章配过),验证进程重启后文件还在。
12. 本章小结 #
deepagents是独立库,pip install -U deepagents单独装,会顺带升级langchain。版本要确认,0.7 改过默认行为。- 选模型有一个硬要求和一个软要求:必须支持工具调用;还得扛得住 11058 字符的大工具面。官方 eval 表里
gpt-5.4单项都不差却总分只有 18%,就是栽在「工具使用」这一列(18%)。 - 本教程用
deepseek-v4-flash是有依据的:官方 eval 里它总分 81%、文件操作 100%、工具使用 90%,正好覆盖本章要观察的两种行为。 - 官方对 eval 表有句克制的说明:通过它是必要条件,但不足以保证复杂任务上表现好。排除烂模型可以靠它,选出最优还得自己评测(第 29~38 章)。
- 三行代码就能跑通。都不给工具时,
create_deep_agent说「我能读写文件、派子智能体」,create_agent说「我能陪你聊天」——这两段自我介绍并排放着,就是 harness 和 framework 差别最直观的样子。 - 模型还主动声明「我没有执行系统命令的能力」,准确印证了第 39 章 §3.3 的发现:
execute被注册进ToolNode但没暴露给模型(默认StateBackend不支持执行)。一边是数代码数出来的,一边是模型自己读出来的,对上了。 - 搜索工具三种接法:供应商内置(传字典,最省事但绑定供应商)、Tavily(装包配 Key,跨供应商)、自建工具(零依赖、可复现,也最接近企业实际)。本章主线用第三种。
- 一次完整运行的轨迹只有九步:并行查三次资料 →
ls探一眼 →write_file落盘 → 汇报。其中ls那一步没人教它做,来自工具说明书里的使用纪律。 state["files"]的值是字典不是字符串,键为content/encoding/created_at/modified_at,正文要从content取。直接打印会得到一坨带元数据的字典。- 默认它不会派子智能体,这是对的。
task的说明书要求任务「复杂、多步、上下文密集」,而一个装得下的任务派出去只有开销没有收益。 - 要让它派,得改两处:系统提示词里写清委派策略,任务描述成「彼此独立的多个子课题」。改完实测真的一条消息里并行发了三个
task。 - 子智能体只回一段几百字的摘要:三个子智能体各写了 5800~9000 字的文件,返回给主智能体的却只有 474~687 字。上下文隔离在这里变得可见。
- 代价是 11 倍:13.8 秒 → 153.4 秒,16403 → 190972 输入 tokens。而且这只是主智能体的账,子智能体的开销还在外面——所以这一章开始必须上 LangSmith。
- 子智能体无状态,只能单次交接。派它时要把细节一次写全并说清要返回什么,指望之后追加要求是行不通的。
- 文件要跨轮次保留必须配
checkpointer+ 同一个thread_id。不配的话第一轮写的文件第二轮就找不到,模型会老老实实说「没找到」然后重做一遍。 - 流式输出里只有
model和tools两个节点交替,动态验证了第 39 章那个静态结论:所有内置能力都以工具形式存在,不新增节点。 - 要调模型参数就传实例(
init_chat_model)而不是字符串;要让用户运行时选模型,用wrap_model_call+context_schema。
下一章开始逐个拆解四组能力,从工具与 MCP 接入讲起:tools= 的三种形态、怎么把第 17 章那些 MCP 工具接进来,以及怎么用 excluded_tools 和 FilesystemMiddleware(tools=[...]) 把工具面收窄——正好解决 §7.2 里「它多做了一堆动作」那个问题。