企业官网建设流程全解析

看到标题里的“吊打付费”四个字，先别急着点收藏。LangGraph 这个框架真正硬核的地方，不在于它有多“高级”，而在于它把 AI Agent 从“单轮问答玩具”推进到了“可编排、可控制、可恢复的工程化工作流”。我结合最近给多个业务方落地多智能体系统的经验，把 LangGraph 的核心组件、多智能体架构、条件路由、子图、并行分支、状态记忆逐一拆开，配一份能直接跑起来的实战代码。不管你是刚接触 AI 大模型应用开发，还是已经在用 LangChain 写链式调用，这篇文章都能让你少走很多弯路。

1. 背景与核心概念

1.1 LangGraph 到底是什么？

LangGraph 是 LangChain 社区推出的一个专门用于构建有状态 Agent 工作流的编排框架。它的名字里有两个关键词：Lang 代表它和 LangChain 生态天然打通，Graph 则揭示了它的底层模型——把整个智能体任务看成一张有向图（Directed Graph）。

在上面的图里，每个节点（Node）代表一个可执行单元，比如“调用大模型”“读取数据库”“调用外部 API”“执行 Python 函数”；每条边（Edge）代表控制流，也就是从一个节点运行完之后下一步该去哪里。和传统 Chain 那种“一条线走到黑”的固定管道不同，LangGraph 允许中间节点根据当前状态动态决定下一步走向，这就是“条件路由（Conditional Edge）”的核心能力。

用一个通俗的比喻来理解：

• LangChain Chain：像一条传送带，物料从入口放上去，沿着固定路线依次经过加工台，最后从出口出来。
• LangGraph：像一间工厂的中央调度室，每个工位都有自己的职责，调度员会根据当前物料的状态、质检结果、订单类型，动态决定这个物料下一步去哪条产线、要不要返工、要不要几个工位并行处理。

对 AI 大模型应用开发者来说，这意味着你不再只能写“用户提问 -> 调用模型 -> 返回答案”这种固定流程，而是可以搭建真正的多智能体系统：一个 Agent 负责意图识别，一个 Agent 负责工具调用，一个 Agent 负责结果校验，它们之间通过状态共享和条件路由协同工作。

1.2 LangGraph 解决的真实痛点

在实际项目里，单次大模型调用往往不能满足业务需求。以智能客服系统为例：

• 用户先问“我的订单什么时候发货？”
• 客服 Agent 需要查订单系统，发现订单已发货，但用户又追问“那运单号是多少？”
• Agent 需要记住这是同一个用户、同一个订单上下文，再去查物流系统。
• 如果用户对答案不满意，Agent 需要自动转入人工投诉流程。

这个过程涉及多轮对话状态维护、工具调用、条件判断、异常降级、人机切换，如果纯粹靠 if-else 胶水代码去拼，很快就会失控。LangGraph 通过状态对象（State）在不同节点之间传递数据，通过 Checkpointer 实现状态持久化和断点恢复，让这类复杂流程变得可编排、可观测、可恢复。

1.3 LangGraph 与 LangChain 的区别

这是很多初学者最容易混淆的地方。我直接用一张表说明：

简单说，LangChain 解决的是“怎么让模型调用工具更简单”，LangGraph 解决的是“怎么把多个模型调用和工具调用编排成一套健壮的分布式流程”。新项目中如果涉及多 Agent 协作，我建议直接基于 LangGraph 起步，而不是先搭好 LangChain 再迁移。

1.4 多智能体的常见交互模式

在正式写代码之前，有必要了解一下多智能体系统中常见的四种交互模式，这能帮助你判断自己的业务场景该用哪种图结构：

1. 顺序协作模式（Sequential）：Agent A 的输出作为 Agent B 的输入，适合流水线式处理。比如先做意图分类，再做实体抽取，最后生成回复。
2. 路由分发模式（Router）：一个主导 Agent 根据输入内容，把任务分发给不同的专用 Agent。这是客服、工单系统最常见的模式。
3. 并行协作模式（Parallel）：多个 Agent 同时处理同一任务的不同子任务，最后汇总。比如同时生成多方言版本的回复，或者同时检查代码和检查安全风险。
4. 层级监督模式（Hierarchical）：有一个“主管 Agent”负责任务拆解和结果验收，多个“执行 Agent”负责具体执行。这是复杂项目中最接近真实团队协作的模式，但实现成本也最高。

