4、多智能体架构模式：多智能体系统的常见架构及LangGraph实现

多智能体系统的价值不在于简单地把多个模型调用堆叠在一起，而在于通过合理的架构设计，让多个智能体之间形成有效的分工、协作与制衡。架构决定了信息如何流动、任务如何分配、冲突如何消解，也直接决定了系统的扩展性、容错性和可维护性。LangGraph 作为专为有状态、可编排的智能体应用设计的框架，提供了对多种多智能体架构模式的原生支持。本章将系统梳理多智能体系统的四种核心架构模式——协作、竞争、层级、混合，并深入讲解每种模式在 LangGraph 中的具体实现方式、适用场景与工程注意事项。

4.1 协作模式：任务分解与结果聚合

核心原理

协作模式（Collaborative Architecture）是最直观的多智能体组织方式。其核心思想是将一个复杂任务拆解为多个相对独立的子任务，由不同的智能体并行或串行执行，最后通过聚合器将各子结果整合为最终输出。这种模式模仿了人类团队分工协作的方式，每个智能体只关注自己擅长的领域，从而降低单次调用的复杂度，提升整体系统的专业性和并行处理能力。

在协作模式中，通常存在一个规划者（Planner）角色，负责理解用户意图、拆解任务、分配子任务；若干执行者（Worker）角色，各自处理被分配的子任务；以及一个聚合器（Aggregator）角色，负责收集所有执行者的输出，进行去重、排序、冲突消解和最终整合。LangGraph 中，协作模式可以通过 Send API 实现动态任务分发，配合并行分支（Parallel Branches）和状态合并机制（Reducer）完成结果聚合。

LangGraph实现方式

在 LangGraph 中实现协作模式，核心是利用 Send 函数实现动态并行。以下是一个典型的协作模式实现框架：

from langgraph.graph import StateGraph, Send
from typing import TypedDict, List, Annotated
import operator

class AgentState(TypedDict):
    task: str
    subtasks: List[dict]
    results: Annotated[List[str], operator.add]

def planner_node(state: AgentState):
    # 调用LLM将任务拆解为多个子任务
    subtasks = llm_split_task(state["task"])
    return {"subtasks": subtasks}

def worker_node(state: AgentState):
    # 每个worker处理一个子任务
    subtask = state["subtasks"][0]  # 实际通过Send传入
    result = llm_execute_subtask(subtask)
    return {"results": [result]}

def aggregator_node(state: AgentState):
    # 聚合所有结果
    final_answer = llm_aggregate(state["results"])
    return {"final": final_answer}

def continue_to_workers(state: AgentState):
    # 动态分发：为每个子任务创建一个worker实例
    return [Send("worker", {"subtasks": [st]}) for st in state["subtasks"]]

graph = StateGraph(AgentState)
graph.add_node("planner", planner_node)
graph.add_node("worker", worker_node)
graph.add_node("aggregator", aggregator_node)
graph.add_edge("planner", "continue_to_workers")
graph.add_conditional_edges("planner", continue_to_workers, ["worker"])
graph.add_edge("worker", "aggregator")
graph.set_entry_point("planner")

案例拆解：多源信息聚合报告生成

假设我们需要构建一个"行业竞争分析报告"生成系统。输入为"请分析新能源汽车行业2024年竞争格局"。规划者将任务拆解为：市场数据收集、主要玩家分析、技术趋势研判、政策环境解读四个子任务。四个worker分别调用不同的数据源API和LLM进行专项分析。聚合器收集四份子报告后，进行交叉验证、去重、生成统一的报告框架和结论。这种模式将原本需要超长上下文的一次性生成，转化为多个短上下文并行处理，显著提升了生成质量和速度。

4.2 竞争模式：多路尝试与择优输出

核心原理

竞争模式（Competitive Architecture）的核心思想是"多个候选方案并行生成，通过评审机制选择最优结果"。这种模式特别适用于对输出质量要求高、单次生成可能不稳定的场景。多个智能体（或同一智能体的多次采样）针对同一任务给出不同答案，评审器（Evaluator）根据预设标准（如正确性、完整性、风格匹配度）对候选结果进行评分，最终输出得分最高的方案。

