为什么一个Agent不够用?

在之前的系列文章中,我们构建的大多是单Agent系统:依托一个大语言模型,搭配一组工具集,维护一条对话历史。这套架构足以解决日常开发中80%的问题,但面对一些天然需要"多专家协作"的复杂任务时,单Agent的局限性会快速显现。

比如撰写一篇技术文章,需要研究员收集资料、写手起草内容、编辑润色打磨,三个角色对应三种完全不同的思维方式;处理用户工单,需要依次完成意图识别、知识库查询、回复生成三个独立可测的阶段;进行代码评审,则需要从静态分析、安全扫描、可读性审查三个互不干扰的维度展开。

用单Agent处理这些任务并非不可行,但最终会陷入"System Prompt越写越长,输出质量越来越不稳定"的困境——本质上是在强迫一个模型同时扮演多个角色,分散了它的注意力和专业度。而多Agent系统的核心价值,正是实现职责分离,让每个Agent只专注于一件事,把这件事做到极致。

多Agent系统两种主流架构模式

多Agent系统存在多种拓扑结构,其中应用最广泛、落地最成熟的是Supervisor模式(动态路由)和Pipeline模式(固定顺序)。二者并非竞争关系,而是分别适配不同的业务场景。

Supervisor模式(动态路由)

Supervisor模式的核心是存在一个"指挥中心",负责根据任务状态决定下一步调用哪个Worker Agent。其典型执行流程为:任务分类 → Supervisor调度 → 对应Worker执行 → 结果返回Supervisor → 循环直至任务完成。

该模式的最大挑战是路由可靠性。如果完全依赖LLM每一步自主决定"下一个调谁",极易出现重复调用同一个Worker、忘记记录已调用Worker、无法判断任务终止时机等问题。因此更优的设计是采用两阶段混合路由

  1. 第一阶段:由LLM完成一次任务分类,明确任务类型
  2. 第二阶段:由Python代码根据分类结果和已调用Worker列表做确定性路由

让擅长理解语义的LLM做"任务判断",让擅长逻辑执行的Python做"流程控制",兼顾灵活性与可靠性。

LangGraph实现示例
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages

class SupervisorState(TypedDict):
    messages: Annotated[list, add_messages]
    task: str
    task_type: str  # "simple_fact" or "full_article"
    called: list[str]
    next: str

def classify_node(state: SupervisorState) -> SupervisorState:
    """LLM做一次任务分类,结果写入全局状态"""
    decision = _ask(
        "Classify this task:\n"
        "  simple_fact  — a factual question with a direct short answer\n"
        "  full_article — needs research, writing, and editorial review\n"
        "Output one word only: simple_fact / full_article",
        f"Task: {state['task']}",
    ).strip().lower()
    task_type = "full_article" if "full_article" in decision else "simple_fact"
    return {**state, "task_type": task_type}

def supervisor_node(state: SupervisorState) -> SupervisorState:
    """纯Python实现确定性路由,无LLM调用,避免循环"""
    called = state["called"]
    task_type = state["task_type"]
    
    if "researcher" not in called:
        next_worker = "researcher"
    elif task_type == "simple_fact":
        next_worker = "FINISH"
    elif "writer" not in called:
        next_worker = "writer"
    elif "reviewer" not in called:
        next_worker = "reviewer"
    else:
        next_worker = "FINISH"
    
    return {**state, "next": next_worker}

# 构建图拓扑结构
g = StateGraph(SupervisorState)
g.set_entry_point("classify")
g.add_edge("classify", "supervisor")
g.add_conditional_edges(
    "supervisor",
    route_supervisor,
    {"researcher": "researcher", "writer": "writer", 
     "reviewer": "reviewer", "FINISH": END},
)
g.add_edge("researcher", "supervisor")
g.add_edge("writer", "supervisor")
g.add_edge("reviewer", "supervisor")

最终形成classify → supervisor → [workers] → supervisor → ... → FINISH的可控循环。以任务"Write a short article about Python list comprehensions"为例,执行流程为:

[classify] task_type = full_article
[supervisor] → researcher
[researcher] working...
[supervisor] → writer
[writer] working...
[supervisor] → reviewer
[reviewer] working...
[supervisor] → FINISH
Workers called: ['researcher', 'writer', 'reviewer']

Pipeline模式(固定顺序)

Pipeline模式的执行路径完全硬编码在代码中,每个Agent只知道自己的输入上下文和下一个执行节点,无需任何路由决策。其典型流程为:节点1 → 节点2 → 节点3 → ... → 结束,前一个节点的输出直接作为后一个节点的输入。

该模式的优势是实现简单、可预测性强、调试成本低,非常适合步骤固定的标准化任务。

LangGraph实现示例
class PipelineState(TypedDict):
    topic: str
    outline: str
    draft: str
    polished: str
    stage_log: list[str]