了解这些交互模式后，你会发现它们几乎都可以用 LangGraph 的节点 + 条件边 + 子图 + 并行分支组合出来。接下来我们直接进入实操环节。

2. 环境准备与版本说明

2.1 运行环境说明

本文的示例代码基于以下环境，版本可以根据你的项目实际情况调整，重点演示的是配置和实现思路：

• 操作系统：Windows 10/11、macOS、Linux 均可，本文以 Linux 终端命令为例
• Python 版本：3.10 或更高（建议 3.11）
• 包管理工具：pip 或 uv
• 核心依赖：langgraph、langchain-core、langchain-openai（或你本地部署模型的 SDK）
• 可选：langgraph-cli、langsmith

需要提醒的是，LangGraph 的 API 在 0.1.x、0.2.x、0.3.x 之间有少量调整。本文示例以 0.2.x 和 0.3.x 通用写法为主，如果遇到兼容性问题，优先查看官方文档和 changelog。

2.2 安装 LangGraph

建议创建一个干净的虚拟环境：

python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install --upgrade pip pip install langgraph langchain-core langchain-openai

如果你同时需要 LangGraph 的开发调试工具，可以安装 CLI：

pip install langgraph-cli

安装完成后，可以用下面的命令验证版本：

python -c "import langgraph; print(langgraph.__version__)"

我这里演示时安装的是0.3.x版本的 LangGraph。如果你后续引入了langgraph dev相关命令，它本质上会启动一个带持久化存储的后端服务；而直接uvicorn app:app启动普通的 FastAPI 应用则不会自动附加 LangGraph 的持久化和热重载能力。生产环境部署时，根据是否需要状态管理来选择启动方式。

2.3 示例项目结构

为了后续实战代码更清晰，推荐按下面的目录结构组织项目：

langgraph_agent_demo/ ├── agent_system/ │ ├── __init__.py │ ├── state.py # 定义全局 State │ ├── tools.py # 工具函数，模拟订单/物流查询 │ ├── nodes.py # 各个节点函数 │ ├── subgraphs.py # 子图定义 │ └── graph.py # 主图组装 ├── .env # 存放模型 API Key └── main.py # 程序入口

这样拆分的好处是，每一个节点函数都可以独立测试，后续扩展新 Agent 时不需要改主图逻辑，只需要加节点、加边。

3. LangGraph 核心组件深度拆解

3.1 State：全局状态

State 是 LangGraph 的灵魂。它决定了节点之间如何传递数据。在实际工程中，State 通常是一个 TypedDict 或者 dataclass。下面是一个典型的定义：

# 文件路径：agent_system/state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages def merge_lists(left: list, right: list) -> list: """自定义状态合并函数，用于多个节点并行写入同一字段时进行合并。""" return left + right class AgentState(TypedDict): # 用户原始输入 user_input: str # 意图识别结果 intent: str # 订单号 order_no: str # 节点间的中间消息列表 messages: Annotated[List[str], add_messages] # 最终回复 reply: str # 当前重试次数 attempts: int # 并行任务结果汇总 parallel_results: Annotated[List[str], merge_lists]

这里的关键点在于Annotated[List[str], add_messages]这种写法。add_messages是 LangGraph 内置的状态归约器（Reducer），当多个节点试图写入同一个messages字段时，会用追加的方式合并，而不是直接覆盖。同理，自定义的merge_lists函数也能控制并行分支回填状态的方式。

为什么需要 Reducer？因为在并行分支场景下，多个节点会同时更新同一个状态字段，如果没有合并策略，后写入的会覆盖先写入的，导致数据丢失。实际项目中，凡是会被多个节点写入的列表字段，都应该定义合适的 Reducer。

3.2 Node：工作流节点

Node 就是普通的 Python 函数，签名统一为(state: AgentState) -> dict[str, Any]，返回值会被合并到全局状态中。下面是一个最小节点示例：