竞争模式与协作模式的核心区别在于：协作模式中每个智能体处理不同的子任务，而竞争模式中每个智能体处理相同的完整任务。竞争模式通过冗余计算换取可靠性，代价是更高的计算成本。

LangGraph实现方式

LangGraph 中实现竞争模式，需要创建多个并行分支执行相同任务，然后通过一个评审节点进行条件汇总。关键点在于使用 add_redundant 或自定义Reducer来保留所有候选结果，而不是覆盖。

class CompetitionState(TypedDict):
    task: str
    candidates: Annotated[List[str], operator.add]  # 保留所有候选
    best_answer: str

def candidate_a(state: CompetitionState):
    return {"candidates": [llm_with_temp(state["task"], temperature=0.7)]}

def candidate_b(state: CompetitionState):
    return {"candidates": [llm_with_temp(state["task"], temperature=0.3)]}

def candidate_c(state: CompetitionState):
    return {"candidates": [llm_with_temp(state["task"], temperature=0.5)]}

def evaluator_node(state: CompetitionState):
    # 评审器对每个候选打分并选择最优
    scores = [score_candidate(c) for c in state["candidates"]]
    best_idx = scores.index(max(scores))
    return {"best_answer": state["candidates"][best_idx]}

# 构建并行分支
graph.add_node("candidate_a", candidate_a)
graph.add_node("candidate_b", candidate_b)
graph.add_node("candidate_c", candidate_c)
graph.add_node("evaluator", evaluator_node)
# 三个候选节点并行执行，全部完成后进入evaluator
graph.add_edge("candidate_a", "evaluator")
graph.add_edge("candidate_b", "evaluator")
graph.add_edge("candidate_c", "evaluator")

案例拆解：代码生成与Bug修复

在代码生成场景中，竞争模式尤为有效。例如，要求智能体"编写一个Python函数实现快速排序"。三个候选智能体分别以不同temperature和不同prompt策略生成代码。评审器不仅检查代码语法正确性，还会运行单元测试，根据测试通过率和代码风格评分。如果三个候选都未通过测试，评审器可以触发"二次竞争"——将测试失败信息反馈给候选智能体进行修正。这种模式将代码生成的准确率从单次调用的约70%提升到95%以上。

4.3 层级模式：主管调度与子智能体执行

核心原理

层级模式（Hierarchical Architecture）引入了明确的上下级关系。一个主管智能体（Supervisor）负责理解全局任务、制定执行计划、调度子智能体（Sub-agent），并汇总子智能体的执行结果。子智能体之间不直接通信，所有信息交互都通过主管进行。这种模式适用于任务具有天然层次结构、需要全局规划与局部执行相结合的场景。

层级模式与协作模式的区别在于：协作模式中的规划者通常在任务开始时一次性完成拆解，然后不再干预执行过程；而层级模式中的主管智能体在整个执行过程中持续监控、动态调整计划、处理异常情况。主管智能体拥有全局状态视图，可以决定是否重新分配任务、是否终止某个子智能体的执行。

LangGraph实现方式

LangGraph 实现层级模式最优雅的方式是使用子图（Subgraph）嵌套。主管智能体运行在父图中，每个子智能体是一个独立的子图，通过 invoke 或 ainvoke 进行调用。父图的状态可以传递给子图，子图的输出也可以合并回父图状态。

from langgraph.graph import StateGraph

# 定义子智能体图（子图）
def create_sub_agent(system_prompt: str):
    sub_graph = StateGraph(SubAgentState)
    sub_graph.add_node("agent", lambda state: call_llm(system_prompt, state["input"]))
    sub_graph.set_entry_point("agent")
    return sub_graph.compile()

# 主管智能体图（父图）
class SupervisorState(TypedDict):
    task: str
    plan: List[str]
    sub_results: Annotated[List[str], operator.add]
    final_output: str

def supervisor_plan(state: SupervisorState):
    plan = llm_create_plan(state["task"])
    return {"plan": plan}

def execute_sub_agents(state: SupervisorState):
    results = []
    for step in state["plan"]:
        # 根据步骤内容选择合适的子智能体
        sub_agent = select_sub_agent(step)
        result = sub_agent.invoke({"input": step})
        results.append(result["output"])
    return {"sub_results": results}

