多智能体（Multi-Agent）编排实战：用 LangGraph 构建生产级 AI 系统

1. 引言：为什么需要多智能体编排

随着大语言模型（LLM）能力的持续提升，单一智能体在复杂业务场景中逐渐暴露出局限性：上下文窗口有限、工具调用链路过长、职责边界模糊、错误难以隔离。多智能体（Multi-Agent）架构通过将复杂任务拆解为多个专业智能体，各自负责特定子任务，再通过编排层协调协作，从而提升系统的可维护性、可扩展性和鲁棒性。

LangGraph 是 LangChain 团队推出的低层级编排框架，专为构建有状态、可循环、可细粒度控制的智能体工作流而设计。与 LangChain 的 Chain 相比，LangGraph 更贴近图计算模型，支持条件分支、循环、人工介入、持久化等生产级特性，是多智能体编排的理想底座。

本文将从零开始，围绕一个「智能客服工单处理系统」的实战案例，逐步讲解如何用 LangGraph 构建生产级多智能体系统，涵盖状态管理、图编排、智能体间通信、人工审批、持久化与可观测性等核心主题。

2. LangGraph 核心概念

在动手编码之前，先厘清 LangGraph 的五个核心抽象，它们是理解后续所有代码的基础。

• State（状态）：贯穿整个图执行过程的共享数据结构，可以是 TypedDict、Pydantic 模型或自定义类。每个节点读写 State 的指定字段，实现数据流转。
• Node（节点）：图中的一个执行单元，通常是一个 Python 函数，接收 State 并返回 State 的部分更新。
• Edge（边）：连接节点的有向边，定义执行顺序。普通边表示无条件流转，条件边根据 State 内容动态选择下一个节点。
• Graph（图）：由节点和边组成的执行拓扑，LangGraph 支持 StateGraph 和 MessageGraph 两种构建方式。
• Checkpointer（检查点）：持久化 State 快照的组件，支持断点续跑、时间旅行和人工介入。

下图展示了本文实战案例的整体编排拓扑：

flowchart TD A[入口: 工单接收] --> B{意图分类} B -->|技术问题| C[技术专家 Agent] B -->|账单问题| D[账单 Agent] B -->|通用咨询| E[通用客服 Agent] C --> F{是否需要升级} D --> F E --> G[直接回复] F -->|是| H[人工审批节点] F -->|否| I[生成最终回复] H --> I I --> J[结束] G --> J

3. 环境准备与项目结构

首先创建虚拟环境并安装依赖。建议使用 Python 3.10 及以上版本。

python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install langgraph langchain langchain-openai langchain-community pip install redis # 用于持久化 Checkpointer pip install langsmith # 用于可观测性追踪

项目目录结构如下：

multi_agent_system/ ├── main.py # 入口脚本 ├── state.py # State 定义 ├── agents/ │ ├── __init__.py │ ├── triage.py # 意图分类 Agent │ ├── technical.py # 技术专家 Agent │ ├── billing.py # 账单 Agent │ └── general.py # 通用客服 Agent ├── graph/ │ ├── __init__.py │ ├── builder.py # 图构建逻辑 │ └── edges.py # 条件边路由函数 ├── tools/ │ ├── __init__.py │ └── knowledge_base.py # 知识库检索工具 └── config.py # 全局配置

4. 定义全局状态（State）

State 是多智能体协作的「共享黑板」。我们使用 TypedDict 定义字段，并通过 Annotated 指定消息列表的合并策略。

# state.py from typing import Annotated, TypedDict, List, Optional from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class AgentState(TypedDict): """全局共享状态""" # 原始工单内容 ticket_id: str customer_name: str ticket_text: str # 消息历史，使用 add_messages 自动合并 messages: Annotated[List[BaseMessage], add_messages] 意图分类结果 intent: Optional[str] 当前处理 Agent 的名称 current_agent: Optional[str] 是否需要人工审批 needs_human_approval: bool 最终回复 final_response: Optional[str]

关键点说明：

• messages字段使用add_messages归约器，LangGraph 会自动将各节点返回的新消息追加到历史列表，避免手动拼接。
• intent、current_agent等字段用于条件边路由决策。
• needs_human_approval是人工介入的开关，当技术 Agent 判定问题复杂时置为 True。

