在构建复杂的 AI 应用时，单体大模型（Single Agent）往往会显得力不从心。它可能会忘记上下文、在复杂流程中出现偏移，或者无法同时协调多类专业工具。这时，多智能体系统（Multi-Agent System, MAS） 就成为一种更合适的组织方式。

在多智能体架构中，一种常见的设计模式是**“主从架构（Main-Sub Architecture）”**：一个主 Agent 负责全局规划和任务分发，多个 SubAgent 负责具体执行。

本文以 LangGraph 为例，逐步说明主 Agent 如何动态派生（Dynamic Spawning） SubAgent。

---

为什么要“动态派生”？

在早期的 LangChain 或简单的多 Agent 框架中，很多 Agent 之间的连线是静态的。比如：“如果意图是写代码，就交给 Coder Agent；如果是查资料，就交给 Search Agent”。

但真实世界的任务往往更复杂：
“请帮我调研五家主流的新能源车企，并分别写一份财报分析。”

主 Agent 面临的不仅是对“专业工种”的选择，而是需要在运行时才知道**“要创建多少个”类似的 SubAgent 并行或串行工作。这就是动态派生**的应用场景：主 Agent 根据输入内容，在运行时动态生成任意数量的子任务，并挂载相应的 SubAgent 去独立执行。

上图是概念图：它表达的是“主 Agent 在运行时把一个请求拆成多个执行分支”。在 LangGraph 的静态图定义里，通常只有一个 worker 节点；图中的 Worker 1 / 2 / N 更准确地说，是这个节点在运行时被多次调度执行的效果。

---

理解 LangGraph 中的 Send API

LangGraph 的核心思想是将工作流建模为一张状态图（State Graph）。在图中，节点代表函数或 Agent，边代表流程走向。

要在 LangGraph 中实现动态派生（一对多的并行分支），核心在于其提供的 Send 以及 Map-Reduce 机制。
传统的一条边只是将当前状态单纯地传递给下一个节点；而通过返回多个 Send 对象，我们可以在运行时决定把哪些数据分发给哪些节点实例，从而实现动态并行。

---

实战演练：构建一个任务拆解与分发系统

我们假设一个场景：用户输入一个需要拆分的复合需求。
主 Agent（Planner） 负责把需求拆分成多个子任务。
SubAgent（Worker） 负责独立完成每一个子任务。

1. 定义状态 (State)

多智能体沟通的桥梁是状态。我们需要一个全局状态来保存总体结果，也需要一个属于 SubAgent 的局部状态。

from typing import Annotated, List
from typing_extensions import TypedDict
import operator

# SubAgent 的状态定义（局部状态）
class WorkerState(TypedDict):
    task_id: str
    task_desc: str

# 全局状态定义
class GlobalState(TypedDict):
    main_request: str
    # 存放子任务的详细信息
    tasks: List[WorkerState]
    # 收集并行 SubAgent 执行完毕后的结果
    # 使用 operator.add 确保并行结果可以安全地合并（Reduce）到列表中
    results: Annotated[List[str], operator.add]

2. 构建主 Agent (Planner)

主 Agent 的任务很纯粹：拿到 main_request，动用大模型进行全局理解，生成任务清单，组装成字典列表并返回，更新 tasks 状态。

from langchain_openai import ChatOpenAI
from langchain_core.pydantic_v1 import BaseModel, Field

llm = ChatOpenAI(model="gpt-4o")

class TaskList(BaseModel):
    task_desc_list: List[str] = Field(description="List of sub-tasks")

def planner_node(state: GlobalState):
    request = state["main_request"]
    prompt = f"Break down the following request into distinct, actionable sub-tasks: {request}"
    
    # 强制大模型以 JSON 格式输出独立子任务列表
    structured_llm = llm.with_structured_output(TaskList)
    response = structured_llm.invoke(prompt)
    
    # 构造将要派生给 Worker 的状态列表
    tasks = [{"task_id": f"t_{i}", "task_desc": desc} for i, desc in enumerate(response.task_desc_list)]
    
    return {"tasks": tasks}