# 文件路径：agent_system/nodes.py from .state import AgentState def receive_input(state: AgentState) -> dict: """接收用户输入，并初始化相关字段。""" print(f"收到用户问题：{state['user_input']}") return { "messages": [f"用户输入：{state['user_input']}"], "attempts": 0, } def classify_intent(state: AgentState) -> dict: """ 意图识别节点。 真实项目中这里应该调用大模型，本示例用简单规则代替，方便演示。 """ user_input = state.get("user_input", "") if "订单" in user_input or "发货" in user_input: intent = "after_sales" elif "投诉" in user_input or "不满" in user_input: intent = "complaint" else: intent = "consult" print(f"意图识别结果：{intent}") return {"intent": intent, "messages": [f"意图：{intent}"]}

注意，Node 函数不直接修改外部全局变量，而是返回一个字典告诉 LangGraph 哪些状态字段需要更新。这种设计保证了每个节点都是纯函数式的，便于测试、回放和并行执行。

3.3 Edge：连接与路由

Edge 用来定义节点之间的连接关系。基础用法很简单：

from langgraph.graph import StateGraph, START, END # 创建状态图 graph = StateGraph(AgentState) # 添加节点 graph.add_node("receive_input", receive_input) graph.add_node("classify_intent", classify_intent) # 添加边 graph.add_edge(START, "receive_input") graph.add_edge("receive_input", "classify_intent") graph.add_edge("classify_intent", END)

START是图入口，END是图出口。这种写法形成了一条“开始 -> 接收输入 -> 识别意图 -> 结束”的线性链路。

如果所有流程都这么线性，LangGraph 和普通 Chain 就没区别了。它的威力体现在add_conditional_edges上。

3.4 Conditional Edge：条件路由

条件路由允许你根据当前状态动态决定下一步走向。这是实现 Router 模式、多智能体分工的基础。用法如下：

from typing import Literal from .state import AgentState def route_after_classify(state: AgentState) -> Literal["consult", "after_sales", "complaint"]: """根据意图，将任务路由到不同 Agent。""" intent = state.get("intent", "consult") # 这里返回的是下一步节点的名称 if intent == "after_sales": return "after_sales_agent" elif intent == "complaint": return "complaint_agent" return "consult_agent" # 在主图中添加条件路由 graph.add_conditional_edges( "classify_intent", route_after_classify, { "consult": "consult_agent", "after_sales": "after_sales_agent", "complaint": "complaint_agent", } )

关键点：

• 第一个参数是源节点名，也就是执行完哪个节点后调用这个路由函数。
• 第二个参数是路由函数，它接收当前 State，返回一个目标节点标识。
• 第三个参数字典是可选的，它把路由函数的返回值映射到实际的节点名。如果路由函数返回的值和节点名一致，可以省略这个字典。

在实际项目中，路由函数通常由大模型驱动。比如在意图识别节点里，模型返回 JSON{"intent": "refund"}，然后路由函数读取这个字段决定走退款流程还是人工客服。

3.5 Checkpointer：记忆与断点

Checkpointer 是 LangGraph 相对较难理解但价值极高的组件。它的核心作用是保存图的运行状态快照，从而支持三类能力：

1. 多轮对话记忆：同一线程（thread_id）的多轮交互共用同一份状态。
2. 断点恢复：比如人工审批节点执行完毕后，可以从断点继续往下走。
3. 时间旅行：可以回放历史状态，调试复杂工作流。

LangGraph 提供了多种持久化后端，常用的有：

• MemorySaver：内存存储，适合开发测试。
• SqliteSaver：SQLite 文件存储，适合单机生产。
• PostgresSaver：PostgreSQL 存储，适合多实例部署。

用法上，只需要在编译时传入一个 Checkpointer 实例：

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() # 编译图时传入 checkpointer app = graph.compile(checkpointer=memory) # 运行时传入 thread_id，用于区分不同会话 config = {"configurable": {"thread_id": "user-001"}} result = app.invoke({"user_input": "我想查一下订单发货进度"}, config=config)