5. 构建基础工具：知识库检索

为了让 Agent 具备领域知识，我们实现一个简单的知识库检索工具。生产环境可替换为向量数据库（如 Pinecone、Milvus）或 Elasticsearch。

# tools/knowledge_base.py from typing import List, Dict 模拟知识库数据，生产环境替换为向量检索 KNOWLEDGE_BASE = [ { "id": "KB001", "category": "technical", "keywords": ["登录", "密码", "认证失败"], "content": "如果用户遇到登录失败，请先引导其检查网络连接，然后尝试重置密码。若仍无法登录，可能是账号被锁定，需联系安全团队解锁。" }, { "id": "KB002", "category": "billing", "keywords": ["退款", "账单", "扣费"], "content": "退款申请需在账单生成后 30 天内提交。确认订单状态为已支付后，财务团队将在 3-5 个工作日内处理退款。" }, { "id": "KB003", "category": "general", "keywords": ["使用", "教程", "入门"], "content": "新用户可参考官方快速入门文档，包含账号注册、项目创建和第一个 API 调用的完整指引。" } ] def search_knowledge_base(query: str, category: str = None, top_k: int = 2) -> List[Dict]: """基于关键词匹配的知识库检索""" results = [] for item in KNOWLEDGE_BASE: if category and item["category"] != category: continue # 简单关键词命中评分 score = sum(1 for kw in item["keywords"] if kw in query) if score > 0: results.append({"score": score, **item}) results.sort(key=lambda x: x["score"], reverse=True) return results[:top_k]

在实际生产系统中，建议使用 Embedding 模型将查询和知识条目向量化，通过余弦相似度检索，召回效果远优于关键词匹配。

6. 实现意图分类 Agent

意图分类 Agent 是整个工作流的入口，负责判断工单属于技术问题、账单问题还是通用咨询。这里使用结构化输出（Structured Output）确保返回结果可被程序解析。

# agents/triage.py from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from langchain_core.pydantic_v1 import BaseModel, Field from state import AgentState class IntentOutput(BaseModel): """意图分类结构化输出""" intent: str = Field(description="工单意图类别: technical / billing / general") confidence: float = Field(description="置信度 0-1") reasoning: str = Field(description="分类依据简述") def triage_agent(state: AgentState) -> AgentState: """意图分类节点""" llm = ChatOpenAI(model="gpt-4o", temperature=0) structured_llm = llm.with_structured_output(IntentOutput) system_prompt = """ 你是工单意图分类专家。请根据用户工单内容，判断其属于以下哪一类： - technical: 技术故障、报错、功能异常 - billing: 账单、扣费、退款、发票 - general: 产品咨询、使用教程、其他 只输出 JSON 格式结果。 """ result = structured_llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=f"工单内容: {state['ticket_text']}") ]) return { "intent": result.intent, "messages": [HumanMessage(content=f"[意图分类] 判定为 {result.intent}，置信度 {result.confidence:.2f}")] }

这里使用with_structured_output让模型直接输出 Pydantic 对象，避免手写 JSON 解析的脆弱性。生产环境建议增加置信度阈值判断，低于阈值时直接转人工。

7. 实现专业 Agent（技术 / 账单 / 通用）

三个专业 Agent 结构相似，均接收工单内容，结合知识库检索结果生成回复。以技术专家 Agent 为例：