3. 实现动态派生逻辑 (Dynamic Routing)

这一步是实现动态派生的关键。我们不将 planner_node 静态地连接到 worker_node，而是通过一条条件边（Conditional Edge），动态生成并行分支。

利用 LangGraph 的 Send API，我们可以把一个含有多个实体的数组，映射成多个并发的流：

from langgraph.types import Send

def trigger_subagents(state: GlobalState):
    # 遍历主 Agent 刚刚拆分好的任务清单
    tasks = state.get("tasks", [])
    
    # 为每一个任务动态派生一个名为 "worker" 的目标节点实例，并注入专属于该任务的 WorkerState
    return [Send("worker", task) for task in tasks]

这里的 Send("worker", task) 表示“把 task 作为局部输入发送给 worker 节点执行”。如果 tasks 中有 5 项内容，LangGraph 会在同一轮里调度 5 次对 worker 的执行；这些执行通常可以并发推进，但它们本质上仍然是同一个节点的多次运行，而不是静态图中预先存在的 5 个不同节点。

4. 构建 SubAgent (Worker)

SubAgent 接收的是经过 Send 拆分后的局部数据，即 WorkerState。它执行完毕后，将结果放入 results 字段中。LangGraph 的 operator.add reducer 会把这些独立执行结果汇总（Reduce）到一个公共集合里。

def worker_node(state: WorkerState):
    task_id = state["task_id"]
    task_desc = state["task_desc"]
    
    # 实际场景中这里通常是另一个完整的 LangGraph 子图 (CompiledGraph) 或带 Tools 的 Agent
    # 为方便演示，我们直接调用 LLM 处理
    result = llm.invoke(f"Execute specific task [{task_id}]: {task_desc}")
    
    # 注意，由于 GlobalState 中 results 定义了 operator.add 归约，这里返回由单个元素组成的列表即可
    return {"results": [f"[{task_id}]: {result.content}"]}

5. 编排并编译状态图

把所有的节点和边粘合到 StateGraph 的画布上。

from langgraph.graph import StateGraph, START, END

# 使用 GlobalState 初始化主图
builder = StateGraph(GlobalState)

# 注册节点
builder.add_node("planner", planner_node)
builder.add_node("worker", worker_node, input_schema=WorkerState)

# 设置逻辑连线
builder.add_edge(START, "planner")

# 添加动态派生所需的条件边
builder.add_conditional_edges("planner", trigger_subagents, ["worker"])

# 所有并行的 worker 实例运行结束后，结束整体图
# 也可以在这里承接一个汇总节点总结全盘 (summary_node)
builder.add_edge("worker", END)

# 编译生成可执行图
multi_agent_app = builder.compile()

上面的代码在静态定义层只有一个 planner 和一个 worker 节点；但在运行时展开时，会表现出“一对多派发”的效果。下面这张图更准确地说，是运行时展开示意图，不是 get_graph() 直接画出来的唯一静态结构：

如果你用 LangGraph 的图可视化能力去查看静态图，你看到的更接近下面这种结构：

二者并不矛盾：前者强调运行时 fan-out 效果，后者强调编译后的节点定义。

运行过程

当我们向框架输入一个综合请求：

inputs = {"main_request": "为一次三天两夜的北京求职兼旅游做计划，包含：规划交通，列出必打卡的美食，整理一份互联网大厂面试要点。"}
final_state = multi_agent_app.invoke(inputs)

for res in final_state["results"]:
    print(res)

在执行过程中，LangGraph 会按如下顺序推进：

