如何基于 LangGraph 设计多智能体架构

一、为什么需要多智能体架构

单 LLM 调用存在天然天花板：一次对话只能承载一个视角、一种推理路径、一份上下文。当问题复杂度上升——例如需要同时分析财务数据、扫描风险、评估市场情绪、生成调仓建议——单一 LLM 会话的答案往往是浅层的汇总，而非多维度交叉验证后的结论。

多智能体架构解决的核心问题是：将一个复杂决策拆解为多个独立分析任务，由不同专长的 Agent 并行执行，再通过合成层统一整合，形成多维交叉验证的综合结论。

类比人类团队：一个资深分析师无法同时精通基本面分析、技术分析、风险控制和宏观策略。但一个由四位专长互补的分析师组成的团队，可以在各自领域深入分析后，由研究主管汇总形成集体决策——这才是应对复杂问题的合理组织方式。

---

二、LangGraph 在多智能体架构中的角色

2.1 为什么是 LangGraph，而不是简单的 Prompt Chain

写过多 Agent 系统的开发者都会遇到几个共性问题：

• 条件路由：不同用户、不同场景需要走不同的 Agent 组合路径。用 if-else 硬编码，分支数量随场景数指数增长。
• 并行执行：多个 Agent 需要同时运行以控制在可接受的响应时延内。自行管理 asyncio 容易死锁或资源泄露，在复杂流程中维护成本陡增。
• 状态管理：每个节点的输入输出需要在多步流程中可靠传递。参数传递链路一长，手动管理极易出错。
• 可观测性：复杂流程的调试如同黑箱——每个步骤吃了什么输入、输出了什么、耗时多少，缺少系统化的记录。

LangGraph 的 StateGraph 针对这四个问题给出了原语级支持：

问题	LangGraph 解决方案
条件路由	add_conditional_edges —— 原生支持条件边，根据运行时状态动态分流到不同节点
并行执行	Send API —— 将多个子任务分发到同一节点的多个实例并行执行，框架管理生命周期
状态管理	AgentState —— 统一的 TypedDict 在节点间自动传递，每个节点的输出合并回共享状态
可观测性	每个节点/边的输入、输出、耗时自动记录，配合 LangSmith 实现全链路追踪

换句话说，LangGraph 的价值不是"能做别人做不了的事"，而是"把所有人都要做的事做得足够可靠"，让你把精力放在 Agent 的能力设计上，而非流程控制的工程细节上。

2.2 StateGraph 的核心概念

LangGraph 的核心抽象是一个有向图，其中：

• 节点（Node）：一个可执行单元，通常是 async 函数，接收 State 并返回 State 的部分更新
• 边（Edge）：连接节点的有向路径
  • 普通边：固定从节点 A → 节点 B
  • 条件边：根据当前 State 的某个字段值，动态决定下一步走向哪个节点
• State：一个 TypedDict，定义了图中流转的数据结构，所有节点共享同一个 Schema

from typing import TypedDict, List, Optional
from langgraph.graph import StateGraph

class AgentState(TypedDict):
    user_query: str
    intent: str
    user_level: str
    profile_context: dict
    agent_results: List[dict]
    final_report: Optional[str]

这张图的本质是：用声明式的方式描述"根据状态做什么决策、走什么路径"，而不是用控制流语句在代码中硬编码这些决策。

---

三、多智能体架构的核心设计模式

以下基于一个实际的多 Agent 系统设计经验，提炼出四个可复用的设计模式。

3.1 模式一：条件化路由 —— 一份编排，多种路径

适用场景：同一个入口，需要根据用户属性或请求特征走不同的处理路径。

核心思路：将 “应该走哪条路” 的判断从代码中提取到条件边的路由函数里。路由函数是纯函数——输入当前 State，输出下一个节点名——因此易于测试和修改。

def route_by_level(state: AgentState) -> str:
    """根据用户等级选择执行路径"""
    level = state["user_level"]
    if level == "A":
        return "multi_agent_parallel"    # 4 Agent 并行
    elif level == "B":
        return "dual_agent_parallel"     # 2 Agent 并行
    else:
        return "single_agent"            # 单 Agent 串行

图结构定义：