这里需要特别说明一个热词：InMemorySaver 中的内容如何参与大模型上下文。调用app.invoke()时，State 里的messages字段会自动带上历史消息。假设你将messages设计为消息列表，那么后续节点可以直接把state["messages"]拼接成提示词发送给大模型。步骤如下：

def build_prompt(state: AgentState) -> str: # 从 Checkpointer 恢复的消息历史 history_messages = state.get("messages", []) # 把历史消息格式化成大模型可用的对话上下文 context = "\n".join([f"{msg}" for msg in history_messages]) prompt = f"以下是历史对话记录：\n{context}\n\n当前用户问题：{state['user_input']}" return prompt

换句话说，Checkpointer 负责保存状态，而状态里哪些字段需要作为大模型上下文，完全由你决定。

3.6 编译与执行

图搭建完成后，调用compile()得到一个可调用的CompiledStateGraph对象：

app = graph.compile(checkpointer=memory) # 同步调用 result = app.invoke({"user_input": "我的订单怎么还没发货？"}, config=config) # 异步调用 # result = await app.ainvoke({"user_input": "..."}, config=config) # 流式输出 # for event in app.stream({"user_input": "..."}, config=config): # print(event)

compile()底层会做一次拓扑校验，如果图中有不可达节点、环上缺少条件边、或节点函数签名不符合要求，会在编译阶段报错。这也是 LangGraph 适合工程化的原因——很多结构性问题在运行前就能暴露。

4. 完整实战案例：智能客服工单多智能体系统

4.1 需求分析

我们构建一个“智能客服工单多智能体系统”，完整覆盖以下几种真实场景：

• 咨询类问题：用户询问退换货政策。系统调用咨询 Agent 直接回答。
• 售后类问题：用户询问订单发货进度。系统进入售后子图，该子图尝试查询订单和物流信息，如果查询失败则自动重试，最多 3 次，失败后转入人工处理。
• 投诉类问题：用户表达严重不满。系统同时启动“安抚话术生成”和“补偿方案生成”两个并行节点，汇总后生成最终回复。
• 所有对话记录通过 Checkpointer 持久化，支持多轮追问联系上下文。

整个系统的主图结构如下：

START | v receive_input | v classify_intent | v route_after_classify (条件路由) |--- consult_agent ----------> merge_result ---> END |--- after_sales_subgraph ---> merge_result ---> END |--- complaint_agent ---------> merge_result ---> END

其中after_sales_subgraph是一个子图，complaint_agent内部有并行分支。

4.2 创建项目结构

mkdir -p langgraph_agent_demo/agent_system cd langgraph_agent_demo touch agent_system/__init__.py

后续代码都放在agent_system目录下。

4.3 定义状态

先完善state.py：

# 文件路径：langgraph_agent_demo/agent_system/state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages def merge_lists(left: list, right: list) -> list: """并行节点结果合并函数。""" return left + right class AgentState(TypedDict): user_input: str intent: str order_no: str attempts: int messages: Annotated[List[str], add_messages] parallel_results: Annotated[List[str], merge_lists] reply: str

4.4 编写基础节点

下面编写nodes.py，包含主图的几个基础节点：

# 文件路径：langgraph_agent_demo/agent_system/nodes.py from .state import AgentState def receive_input(state: AgentState) -> dict: """接收用户输入。""" print(f"[receive_input] {state['user_input']}") return { "messages": [f"用户：{state['user_input']}"], "attempts": 0, } def classify_intent(state: AgentState) -> dict: """ 意图识别节点。 简化版：用关键词判断，真实项目可替换为 LLM 调用。 """ text = state.get("user_input", "") if "投诉" in text or "不满" in text or "差评" in text: intent = "complaint" elif "订单" in text or "发货" in text or "物流" in text or "退款" in text: intent = "after_sales" else: intent = "consult" print(f"[classify_intent] 意图：{intent}") return {"intent": intent, "messages": [f"系统：识别到意图 -> {intent}"]} def consult_agent(state: AgentState) -> dict: """咨询 Agent，直接基于知识库规则回答。""" reply = "我们的退换货政策是：签收后 7 天内支持无理由退换货。请问还需要了解其他内容吗？" print(f"[consult_agent] 咨询 Agent 回复") return {"reply": reply, "messages": [f"咨询Agent：{reply}"]} def complaint_agent(state: AgentState) -> dict: """投诉 Agent 的入口节点。""" print("[complaint_agent] 启动投诉处理流程") # 这里只写入一个占位消息，真正的并行逻辑在子图节点中处理 return {"messages": ["系统：已进入投诉处理流程"]} def merge_result(state: AgentState) -> dict: """汇总节点，把多个节点的输出合并成最终回复。""" results = state.get("parallel_results", []) if results: final_reply = "；".join(results) else: final_reply = state.get("reply", "抱歉，暂时无法处理，已转接人工客服。") print(f"[merge_result] 最终回复：{final_reply}") return {"reply": final_reply, "messages": [f"系统：{final_reply}"]}