# agents/technical.py from langchain_openai import ChatOpenAI from langchain_core.messages import SystemMessage, HumanMessage from state import AgentState from tools.knowledge_base import search_knowledge_base def technical_agent(state: AgentState) -> AgentState: """技术专家 Agent""" llm = ChatOpenAI(model="gpt-4o", temperature=0.3) # 检索相关知识 kb_results = search_knowledge_base(state["ticket_text"], category="technical") kb_context = "\n\n".join( f"[知识条目 {r['id']}] {r['content']}" for r in kb_results ) if kb_results else "知识库中未找到直接匹配条目。" system_prompt = f""" 你是资深技术支持工程师。请基于以下知识库内容，结合你的技术经验，为用户工单提供解决方案。 知识库参考： {kb_context} 要求： 先复述用户问题，确认理解正确 给出分步骤的解决方案 如果问题超出知识库范围且涉及账号安全、数据丢失等高风险场景，将 needs_human_approval 置为 True """ response = llm.invoke([ SystemMessage(content=system_prompt), HumanMessage(content=f"用户工单: {state['ticket_text']}") ]) 简单判断是否需要人工介入（生产环境可用 LLM 判断或规则引擎） needs_human = any(kw in state["ticket_text"] for kw in ["数据丢失", "账号被盗", "无法恢复"]) return { "current_agent": "technical", "needs_human_approval": needs_human, "messages": [HumanMessage(content=f"[技术专家] {response.content}")] }

账单 Agent 和通用客服 Agent 的实现模式完全一致，仅替换系统提示词和知识库分类参数。为节省篇幅，这里不再重复贴出，完整代码可在文末获取。

8. 实现人工审批节点

人工审批是多智能体系统生产化的关键能力。LangGraph 的interrupt机制允许图在特定节点暂停执行，等待人工输入后恢复。

# agents/human_approval.py from langgraph.types import interrupt, Command from state import AgentState def human_approval_node(state: AgentState) -> AgentState: """人工审批节点：暂停图执行，等待人工决策""" # 组装需要人工审核的信息 approval_request = { "ticket_id": state["ticket_id"], "customer": state["customer_name"], "ticket_text": state["ticket_text"], "agent_response": state["messages"][-1].content if state["messages"] else "", "question": "该工单涉及高风险操作，请审核是否批准以下回复？" } # 暂停执行，将审批请求发送给前端 user_decision = interrupt(approval_request) 根据人工决策更新状态 if user_decision.get("approved"): return { "needs_human_approval": False, "messages": [HumanMessage(content="[人工审批] 已批准该回复")] } else: # 人工驳回，可附带修改意见 feedback = user_decision.get("feedback", "请重新生成回复") return { "needs_human_approval": True, "messages": [HumanMessage(content=f"[人工审批] 已驳回，意见: {feedback}")] }

当图执行到interrupt时，LangGraph 会保存检查点并暂停。外部系统通过graph.invoke(Command(resume=...))恢复执行，传入人工决策结果。

9. 组装多智能体图

现在将上述节点组装成完整的 StateGraph。这是整个系统的核心编排逻辑。

# graph/builder.py from langgraph.graph import StateGraph, START, END from state import AgentState from agents.triage import triage_agent from agents.technical import technical_agent from agents.billing import billing_agent from agents.general import general_agent from agents.human_approval import human_approval_node def route_by_intent(state: AgentState) -> str: """根据意图分类结果路由到对应 Agent""" intent = state.get("intent") if intent == "technical": return "technical" elif intent == "billing": return "billing" else: return "general" def route_after_agent(state: AgentState) -> str: """判断是否需要人工审批""" if state.get("needs_human_approval"): return "human_approval" return "finalize" def build_graph(): """构建多智能体图""" graph = StateGraph(AgentState) # 添加节点 graph.add_node("triage", triage_agent) graph.add_node("technical", technical_agent) graph.add_node("billing", billing_agent) graph.add_node("general", general_agent) graph.add_node("human_approval", human_approval_node) graph.add_node("finalize", finalize_node) 添加边 graph.add_edge(START, "triage") graph.add_conditional_edges("triage", route_by_intent, { "technical": "technical", "billing": "billing", "general": "general" }) graph.add_conditional_edges("technical", route_after_agent, { "human_approval": "human_approval", "finalize": "finalize" }) graph.add_conditional_edges("billing", route_after_agent, { "human_approval": "human_approval", "finalize": "finalize" }) graph.add_conditional_edges("general", route_after_agent, { "human_approval": "human_approval", "finalize": "finalize" }) graph.add_edge("human_approval", "finalize") graph.add_edge("finalize", END) return graph.compile()

其中finalize_node负责从消息历史中提取最终回复并写入final_response字段：