def outline_agent(state: PipelineState) -> PipelineState:
    outline = _ask("创建一个5点大纲,涵盖核心知识点", state["topic"])
    return {**state, "outline": outline, "stage_log": ["outline_completed"]}

def draft_agent(state: PipelineState) -> PipelineState:
    draft = _ask("基于大纲撰写200字左右的草稿", state["outline"])
    return {**state, "draft": draft, "stage_log": state["stage_log"] + ["draft_completed"]}

def polish_agent(state: PipelineState) -> PipelineState:
    polished = _ask("润色草稿,优化语言表达和逻辑结构", state["draft"])
    return {**state, "polished": polished, "stage_log": state["stage_log"] + ["polish_completed"]}

# 构建线性拓扑结构
g = StateGraph(PipelineState)
g.add_node("outline_agent", outline_agent)
g.add_node("draft_agent", draft_agent)
g.add_node("polish_agent", polish_agent)
g.add_edge("outline_agent", "draft_agent")
g.add_edge("draft_agent", "polish_agent")
g.add_edge("polish_agent", END)

执行结果呈现出清晰的内容递进关系:

[outline_agent] 957 chars
[draft_agent] 1846 chars
[polish_agent] 2168 chars

两种模式核心能力对比

维度 Pipeline模式 Supervisor模式
执行路径 固定,硬编码在代码中 动态,由任务分类驱动
最适合场景 ETL流水线、标准化文档处理 研究类任务、开放问答、混合任务
可调试性 高(线性追踪) 中(路径随任务动态变化)
LLM调用次数 N(每阶段一次) N+1(多一次分类调用)
灵活性
可预测性 较低
实现复杂度 简单 中等

经验法则:如果明确知道任务需要执行哪几个固定步骤,选择Pipeline模式;如果需要根据任务的复杂程度和类型动态调整执行步骤,选择Supervisor模式。

Supervisor模式的独特优势:自适应执行路径

Supervisor模式最直观的价值在于,无需修改代码,就能让不同复杂度的任务自动走不同的执行路径。

例如用同一个Supervisor图处理简单事实问题"What year was Python created?",执行流程会自动简化为:

[classify] task_type = simple_fact
[supervisor] → researcher
[researcher] working...
[supervisor] → FINISH
Workers called : ['researcher']

对比两次运行结果:

  • 复杂文章任务:researcher→writer→reviewer(3个步骤)
  • 简单事实问题:researcher(1个步骤)

这种自适应能力是Pipeline模式无法实现的,后者的执行路径在代码编写阶段就已完全确定。

多Agent系统设计Checklist

架构选型

  • 任务步骤固定且已知 → Pipeline;需要动态决策 → Supervisor
  • Worker数量≤3且职责清晰时,两种模式均可
  • Worker数量>5时,考虑分层Supervisor架构(Supervisor of Supervisors)

State设计

  • 每个Worker只读自己需要的字段,只写自己生成的字段
  • Supervisor状态必须包含called: list[str],用于路由去重
  • Pipeline状态按阶段命名(如outlinedraftpolished),便于调试

路由可靠性

  • 避免纯LLM路由(LLM无法可靠追踪调用历史)
  • 推荐采用"LLM一次分类+Python确定性路由"的混合方案
  • 设置recursion_limit(建议20-30),防止意外循环

Worker设计

  • 每个Worker只做一件事,输入输出明确
  • Worker之间通过State传递结果,不直接相互调用
  • 编写清晰的Worker System Prompt,避免模型猜测上下文

可观测性

  • 在每个节点打印执行日志(Worker名称、输入/输出摘要)
  • 记录called列表,方便回溯路由决策
  • 对异常分支(如提前触发FINISH)添加警告日志

核心总结

  1. 多Agent的本质是职责分离:不是为了炫技,而是当单Agent的Prompt开始失控时,将任务拆分为多个专注的角色,提升输出质量和稳定性。
  2. Pipeline模式胜在简单可预测:执行路径写在代码里,追踪是线性的,测试和调试成本最低,适合标准化流程。
  3. Supervisor模式胜在自适应:同一张架构图,简单问题一步到位,复杂任务全程走完,无需修改代码。
  4. LLM路由+Python执行是最佳搭档:让擅长语义理解的LLM做任务分类,让擅长逻辑执行的Python做流程控制,兼顾灵活性与可靠性。
  5. called列表是Supervisor State的关键字段:是实现路由确定性的基础,缺少它极易出现重复调用或死循环问题。

在实际搭建多Agent系统的过程中,除了架构设计的选型,高效稳定的API调用管理也是影响系统性能的关键环节。TreeRouter作为轻量高效的API中转站,降低多Agent系统的运维复杂度,提升整体运行的稳定性与可扩展性,是多Agent架构落地的优质配套工具。