为了让代码更接近真实场景，我们在tools.py中准备几个模拟工具函数，用于模拟订单查询、物流查询等外部系统：

# 文件路径：langgraph_agent_demo/agent_system/tools.py import random def query_order(order_no: str) -> dict: """模拟查询订单信息。""" print(f"调用订单系统：order_no={order_no}") status_list = ["已支付", "已发货", "运输中", "已签收"] return { "order_no": order_no, "status": random.choice(status_list), "logistics_company": "顺丰速运", "tracking_no": f"SF{order_no}123456", } def simulate_success(probability=0.6) -> bool: """模拟外部接口调用成功率，用于演示重试机制。""" return random.random() < probability

4.5 编写售后子图

子图（Subgraph）是 LangGraph 实现模块化的重要手段。你可以把一个负责“售后进度查询 + 重试机制”的流程封装成子图，然后在主图中当作一个普通节点来使用。下面是子图的完整实现：

# 文件路径：langgraph_agent_demo/agent_system/subgraphs.py from typing import TypedDict from langgraph.graph import StateGraph, START, END from .state import AgentState from .tools import query_order, simulate_success class AfterSalesState(TypedDict): order_no: str query_result: str attempts: int status: str # success / failed / max_retry def query_order_node(state: AfterSalesState) -> dict: """查询订单与物流信息，模拟可能失败。""" order_no = state.get("order_no", "UNKNOWN") attempts = state.get("attempts", 0) + 1 if not simulate_success(probability=0.5): print(f"[售后子图] 第 {attempts} 次查询订单失败") return { "query_result": f"第 {attempts} 次查询失败", "attempts": attempts, "status": "failed", } result = query_order(order_no) text = f"订单 {order_no} 当前状态：{result['status']}，物流公司：{result['logistics_company']}，运单号：{result['tracking_no']}" print(f"[售后子图] 查询成功：{text}") return { "query_result": text, "attempts": attempts, "status": "success", } def check_retry(state: AfterSalesState) -> str: """判断是否需要重试。""" if state.get("status") == "success": return "success" if state.get("attempts", 0) >= 3: return "max_retry" return "retry" def build_after_sales_subgraph(): """构建售后处理子图。""" subgraph = StateGraph(AfterSalesState) # 添加节点 subgraph.add_node("query_order", query_order_node) # 添加入口和出口 subgraph.add_edge(START, "query_order") # 条件路由：支持循环重试 subgraph.add_conditional_edges( "query_order", check_retry, { "success": END, "retry": "query_order", # 回到自身节点，形成循环 "max_retry": END, }, ) return subgraph.compile()

这个子图体现了 LangGraph 的循环检测能力。由于check_retry是一个条件边，LangGraph 能识别出“从 query_order 到 query_order”的循环边，只要循环存在出口（本例中的success和max_retry），图就能正常编译和运行，不会进入死循环。

4.6 编写并行分支节点

在投诉处理场景中，我们希望同时生成“安抚话术”和“补偿方案”，然后在汇合节点统一汇总。LangGraph 天然支持这种扇出（fan-out）汇聚（fan-in）结构：

