DeepAgents 入门指南:从零构建你的第一个多智能体系统

一、DeepAgents 是什么?
先给一句能记住的话:
DeepAgents 是一个 Python 框架,让你像搭积木一样,把多个 AI 组合起来协作干活。
你可能已经用过 ChatGPT 的 API——「一个模型 + 一段提示词」就能对话。这在简单场景够用,但真实需求往往更复杂:
- AI 需要查外部数据(调数据库、访问 API、读文件)
- AI 需要分工(一个负责调研、一个负责写代码、一个负责审校)
- AI 需要记住上文(多轮对话里,"那它呢"能接得上)
这些用裸 prompt 都很难优雅地实现。DeepAgents 就是来补这块的。它建立在 LangGraph 之上,替你封装好了三件核心能力:
| 能力 | 解决什么 |
|---|---|
| 工具调用 | 让 AI 能调用你写的 Python 函数——查库、调接口、读文件都行 |
| 子智能体 | 让一个"主管 AI"把任务分派给多个"专业 AI",各司其职 |
| 记忆持久化 | 让 AI 跨轮次记住上下文,刷新页面、重启服务都不丢 |
通俗比喻:如果单个 AI 模型是一台引擎,DeepAgents 就是变速箱 + 方向盘 + 刹车——它让引擎真正能"开上路"。
先理清:DeepAgents、LangGraph、LangChain 三者什么关系?
新手常被这三个名字绕晕。其实它们是同一套技术栈里由高到低的三层——DeepAgents(最高层)建立在 LangGraph(中间层)之上,LangGraph 又建立在 LangChain(基础层)之上:
你写的应用
│ 调用
▼
[ DeepAgents ] ← 最高层:开箱即用的"深度 Agent"范式
规划 + 子 Agent 委派 + 记忆,一行 create_deep_agent 就有
│ 基于
▼
[ LangGraph ] ← 中间层:把流程编排成"状态图"
节点 + 边 + 条件边 + checkpointer,能画循环/分支/回退
│ 基于
▼
[ LangChain ] ← 基础层:与大模型打交道的地基
统一的模型接口、消息类型、工具定义、Prompt 模板
| 层 | 定位 | 你会用到它的什么 |
|---|---|---|
| LangChain | 和大模型对话的基础工具箱 | 统一的模型调用接口(ChatOpenAI 等)、消息类型(HumanMessage/AIMessage)、工具定义、回调 |
| LangGraph | 把多步骤流程组织成状态图的编排引擎 | 节点、边、条件边、checkpointer、stream()、get_state() |
| DeepAgents | 在 LangGraph 之上封装好的 Agent 范式 | create_deep_agent()、subagents、async_subagents |
一句话理清:
- LangChain 负责"怎么跟一个模型说话"(最底层的地基);
- LangGraph 负责"怎么把多个步骤串成可控的流程"(在 LangChain 之上加了状态图);
- DeepAgents 负责"直接给你一个能规划、能委派、能记忆的成品 Agent"(在 LangGraph 之上再封装一层)。
层层向下兼容:你用 DeepAgents 时,随时能"往下捅一层"——比如它的
checkpointer、stream()其实是 LangGraph 的能力,消息类型、模型接口又是 LangChain 的能力。越往上越省心,越往下越自由。这也是为什么本文第七章会讲"什么时候该从 DeepAgents 下沉到 LangGraph"。
DeepAgents 站在多智能体世界的哪个位置?
业界把多智能体系统大致分成五种架构。先有个全景印象:
| 架构 | 一句话描述 | 生活类比 |
|---|---|---|
| 单智能体 | 一个 AI 包办一切:思考 + 调工具 + 回复 | 全能的自由职业者 |
| 主从 / Supervisor | 一个主管 AI 调度多个下属 AI,主管说了算 | 项目经理 + 执行团队 |
| 网络 / Swarm | AI 之间平等,可以互相"喊话"交接任务 | 急诊室各科室会诊 |
| 层级 / Hierarchical | 树状:主 → 子 → 孙,层层往下委派 | 总部 → 大区 → 门店 |
| 角色 / 流水线 | 严格顺序:A 的输出喂给 B,确定性流程 | 工厂流水线:装配 → 质检 → 包装 |
DeepAgents 的原生舒适区是前两种——单智能体和主从(Supervisor)。层级(主→子→孙)也不在话下,因为子 Agent 本身就是完整 Agent,可以再带自己的下属。
它不擅长的是后两种——网络式 Swarm(平级互喊)和严格流水线(确定性顺序)。这不是能力缺陷,而是设计取舍:DeepAgents 把编排固定成了「规划 + 子 Agent 委派」这一种模式,执行顺序由 LLM 临场决定。真需要严格流程时,应该下沉到底层的 LangGraph(详见第七章)。
一句话记忆:凡是能表达成「主管派活给下属(下属还能再派给下属)」的,DeepAgents 直接支持;凡是需要「同事平起平坐互相喊话」或「严格按剧本一步步走」的,就得回到 LangGraph 手动编排。
别搞混:DeepAgents ≠ OpenClaw(龙虾)
如果你听说过 OpenClaw(龙虾),要注意它和 DeepAgents 根本不是一类东西:
| DeepAgents | OpenClaw(龙虾) | |
|---|---|---|
| 定位 | 多智能体编排框架 | 单 Agent 操控真实世界的运行时 |
| 一句话 | 多个 AI 脑子开会,互相委派 | 一只龙虾帮你操作电脑,直接干活 |
| 核心能力 | 工具调用、子 Agent 委派、多轮记忆 | 操控 OS、浏览器、文件系统、GUI |
| 范式 | 思考型——对话、推理、委派 | 行动型——直接操作真实软件 |
两者不冲突,甚至能组合:DeepAgents 在上层做任务规划与分派,OpenClaw 在底层做桌面操作——一个动脑,一个动手。
本章小结:DeepAgents = 建立在 LangGraph 之上、以「工具调用 + 子 Agent 委派 + 记忆」为核心的高层框架,最擅长 Supervisor 式的多智能体协作。
二、快速上手:5 分钟跑通第一个 DeepAgent
环境准备
# 安装 uv(现代 Python 包管理器,比 pip 快很多)
pip install uv
# 创建项目并安装依赖
uv init deepagents-demo
cd deepagents-demo
uv add deepagents langchain-openai
在项目目录建一个 .env,填入你的模型和 Key:
DEEPAGENTS_MODEL=anthropic:claude-sonnet-4-5-20250929
ANTHROPIC_API_KEY=你的key
主流模型都支持:想用 OpenAI 换成
openai:gpt-4o,想用 DeepSeek 换成deepseek:deepseek-chat即可,代码其余部分不用动。
第一个 Agent:会查天气的助手
from deepagents import create_deep_agent
# 1. 定义一个工具——它就是个普通 Python 函数
def get_weather(city: str) -> str:
"""返回指定城市的天气信息。"""
weather_data = {
"北京": "晴,26°C",
"上海": "多云,28°C",
"深圳": "雷阵雨,30°C",
}
return weather_data.get(city.strip(), f"暂无 {city} 的天气数据")
# 2. 把工具交给 Agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-5-20250929", # 换成你的模型
tools=[get_weather],
system_prompt="你是一个乐于助人的中文助手,需要时可调用工具。",
)
# 3. 开始对话
result = agent.invoke(
{"messages": [{"role": "user", "content": "北京今天天气怎么样?"}]}
)
print(result["messages"][-1].content)
# 输出类似:北京今天晴,26°C,很适合出门~
这背后发生了什么?
你只是问了句"北京天气怎么样",Agent 内部自动完成了这一串动作:
用户提问
→ Agent 判断:这需要查天气,我有 get_weather 工具
→ 自动调用 get_weather("北京")
→ 拿到结果 "晴,26°C"
→ 用自然语言组织成回复:"北京今天晴,26°C,很适合出门~"
注意:你没有写任何 if "天气" in question: ... 的判断。什么时候调工具、传什么参数、怎么把结果说成人话——全是 Agent 自己决定的。这就是 Agent 和普通函数调用的本质区别。
本章小结:给 create_deep_agent 传一个 tools 列表,Agent 就获得了自主调用这些函数的能力。
三、子智能体:让多个 AI 各司其职
单个 Agent 什么都干,功能一多就臃肿——工具塞几十个、prompt 写几百行,既难维护又容易出错。更专业的做法:一个主管 Agent 负责调度,多个子 Agent 各管一摊。
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver
# 定义一个子 Agent——专职天气播报员
weather_reporter = {
"name": "weather-reporter",
"description": "当需要查询天气时委派给它,它会调用工具并生动播报。",
"system_prompt": "你是天气播报员,收到城市名后调用工具查询,并用生动的中文播报天气。",
"tools": [get_weather], # 子 Agent 有自己的工具
}
# 主 Agent 自己不带天气工具,只能通过内置 task 工具委派给子 Agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-5-20250929",
tools=[], # 主 Agent 工具为空
system_prompt="你是调度助手。涉及天气的问题,必须委派给 weather-reporter 处理。",
subagents=[weather_reporter], # 挂上子 Agent
checkpointer=InMemorySaver(), # 顺便开个记忆(下一章细讲)
)
# 同一个 thread_id = 同一段会话,记忆跨轮次保留
config = {"configurable": {"thread_id": "demo-1"}}
# 第一轮
agent.invoke({"messages": [("user", "北京今天天气怎么样?")]}, config)
# → 主 Agent 判断这是天气问题 → 委派 weather-reporter 子 Agent → 子 Agent 查询并播报
# 第二轮:"那上海呢" 能被正确理解,因为上文还记着
agent.invoke({"messages": [("user", "那上海呢?")]}, config)
# → Agent 知道"那"指的是天气
三个关键点:
- 委派是自动的:主 Agent 通过内置的
task工具委派,你不用写任何"该派给谁"的分配逻辑——主管自己根据子 Agent 的description判断。 - 子 Agent 有独立工具:
weather-reporter配了get_weather,而主 Agent 自己没有。这就是"专业分工"。 - 可以无限嵌套:子 Agent 本身也是完整 Agent,能再配自己的
subagents,形成主管 → 子 → 孙的树状层级。
这正是经典的 Supervisor(管理者-工作者)模式,也是 DeepAgents 最原生、最推荐的多智能体范式。
本章小结:给主 Agent 传一个 subagents 列表,它就能自主把任务委派给合适的下属,实现分工协作。
四、异步子智能体:不阻塞的并行任务
上一章的子 Agent 有个隐藏局限:主管委派后必须原地死等结果。子 Agent 跑 5 秒,主管就卡 5 秒,期间没法响应用户,也没法同时干别的。
异步子智能体打破了这个限制,把模型从「委派 → 等待」变成 「启动 → 立即返回任务 ID → 后台执行 → 随时干预」。
主 Agent 因此获得五类新工具:
| 工具 | 作用 |
|---|---|
start_async_task |
启动后台任务,立即返回任务 ID,主 Agent 不阻塞 |
check_async_task |
查看某个任务的状态、拿结果 |
update_async_task |
给正在跑的任务追加新指令(中途改需求) |
cancel_async_task |
取消正在跑的任务(用户反悔了) |
list_async_tasks |
列出所有在跟踪的任务 |
同步 vs 异步,一张图看懂
假设主 Agent 要同时做三件事:调研架构、写代码、写文章。
同步模式(串行,一个接一个):
调研 ████████ (3s)
写代码 ██████ (2.5s)
写文章 █████ (2s)
总耗时 ≈ 7.5s 主 Agent 全程被阻塞
异步模式(并行,三个同时跑):
调研 ████████ (3s)
写代码 ██████ (2.5s) ← 同时启动
写文章 █████ (2s)
总耗时 ≈ 3s 主 Agent 立即就能响应用户
并行只是第一个好处。 异步模式真正强大的地方在于——任务跑起来后,主 Agent 还能中途干预:
- 查进度:“researcher 那边调研得怎么样了?”
- 改需求:“跟 coder 说一声,代码再加个 Pydantic 校验”
- 喊停:“writer 那篇不用写了,取消掉”
这些在同步模式下统统做不到(同步模式下主 Agent 只能干等)。
核心代码(简化版)
from deepagents import AsyncSubAgent, create_deep_agent
# 定义异步子 Agent(graph_id 对应服务端已部署的执行图)
researcher = AsyncSubAgent(
name="researcher",
description="负责调研任务",
graph_id="research",
)
coder = AsyncSubAgent(
name="coder",
description="负责编码任务",
graph_id="coding",
)
# 用 async_subagents 参数挂上异步子 Agent
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-5-20250929",
system_prompt="你是主协调 Agent,把合适的任务派发给异步子 Agent。",
async_subagents=[researcher, coder],
)
# 主 Agent 对话时,会自动使用上面那五类异步工具
agent.invoke({"messages": [("user", "同时调研架构并写段代码")]})
本章小结:异步子智能体是 Supervisor 模式的演进——控制权依然在主 Agent,但执行从「串行死等」变成「并行 + 可中途干预」。
五、多轮记忆:让 Agent 不"失忆"
默认情况下,每次 agent.invoke() 都是全新的一轮,Agent 不记得上次说过什么。想让它"记住",加一个 checkpointer(状态存档器) 即可。
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3
# SQLite 持久化:程序关掉再开,对话历史还在
conn = sqlite3.connect("conversations.db", check_same_thread=False)
saver = SqliteSaver(conn)
agent = create_deep_agent(
model="anthropic:claude-sonnet-4-5-20250929",
tools=[get_weather],
system_prompt="你是一个乐于助人的助手。",
checkpointer=saver, # ← 关键就这一行
)
# thread_id 用来区分不同对话(不同用户、不同话题各用一个)
config = {"configurable": {"thread_id": "user-001"}}
# 带同一个 thread_id 的多次调用,共享同一段记忆
agent.invoke({"messages": [("user", "我叫小明")]}, config)
agent.invoke({"messages": [("user", "我叫什么?")]}, config)
# → "你叫小明" ← Agent 记住了!
原理一句话:每执行完一步,LangGraph 就自动把当前全部状态(消息历史、中间结果、执行到哪一步了)存进 checkpointer;下次用同一个 thread_id 调用时,自动从存档恢复。你不用手写任何"存历史、拼历史"的代码。
两种存档器,按需选择:
| InMemorySaver | SqliteSaver | |
|---|---|---|
| 存哪 | 内存 | SQLite 文件 |
| 重启后 | 丢失 | 保留 |
| 适用 | 单次脚本、快速测试 | Web 服务、生产环境 |
一个重要提醒:checkpointer 解决的是"记住",但不解决"记太多"。对话轮数一多,消息总量会撑爆模型的上下文窗口(context window)。怎么控制记忆膨胀——摘要、裁剪、限轮——正是下一章的主题。
本章小结:传一个 checkpointer + 固定 thread_id,Agent 就有了跨轮次、可持久化的记忆。
六、记忆控制:长对话如何不爆上下文
上一章让 Agent “记住了”,但 checkpointer 只管存、不管裁。如果用户跟你聊了一百轮,把全部消息一股脑发给模型,就会超出上下文窗口(比如 DeepSeek 的 64K token),直接报错。
有三种控制手段,从"智能"到"粗暴"排列:
方案一:自动摘要(生产环境首选)
DeepAgents 自带摘要中间件,token 超阈值时自动把旧消息压缩成一段摘要:
from deepagents.middleware.summarization import SummarizationMiddleware
agent = create_deep_agent(
model=model,
tools=[get_weather],
system_prompt="你是乐于助人的助手。",
checkpointer=SqliteSaver(conn),
middleware=[
SummarizationMiddleware(
model=model, # 用哪个 LLM 来生成摘要
trigger=("fraction", 0.85), # 消息占到上下文窗口 85% 时触发
keep=("messages", 6), # 保留最近 6 条不压缩
),
],
)
工作流程:每次调 LLM 前先检查 token 数 → 超过阈值就把最旧的一批消息发给 LLM 做摘要 → 用一段摘要替换掉那批旧消息 → token 回落到安全水位。整个过程对用户完全透明,且旧信息的语义不丢。
方案二:轮数限制(简单场景够用)
不想额外花钱做摘要?那就直接砍——只保留最近 N 轮:
from langgraph.graph import RemoveMessage
def trim_history(agent, config, max_rounds=3):
"""只保留最近 max_rounds 轮对话,更早的删掉"""
state = agent.get_state(config)
messages = (state.values or {}).get("messages", []) if state else []
# 找到所有"用户发言"的位置(一次用户发言 = 一轮的起点)
user_indices = [i for i, m in enumerate(messages) if m.type == "human"]
if len(user_indices) <= max_rounds:
return # 还没超,不用裁
# 从倒数第 max_rounds 个用户发言处切开,之前的全删
cutoff = user_indices[len(user_indices) - max_rounds]
removes = [RemoveMessage(id=m.id) for m in messages[:cutoff]]
agent.update_state(config, values={"messages": removes})
# 使用:每次对话前先裁到最近 3 轮
config = {"configurable": {"thread_id": "demo-1"}}
trim_history(agent, config, max_rounds=3)
agent.invoke({"messages": [("user", "继续")]}, config)
RemoveMessage是 LangGraph 的消息"墓碑"——写进去后,框架会自动把对应 ID 的消息从状态里删除。
方案三:换 thread_id(干脆重开)
最粗暴的"清空记忆":不删任何数据,只是换一个新的 thread_id。对模型来说,这就是一段零上下文的崭新对话;旧对话仍然躺在 SQLite 里,用旧 thread_id 随时能翻出来。
三种方案怎么选
| 自动摘要 | 轮数限制 | 换 thread_id | |
|---|---|---|---|
| 旧内容能回溯吗 | 能(摘要保留语义) | 不能(永久丢弃) | 能(旧 thread 还在) |
| 额外 LLM 开销 | 每次摘要一次 | 无 | 无 |
| 适合场景 | 长对话 Web 应用 | 简单 demo、限轮需求 | 彻底重开新话题 |
本章小结:长对话必须主动控制记忆——生产环境上"自动摘要",简单需求用"轮数限制",重开话题就"换 thread_id"。
七、DeepAgents 与 LangGraph:自动挡与手动挡
读到这里,你可能冒出一个问题:DeepAgents 到底能不能自己画流程图(图编排)? 答案要分两层看。
底层:DeepAgents 本身就是一张 LangGraph 图
create_deep_agent(...) 内部其实已经替你编译好了一张 LangGraph 状态图(模型节点 + 工具节点 + 子 Agent 委派循环)。你前面用到的 agent.stream()、agent.get_state()、checkpointer——全都是这张底层图的能力。
所以 DeepAgents 不是"没有图",而是图已经被封装好、藏在了内部。
应用层:它不开放"让你自己画图"的入口
你没法在 DeepAgents 的 API 里显式地定义节点、连边、加条件边,去规定「A 跑完必须走 B,B 不通过就回到 A」这种确定性流程。它的编排是固定范式:主 Agent 用 task 工具自主委派——走哪条路、调谁、调几次,全由 LLM 临场决定,而不是你用边写死。
一个精准的比喻
| DeepAgents | 原生 LangGraph | |
|---|---|---|
| 比喻 | 自动挡:省心,换挡逻辑车替你定 | 手动挡:每一挡自己踩 |
| 谁决定执行顺序 | 主 Agent 的 LLM 临场自主 | 开发者用边 / 条件边写死 |
| 会不会跳步漏步 | 可能(LLM 自由发挥) | 不会,流程 100% 确定 |
| 分支 / 循环 / 回退 | 不易保证 | 用条件边天然支持 |
| 开发成本 | 低(一行 API 搞定) | 较高(手画节点和边) |
顺带回答:普通代码也能写流程,为啥非要用 LangGraph?
很实在的疑问——像「生成 → 评审 → 修订」,用几行 if/while 就能写,何必上 LangGraph?
结论先行:简单场景,普通代码更好,上 Graph 反而是过度设计。 Graph 的价值不在"控制流程"本身,而在于——一旦把流程变成框架能接管的数据结构,它就白送你一批工程化能力:
| 能力 | 普通代码要自己造 | LangGraph 直接给 |
|---|---|---|
| ① 持久化 / 断点续跑 | 自己设计存档、每步存 DB、崩溃后手动恢复 | checkpointer 自动存,崩了从断点继续 |
| ② 人工介入审批 | 自己实现"暂停 → 等审批 → 唤醒"并保存现场 | 任意节点 interrupt 暂停,人批完再继续 |
| ③ 流式输出 | 自己在每步埋点、拼接、往前端推 | .stream() 统一吐出每步中间态 |
| ④ 状态可观测 / 回放 | 自己打日志、手动串调用链 | get_state() + 时间旅行,可回退重跑 |
| ⑤ 并行 & 汇合 | 自己写 asyncio.gather + 手动合并 |
声明式并行分支,自动汇合 |
| ⑥ 复杂拓扑 | 嵌套 if/while 越写越乱 |
节点 + 边解耦,加分支不牵连全局 |
本质:普通代码把"流程"写死在函数调用栈里;LangGraph 把"流程"变成一个显式、可序列化的状态机。一旦流程是"数据"而非"调用栈",框架就能对它存档、暂停、恢复、观测、回放、可视化——这些是普通调用栈做不到的(函数一旦在跑,中间状态没法冻结存盘、换台机器再解冻续跑)。
务实的判断标准:
| 你的场景 | 推荐 |
|---|---|
| 流程短、跑一趟就完、不用中途停 | 普通代码,别上框架 |
| 任务久、会中断、要能续跑 | DeepAgents / LangGraph(checkpointer) |
| 某步要等人确认才继续 | LangGraph(interrupt) |
| 前端要实时看进度、要能调试每步 | DeepAgents / LangGraph(stream) |
| 灵活任务、让 AI 自主决定派谁 | DeepAgents(一行 API) |
| 确定性流程:A→B→C 不许乱、不过就回退 | 原生 LangGraph(节点 + 条件边) |
深入一点:LangGraph 里的"节点"究竟是什么?
前面反复提到"节点 + 边"。新手最常见的误区是把**「一个节点」等同于「一次模型调用」**。其实不是——在 LangGraph 里,节点就是一个普通函数:输入 state → 返回部分新 state,里面干什么完全由你决定:
| 节点类型 | 里面装的是 | 会调模型吗 |
|---|---|---|
| 模型节点 | 一次 LLM 调用 | ✅ |
| 工具节点 | 执行一个函数(查库、调 API) | ❌ |
| 纯逻辑节点 | 数据处理、格式转换、拼 prompt | ❌ |
| 路由节点 | 判断下一步走哪(配合条件边) | 通常否 |
| 子图节点 | 内部嵌套另一整张图 | 可能多次 |
| Agent 节点 | 一整个子 Agent(自带工具、自主决策) | ✅ 可能多次 |
划分节点的标准,不是"一次模型调用",而是"一个你希望框架能观测 / 暂停 / 存档的有意义步骤"。 尤其重要的是——一个节点可以塞进一整个子 Agent:
# 每个子 Agent 都是完整 Agent,能自主调工具、多轮推理
researcher = create_deep_agent(model=..., tools=[search])
writer = create_deep_agent(model=..., tools=[])
def node_research(state):
result = researcher.invoke({"messages": state["messages"]})
return {"research": result["messages"][-1].content}
def node_write(state):
result = writer.invoke({"messages": [{"role": "user",
"content": f"根据资料写文章:{state['research']}"}]})
return {"article": result["messages"][-1].content}
g.add_edge(START, "research")
g.add_edge("research", "write") # 用边把顺序写死
g.add_edge("write", END)
这就是用 LangGraph 手工搭多智能体系统的标准姿势:把每个 Agent 做成一个节点,再用边规定它们怎么交接。五种架构都能这么落地:
| 架构 | 怎么搭 |
|---|---|
| 主从 / Supervisor | 一个 supervisor 节点用条件边路由到多个 worker 节点 |
| 网络 / Swarm | 多个 Agent 节点用条件边互相 handoff |
| 层级 | Agent 节点内部再嵌一张子图 |
| 角色流水线 | researcher → writer → reviewer 三个节点串成边 |
DeepAgents 的委派 = LLM 临场决定调谁;LangGraph 的 Agent 节点 = 开发者用边写死顺序。区别还是那句——「谁来控制流程」。
实例:用 LangGraph 搭一个 Swarm(去中心化协作)
Swarm 和 Supervisor 最大的不同:没有中心主管。Agent 之间平级,每个 Agent 自己决定"下一步交给谁"——这个动作叫 handoff(交接)。核心实现:
class SwarmState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages] # 所有 Agent 共写同一段对话
active: str # 当前该谁干活
steps: int # 步数计数,防止无限踢皮球
def route(state):
"""条件边的路由函数——这是 Swarm 的心脏"""
if state["steps"] >= MAX_STEPS:
return "end" # 兜底:防两个 Agent 互相推来推去
return state["active"] # 路由到当前活跃的 Agent
# 每个 Agent 在输出里带上 "HANDOFF: <目标>",表示"我决定交给谁"
# 条件边读 active 字段,把控制权平级转给对应 Agent 节点
g.add_conditional_edges("triage", route, routes)
g.add_conditional_edges("weather_expert", route, routes)
| Supervisor(主从) | Swarm | |
|---|---|---|
| 谁决定下一个 | 中心主管统一派 | 当前 Agent 自己交接 |
| 拓扑形状 | 星形(主 → 各下属) | 网状(平级互转) |
| 控制权 | 始终回到主管 | 在 Agent 间流转,无中心 |
生产环境直接用官方
langgraph-swarm库(create_swarm+create_handoff_tool开箱即用);手写只是为了让你看清底层原理。
怎么观测和调试:Graph 给你的四层"透视镜"
"能看清内部运行"正是用 LangGraph 而非裸代码的一大价值。从简单到专业分四层:
| 层次 | 方法 | 能看到什么 |
|---|---|---|
| ① 逐节点观测 | graph.stream(..., stream_mode="updates") |
每个节点改了哪些状态(debug 最常用) |
| ② 结构可视化 | graph.get_graph().draw_ascii() |
图长什么样、有哪些节点和分支 |
| ③ 历史回放 | checkpointer + get_state_history() |
每一步的完整快照,可"时间旅行"回退 |
| ④ 生产级追踪 | 设 LANGSMITH_TRACING=true |
每次 LLM 调用的 prompt、耗时、token、费用 |
# ① 最常用:看每一步到底改了什么
for chunk in graph.stream({"topic": "多智能体"}, stream_mode="updates"):
print(chunk) # {"draft": {"draft": "文章草稿内容..."}}
# ③ 时间旅行:逐步回看,甚至从中间某步重跑
for snap in graph.get_state_history(config):
print(snap.next, list(snap.values.keys()))
平时调流程,
stream_mode="updates"最直接;要排查 prompt / 性能 / 费用,上 LangSmith(设个环境变量即可,代码一行都不用改)。
本章小结:DeepAgents 是封装好的"自动挡",底层就是 LangGraph;需要确定性流程、人工审批、复杂拓扑时,就切到"手动挡"直接用 LangGraph。
八、总结
从一个"查天气小助手"起步,我们一路走过了工具调用、子 Agent 分工、异步并行、多轮记忆、记忆控制,最后摸到了底层的 LangGraph。DeepAgents 让你用很少的 Python 代码,就能把多个 AI 组织成一支可协作、可记忆、可干预的团队。
核心能力速查
| 能力 | 一行代码 |
|---|---|
| 单 Agent + 工具 | create_deep_agent(model=..., tools=[...]) |
| 子 Agent 委派 | create_deep_agent(..., subagents=[...]) |
| 异步子 Agent | create_deep_agent(..., async_subagents=[AsyncSubAgent(...)]) |
| 多轮记忆 | create_deep_agent(..., checkpointer=SqliteSaver(conn)) |
| 自动摘要 | create_deep_agent(..., middleware=[SummarizationMiddleware(...)]) |
什么时候该用 DeepAgents?
| 场景 | 用吗 |
|---|---|
| 一个模型 + 简单 prompt 就能搞定 | ❌ 杀鸡不用牛刀 |
| 要调外部工具(API / 数据库 / 文件) | ✅ |
| 要多个 AI 角色分工协作 | ✅ |
| 要跨轮次的长对话记忆 | ✅ |
| 要并行跑多任务、还能中途干预 | ✅ |
什么时候该切到 LangGraph?
DeepAgents 是"自动挡",够用且省心。但遇到下面这些,就该下沉到 LangGraph 手动编排(详见第七章):
- 确定性流程:A 之后必须 B、评审不通过就回退到 A
- 人工审批节点:某步要等人点头才能继续
- 复杂拓扑:多分支、多循环、去中心化 Swarm
除此之外,一行 create_deep_agent 就能把你带得很远。
- 点赞
- 收藏
- 关注作者
评论(0)