def supervisor_synthesize(state: SupervisorState):
    final = llm_synthesize(state["task"], state["sub_results"])
    return {"final_output": final}

# 构建父图
supervisor_graph = StateGraph(SupervisorState)
supervisor_graph.add_node("plan", supervisor_plan)
supervisor_graph.add_node("execute", execute_sub_agents)
supervisor_graph.add_node("synthesize", supervisor_synthesize)
supervisor_graph.add_edge("plan", "execute")
supervisor_graph.add_edge("execute", "synthesize")

案例拆解：企业级智能客服系统

以一个企业级智能客服系统为例。主管智能体负责意图识别和对话管理。当用户提出"我的订单延迟了，想退款"时，主管智能体将任务分解为：订单查询（调用订单子智能体）、退款政策查询（调用政策子智能体）、情绪安抚话术生成（调用对话优化子智能体）。三个子智能体依次或并行执行，主管智能体收集结果后，结合对话历史生成最终回复。如果订单子智能体发现订单已发货，主管智能体需要动态调整计划，改为引导用户走退货流程。这种动态调整能力是层级模式相对于协作模式的核心优势。

4.4 混合模式：动态路由与按需组合

核心原理

混合模式（Hybrid Architecture）是生产级多智能体系统的常态。它不局限于某一种固定架构，而是根据任务类型、上下文状态、资源约束等因素，动态选择最合适的架构模式。例如，对于简单任务直接单智能体处理；对于中等复杂任务采用协作模式；对于高可靠性要求任务采用竞争模式；对于复杂项目型任务采用层级模式。

混合模式的核心是路由决策机制。LangGraph 的条件边（Conditional Edges）和动态图构建能力，使得系统可以在运行时根据状态决定下一步走向哪个子图或哪种模式。这种灵活性带来了极大的设计空间，但也对状态管理和错误处理提出了更高要求。

LangGraph实现方式

混合模式在LangGraph中通常表现为一个"路由节点"加上多个模式子图。路由节点分析当前状态，返回下一个要进入的节点或子图名称。LangGraph会根据返回值自动路由。

def router_node(state: AgentState):
    # 分析任务特征，决定使用哪种模式
    complexity = estimate_complexity(state["task"])
    reliability = check_reliability_requirement(state["task"])
    
    if complexity < 0.3:
        return "single_agent"       # 简单任务直接处理
    elif reliability > 0.8:
        return "competition_mode"   # 高可靠性要求进入竞争模式
    elif complexity < 0.7:
        return "collaboration_mode" # 中等复杂度进入协作模式
    else:
        return "hierarchy_mode"     # 复杂任务进入层级模式

# 在图中添加条件路由
graph.add_conditional_edges("router", router_node, {
    "single_agent": "single_agent_node",
    "competition_mode": "competition_subgraph",
    "collaboration_mode": "collaboration_subgraph",
    "hierarchy_mode": "hierarchy_subgraph"
})

案例拆解：智能数据分析平台

假设我们构建一个数据分析平台，用户输入可以是"帮我算一下这个月的销售额"（简单查询），也可以是"分析过去一年各区域销售趋势并预测下季度"（复杂分析），还可以是"对比三种促销方案的效果并给出推荐"（需要竞争评估）。路由节点首先判断任务类型：简单查询直接调用SQL生成智能体；复杂分析进入协作模式，拆解为数据清洗、趋势分析、预测建模三个子任务；方案对比进入竞争模式，三个候选方案分别模拟执行，评审器根据ROI、风险、实施难度综合评分。这种混合架构使得系统能够以最低成本处理简单任务，同时保证复杂任务的处理质量。

4.5 架构选型对比与工程实践建议

工程实践建议

• 状态设计先行：无论采用哪种架构，都要先明确状态的数据结构、Reducer策略和共享字段。LangGraph的状态是全局的，设计不当会导致数据竞争和覆盖。
• 超时与重试机制：多智能体系统中，单个智能体的失败可能被放大。建议为每个节点设置超时时间，并实现重试或降级策略。
• 可观测性：使用LangGraph的get_state()和update_state()方法进行调试，记录每个节点的输入输出，便于回溯问题。
• 渐进式演进：不要一开始就设计复杂的混合架构。先实现单智能体，再根据瓶颈逐步引入协作、竞争或层级模式。