# 文件路径：langgraph_agent_demo/agent_system/parallel_nodes.py from .state import AgentState def generate_comfort_message(state: AgentState) -> dict: """并行分支1：生成安抚话术。""" comfort = "非常抱歉给您带来了不好的体验，我们已经高度重视您的问题。" print(f"[并行分支1] 安抚话术生成完毕") return { "parallel_results": [f"安抚话术：{comfort}"], "messages": ["系统：安抚话术生成完毕"], } def generate_compensation(state: AgentState) -> dict: """并行分支2：生成补偿方案。""" compensation = "针对本次问题，我们为您提供 20 元无门槛优惠券作为补偿。" print(f"[并行分支2] 补偿方案生成完毕") return { "parallel_results": [f"补偿方案：{compensation}"], "messages": ["系统：补偿方案生成完毕"], }

这里的parallel_results字段配置了merge_lists归约器，所以两个并行节点返回的列表会被合并，而不是后者覆盖前者。这是 LangGraph 并行分支中最需要注意的细节。

4.7 组装主图

现在把所有部分组装成完整的主图graph.py：

# 文件路径：langgraph_agent_demo/agent_system/graph.py from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from typing import Literal from .state import AgentState from .nodes import ( receive_input, classify_intent, consult_agent, complaint_agent, merge_result, ) from .parallel_nodes import generate_comfort_message, generate_compensation from .subgraphs import build_after_sales_subgraph # 构建售后子图 after_sales_subgraph_app = build_after_sales_subgraph() def route_after_classify(state: AgentState) -> Literal["consult_agent", "after_sales_subgraph", "complaint_agent"]: """根据意图分发到不同 Agent。""" intent = state.get("intent", "consult") if intent == "after_sales": return "after_sales_subgraph" elif intent == "complaint": return "complaint_agent" return "consult_agent" def build_main_graph(): graph = StateGraph(AgentState) # 添加主图节点 graph.add_node("receive_input", receive_input) graph.add_node("classify_intent", classify_intent) graph.add_node("consult_agent", consult_agent) graph.add_node("complaint_agent", complaint_agent) # 子图作为节点加入主图 graph.add_node("after_sales_subgraph", after_sales_subgraph_app) # 并行分支节点 graph.add_node("generate_comfort_message", generate_comfort_message) graph.add_node("generate_compensation", generate_compensation) # 汇总节点 graph.add_node("merge_result", merge_result) # 入口 -> 接收输入 -> 意图识别 graph.add_edge(START, "receive_input") graph.add_edge("receive_input", "classify_intent") # 条件路由：分发到不同 Agent graph.add_conditional_edges( "classify_intent", route_after_classify, { "consult_agent": "consult_agent", "after_sales_subgraph": "after_sales_subgraph", "complaint_agent": "complaint_agent", }, ) # 咨询 Agent 和售后子图完成后都进入汇总节点 graph.add_edge("consult_agent", "merge_result") graph.add_edge("after_sales_subgraph", "merge_result") # 投诉 Agent：启动并行分支 graph.add_edge("complaint_agent", "generate_comfort_message") graph.add_edge("complaint_agent", "generate_compensation") # 并行分支汇合到汇总节点 graph.add_edge("generate_comfort_message", "merge_result") graph.add_edge("generate_compensation", "merge_result") # 汇总节点输出 graph.add_edge("merge_result", END) return graph.compile(checkpointer=MemorySaver()) main_app = build_main_graph()

这里其实有一个小的补充：如果希望售后子图的输出直接回填到主图的reply字段，需要把主图节点函数和子图 State 的结构对齐。在上面的示例中，子图内部只维护了query_result字段，而主图状态中的reply没有被更新，因此汇总节点的兜底逻辑会生效：“抱歉，暂时无法处理”。在实际项目中，建议在子图内部返回一个与主图 State 对应的字段（比如reply），或者通过封装节点函数做字段映射。

4.8 运行与验证

在项目根目录创建main.py：

