多 Agent 与子图:把复杂问题拆成团队

什么时候要多 Agent

• 工具太多:单 Agent 给 30+ 工具 → LLM 选错、prompt 爆炸
• 职责分明:研究员、程序员、QA——prompt 和工具集差别大
• 独立演进:SQL Agent 归数据团队、邮件 Agent 归营销团队,各自发版
• 分层规划:Planner 先拆任务,Executor 再干活

三种编排模式

模式一:Supervisor

from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.types import Command
from typing import Literal

# 1) 造两个专员
researcher = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[search_web, search_kb],
    prompt="你是资料员,只负责检索和总结资料,不要做决策。",
    name="researcher",
)

coder = create_react_agent(
    model="openai:gpt-4o-mini",
    tools=[write_file, run_python],
    prompt="你是程序员,只负责写/跑代码。",
    name="coder",
)

# 2) Supervisor 节点:决定派给谁
def supervisor(state) -> Command[Literal["researcher", "coder", END]]:
    msg = state["messages"][-1].content
    decision = llm.invoke([
        ("system", "根据用户最新消息,回答 researcher / coder / FINISH 三选一"),
        ("user", msg),
    ]).content.strip()

    if decision == "FINISH":
        return Command(goto=END)
    return Command(goto=decision)

# 3) 拼图
g = StateGraph(MessagesState)
g.add_node("supervisor", supervisor)
g.add_node("researcher", researcher)
g.add_node("coder", coder)

g.add_edge(START, "supervisor")
g.add_edge("researcher", "supervisor")   # 干完回来复命
g.add_edge("coder", "supervisor")

app = g.compile()

图示

模式二:Swarm(互相 handoff)

没有中心调度,每个 Agent 自己决定"交给谁":

from langchain_core.tools import tool
from langgraph.prebuilt import InjectedState
from typing_extensions import Annotated

@tool
def transfer_to_billing() -> str:
    """当问题涉及账单、发票、付款时,转给 billing agent。"""
    return Command(
        goto="billing",
        update={"messages": [ToolMessage(content="已转账单部", tool_call_id=...)]},
        graph=Command.PARENT,   # 关键:跳出当前 Agent,回父图再跳
    )

@tool
def transfer_to_support() -> str:
    """当问题涉及技术故障时,转给 support agent。"""
    return Command(goto="support", graph=Command.PARENT, ...)

# 每个 agent 带上对应的 handoff 工具
greeter = create_react_agent(
    model="...",
    tools=[transfer_to_billing, transfer_to_support],
    prompt="你是前台,简单问题自己答,复杂的转对应专员。",
    name="greeter",
)

LangGraph 官方还有个 langgraph-swarm 包把这模式打包好了,create_swarm([a, b, c], default_active="greeter") 一行搞定。

Subgraph:把图当积木

独立 state 的子图

子图有自己的 state schema,父图通过"桥接节点"转换数据:

# 子图:专门做报表生成,state 是自己的
class ReportState(TypedDict):
    query: str
    sql: str
    rows: list
    markdown: str

sub = StateGraph(ReportState)
sub.add_node("gen_sql", gen_sql)
sub.add_node("run_sql", run_sql)
sub.add_node("format", format_md)
sub.add_edge(START, "gen_sql"); sub.add_edge("gen_sql", "run_sql")
sub.add_edge("run_sql", "format"); sub.add_edge("format", END)
report_subgraph = sub.compile()

# 父图里用桥接节点调子图
class ParentState(TypedDict):
    messages: Annotated[list, add_messages]
    last_report: str

def call_report(state: ParentState):
    q = state["messages"][-1].content
    out = report_subgraph.invoke({"query": q})
    return {
        "messages": [AIMessage(content=out["markdown"])],
        "last_report": out["markdown"],
    }

parent.add_node("report", call_report)

共享 state 的子图

如果 schema 完全一样,可以直接把子图当节点:

parent.add_node("research_flow", research_subgraph)
# state 直接共用,但要注意 reducer 行为

跨图跳转:Command(graph=...)

默认 Command(goto=...) 只在当前图内跳。想从子图跳回父图:

from langgraph.types import Command

def child_node(s):
    if s["should_escalate"]:
        return Command(
            goto="supervisor",
            update={"reason": "out of scope"},
            graph=Command.PARENT,   # 关键:去父图找 supervisor
        )
    return {"answer": "..."}

多 Agent 的 state 设计陷阱

坑 1:messages 串掉了角色

不同 Agent 共用一个 messages,容易互相看到对方的工具调用,prompt 被污染。解法:

# 选项 A:每个 Agent 独立 messages 字段
class State(TypedDict):
    user_messages: Annotated[list, add_messages]
    researcher_messages: Annotated[list, add_messages]
    coder_messages: Annotated[list, add_messages]

# 选项 B:喂给 LLM 前过滤
def researcher_node(s):
    filtered = [m for m in s["messages"] if m.name != "coder"]
    ...

坑 2:两个 Agent 并行改同一字段

没 reducer 会覆盖。给字段加 operator.add 或自定义 reducer:

def merge_findings(a: list, b: list) -> list:
    return list({x["id"]: x for x in a + b}.values())    # 按 id 去重

class State(TypedDict):
    findings: Annotated[list, merge_findings]

Planner + Executor 分层

# Planner 产出 TODO 列表
class Plan(BaseModel):
    steps: list[str]

def planner(s):
    plan = llm.with_structured_output(Plan).invoke(...)
    return {"todo": plan.steps, "done": []}

# Executor 每次做一步
def executor(s):
    step = s["todo"][0]
    result = do(step)
    return {"todo": s["todo"][1:], "done": s["done"] + [result]}

def more_to_do(s):
    return "executor" if s["todo"] else "reporter"

g.add_conditional_edges("planner", more_to_do)
g.add_conditional_edges("executor", more_to_do)

多 Agent 的 7 个设计原则

1. 先单 Agent 再多 Agent:能单不多,每多一个调试成本 x2
2. 角色定义要锐利:"做研究"不如"只用搜索工具生成 1 段 200 字摘要"
3. Supervisor 要简单:路由判断别塞业务,塞了就难以调整
4. 给每个 Agent 加 name:日志/trace 里好找
5. Handoff 要有理由:工具返回值带上"为什么转",下一个 Agent 才有上下文
6. 限制递归:recursion_limit 调低些,Agent 互相踢皮球很常见
7. 评估单独搞:每个 Agent 单独跑 eval,先通过才拼系统

常见坑

本章小结