1. Planner 扫描需求，将其拆解为 3 项明确的子任务，并生成了字典结构。
2. trigger_subagents 条件函数读取这些任务，动态生成了 3 个 Send 对象。
3. LangGraph 捕捉到这些 Send 对象投递的目标均为 “worker”，随后会对同一个 worker 节点调度 3 次独立执行；这些执行各自携带自己的局部输入，因此效果上就像 3 个拥有独立上下文的 Worker 在并行工作。
4. 各次 worker 执行完成后，结果会依据 Annotated[..., operator.add] 自动规约到状态中的 results 列表。

---

官方语义 vs 直觉理解

在学习 LangGraph 时，很多概念会先以“直觉模型”的方式进入脑中，这有助于快速理解；但在实际实现和排查问题时，最好再切换回更接近官方定义的表述。下面是一张对照表：

直觉理解	更接近官方语义的表述
主 Agent 创建了多个 worker 节点	静态图中通常只有一个 worker 节点；运行时会对它进行多次调度执行
Send 会“实例化一个新 Agent”	Send 的作用是把一个局部输入发送到指定节点，在下一步触发该节点执行
每个 SubAgent 都拥有一份完整全局状态	被 Send 发送的输入可以只是局部状态，且可以与主图状态结构不同
条件边就是普通 if/else 跳转	条件边返回的是下一步要走的目标，可以是单个节点、多个节点，或多个 Send
图里画出 worker1 / worker2 / worker3 就代表真实静态结构	这种画法通常是运行时 fan-out 的示意，不一定等于 get_graph() 看到的静态编译图
Reducer 只是结果拼接技巧	Reducer 是状态合并机制，用于定义多个节点更新同一 state key 时如何聚合
多个 Send 一定意味着绝对并行	更准确地说，是同一轮中的多任务调度；通常可并发推进，但具体执行细节还取决于运行时与配置
worker 一定是一个独立 Agent 对象	在 LangGraph 里，worker 更常见的形态是一个节点函数、一个子图，或一个可调用对象

如果只是为了建立第一层理解，使用“主 Agent 派生多个 SubAgent”这样的说法没有问题；但如果要进一步理解可视化、调试、状态规约、子图嵌套等机制，上表右侧的表述会更准确。

---

进阶探讨：设计哲学

通过上文基于 LangGraph 的实现例程，我们可以提炼出动态派生 SubAgent 的三个关键要素：

1. 状态隔离 (State Isolation & Scalability)
   动态派生的每个 SubAgent 仅能接收和看见自己专有的 WorkerState 数据，并不需要加载庞大冗杂的 GlobalState。这除了节省 API Key 的 Token 开销，更极大地收拢了 LLM 的注意力重点，避免上下文幻觉。

2. 自动伸缩 (Dynamic Map-Reduce)
   这种设计不需要把执行分支数量写死在图结构里。主 Agent 规划出 2 个子任务，就调度 2 次执行；规划出 100 个子任务，就调度 100 次执行。相比硬编码的条件控制结构，这种方式更适合任务数量在运行时才确定的场景。

3. 智能体分形 (Fractal Complexity)
   文中的 worker 只是一个简单的 LLM 调用示例。在实际项目中，worker_node 也可以是一个带工具调用能力的 Agent，或者一个已经编译好的 CompiledGraph 子图。也就是说，运行时被调度的执行单元，本身还可以继续包含更复杂的内部流程。

结语

多智能体系统的难点，往往不只在模型能力本身，也在于任务拆分、路由、状态管理和结果合并。LangGraph 通过 StateGraph、条件边、Send 和 reducer 提供了一套较清晰的实现方式，用来表达“动态路由与派发”这类模式。

如果把握住以下几点，这一机制就比较容易理解：

• 静态图定义的是节点与边；
• 运行时 fan-out 体现的是同一节点被多次调度；
• Send 负责把局部输入送到目标节点；
• reducer 负责把多路结果合并回共享状态。

从这个角度看，“主 Agent 动态派生 SubAgent”可以被理解为一种工程表达；而在 LangGraph 的官方语义中，它更接近于：一个状态图在运行时根据条件边返回的 Send，对某个节点或子图发起多次局部输入驱动的执行。