intent_recognition  →  profile_loading  →  level_routing (条件边)
                                               ├── A/B → multi_agent_parallel
                                               └── C/D/E → single_agent

为什么这个模式重要：条件边让你用一份编排代码覆盖所有场景。新增一个用户等级只需修改路由函数，不需要复制粘贴整条流程。对于多等级、多场景的系统，这是控制复杂度的关键杠杆。

3.2 模式二：并行分发 + 结果汇聚

适用场景：一个问题需要多个不同专长的 Agent 同时给出分析，最后汇总为综合结论。

核心思路：利用 LangGraph 的 Send API 将多个子任务并行分发，每个任务在各自的上下文中独立执行，执行完毕后由汇聚节点统一收集和合成。

from langgraph.graph import StateGraph
from langgraph.constants import Send

class ConsultationState(TypedDict):
    user_query: str
    profile: dict
    agents_to_run: List[str]    # 本次需要执行的 Agent 列表
    agent_results: List[dict]   # 各 Agent 返回结果
    final_report: str

def dispatch_agents(state: ConsultationState):
    """分发节点：为每个 Agent 生成一个 Send 指令"""
    agent_names = state["agents_to_run"]
    return [
        Send("agent_executor", {"agent_name": name, "query": state["user_query"]})
        for name in agent_names
    ]

async def agent_executor(state: ConsultationState):
    """各个 Agent 的并行执行节点（同一函数，不同 agent_name）"""
    agent = get_agent(state["agent_name"])
    result = await agent.run(query=state["query"], profile=state["profile"])
    return {"agent_results": [result]}

async def synthesize_results(state: ConsultationState):
    """汇聚节点：合成多份 Agent 报告为统一结论"""
    reports = state["agent_results"]
    synthesized = await research_manager.synthesize(reports, state["profile"])
    return {"final_report": synthesized}

图结构：

dispatch_agents ──→ [agent_executor (副本1)]
                ──→ [agent_executor (副本2)]   ← 并行执行
                ──→ [agent_executor (副本3)]
                          │
                          └──→ synthesize_results → 输出

关键细节：

• agent_executor 节点每次调用返回的 agent_results 字段是追加而非覆盖——LangGraph 的默认 reducer 将同名字段的新值累加到列表中。
• 不同 Agent 的执行时长可能相差数倍（比如一个做深度推理的 Agent 耗时 5 秒，一个做文本摘要的 Agent 耗时 0.5 秒）。并行模式下，总耗时 = max(各 Agent 耗时)，而非 sum。
• 如果某些 Agent 的输出是另一些 Agent 的输入，不要并行——拆分到不同的图阶段中顺序执行。

3.3 模式三：合成层（Research Manager）

适用场景：多个 Agent 并行分析后，各自的结论可能相互矛盾（例如选股 Agent 推荐买入某标的，但风险 Agent 对该标的所在的板块发出了高风险预警）。需要一个专门的合成层来处理冲突、加权仲裁、生成统一结论。

核心设计要点：

1. 冲突检测：识别不同 Agent 报告中互相矛盾的结论。不是简单的关键词匹配，而是让合成层 LLM 理解各报告的结论方向（正向/负向/中性），找出方向相反的判断对。

2. 优先级仲裁：不同 Agent 的权重不同。风险类分析的警告应具有较高的优先级——"有一个 Agent 说危险"比"三个 Agent 说没问题"更需要引起注意。这需要在合成提示词中明确定义仲裁规则。

3. 逻辑串联：不是简单地把四份报告粘在一起，而是提取每份报告的核心论据，按照"问题 → 分析 → 风险 → 建议"的逻辑线重新组织。

4. 用户适配：最终报告的表述风格应根据用户特征调整——对风险厌恶型用户优先呈现风险分析，对激进型用户可适当突出机会面。

合成层提示词的设计是这一模式中最关键的部分。需要明确告诉合成 LLM：每个子 Agent 的角色是什么、遇到矛盾时的处理规则、优先级排序标准、以及输出格式。

3.4 模式四：Agent 内部的轻重模型分工

适用场景：单个 Agent 内部的多个子任务认知复杂度差异显著——有些需要深度推理（如基本面分析、多步因果推理），有些只需要轻量处理（如数据摘要、格式化输出）。