# graph/builder.py 中补充 def finalize_node(state: AgentState) -> AgentState: """汇总最终回复""" # 取最后一个 Agent 的回复作为最终结果 for msg in reversed(state["messages"]): if msg.content.startswith("[技术专家]") or \ msg.content.startswith("[账单专家]") or \ msg.content.startswith("[通用客服]"): return {"final_response": msg.content} return {"final_response": "抱歉，暂时无法处理该工单，已转交人工客服。"}

10. 持久化与断点续跑

生产级系统必须支持持久化，以便在进程重启后恢复执行状态。LangGraph 提供 Checkpointer 抽象，这里以 Redis 为例：

# main.py 片段 from langgraph.checkpoint.redis import RedisSaver import redis 初始化 Redis 连接 redis_client = redis.Redis(host="localhost", port=6379, db=0) checkpointer = RedisSaver(redis_client) 编译图时传入 checkpointer app = build_graph() app = app.compile(checkpointer=checkpointer)

执行图并指定线程 ID，实现会话级状态隔离：

# main.py 片段 config = {"configurable": {"thread_id": "ticket-1001"}} initial_state = { "ticket_id": "ticket-1001", "customer_name": "张三", "ticket_text": "我无法登录账号，提示密码错误，但重置密码后仍然无法登录，怀疑账号被锁定。" } 首次执行 result = app.invoke(initial_state, config=config) print("意图:", result["intent"]) print("最终回复:", result["final_response"])

当图在人工审批节点暂停后，可通过Command(resume=...)恢复：

# 恢复执行，传入人工决策 from langgraph.types import Command resume_result = app.invoke( Command(resume={"approved": True}), config=config ) print("审批后最终回复:", resume_result["final_response"])

11. 可观测性与追踪

生产环境必须能观测每个 Agent 的输入输出、耗时和 Token 消耗。LangGraph 原生集成 LangSmith，只需配置环境变量即可自动追踪：

export LANGCHAIN_TRACING_V2=true export LANGCHAIN_API_KEY=your_langsmith_api_key export LANGCHAIN_PROJECT=multi_agent_system

此外，可以在节点函数中手动记录关键指标：

# 在节点中记录耗时 import time from langchain_core.callbacks import CallbackManager def technical_agent_with_metrics(state: AgentState) -> AgentState: start = time.time() result = technical_agent(state) elapsed = time.time() - start print(f"[Metrics] technical_agent 耗时: {elapsed:.2f}s") return result

12. 生产化最佳实践

将多智能体系统部署到生产环境，需要关注以下关键点：

• 超时与重试：为每个节点设置超时时间，LLM 调用失败时自动重试，避免单点故障拖垮整个工作流。
• 限流与配额：对 LLM API 调用做限流，防止并发过高导致成本失控。
• 状态版本控制：State 结构变更时做好迁移策略，避免旧检查点无法恢复。
• 安全与权限：Agent 调用外部工具时，必须做权限校验和敏感信息脱敏。
• 灰度发布：新 Agent 或新提示词上线前，先在小流量灰度验证效果。
• 成本控制：为每个 Agent 设置 Token 预算上限，超限自动降级到简单回复。

13. 总结与扩展方向

本文通过一个完整的智能客服工单处理系统，系统讲解了如何用 LangGraph 构建生产级多智能体系统。核心要点总结如下：

• 使用StateGraph定义节点和边的执行拓扑，通过条件边实现动态路由。
• 利用Annotated归约器管理消息历史的自动合并。
• 通过interrupt机制实现人工审批，结合 Checkpointer 支持断点续跑。
• 使用 Redis 等外部存储实现状态持久化，保证系统可恢复性。
• 集成 LangSmith 实现全链路可观测性。

后续可扩展的方向包括：引入 ReAct 循环让 Agent 自主调用工具、使用 Supervisor 模式实现层级化编排、结合向量数据库构建更强大的知识检索、以及通过 LangGraph Cloud 实现一键部署和自动扩缩容。

多智能体编排是通往复杂 AI 应用的关键技术栈，LangGraph 提供了灵活且生产友好的基础设施。建议读者在本文示例基础上，结合自身业务场景逐步迭代，构建真正可落地的智能体系统。