从 LangChain 到 LangGraph,Agent 开发到底变了什么?这是很多读者最近反复问我的问题。如果你已经写过几个 LangChain 版本的 Agent 小程序,大概率会碰到这样几个场景:明明只是调了个工具,Agent 却反复横跳;对话一多,记忆就开始错乱;流程稍微复杂一点,代码就变成了 if-else 套娃;更别提从开发到上线的部署运维,几乎全靠手工。
LangGraph 的出现,正是为了解决这些问题。它不只是一个“新的 Agent 框架”,而是把 Agent 开发从“自由发挥”拉回到了“工程化、可控、可恢复”的轨道上。本文会从企业级 Agent 开发的真实痛点出发,讲清楚 LangGraph 与 LangChain 的核心差异、状态图模型、多智能体架构、核心组件,并带你把一个支持工具调用、多轮记忆、并行检索的 Agent 完整跑通。无论你是刚接触 Agent 开发的新手,还是正在做 Agent 落地评估的技术负责人,这篇文章应该都能提供一些有信息增量的参考。
如果只用一句话来概括 LangGraph 的价值,我会说:它把 Agent 从“一个聪明的函数”变成了“一条可编排、可恢复、可监控的工作流”。
1. 这篇文章真正要解决的问题
先说一个我在技术社区里经常看到的场景:一个团队用 LangChain 做了个客服 Agent,最开始只有一个意图识别加一个知识库检索,跑得挺好。后来业务要求增加权限控制、支持多轮澄清、需要调用内部订单接口,还要在 Agent 出错时自动重试。结果代码越写越复杂,链路越来越不可控,线上出了问题只能靠人工看日志。
问他们为什么不用 LangGraph,回答通常是这两种:
一是“LangGraph 不就是给 LangChain 画了个图吗?我用 LangChain 也能实现。”
二是“图编排我懂,但一直没搞清楚 State、Node、Edge 到底怎么设计,学起来成本高。”
这两种回答其实都站得住脚,也恰恰说明 LangGraph 的定位被很多人低估了。
LangGraph 和 LangChain 的区别,本质上不是“有没有图”,而是有没有状态、有没有流程控制、有没有持久化恢复能力。LangChain 提供的是组件和工具链,LangGraph 提供的是 Agent 的运行时和编排层。Enterprise Agent 需要的不是“能回答问题”,而是“在复杂流程里不出错、能回溯、可监管”。
本文要解决的核心问题有三类:
- 理解问题:LangGraph 的 StateGraph、Node、Edge、Checkpointer、Send API 在真实 Agent 里到底是什么角色,和 LangChain 的 Chain、AgentExecutor 有什么本质差异。
- 实践问题:如何从零搭建一个可运行的企业级 Agent,包括工具调用、多轮记忆、并行任务分发和条件路由。
- 工程问题:多智能体架构怎么拆分、状态 Schema 怎么设计、生产环境如何做权限管控和监控回溯。
读完这篇文章,你应该能独立设计一个中等复杂度的 LangGraph Agent,并知道下一步该往哪个方向深入。
2. LangGraph 基础概念与核心原理
2.1 为什么 LangChain 不够用了
很多人问:LangChain 已经很方便了,为什么还要再学一个新框架?
LangChain 的优秀之处在于抽象了 LLM 调用、Prompt 模板、Retriever、Tool 等标准化组件,让开发者可以用很短代码组合出一个 Agent。它是一个非常出色的组件库和工具链。
但 Agent 一旦进入业务系统,问题就变了。业务系统要求的不是“模型自由发挥”,而是“流程可控、状态可查、异常可恢复”。LangChain 的 AgentExecutor 虽然也维护了 Agent 循环,但它缺乏对中间态的显式建模,流程一旦复杂,分支和循环就很难管理。比如:
- 多步工具调用之间共享状态如何设计?
- 中间某一步失败,能否回退重试?
- 出现了人类审批节点,怎么暂停流程并恢复?
- 多个子任务并行执行,怎么汇总结果?
这些需求在 LangChain 的 Chain 模型里写出来非常别扭,但在 LangGraph 里,它们就是图的基本能力。
2.2 核心抽象:StateGraph(状态图)
LangGraph 的核心是一个状态机模型。开发者把 Agent 的完整执行流程定义为一个图,图中每个节点是一个处理逻辑(LLM 调用、工具函数、API 请求、人工审核),每条边是节点之间的流转条件。
这个模型借鉴的是经典计算机科学里的状态机思想和图计算框架,但它不抽象、不绕弯。LangGraph 把它做到了 Agent 领域最容易理解的程度:节点就是函数,边就是路由,状态就是共享数据。你可以把它想象成把传统工作流引擎的控制逻辑、状态管理与 LLM 的自由推理能力做了一个融合。
State 是 LangGraph 里最重要的概念。它是一个共享的数据对象,贯穿整个图执行过程。每个节点接收 State 作为输入,处理完后返回 State 的增量更新。这种模式非常接近 Redux 或 Flux 的前端状态管理思想,对写过前端状态管理的开发者非常友好。
在 LangGraph 中,State 通常用 TypedDict 定义,例如包含messages、todos、user_info等字段。所有节点共享这个对象,节点之间不直接传参,而是通过修改 State 来通信。这个设计带来的最大好处是:每个节点都是纯函数式的,便于测试、便于回放、便于追踪。
2.3 节点、边与条件边
- Node(节点):一个异步或同步函数,输入是 State,输出是 State 的部分更新。
- Edge(边):连接两个节点,表示无条件的流转。
- Conditional Edge(条件边):根据 State 或外部结果动态决定走哪个分支。
节点的设计原则是“单一职责”。一个节点只做一件事,要么调用一次模型、要么调用一个工具、要么执行一段规则逻辑。相比把大量逻辑塞进一个 Chain 里,这种细粒度拆分让每个环节都变得可测试。
2.4 LangGraph 和 LangChain 的对比
| 维度 | LangChain | LangGraph |
|---|---|---|
| 定位 | 模型组件与工具链 | Agent 编排与状态运行时 |
| 流程控制 | 以 Chain 为主,流程固定 | 图模型,支持循环、分支、并行 |
| 状态管理 | 隐式、简单 | 显式 State,全局共享 |
| 持久化 | 默认不具备 | Checkpointer,支持恢复 |
| 失败重试 | 手工实现 | 图级别支持回退 |
| 多智能体 | 支持有限 | 原生支持,通过图或 Send 实现 |
| 适合场景 | 快速原型、固定流程 | 复杂业务系统、企业级 Agent |
这里不是说 LangChain 该被替代,而是工具分工不同。实际项目中二者经常配合使用:用 LangChain 的组件能力,用 LangGraph 做流程编排。
2.5 LangGraph 中的 Checkpointer 与记忆
记忆是 Agent 开发里的老大难问题。模型本身不保留上下文,多轮对话里为了保持一致性,只能把历史消息塞进 Prompt。一旦历史无限增长,Token 成本随之上升,还会冲淡核心指令。
LangGraph 的 Checkpointer 机制提供了两层能力:
- 状态持久化:把图执行的中间状态存到持久层,进程重启后可以从断点恢复。这对人机协同审批、任务中断、故障恢复极其重要。
- 对话级记忆:结合
langgraph-checkpoint可以在多轮会话中维护历史消息,而不需要每次手动拼接全部上下文。
在企业级 Agent 里,记忆不只是“聊天记录”,而是业务上下文。比如一个售后 Agent 需要记住订单号、用户等级、当前处理到哪一步。这些信息放 State 里,比让模型从历史消息里猜,要可靠得多。
2.6 LangGraph 如何支撑多智能体架构
多智能体架构听起来高深,其实就是“把大任务拆给多个专业 Agent 协作”。
LangGraph 支持两种典型的多智能体模式:
- Supervisor(路由模式):一个主 Agent 负责任务理解与分发,多个子 Agent 各司其职。类似公司的部门负责人接需求后拆解任务派发给专业团队。
- Network(图协作模式):多个 Agent 按业务规则形成协作网络,由 LangGraph 的图来约束流程。
用 LangGraph 做多智能体,真正的好处不是“多个模型一起聊天”,而是流程有了法律》般的边界。谁先执行、谁后执行、什么条件下切换,都是显式定义好的。
3. LangGraph 环境准备与前置条件
在进入代码之前,先把环境准备好。下面的说明以当前主流版本为基础,具体的版本号以你实际安装的结果为准。
3.1 系统与运行环境
- 操作系统:Windows / macOS / Linux 均可。
- Python:建议 3.10 或更高版本,LangGraph 使用了很多新的 Python 语法特性,低版本会遇到兼容问题。
- 包管理:推荐使用
uv或poetry。pip 也可以用,但建议用虚拟环境隔离项目依赖。
3.2 安装 LangGraph
# 创建虚拟环境(可选但强烈推荐) python -m venv .venv source .venv/bin/activate # Windows 下是 .venv\Scripts\activate # 安装核心库 pip install langgraph langchain langchain-openai # 如果需要本地方言模型支持(可选) pip install ollama如果你使用的是新版本 LangGraph 的快速开发工具,官方还提供了一个基于 JSON Schema 的开发方式,可以用langgraph dev在本地启动开发服务。这个工具会把图编译成 API,方便前端联调和调试。
# 安装 LangGraph CLI pip install "langgraph-cli[inmem]" # 启动开发模式 langgraph dev3.3 LLM 服务配置
LangGraph 本身不绑定具体模型服务。你可以用 OpenAI、Anthropic、Google Gemini,也可以用本地模型。
以 OpenAI 为例,需要设置环境变量:
export OPENAI_API_KEY="你的Key"如果你没有 OpenAI 的 Key,可以用 Ollama 跑本地模型替代:
ollama pull llama3.1然后在代码里用ChatOllama来替代ChatOpenAI。
3.4 目录结构建议
langgraph-demo/ ├── agent/ │ ├── __init__.py │ ├── state.py # 状态定义 │ ├── nodes.py # 节点函数 │ ├── tools.py # 工具函数 │ ├── graph.py # 图定义与编译 │ └── config.py # 模型与 API 配置 ├── tests/ ├── .env └── pyproject.toml从项目一开始就按目录拆开维护,后面加节点、改配置会轻松很多。
4. LangGraph 核心流程拆解:从需求到图设计
这一节我们走一遍 LangGraph Agent 的完整设计流程。以一个“电商售后 Agent”为例,需求如下:
- 用户咨询售后问题。
- Agent 需要判断是否需要查订单。
- 如果需要,调用订单查询工具。
- 根据查询结果判断是否需要升级到人工客服。
- 多轮对话,能记住用户已经查过哪个订单。
- 支持同时查询多个订单(并行)。
4.1 第一步:定义状态 Schema
状态是图的“数据库”,所有节点共享。状态设计的质量直接决定后续开发的复杂度。
# agent/state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages class AgentState(TypedDict): # 对话消息列表,add_messages 会做消息追加而不是覆盖 messages: Annotated[List[dict], add_messages] # 当前正在处理的订单号 current_order_id: str # 查询出的订单信息 order_info: dict # 是否已升级人工客服 escalated: bool # 待查询订单列表 pending_orders: List[str] # 并行查询结果汇总 order_results: dict这里最值得注意的代码是messages字段上的Annotated[List[dict], add_messages]。这个注解告诉 LangGraph:当不同节点都返回新消息时,不是覆盖旧消息,而是追加。这是对话类 Agent 特别常用的模式。
4.2 第二步:设计节点
根据需求拆节点:
llm_node:调用模型,让模型决定下一步动作。tools_node:执行模型选择的工具。check_already_query:检查订单是否查过,避免重复查询。escalate_node:执行人工客服升级流程。parallel_query_node:并行查询多个订单。
每个节点都是独立的函数,输入 State,返回 State 的更新片段。
这种设计的好处是:你在调试时可以单独测试每个节点,而不必启动整个图。
4.3 第三步:设计边与条件路由
图的路由逻辑决定了 Agent 在不同情况下走什么分支。条件路由通常依赖模型输出的结构化结果,比如模型返回一个 JSON,其中包含next_action字段。
4.4 第四步:编译与运行
图创建完成之后,通过编译得到一个可执行的 Runnable 对象。
app = workflow.compile(checkpointer=memory)这里的checkpointer参数很关键,它让图具备了状态持久化和恢复能力。
5. LangGraph 完整示例与代码实现
下面进入完整示例。我会把上面的售后 Agent 用最小可用代码跑通。
5.1 第一步:定义工具函数
# agent/tools.py # 模拟订单查询工具 from datetime import datetime, timedelta def query_order(order_id: str) -> dict: """ 查询订单信息。 实际项目中这里应该调用内部订单系统 API,并做好权限校验和限流。 这里用一个简化实现来演示。 """ # 模拟数据,实际项目中请勿这样做 mock_orders = { "A1001": { "order_id": "A1001", "status": "已发货", "item": "无线鼠标", "delivery_time": (datetime.now() - timedelta(days=2)).isoformat(), "can_return": True, }, "A1002": { "order_id": "A1002", "status": "派送中", "item": "机械键盘", "delivery_time": datetime.now().isoformat(), "can_return": False, }, "B2001": { "order_id": "B2001", "status": "已签收", "item": "显示器支架", "delivery_time": (datetime.now() - timedelta(days=30)).isoformat(), "can_return": False, }, } return mock_orders.get(order_id, {"error": "订单不存在或无权访问"}) def create_return_request(order_id: str, reason: str) -> dict: """创建退货申请""" return { "success": True, "ticket_id": f"RMA-{order_id}-{int(datetime.now().timestamp())}", "message": "退货申请已创建,请等待审核", }这个工具函数有几个细节值得注意:
- 真实场景中,工具调用应该带有用户身份信息,不能只凭一个订单号就返回数据,必须校验当前用户是否有权访问该订单。
- 工具应该具备幂等性,重试时不会产生重复订单操作。
- 返回结构最好统一,方便后续节点解析。
5.2 第二步:绑定工具到 LLM
# agent/config.py from langchain_openai import ChatOpenAI from agent.tools import query_order, create_return_request def build_llm(): """构建带有工具调用能力的模型""" llm = ChatOpenAI( model="gpt-4o-mini", temperature=0, ) return llm.bind_tools([query_order, create_return_request])bind_tools是 LangChain 提供的能力,它会把工具的结构化描述(函数名、参数、说明)传给模型,让模型在需要时返回一个工具调用请求。注意temperature=0,在 Agent 场景里,工具调用决策需要确定性,温度过高会导致模型“发挥不稳定”。
如果你使用的是 Ollama 本地模型,可以改用:
from langchain_ollama import ChatOllama llm = ChatOllama(model="llama3.1", temperature=0) tools = [query_order, create_return_request] llm_with_tools = llm.bind_tools(tools)5.3 第三步:定义图结构
# agent/graph.py from typing import Literal from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver from agent.state import AgentState from agent.config import build_llm from agent.tools import query_order, create_return_request # 使用内存版 Checkpointer 做持久化 memory = MemorySaver() def llm_node(state: AgentState) -> dict: """主 LLM 节点:决定下一步动作""" llm = build_llm() response = llm.invoke(state["messages"]) return {"messages": [response]} def tools_node(state: AgentState) -> dict: """工具执行节点:执行模型发起的工具调用""" last_message = state["messages"][-1] if not hasattr(last_message, "tool_calls") or not last_message.tool_calls: return {"messages": []} updated_messages = [] for tool_call in last_message.tool_calls: tool_name = tool_call["name"] tool_args = tool_call["args"] if tool_name == "query_order": result = query_order(**tool_args) if "error" not in result: state["order_info"] = result elif tool_name == "create_return_request": result = create_return_request(**tool_args) else: result = {"error": f"未知工具: {tool_name}"} updated_messages.append({ "role": "tool", "content": str(result), "tool_call_id": tool_call["id"], }) return {"messages": updated_messages} def should_continue(state: AgentState) -> Literal["tools", END]: """路由函数:模型是否请求调用工具""" last_message = state["messages"][-1] if hasattr(last_message, "tool_calls") and last_message.tool_calls: return "tools" return END def escalate_node(state: AgentState) -> dict: """升级人工客服""" return { "messages": [{"role": "ai", "content": "您的需求已记录,系统正在为您转接人工客服,请稍候。"}], "escalated": True, } def build_graph(): # 1. 创建状态图,指定 State 类型 workflow = StateGraph(AgentState) # 2. 注册节点 workflow.add_node("llm", llm_node) workflow.add_node("tools", tools_node) workflow.add_node("escalate", escalate_node) # 3. 定义入口 workflow.add_edge(START, "llm") # 4. 定义条件边:LLM 决定是否调工具 workflow.add_conditional_edges("llm", should_continue, {"tools": "tools", END: END}) # 5. 工具执行完返回 LLM workflow.add_edge("tools", "llm") # 6. 编译图,注入 Checkpointer app = workflow.compile(checkpointer=memory) return app这段代码是 LangGraph 入门最重要的骨架,我逐块解释一下。
StateGraph(AgentState)创建了一个以AgentState为状态模型的状态图。add_node注册节点,每个节点对应一个函数。add_edge(START, "llm")定义图执行的入口,所有流程从llm节点开始。
add_conditional_edges是核心路由逻辑:在llm节点执行完后,调用should_continue函数,根据返回结果决定进入tools节点还是结束流程。这里的返回类型Literal["tools", END]是让路由明确化,避免拼写错误。
workflow.compile(checkpointer=memory)编译出一个可执行对象。MemorySaver把状态保存在内存中,适合开发和测试。生产环境可以替换为langgraph-checkpoint-postgres或langgraph-checkpoint-redis等持久化实现。
5.4 第四步:加一个老手最容易忽略的纠错节点
仅靠上面的代码,这个 Agent 已经可以运行了。但实际使用中很快会遇到一个问题:模型可能连续多次调用同一个工具,或者在一次工具返回错误后陷入死循环。
更稳的做法是加入一个“重复检测”节点,统计同一工具在最近 N 次轮次中的调用次数,如果超过阈值,就强制走人工升级。
# agent/graph.py 追加 def anti_loop_node(state: AgentState) -> dict: """防循环节点:检测连续重复工具调用""" messages = state["messages"] if len(messages) < 4: return {"escalated": False} # 统计最近 6 条消息里的工具调用名 tool_calls = [] for msg in messages[-6:]: if hasattr(msg, "tool_calls"): for tc in msg.tool_calls: tool_calls.append(tc["name"]) if len(tool_calls) >= 3 and len(set(tool_calls)) <= 1: return {"escalated": True, "messages": [{ "role": "ai", "content": "系统检测到重复操作,已自动为您转接人工客服。" }]} return {"escalated": False}然后在图中注册节点并调整边的方向:
workflow.add_node("anti_loop", anti_loop_node) workflow.add_edge("tools", "anti_loop") workflow.add_edge("anti_loop", "llm")这样执行路径变成了llm -> tools -> anti_loop -> llm,循环被显式约束,不再是模型的自由发挥。
5.5 第五步:并行查询能力
售后场景中,用户可能同时反馈多个订单的问题。顺序查询会变慢,而且会让用户等很久。LangGraph 的SendAPI 可以在同一个图中动态创建多个并行分支。
这里直接回答很多读者关于Send(node_name, state)的疑问:它和普通add_edge不同。普通边是在编译时固定好的路径,而Send是在运行时根据当前状态动态创建“多个新任务”,每个任务携带独立的 state 快照,进入同一个节点并行执行。
from langgraph.types import Send def parallel_query_node(state: AgentState) -> dict: """将待查订单拆成多个并行任务并合并结果""" results = {} pending = state.get("pending_orders", []) for order_id in pending: # 真实项目中,这里应该通过异步或固定并发窗口执行 # 例如使用 asyncio.gather 配合限流 results[order_id] = query_order(order_id) return {"order_results": results, "pending_orders": []} def dispatch_parallel(state: AgentState): """动态分发并行任务""" # 使用 Send 为每个订单创建一个并行分支 return [ Send("query_one_order", {"order_id": order_id}) for order_id in state.get("pending_orders", []) ] def query_one_order(state: dict) -> dict: """单个订单查询节点""" order_id = state["order_id"] return {"order_results": {order_id: query_order(order_id)}}关于Send的进一步理解:普通边在编译期就确定了“谁连谁”,而Send是在运行期动态决定“发多少任务给谁”。如果业务场景是“把列表拆成多个子任务并发执行”,用Send是最合适的方案。
但这里有个工程细节提醒你:并发不是免费的。如果每个子任务都会调用外部 API,你需要注意限流、超时和错误隔离。LangGraph 对并发的支持很好,但下游服务的承受能力不一定跟得上。建议用信号量或批处理来控制并发窗口。
5.6 完整主流程
# main.py from agent.graph import build_graph def main(): app = build_graph() # config 里的 thread_id 是会话标识 config = {"configurable": {"thread_id": "user-123-session-456"}} # 第一轮对话 result1 = app.invoke( {"messages": [{"role": "user", "content": "你好,我想查一下订单 A1001 到哪里了"}]}, config=config, ) print("第一轮回复:", result1["messages"][-1].content) # 第二轮对话:不传历史,因为有 thread_id,状态自动保留 result2 = app.invoke( {"messages": [{"role": "user", "content": "这个鼠标可以退货吗"}]}, config=config, ) print("第二轮回复:", result2["messages"][-1].content) if __name__ == "__main__": main()运行方式:
python main.py这里最关键的一点是thread_id。它相当于一次业务会话的标识。只要同一个thread_id,LangGraph 会通过 Checkpointer 自动把之前的状态和消息历史加载出来。这也意味着,就算是 serverless 环境,APP 实例被回收,只要 Checkpointer 和 thread_id 不变,对话就能无缝恢复。
6. LangGraph 运行结果与效果验证
6.1 预期成功表现
正常运行时,你会看到类似下面的轮次:
- 用户输入“我想查一下订单 A1001 到哪里了”。
- LLM 节点判断需要调用
query_order工具。 tools节点执行工具函数并返回订单状态。llm节点根据工具返回结果生成自然语言回复。- 图中终止,返回最终结果。
第二次对话时,即使你没有显式传入历史消息,Agent 也会记得第一轮查的是 A1001。这是因为MemorySaver保存了整张图的状态,第二轮 LLM 节点看到的是“完整历史 + 新提问”。
6.2 如何判断图执行是否正确
LangGraph 最实用的调试方式是打开调试输出,查看每一步节点名和状态变化:
from langchain.globals import set_debug set_debug(True) app.invoke( {"messages": [{"role": "user", "content": "我想查订单 A1001"}]}, config={"configurable": {"thread_id": "debug-session"}}, )在 debug 模式下,控制台会打印出每一步的节点执行顺序、工具调用输入输出和状态更新情况。这是定位流程异常的第一手段。
6.3 状态回溯与断点检查
如果你需要人工审批流程,LangGraph 支持在执行到指定节点前暂停:
# 在 escalate 节点执行前暂停 app.invoke( {"messages": [{"role": "user", "content": "我要投诉"}]}, config={"configurable": {"thread_id": "human-in-loop", "stop": "escalate"}}, )这种能力对企业级场景非常关键:流程可以停在人工审批点,等待用户确认后再继续。多智能体之间如果有人工审核节点,这套机制可以确保每步操作都有记录、有授权、可追溯。
6.4 常见失败现象与处理
如果工具返回“订单不存在”,模型可能一直追问用户,而不是改变策略。这是 LangGraph 开发中非常典型的问题:Agent 陷入“反复尝试同一个失败路径”的循环。
解决思路有两个方向:
一是靠防循环节点做硬限制(上面已经实现)。
二是在工具返回错误信息时附带下一步建议,比如“如果订单不存在,请提示用户核对订单号,不要重复查询”。
这些细节决定了 Agent 在真实场景中是“智能”还是“智障”。
7. LangGraph 常见问题与排查思路
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 模型不返回 tool_calls | 模型不支持工具调用,或 temperature 设置不合理 | 检查模型版本和参数 | 换用工具调用兼容模型,调低 temperature |
| 工具调用循环不停 | 路由逻辑缺少终止条件 | 打开 debug 日志观察边走向 | 增加防循环节点,设置最大迭代次数 |
| 多轮对话丢失记忆 | 忘记配置 Checkpointer,或 thread_id 不一致 | 检查编译时 checkpointer 参数 | 配置持久化 Checkpointer,统一 thread_id |
| Send 并行任务结果丢失 | 状态合并逻辑写错 | 检查子节点返回值 key 是否一致 | 统一状态更新字段,使用合并函数 |
| 图执行很慢 | 节点串行执行,模型调用次数多 | 查看 debug 日志统计调用次数 | 合并节点、减少模型往返、引入并行机制 |
| 第三方 API 超时 | 下游服务响应慢 | 查看工具函数日志 | 设置请求超时,增加重试,配置排队机制 |
下面挑两个最常遇到的问题展开说明。
问题一:模型不调用工具,反复输出文字。
如果用的是 OpenAI 早期模型或某些本地模型,工具调用能力比较弱。解决方案是换用支持 function calling 的模型,或者改用 JSON Output 模式,先让模型输出结构化 JSON,再由代码解析判断下一步动作。
问题二:Agent 陷入死循环。
这是所有 Agent 框架都会遇到的问题,不只是 LangGraph。常见的防循环手段:
- 在 State 里增加
step_count,节点每次执行时加一,超过阈值强制走人工升级。 - 检测到重复调用同一工具且参数相同,直接返回错误。
- 给 LLM 的 Prompt 里明确写“不要重复调用同一工具,除非参数发生变化”。
在企业级 Agent 里,死循环不只是技术问题,还可能导致下游系统被重复调用,产生脏数据或重复订单。必须在图层面做好硬限制。
8. LangGraph 工程化最佳实践与多智能体架构建议
8.1 状态 Schema 设计原则
状态是 LangGraph Agent 的核心,设计时建议遵循三个原则:
第一,能收则收,能简则简。不要把一整个第三方接口返回值都塞进 State,只保留节点之间真正需要传递的信息。State 里的数据越杂,节点之间的耦合就越高,调试越困难。
第二,区分短期状态与长期状态。比如order_info这类业务数据属于短期状态,随会话结束可以清理;用户偏好、历史行为等属于长期状态,应放独立的记忆服务,不要全部塞进 State。
第三,显式定义合并逻辑。使用Annotated类型标注字段的合并方式。默认是后续值覆盖前面的值;对话消息是追加合并;列表字段可能需要去重合并。
from typing import Annotated, List from langgraph.graph.message import add_messages def merge_deduplicate(left: List[str], right: List[str]) -> List[str]: """列表去重合并""" return list(dict.fromkeys(left + right)) class AgentState(TypedDict): messages: Annotated[List[dict], add_messages] order_ids: Annotated[List[str], merge_deduplicate]8.2 工具设计与权限控制
工具是 Agent 调动系统的“手”,权限边界必须收敛。真实项目中建议做到:
- 每个工具接入前先审查,确认它只做该做的事。
- 工具内部必须校验“当前用户是否有权操作该资源”,不能只靠 Agent 的 Prompt 来约束。
- 写操作(创建退货单、改动订单、发送通知)应设计确认环节。
- 记录工具调用的全量日志:入参、出参、耗时、发起节点、会话 ID。
这里特别提醒:很多 Agent 事故不是模型“变坏”了,而是工具权限过大,模型在正常推理过程中拿到了超出范围的权限。LangGraph 可以在工具层加一层权限过滤,确保任何模型都无法调用未注册的工具。
8.3 模型 Cost 与延迟控制
企业级 Agent 必须关注成本。一个简单的 FAQ 问题,如果每次都调用工具、每次调用模型,成本会非常高。常用的优化手段:
- 在图中加“快速通道”节点:先判断问题是否命中常见问题库,命中则直接返回,不经过模型。
- 对简单任务使用小模型(如
gpt-4o-mini),对复杂任务切换大模型。 - 缓存重复的工具调用结果,同一订单在短时间内不用重复查询。
- 设定单次会话最大模型调用次数,超限后转人工。
8.4 多智能体架构设计的经验
多智能体不是越“多”越好。很多场景下,一个设计良好的单 Agent 比三个混乱的 Agent 更可靠。什么时候才需要多智能体?
一种情况是领域隔离需求强,比如售前 Agent 和售后 Agent 使用不同的知识库、不同的工具,放一起会导致 Prompt 互相污染。
另一种情况是流程复杂度高,比如客服 Agent 需要后端系统权限,但权限策略不同,拆成多个 Agent 有助于安全隔离。
LangGraph 的多智能体实现通常有两种方式:
# Supervisor 模式简化示例(伪代码) def supervisor_node(state): # 让主模型决定调用哪个子 Agent # 子 Agent 也是一个编译后的 LangGraph app return {"next_agent": "after_sales"} workflow.add_node("supervisor", supervisor_node) workflow.add_conditional_edges( "supervisor", lambda state: state["next_agent"], { "after_sales": "after_sales_agent", "pre_sales": "pre_sales_agent", END: END, }, )在真实项目中,每个子 Agent 建议独立维护 Prompt、工具和状态,避免相互影响。子 Agent 之间通过主 Agent 的 State 传递最终结果,而非直接共享内部状态。
8.5 生产环境的 Checkpointer 选型
本文示例使用的是MemorySaver,进程重启后状态清空,只适合开发测试。
生产环境需要可持久化的 Checkpointer。LangGraph 支持 PostgreSQL、Redis 等后端:
langgraph-checkpoint-postgres:适合需要强一致性和事务保障的场景。langgraph-checkpoint-redis:适合高吞吐、低延迟的会话状态场景。
选择标准取决于你的 Agent 对状态丢失的容忍度。如果 Agent 涉及交易、审批、工单流转,建议使用 PostgreSQL 这种具备持久化保障的存储。
8.6 可观测性与日志
Agent 的可观测性比传统应用更重要,因为它的执行路径不固定。
至少需要记录:
- 每个节点的执行时间与状态。
- 工具调用的入参与出参。
- LLM 的 Prompt 与响应(注意脱敏)。
- 条件路由的判断依据。
- 会话 ID 与最终结果。
LangGraph 提供了回调机制可以接入日志和追踪系统。建议从第一天就把日志规范化,不要等上线后再补。
9. 总结与后续学习方向
LangGraph 不是另一个花哨的 AI 框架,它是 Agent 开发走向工程化的一个重要基础设施。它真正解决的不是“能不能做 Agent”,而是“Agent 能不能稳定地跑在企业系统里”。
有几个判断值得记住:
第一,LangChain 解决的是“模型和工具怎么接”,LangGraph 解决的是“Agent 流程怎么管”。二者不是替代关系,而是分工关系。
第二,State 是 LangGraph 的灵魂。很多从 LangChain 转过来的开发者,入门最快的方法是先忘记 Chain,记住“节点就是函数、状态就是数据、边就是路由”这句话。
第三,Checkpointer 是 LangGraph 打开企业级场景的钥匙。有了它,Agent 不再是一个无状态的函数调用,而是一个可以被暂停、恢复、跟踪的业务流程。
后续你可以从这几个方向继续深入:
- 学习
langgraph dev工具链,用官方开发工具提高迭代效率。 - 深入研究 Checkpointer 的持久化实现,理解断点恢复和人工审批集成。
- 阅读 LangGraph 源码里关于
Send和图执行器(Pregel)的实现原理,这对理解多智能体并行非常有帮助。 - 把示例中的
MemorySaver替换为 PostgreSQL 或 Redis 版本,亲手体验生产环境下的状态管理。
最后提醒一句:Agent 开发最重要的是克制。不要一上来就追求大而全的多智能体架构,先从一个用 LangGraph 管理状态和流程的最小 Agent 跑通,再逐步叠加记忆、并行、人工审核这些能力。流程越简单,越容易发现问题的根源。希望这篇文章能帮你顺利跨过 LangGraph 入门到实战的门槛,值得先收藏,等要动手写 Agent 的时候再翻出来对照看看。