# 文件路径：langgraph_agent_demo/main.py from agent_system.graph import main_app def run_demo(): # 模拟三个用户的会话，每个会话都有独立的 thread_id（状态隔离） sessions = [ {"thread_id": "user-001", "user_input": "我想了解一下退换货政策"}, {"thread_id": "user-002", "user_input": "我的订单怎么还没发货？订单号是10086"}, {"thread_id": "user-003", "user_input": "我要投诉！你们服务太差了！"}, ] for session in sessions: thread_id = session["thread_id"] user_input = session["user_input"] config = {"configurable": {"thread_id": thread_id}} print(f"\n==================== 新会话 {thread_id} ====================") result = main_app.invoke( {"user_input": user_input}, config=config, ) print(f"\n最终回复：{result['reply']}") # 第二轮回访，测试 Checkpointer 记忆恢复 if thread_id == "user-002": print("\n--- 用户追问 ---") result2 = main_app.invoke( {"user_input": "那什么时候能送到？"}, config=config, ) print(f"追问回复：{result2['reply']}") if __name__ == "__main__": run_demo()

运行结果示意如下：

==================== 新会话 user-001 ==================== [receive_input] 我想了解一下退换货政策 [classify_intent] 意图：consult [consult_agent] 咨询 Agent 回复 最终回复：我们的退换货政策是：签收后 7 天内支持无理由退换货。请问还需要了解其他内容吗？ ==================== 新会话 user-002 ==================== [receive_input] 我的订单怎么还没发货？订单号是10086 [classify_intent] 意图：after_sales [售后子图] 第 1 次查询订单失败 [售后子图] 第 2 次查询订单失败 [售后子图] 第 3 次查询订单失败 最终回复：抱歉，暂时无法处理，已转接人工客服。 ==================== 新会话 user-003 ==================== [receive_input] 我要投诉！你们服务太差了！ [classify_intent] 意图：complaint [complaint_agent] 启动投诉处理流程 [并行分支1] 安抚话术生成完毕 [并行分支2] 补偿方案生成完毕 最终回复：安抚话术：非常抱歉给您带来了不好的体验，我们已经高度重视您的问题。；补偿方案：针对本次问题，我们为您提供 20 元无门槛优惠券作为补偿。

从运行结果可以看出三点：条件路由正确、售后子图循环重试机制生效、投诉场景并行分支正确合并。如果你第二次运行 user-002 的会话，会发现子图状态从 Checkpointer 恢复，attempts会被重置还是保留取决于子图状态初始化逻辑，这一点需要开发者在设计子图时明确状态初始化时机。

5. 常见问题与排查思路

LangGraph 上手虽然快，但实际开发中还是有不少容易踩的坑。我把高频问题整理成一张排查表，方便你对照处理：

这里单独说一下InvalidUpdateError这个问题。很多初学者搭建第一个高并发 Agent 时会遇到，典型的错误提示是：

langgraph.errors.InvalidUpdateError: Expected list, got str

这个报错的本质是：State 中某个字段被定义为列表，但某个节点返回了字符串。比如：

class State(TypedDict): chat_history: Annotated[list, add_messages] def node_a(state: State): return {"chat_history": "hello"} # 错误：应该返回 ["hello"]

只要保持返回值和 State 类型一致，这类问题就能避免。

6. 最佳实践与工程建议

6.1 状态设计：拒绝“万能 State”

很多开发者习惯把所有中间变量都塞进 State，最终导致状态臃肿、难以追踪。我的建议是：

• State 只保存需要跨节点传递的、会影响路由或最终输出的数据。
• 临时中间结果（比如一批原始 JSON）尽量在节点函数内部消化，不要全部塞进 State。
• 列表字段必须设计 Reducer，避免并行写入覆盖。

6.2 子图拆分：以业务边界为准

不要为了“炫技”把图拆得过于复杂。子图拆分的合理依据是业务边界，比如“售后处理”是一个完整的子流程，内部有重试逻辑；“投诉处理”是另一个完整子流程，内部有并行计算。一个合格的多智能体系统，主图的节点数量通常控制在 5~8 个，更多细节应该沉淀到子图中。

6.3 外部系统调用：重试与降级

在真实项目中，Agent 的节点函数往往要调用订单系统、物流系统、支付系统等外部服务。这些服务不可能 100% 可用，所以每个外部调用节点都应该做三件事：

1. 超时控制：给 HTTP 客户端设置合理的连接超时和读超时。
2. 重试策略：利用 LangGraph 的条件边实现有限次重试，而不是在节点内部写死 while 循环。
3. 降级方案：重试失败后，必须给出降级回复，不能直接抛异常导致整个图崩溃。

