这篇文章只回答一个问题：DeepAgents 到底帮我们隐藏了什么？
为了把调用关系讲清楚，全文使用一个自洽的“资料研究与报告生成”任务。用户给出一个主题，多个子 Agent 分工处理，Coordinator 负责调度，最后生成报告。这里的代码是帮助理解机制的伪代码，不要求复制后直接运行。
一、先看全局关系
01--ScreenShot_2026-08-27_232017_061.png
关键点是：子 Agent C 不直接调用 A、B；它们都通过 Coordinator 被调度。 Coordinator 接收 A、B 的结果，判断依赖是否满足，然后才把合并后的输入交给 C。
把这个系统看成一个有依赖关系的任务图，会比把它看成一串普通函数调用更准确：
02--ScreenShot_2026-08-27_232005_943.png
这张图只保留数据依赖，调度关系见上一张总览图。B 不读取 A 的输出，而是从独立来源核验同一主题，因此 A 和 B 之间没有输入依赖，可以并行；C 同时依赖 A 和 B，必须等待二者都完成；Writer 依赖 C 的分析结果。
知识地图可以先记成一条“调用链”：
Coordinator
│ 调用普通工具，例如 search_docs
│
└─ 调用 task(researcher)
│
└─ researcher Agent 独立完成任务并返回结果
│
└─ Coordinator 再调用 reviewer、analyst 或 writer
上面每句话的具体含义：
| 原句 | 直白解释 | 本文中的例子 |
|---|---|---|
| LLM 能调用工具 | 模型可以决定调用一个函数，并读取函数返回值 | 调用 search_docs 查资料 |
| Agent 能调用工具 | 给模型加上提示词、工具和循环，让它能完成一项工作 | researcher 使用 search_docs
|
Coordinator 能调用 task
|
主 Agent 拥有一个特殊工具，可以把工作交给另一个 Agent | Coordinator 调用 task("researcher")
|
task 能启动子 Agent |
DeepAgents 根据名字找到对应 Agent，启动它的独立执行过程 | 启动 researcher、reviewer 或 analyst |
| 多个子 Agent 通过 Coordinator 协同 | 子 Agent 不直接互相调用，Coordinator 接收结果、检查依赖并决定下一步 | A、B 返回后，Coordinator 才调用 C |
一次完整调用可以这样理解：
1. 用户：研究向量数据库
2. Coordinator：调用 task("researcher") 和 task("reviewer")
3. researcher / reviewer：各自工作，返回结果
4. Coordinator：把两份结果传给 task("analyst")
5. analyst：返回综合分析
6. Coordinator：把分析交给 writer，返回最终报告
二、最小的 DeepAgents 代码
下面是接近真实 DeepAgents API 的高度概括版。它表达结构，不是复制后即可运行的完整示例。
from deepagents import create_deep_agent
from langchain_core.tools import tool
@tool
def search_docs(query: str) -> str:
"""在内部资料库中搜索内容。"""
return search_database(query)
agent_a = {
"name": "researcher",
"description": "负责查找资料并提取事实",
"system_prompt": "你是资料研究员，只输出有来源的事实。",
"tools": [search_docs],
}
agent_b = {
"name": "reviewer",
"description": "从独立来源核验同一主题的事实，寻找遗漏和风险",
"system_prompt": "你是审查员。独立搜索和核验同一主题，不读取 researcher 的输出，不依赖其他 Agent 的结论。",
"tools": [search_docs],
}
agent_c = {
"name": "analyst",
"description": "负责综合 researcher 和 reviewer 的结果",
"system_prompt": "你是分析师，只有在收到上游两份结果后才开始分析。",
}
writer = {
"name": "writer",
"description": "负责把研究结果写成最终报告",
"system_prompt": "你是编辑，只根据上游结果写报告，不要编造事实。",
}
coordinator = create_deep_agent(
model="openai:gpt-4o-mini",
tools=[],
system_prompt="""
你是 Coordinator，负责拆解任务和调度子 Agent。
researcher 与 reviewer 都只依赖用户主题，没有互相依赖，可以并行调用。
analyst 同时依赖 researcher 和 reviewer，必须等两者完成。
writer 依赖 analyst，收到分析结论后再生成最终报告。
传递给下游 Agent 的内容要精简、结构化，并保留来源和失败状态。
""",
subagents=[agent_a, agent_b, agent_c, writer],
)
result = coordinator.invoke({
"messages": [
{"role": "user", "content": "研究向量数据库的优缺点并写一页报告"}
]
})
代码中最重要的只有两处：
subagents=[agent_a, agent_b, agent_c, writer]
这告诉 DeepAgents：Coordinator 可以调度哪些子 Agent。它是一个可调度对象的注册列表，不等于已经执行了这些 Agent。
coordinator = create_deep_agent(...)
这创建的是主 Agent。Coordinator 不一定要是额外定义的 Python 类；它可以是一个拥有子 Agent 调度能力、并根据状态做路由决策的 Agent。
三、等价伪代码：DeepAgents 实际替你做了什么
上面的 create_deep_agent 可以粗略理解成下面这些隐藏步骤：
def create_deep_agent(model, tools, system_prompt, subagents):
# 1. 把子 Agent 放进注册表
registry = {
agent["name"]: build_child_agent(agent)
for agent in subagents
}
# 2. 框架自动创建 task 工具
task_tool = make_task_tool(registry)
# 3. Coordinator 能看到普通工具 + task 工具
coordinator_tools = tools + [task_tool]
# 4. 创建一个能循环执行的主 Agent
return build_agent(
model=model,
tools=coordinator_tools,
system_prompt=system_prompt,
)
重点是这一句：
task_tool = make_task_tool(registry)
你的代码通常看不到 task 的定义，因为 DeepAgents 根据 subagents= 自动创建它。
四、task 到底是什么
从使用角度，task 可以想象成这样一个工具：
def task(
subagent_type: str,
description: str,
context: dict | None = None,
) -> str:
"""让指定的子 Agent 执行一项独立任务。"""
child_agent = registry[subagent_type]
child_input = {
"messages": [
{
"role": "user",
"content": description,
}
],
"context": context,
}
child_result = child_agent.invoke(child_input)
# 把子 Agent 的最终输出作为工具结果返回给 Coordinator
return extract_final_answer(child_result)
注意：这段是等价伪代码，不是 DeepAgents 的源码。它帮助你理解调用关系：
Coordinator 的 LLM 产生 task 工具调用
↓
DeepAgents 根据 subagent_type 找到子 Agent
↓
用 description/context 启动子 Agent
↓
子 Agent 完成任务
↓
结果包装成 tool result 返回 Coordinator
五、Coordinator 如何调度 A、B、C
Coordinator 需要维护的不是简单的调用顺序，而是每个任务的状态和依赖：
jobs = {
"researcher": {
"depends_on": [],
"status": "pending",
},
"reviewer": {
"depends_on": [],
"status": "pending",
},
"analyst": {
"depends_on": ["researcher", "reviewer"],
"status": "pending",
},
"writer": {
"depends_on": ["analyst"],
"status": "pending",
},
}
调度器每轮只派发“依赖已满足”的任务。A 和 B 都只依赖用户主题，因此同时就绪、可以并行；C 必须等待 A、B 都成功：
def ready_jobs(jobs):
return [
name
for name, job in jobs.items()
if job["status"] == "pending"
and all(jobs[parent]["status"] == "succeeded"
for parent in job["depends_on"])
]
while not all_done(jobs):
ready = ready_jobs(jobs)
# researcher 和 reviewer 都没有上游依赖，可以并行派工
for name in ready:
jobs[name]["status"] = "running"
schedule_with_task(
subagent_type=name,
description=make_description(name, results),
context=make_context(name, results),
)
completed = collect_finished_tasks()
for name, outcome in completed:
results[name] = outcome.value
jobs[name]["status"] = (
"succeeded" if outcome.ok else "failed"
)
# 下一轮才会发现 analyst 已经满足两个依赖
调度过程可以用下面的流程图表示：
03---ScreenShot_2026-08-27_231941_566.png
三个概念要区分：
| 概念 | 含义 |
|---|---|
| 注册 | Coordinator 被允许调用某个子 Agent |
| 调度 | Coordinator 决定现在调用哪个子 Agent |
| 依赖 | 某个子 Agent 开始前必须已经得到哪些结果 |
因此，把 C 放进 subagents 列表，只表示 C 可被调用；“C 必须等待 A 和 B”需要由调度逻辑、状态检查或图的边来表达。
运行时可以简化为：Coordinator 派发 A、B 后等待两份结果；如果其中一个失败，则进入重试或恢复分支；两者成功后才派发 C，最后交给 Writer。
六、运行时究竟发生什么
用户输入：
研究向量数据库的优缺点并写一页报告
Coordinator 先产生两个互不依赖的工具调用。A 和 B 都从用户主题开始，概念上可以并行执行：
{
"name": "task",
"args": {
"subagent_type": "researcher",
"description": "查找向量数据库的主要优点、缺点和应用场景",
"context": {
"topic": "向量数据库"
}
}
}
{
"name": "task",
"args": {
"subagent_type": "reviewer",
"description": "从独立来源核验向量数据库的常见优缺点，寻找遗漏和风险",
"context": {
"topic": "向量数据库"
}
}
}
DeepAgents 执行 researcher 和 reviewer 后，分别返回：
Tool Result:
向量数据库适合相似度检索；主要成本是索引维护和存储开销……
Tool Result:
需要区分检索效果、索引维护成本和数据更新延迟；结论应标记证据范围……
Coordinator 收齐两个结果后，才调用依赖它们的 analyst：
{
"name": "task",
"args": {
"subagent_type": "analyst",
"description": "根据 researcher 和 reviewer 的结果分析适用场景、权衡和风险",
"context": {
"topic": "向量数据库",
"research_result": "向量数据库适合相似度检索；主要成本是索引维护和存储开销……",
"review_result": "需要区分检索效果、索引维护成本和数据更新延迟……",
}
}
}
最后再调用 writer：
task(
subagent_type="writer",
description="把研究结果和分析结果整理成一页 Markdown 报告",
context={
"analysis_result": "...",
},
)
七、Coordinator 与子 Agent 如何传递信息
默认情况下，子 Agent 是独立上下文，不会自动看到 Coordinator 的完整聊天记录。
常见的三种传递方式：
方式 1：放进 description
适合短文本和明确任务：
task(
subagent_type="analyst",
description=f"分析以下研究结果：{research_result}",
)
方式 2：放进 context
适合结构化字段：
task(
subagent_type="analyst",
description="分析研究结果",
context={
"topic": "向量数据库",
"research_result": research_result,
},
)
方式 3：通过共享文件
适合大文本、表格、图片或生成的脚本：
researcher 写入 workspace/research.md
↓ 只传文件路径
Coordinator 调用 analyst：请读取 workspace/research.md
共享文件适合传递大文本、表格、图片或脚本；下游 Agent 仍需被明确告知文件位置、格式和读取要求。
八、Coordinator 不是“写死的流程引擎”
下面两种写法要区分：
写死流程：普通程序编排
research = researcher.invoke("查资料")
review = reviewer.invoke("独立核查")
analysis = analyst.invoke({"research": research, "review": review})
report = writer.invoke({"analysis": analysis})
这时流程由程序和固定调用顺序决定，稳定、可测试，但变化任务需要修改程序。
LLM 决策流程：Coordinator 编排
coordinator = create_deep_agent(
subagents=[agent_a, agent_b, agent_c, writer],
system_prompt="根据任务决定调用哪些专家；遵守依赖关系并汇总结果",
)
这时 Coordinator 的 LLM 可以判断：
- 要不要调用 researcher
- 是否需要 reviewer 做独立核查
- 哪些任务可以并行，例如没有互相输入依赖的 researcher 和 reviewer
- analyst 的依赖是否已经满足
- 结果不足时是否重试或补充任务
- 最后是否调用 writer
灵活性更高，但路由结果具有概率性。生产系统通常还会加入结构化输出、允许调用的 Agent 列表、依赖校验、权限控制、超时、重试和失败分支。
九、tool、task、MCP 的关系
@tool
把 Python 函数包装成 LangChain Tool
task
DeepAgents 自动提供的“调用子 Agent”工具
MCP
一个跨进程、跨语言的工具通信协议
在这个例子里：
@tool
def search_docs(query: str) -> str:
...
这是普通 LangChain 工具；而：
subagents=[agent_a, agent_b, agent_c, writer]
会让 DeepAgents 自动向 Coordinator 暴露 task 工具。两者都能被 LLM 调用，但 task 的目标是启动另一个 Agent。
十、把整个机制压缩成 20 行伪代码
registry = {
"researcher": agent_a,
"reviewer": agent_b,
"analyst": agent_c,
"writer": writer,
}
while True:
coordinator_message = coordinator_llm(messages)
if coordinator_message.is_final_answer:
return coordinator_message.content
if coordinator_message.tool_name == "task":
name = coordinator_message.args["subagent_type"]
description = coordinator_message.args["description"]
context = coordinator_message.args.get("context")
child_result = registry[name].invoke(
description=description,
context=context,
)
messages.append(coordinator_message)
messages.append({
"role": "tool",
"name": "task",
"content": child_result,
})
这就是核心思想：Coordinator 是循环调用 LLM 的主循环，task 是这个主循环可以使用的一个特殊工具。
十一、等价的 LangGraph 伪代码
DeepAgents 可以把很多调度细节封装起来；如果把同一个任务显式展开成 LangGraph，通常会看到状态、节点、边和条件路由：
[图片上传失败...(image-3b648a-1787843863847)]
对应的 LangGraph 风格伪代码如下。这里的 Send 表示把同一轮中互不依赖的工作分发出去；具体 API 细节可能随 LangGraph 版本变化。
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
class State(TypedDict, total=False):
topic: str
results: dict
status: dict
phase: str
next_jobs: list[str]
final_report: str
def coordinator_node(state: State) -> dict:
"""根据状态找出就绪任务，并决定下一跳。"""
status = state.get("status", {})
if not status:
return {
"status": {
"researcher": "pending",
"reviewer": "pending",
"analyst": "pending",
"writer": "pending",
},
"phase": "after_ab",
"next_jobs": ["researcher", "reviewer"],
}
if state["phase"] == "after_ab":
# join_ab 已经保证两个分支都返回，这里再检查结果是否成功
if (status["researcher"] != "succeeded"
or status["reviewer"] != "succeeded"):
return {"next_jobs": ["recovery"]}
return {"phase": "after_c", "next_jobs": ["analyst"]}
if state["phase"] == "after_c" and status["analyst"] == "succeeded":
return {"next_jobs": ["writer"]}
return {"next_jobs": ["recovery"]}
def route_from_coordinator(state: State):
"""把就绪任务映射为图中的节点或并行 Send。"""
jobs = state.get("next_jobs", [])
return [
Send(job + "_node", state)
for job in jobs
]
def researcher_node(state: State) -> dict:
result = run_researcher(state["topic"])
return merge_result(state, "researcher", result)
def reviewer_node(state: State) -> dict:
result = run_reviewer(state["topic"])
return merge_result(state, "reviewer", result)
def join_ab_node(state: State) -> dict:
"""汇合 A/B 的结果；只有两个上游分支都完成后才进入。"""
return {}
def analyst_node(state: State) -> dict:
# 只有 coordinator 确认 A/B 成功后，才会进入这个节点
result = run_analyst(
topic=state["topic"],
research=state["results"]["researcher"],
review=state["results"]["reviewer"],
)
return merge_result(state, "analyst", result)
def writer_node(state: State) -> dict:
report = run_writer(
topic=state["topic"],
analysis=state["results"]["analyst"],
)
return {
"final_report": report,
"status": {**state["status"], "writer": "succeeded"},
}
def recovery_node(state: State) -> dict:
"""记录失败原因；实际系统也可以在这里重试或切换替代 Agent。"""
return {
"status": {**state["status"], "workflow": "blocked"},
"error": "上游任务失败，无法满足后续依赖",
}
graph = StateGraph(State)
graph.add_node("coordinator_node", coordinator_node)
graph.add_node("researcher_node", researcher_node)
graph.add_node("reviewer_node", reviewer_node)
graph.add_node("join_ab_node", join_ab_node)
graph.add_node("analyst_node", analyst_node)
graph.add_node("writer_node", writer_node)
graph.add_node("recovery_node", recovery_node)
graph.add_edge(START, "coordinator_node")
graph.add_conditional_edges(
"coordinator_node",
route_from_coordinator,
{
"researcher_node": "researcher_node",
"reviewer_node": "reviewer_node",
"join_ab_node": "join_ab_node",
"analyst_node": "analyst_node",
"writer_node": "writer_node",
"recovery_node": "recovery_node",
},
)
graph.add_edge(["researcher_node", "reviewer_node"], "join_ab_node")
graph.add_edge("join_ab_node", "coordinator_node")
graph.add_edge("analyst_node", "coordinator_node")
graph.add_edge("writer_node", END)
graph.add_edge("recovery_node", END)
app = graph.compile()
result = app.invoke({
"topic": "向量数据库的优缺点",
})
并行节点共同写入状态时，results 和 status 需要使用合并规则，避免 A 的更新覆盖 B 的更新。上面用 merge_result(...) 抽象了这个 reducer 行为；它应该只返回当前节点产生的增量，而不是简单覆盖整个 State。
这份 LangGraph 伪代码把 DeepAgents 隐藏的概念逐一显式化：
| DeepAgents 概念 | LangGraph 中的显式结构 |
|---|---|
| Coordinator Agent |
coordinator_node + 条件路由 |
subagents= 注册表 |
节点注册：add_node(...)
|
task 工具调用 |
路由到目标节点，或使用 Send 分发 |
| 子 Agent 独立执行 |
researcher_node、reviewer_node 等节点 |
| tool result 回传 | 节点返回值合并进 State
|
| C 依赖 A、B |
coordinator_node 检查两个 status 后再路由到 C |
| 最终答案 |
writer_node 写入 final_report 后到 END
|
两者的关系可以压缩成一句话：DeepAgents 提供更高层的 Agent 调度体验；LangGraph 让状态、节点和路由边完全显式。
十二、失败、重试和权限也属于调度设计
真实调度器不能只考虑成功路径，还要回答三个问题：失败后怎么办、谁可以被调用、结果是否可信。
对应的调度约束可以写成：
def can_schedule(name, jobs, permissions):
job = jobs[name]
if name not in permissions.allowed_subagents:
return False, "没有调用权限"
if any(jobs[parent]["status"] != "succeeded"
for parent in job["depends_on"]):
return False, "依赖尚未完成"
if job["attempts"] >= job["max_attempts"]:
return False, "超过重试次数"
return True, "ready"
这也是为什么“让 LLM 自己决定下一步”通常还不够。LLM 可以提出调度意图，但状态机或 Coordinator 外围的约束层应该验证权限、依赖、预算和重试次数。
十三、三个最重要的工程注意点
-
不要把所有历史对话都塞进
context：上下文越大，成本和出错概率越高。 - 尽量传结构化结果或文件路径：比传一大段自然语言更容易被下游 Agent 正确使用。
- 不要把 LLM 的路由判断当成绝对可靠：重要流程应增加允许调用的 Agent 列表、输出 Schema、超时和失败处理。
Mental Model Summary
| 元素 | 压缩理解 |
|---|---|
| Coordinator / Supervisor | 负责判断“下一步找谁做什么”的主 Agent |
| 子 Agent | 负责某一类专业工作的独立 Agent |
task |
DeepAgents 自动提供的子 Agent 调度工具 |
description/context |
Coordinator 传给子 Agent 的任务和数据 |
| tool result | 子 Agent 返回给 Coordinator 的结果 |
subagents= |
告诉 DeepAgents 可以调度哪些子 Agent |
| 依赖图 | 表达哪些任务可以并行、哪些任务必须等待 |
| LangGraph State | 显式保存结果、状态和路由依据 |
Challenge Questions
- 如果
subagents=[]，Coordinator 还会有task工具吗？为什么？ - 为什么大文件更适合通过共享路径传递，而不是直接放进
context？ - 如果 A 和 B 都已完成，但 C 仍然没有被调用，应该优先检查注册表、依赖状态，还是 Writer 的提示词？
- 如果你要求流程必须严格按“研究 → 分析 → 写作”执行，应该使用 Coordinator 自主决策，还是普通程序写死流程？
- 如果 researcher 失败而 reviewer 成功，Coordinator 应该重试 researcher、调用替代 Agent，还是直接终止？请说明你的判断依据。