核心思路：不应让一个 Agent 内的所有 LLM 调用都使用同一级别的模型。将子任务分为两类：

任务类型	特征	使用模型	示例
深度推理	多步推理、跨文档信息融合、复杂判断	大参数量模型	基本面交叉验证、综合诊断报告生成
快速响应	简单统计、信息提取、格式化	轻量模型	K线数据摘要、新闻标题提取、JSON格式化

实际效果：一个 Agent 内约 80% 的 token 消耗在轻量模型上，仅 20% 核心推理使用重量模型。在保持分析质量的前提下，单个 Agent 的推理成本可降低 60% 以上。

class DualLLMAgent:
    def __init__(self, deep_llm, quick_llm):
        self.deep_llm = deep_llm    # 如 DeepSeek-V4
        self.quick_llm = quick_llm  # 如 Qwen-7B

    async def analyze(self, stock_code: str) -> dict:
        # Step 1: 轻量任务 —— 并行获取数据摘要
        tech_summary = await self.quick_llm.summarize(await fetch_kline(stock_code))
        news_briefs = await self.quick_llm.summarize(await fetch_news(stock_code))

        # Step 2: 深度任务 —— 交叉验证 + 综合诊断
        diagnosis = await self.deep_llm.analyze(
            tech_summary=tech_summary,
            news_briefs=news_briefs,
            fundamentals=await fetch_fundamentals(stock_code)
        )
        return diagnosis

---

四、完整图结构设计示例

以下是一个完整的 LangGraph 图结构，展示了上述四种模式如何组合为一个可工作的多 Agent 系统：

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# ===== State 定义 =====
class AdvisoryState(TypedDict):
    user_query: str
    user_id: str
    intent: str
    user_level: str
    profile: dict
    agents_to_run: List[str]
    agent_results: List[dict]
    final_report: str
    needs_review: bool

# ===== 节点函数 =====
async def recognize_intent(state: AdvisoryState) -> dict:
    """意图识别节点"""
    intent = await intent_classifier.classify(state["user_query"])
    return {"intent": intent}

async def load_profile(state: AdvisoryState) -> dict:
    """画像加载节点"""
    profile = await profile_service.load(state["user_id"])
    return {"profile": profile, "user_level": profile["level"]}

def route_by_level(state: AdvisoryState) -> str:
    """条件边：按等级分流"""
    level = state["user_level"]
    intent = state["intent"]

    if level in ("A", "B") and intent in COMPLEX_INTENTS:
        return "dispatch_agents"
    return "single_agent"

async def single_agent(state: AdvisoryState) -> dict:
    """单 Agent 模式：选择最匹配的 Agent 执行"""
    agent = select_agent_for_intent(state["intent"])
    result = await agent.run(state["user_query"], state["profile"])
    return {"agent_results": [result]}

def dispatch_agents(state: AdvisoryState) -> list:
    """并行分发：按等级决定 Agent 数量"""
    capacity = AGENT_CAPACITY[state["user_level"]]  # A=4, B=2
    agents = select_agents_by_intent(state["intent"], capacity)
    return [
        Send("agent_executor", {
            "agent_name": name,
            "query": state["user_query"],
            "profile": state["profile"]
        }) for name in agents
    ]

async def agent_executor(state: AdvisoryState) -> dict:
    """各 Agent 的并行执行器"""
    agent = get_agent(state["agent_name"])
    result = await agent.run(
        query=state["query"],
        profile=state["profile"]
    )
    return {"agent_results": [result]}

async def synthesize(state: AdvisoryState) -> dict:
    """会诊合成节点"""
    if len(state["agent_results"]) > 1:
        report = await research_manager.synthesize(
            state["agent_results"], state["profile"]
        )
    else:
        report = state["agent_results"][0]
    return {"final_report": report}

async def personalize(state: AdvisoryState) -> dict:
    """个性化包装节点"""
    tailored = await personalize_service.adapt(
        state["final_report"], state["profile"]
    )
    return {"final_report": tailored}

def route_review(state: AdvisoryState) -> str:
    """是否需要人工审核"""
    if state["user_level"] == "A":
        return "human_review"
    return "respond"

