多智能体（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 提供了灵活且生产友好的基础设施。建议读者在本文示例基础上，结合自身业务场景逐步迭代，构建真正可落地的智能体系统。

资讯详情

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