本文的售后子图就是一个很好的参考模板。你可以把simulate_success替换成真实的 HTTP 调用，并在query_order_node里捕获外部异常。

6.4 Checkpointer 的正确用法

Checkpointer 在开发调试阶段用MemorySaver最方便，但生产环境要注意：

• 单实例部署可以用SqliteSaver保证服务重启后状态不丢失。
• 多实例部署必须使用PostgresSaver或 Redis 这类共享存储，否则不同实例之间看不到彼此的会话状态。
• 涉及用户隐私的会话数据，建议在 Checkpointer 层做加密或脱敏。

另外一个容易被忽略的点：Checkpointer 保存的 State 会包含模型响应消息、工具调用中间结果等敏感信息。在把 State 写入日志或传给前端时，需要显式过滤。

6.5 安全边界与大模型风险控制

只要涉及大模型调用，就不能忽略安全问题。结合 LangGraph 的实际工程特性，我给出以下清单：

• 提示词注入防护：用户在输入中提到“忽略上述指令”时，节点应拒绝执行敏感操作。
• 权限最小化：子图中的工具调用节点，应使用独立的 API Key，访问范围最小化，避免一个 Agent 拥有全量系统权限。
• 输出校验：模型生成的回复在返回用户之前，建议经过内容安全过滤和关键词校验。
• 人工审批：涉及退款、删除操作时，加入“人工确认”节点。可以用interrupt_before实现挂起，等待人工审批后恢复。

# 编译时挂起示例 app = graph.compile( checkpointer=memory, interrupt_before=["execute_refund"], # 到达退款节点前挂起 )

6.6 日志与可观测性

LangGraph 生态支持 LangSmith 的完整链路追踪，但如果团队没有接入第三方平台，也可以通过自定义 Node 的日志实现基础的可观测性。建议每个节点统一打印结构化日志：

import json import time def log_node_entry(node_name: str, state: dict): print(json.dumps({ "event": "node_entry", "node": node_name, "timestamp": time.time(), "intent": state.get("intent"), "attempts": state.get("attempts"), }, ensure_ascii=False))

6.7 测试策略

多智能体系统的测试和普通单测不太一样。除了单元测试节点函数，还应该做：

• 图结构测试：编译时无异常、节点覆盖率达到预期。
• 路由矩阵测试：构造不同输入，验证意图识别和条件路由结果符合预期。
• 故障注入测试：模拟外部服务异常，验证重试和降级逻辑。
• 状态恢复测试：同一个thread_id下，验证多轮对话上下文正确恢复。

建议用pytest组织这些测试。LangGraph 的纯函数式节点设计，让单测变得非常简单，不需要 mock 复杂的内部状态。

7. 总结与学习路线

写完这套智能客服多智能体系统，你应该已经掌握了 LangGraph 最核心的思维方式和编码套路：State 管理节点间数据、普通边保证线性流程、条件边实现动态路由和循环、子图承载独立业务模块、并行分支提高处理效率、Checkpointer 提供会话记忆和断点恢复。对比纯 LangChain 的链式调用，这种基于图结构的编排方式在面对复杂业务时优势非常明显，尤其是路由、重试、人工审批、多 Agent 协作这些场景，代码的可维护性和可观测性都上了一个台阶。

接下来你可以按这个顺序继续深入：

1. 把示例中的关键词意图识别替换成真实的大模型调用，感受 LLM 驱动路由的实际效果。
2. 将MemorySaver换成SqliteSaver，部署一个带持久化的本地服务。
3. 研究 LangGraph 的interrupt_before/interrupt_after，实现人工审批中断恢复。
4. 尝试写一个层级监督模式的多智能体系统：主管 Agent 拆解任务，多个执行 Agent 并行处理，主管统一验收。

如果本文对你有帮助，可以收藏备用。实际动手时如果遇到新的报错场景，欢迎在评论区把日志贴出来，我们一起排查。技术迭代很快，但 LangGraph 这种“图心态”一旦建立，你在面对任何复杂多智能体应用时都能游刃有余。