# ===== 构建图 =====
builder = StateGraph(AdvisoryState)

# 添加节点
builder.add_node("recognize_intent", recognize_intent)
builder.add_node("load_profile", load_profile)
builder.add_node("single_agent", single_agent)
builder.add_node("dispatch_agents", dispatch_agents)
builder.add_node("agent_executor", agent_executor)
builder.add_node("synthesize", synthesize)
builder.add_node("personalize", personalize)
builder.add_node("human_review", submit_for_review)

# 连接边
builder.set_entry_point("recognize_intent")
builder.add_edge("recognize_intent", "load_profile")

# 核心：条件边 —— 等级分流
builder.add_conditional_edges(
    "load_profile",
    route_by_level,
    {
        "dispatch_agents": "dispatch_agents",
        "single_agent": "single_agent",
    }
)

builder.add_conditional_edges("dispatch_agents", lambda s: "agent_executor", {})
builder.add_edge("agent_executor", "synthesize")
builder.add_edge("single_agent", "synthesize")
builder.add_edge("synthesize", "personalize")

builder.add_conditional_edges(
    "personalize",
    route_review,
    {"human_review": "human_review", "respond": END}
)
builder.add_edge("human_review", END)

# 编译
app = builder.compile(checkpointer=MemorySaver())

图的拓扑结构：

recognize_intent → load_profile
                      │
                 [条件边: route_by_level]
                   ┌──────┴──────┐
                   ▼              ▼
           dispatch_agents    single_agent
                   │              │
               [Send*N]          │
                   ▼              │
            agent_executor       │
                   │              │
                   └──────┬───────┘
                          ▼
                     synthesize
                          │
                     personalize
                          │
                    [条件边: route_review]
                      ┌───┴───┐
                      ▼       ▼
                human_review   END
                      │
                      ▼
                     END

---

五、工程化落地的关键补充

5.1 状态持久化与断点续传

对于长流程（如多 Agent 并行 + 人工审核），整个执行周期可能跨数分钟甚至更长。LangGraph 内置的 checkpointer 机制将每个节点执行后的 State 自动持久化，支持：

• 节点执行到一半时服务重启，从上一个 checkpoint 恢复而非重新开始
• 人工审核节点可以是一个中断点（interrupt），审核完成后从断点继续
• 全链路可审计：每个请求的完整状态变迁历史可回溯

from langgraph.checkpoint.sqlite import SqliteSaver

app = builder.compile(checkpointer=SqliteSaver.from_conn_string("checkpoints.db"))

5.2 模型分层与 LLM Gateway

多 Agent 系统中，不同 Agent 和不同子任务的模型需求差异显著。一个集中的 LLM Gateway 层可以根据调用方传入的 tier 参数动态选择模型：

• 敏感数据请求路由到本地私有化部署的模型，确保数据不出内网
• 公开数据分析可使用云端高性能模型
• 主模型不可用时自动降级到同 tier 备用模型，保证可用性

这层抽象让 Agent 代码不需要感知"用哪个模型"——只声明所需的能力层级，Gateway 负责模型选择、健康检查和容错降级。

5.3 将复杂流程可视化

LangGraph 图结构本身就是一张 DAG。将图导出为可视化表示（如 Mermaid），对于团队沟通、架构评审和新人 onboarding 都有极大价值。节点和边的关系一目了然，条件边的分支逻辑可以在图上直接验证。

---

六、总结

设计多智能体架构的本质是回答三个问题：

1. 任务如何分解：一个复杂问题如何拆成多个独立可执行的分析子任务
2. 路径如何选择：不同场景、不同用户应该走哪条执行路径
3. 结论如何汇聚：多个 Agent 的分析结果如何整合为一致的综合结论

LangGraph 在这三个维度上提供了相应的原语：节点对应任务分解，条件边对应路径选择，StateGraph + Send 对应并行分发与结果汇聚。

四个可复用的设计模式——条件化路由、并行分发+结果汇聚、合成层仲裁、Agent 内部轻重模型分工——构成了一个经过验证的多 Agent 架构蓝图。在此基础上，加上状态持久化、模型分层路由和流程可视化，即可落地一个工程上可靠、业务上可扩展的多智能体系统。