最近我在做模拟项目X的时候一个内部客服工单分类的Agent工作流跑得越来越乱——分支逻辑全靠if-else硬撑会话上下文散落在各个函数里工具调用的顺序稍微一变就要返工。后来我把整个流程迁到了LangGraph上思路一下就顺了。它把Agent的运行方式从一段连续代码变成了一张显式控制的有状态图节点负责执行边负责决定下一步走向状态贯穿全程且可持久化。这篇文章就围绕LangGraph这个框架从入门概念到实战代码聊聊我踩过的坑和最终落地的做法。如果你已经开始用LangChain写Agent或者刚接触LangGraph应该知道这套东西的文档说实话不算友好——概念多示例散版本更新又快。我尽可能用一个完整的实战案例把核心知识点串起来读完你能理解它为什么把“工作流”叫“图”以及有状态、可中断、可恢复这几个能力到底怎么用。1. 为什么是LangGraph从链式调用到有状态编排1.1 链式Agent的结构性缺陷我最早实现的Agent思路很直白用户输入进来走一遍意图识别再走一遍工具调用最后生成答案返回。代码体现出来就是一步步顺序执行每一步接收字符串返回字符串靠最外层一个Session变量记住历史。这种模式在Demo阶段完全没有问题因为每一步之间的依赖关系是线性且固定的。但一旦业务复杂起来缺陷就非常明显。比如意图识别之后根据结果可能要调用不同的工具链这就在“顺序执行”中间撕开了一个分支口又比如某个工具的执行结果需要人工确认后才能进入下一步顺序链就不得不停下来等外部信号。这些情况在普通代码里意味着大量全局状态、回调函数和复杂的错误处理整套逻辑的可维护性会随着业务复杂度指数下滑。1.2 LangGraph的解题思路LangGraph解决的思路是把工作流抽象成一张图。图的每个节点是一段独立的逻辑单元边定义了节点间的流转关系而状态对象是贯穿全局的唯一数据源。LangGraph底层用了一个类似消息传递的执行模型每次运行都是状态对象在节点间流动节点只负责消费和更新状态的某一部分不用关心其他节点怎么运行、运行了几次。这个模型的优势在于执行的控制权完全交还给了开发者。条件分支通过条件边实现循环通过边构成环实现人机交互通过断点和恢复机制实现。整个工作流是显式的图结构而不是隐式的代码顺序这样无论是调试还是扩展都有迹可循。有一点我需要提醒LangGraph本身演进得很快不同版本的API表达差异不小。我下面主要以当前主流版本的习惯用法来写代码如果你看到StateGraph、START、END这些对象说明版本相对较新。如果是旧项目注意迁移时核对API。2. 核心概念逐个击破2.1 State整个图的唯一数据流理解LangGraph的钥匙是State。它通常定义成一个TypedDict或者一个Pydantic模型描述整个工作流运行期间需要维护的数据结构。from typing import TypedDict, Annotated from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): user_input: str intent: str response: str messages: list这个State会在每个节点执行时作为输入传入节点返回一个字典作为状态的增量更新。理解这个机制很关键节点不是返回整个新状态而是只返回自己负责变化的那部分字段。LangGraph对状态字段还有个处理机制叫Reducer用来决定一个字段被多次写入时怎么合并。比如messages这个字段每次节点往里追加消息就要用operator.add这个Reducer来声明合并行为否则后写入的值会覆盖之前的值。import operator from typing import TypedDict, Annotated class AgentState(TypedDict): user_input: str messages: Annotated[list, operator.add]用Annotated给字段标注Reducer这在LangGraph里是一个高频操作。我一开始没加结果多轮的对话历史总是只剩最后一轮排查了半天才发现是Reducer的问题。2.2 Node与Edge图的基本组成Node就是普通的Python函数接收当前状态返回状态更新。比如一个意图识别节点可能长这样def classify_intent(state: AgentState) - dict: # 实际项目里这里会调用模型做分类 intent faq if 退款 in state[user_input] else normal return {intent: intent}Edge分为普通边和条件边。普通边表示固定流转比如节点A结束后必然进入节点B。条件边则需要一个路由函数根据当前状态动态决定下一跳def route_by_intent(state: AgentState) - str: if state[intent] faq: return faq_node return normal_node条件边在构建图时通过add_conditional_edges注册这个函数接收三样东西起始节点、路由函数、路由结果到实际节点名的映射字典。2.3 图的构建与执行图的构建分三步创建StateGraph实例注册节点注册边。最后调用compile()方法获得一个可执行的CompiledGraph对象。builder StateGraph(AgentState) builder.add_node(classify, classify_intent) builder.add_node(faq_node, handle_faq) builder.add_node(normal_node, handle_normal) builder.add_edge(START, classify) builder.add_conditional_edges( classify, route_by_intent, {faq_node: faq_node, normal_node: normal_node} ) builder.add_edge(faq_node, END) builder.add_edge(normal_node, END) graph builder.compile()执行时用invoke方法传入初始状态result graph.invoke({user_input: 我想申请退款, messages: []})这里invoke返回的是最终状态不是某个节点的返回值。这个细节容易让新手困惑——你以为返回的是最后一个节点的输出实际上返回的是整个State对象。如果只想取最后一条消息需要自己从State里取字段。2.4 Checkpointer状态持久化的基座LangGraph真正区别于普通状态机的地方是Checkpointer。它把每一步执行完成后的状态快照保存下来这样执行中断后可以从最近一个快照恢复。官方提供了多种存储实现包括内存存储和基于数据库的持久化方案。from langgraph.checkpoint.memory import InMemorySaver checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: user-123}} result graph.invoke({user_input: 你好}, configconfig)thread_id是一个看似简单但极其重要的参数。它是会话的唯一标识LangGraph根据它来区分哪几轮运行属于同一条记忆线。同一个thread_id下多次invoke会共享状态历史不同thread_id之间则完全隔离。Checkpointer还带来两个高级能力Time Travel回放历史状态和Human-in-the-loop人机协同。前者允许你回到某个中间节点重新执行后者可以在特定节点前暂停等人为确认后再继续。这两块我在实战案例里具体展开。3. 环境准备与AI Agent入门示例3.1 安装与版本确认安装LangGraph建议先建一个干净的虚拟环境避免和旧项目依赖冲突。pip install langgraph langchain-openai装完最好确认一下版本因为不同版本的API有差异遇到网上老教程对不上号时先检查这里import langgraph print(langgraph.__version__)如果你只是想快速试验用在线环境也行但本地跑代码更可控。另外需要准备一个LLM的API密钥LangGraph本身并不绑定特定厂商兼容OpenAI协议的服务都能接进来。3.2 一个最简单的有状态工作流我建议初学第一个示例不要接模型先跑通图本身。下面这个示例包含两个节点节点间传输状态from typing import TypedDict from langgraph.graph import StateGraph, START, END class DemoState(TypedDict): count: int log: str def increment(state: DemoState) - dict: new_count state[count] 1 return {count: new_count, log: fincrement to {new_count}} def double(state: DemoState) - dict: new_count state[count] * 2 return {count: new_count, log: state[log] f, double to {new_count}} builder StateGraph(DemoState) builder.add_node(increment, increment) builder.add_node(double, double) builder.add_edge(START, increment) builder.add_edge(increment, double) builder.add_edge(double, END) graph builder.compile() result graph.invoke({count: 1, log: }) print(result)这个例子跑通后你会直观感受到几个东西State是唯一的流转对象节点之间的调用是隔离的最终输出是整个State。这个简单模式就是后续所有复杂案例的骨架。约10天前在我本地环境跑输出结果是{count: 4, log: increment to 2, double to 4}这是符合预期的。3.3 加入LLM形成一个真正的最小Agent有了图的基础再加入LLM调用就形成了一个最小可用Agent。这里以OpenAI兼容接口为例from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) def call_llm(state: DemoState) - dict: response llm.invoke(state[user_input]) return {response: response.content} # 将这一步做首节点即可形成 用户输入 - LLM - 输出 的流程但这里有一个新手最容易踩的坑LangChain的invoke返回一个AIMessage对象不是字符串。取内容要用.content属性。我在早期的代码里直接把整个Message对象塞进状态后面序列化的时候各种报错。真实世界的Agent不会只有一步LLM调用一定会有工具调用和条件判断接下来我用一个完整的案例把这些串起来。4. 实战构建一个带工具调用和记忆的客服Agent工作流4.1 项目背景与整体设计这个实战案例的需求是这样的做一个客服Agent用户进来后先判断意图是“查订单”还是“提交投诉”根据意图走不同的处理分支。查订单需要调用一个模拟的订单查询工具提交投诉则需要把内容写入一个目标系统并且整个过程要支持多轮对话也就是要记住用户之前说过什么。我把整个工作流设计成五个节点节点名称职责classify意图识别输出intent字段query_order调用订单查询工具只处理查订单意图submit_complaint调用投诉提交工具只处理投诉意图generate_answer汇总工具结果生成给用户的最终回复route_by_intent条件路由不在节点内做业务逻辑整体流转是START进入classify然后条件路由到query_order或submit_complaint两者都汇聚到generate_answer最后到达END。4.2 完整代码实现先定义状态和工具import json from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import InMemorySaver class ServiceAgentState(TypedDict): user_input: str intent: str tool_result: str messages: Annotated[list, operator.add] def query_order_tool(order_id: str) - str: # 模拟查询订单实际项目中这里通常会查数据库或调用第三方接口 fake_db {A1001: 已发货预计3天后送达, A1002: 已签收} return fake_db.get(order_id, 未查询到该订单信息) def submit_complaint_tool(content: str) - str: # 模拟提交到工单系统 return f投诉已受理工单编号T{len(content) % 1000} def classify_intent(state: ServiceAgentState) - dict: text state[user_input] if 订单 in text or 快递 in text: return {intent: order} if 投诉 in text or 不满 in text: return {intent: complaint} return {intent: other} def query_order(state: ServiceAgentState) - dict: # 简单从输入里提取订单号实战中一般由LLM先做实体抽取 order_id state[user_input].split(订单)[-1].strip() result query_order_tool(order_id) return {tool_result: result, messages: [f查单结果{result}]} def submit_complaint(state: ServiceAgentState) - dict: result submit_complaint_tool(state[user_input]) return {tool_result: result, messages: [f投诉结果{result}]} def generate_answer(state: ServiceAgentState) - dict: return {messages: [f最终答复{state[tool_result]}]} def route_by_intent(state: ServiceAgentState) - str: if state[intent] order: return query_order if state[intent] complaint: return submit_complaint return generate_answer builder StateGraph(ServiceAgentState) builder.add_node(classify, classify_intent) builder.add_node(query_order, query_order) builder.add_node(submit_complaint, submit_complaint) builder.add_node(generate_answer, generate_answer) builder.add_edge(START, classify) builder.add_conditional_edges( classify, route_by_intent, {query_order: query_order, submit_complaint: submit_complaint, generate_answer: generate_answer} ) builder.add_edge(query_order, generate_answer) builder.add_edge(submit_complaint, generate_answer) builder.add_edge(generate_answer, END) checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: customer-001}} first graph.invoke({user_input: 查一下订单A1001, messages: []}, configconfig) print(first[messages])4.3 运行效果与设计解读第一次运行状态流转路径是classify识别为order路由到query_order拿到工具结果generate_answer生成答复。四个节点的执行顺序由路由函数在运行时决定而不是写死在代码里。当用户第二轮输入“我要投诉快递太慢”仍然使用同一个thread_id此时状态里已经保留了第一轮的messages记录新消息追加进去形成了一个有记忆的多轮会话。这个例子里比较关键的设计有三个。第一messages字段用了Annotated和operator.add所以多轮信息不会互相覆盖。第二中间节点只负责修改自己的字段比如query_order只写tool_result和messages不碰intent这种职责隔离让每个节点可以独立测试。第三conditional edge的映射表名字和节点名保持一致看起来重复但这是LangGraph的解析约定省去一层间接的查表逻辑。4.4 引入真正的LLM和ReAct风格循环模拟版本跑通后进阶的方向是把“意图识别”和“实体抽取”从规则代码替换成LLM。ReAct风格的Agent本质上就是一个循环模型思考决定调用哪个工具观察工具返回再思考直到不需要调用工具为止。和上面这个客服案例相比它的路由不是一次性的而是一个循环。LangGraph官方提供了一个封装好的Agent模式可以用create_react_agent快速构建。但项目稍微复杂之后我建议仍然要理解它背后的图结构。官方实现里通常有agent节点和tools节点agent节点负责模型推理和工具选择tools节点负责执行工具并返回结果两者之间靠条件边形成循环——当模型决定不再需要工具时循环跳出走向END。from langgraph.prebuilt import create_react_agent # 假设 tools 是一个已定义好的工具列表 agent create_react_agent(llm, tools, checkpointercheckpointer) result agent.invoke({messages: [(user, 请帮我查一下订单A1001)]}, configconfig)这种封装的写法启动快但如果你想精细控制“哪几步允许中断”“工具返回之后要不要做额外的规则校验”还是需要手动构图。我的经验是快速原型用预置Agent正式项目手工构图。5. 进阶技巧Human-in-the-loop与在线记忆扩展5.1 在关键节点前暂停等待人工确认有状态工作流最有价值的落地场景之一就是人机协同。某些操作比如退款、删除数据、批量发送消息不适合让Agent自动完成需要在操作前停下来问用户“确认执行吗”。LangGraph通过interrupt函数实现这个能力配合Checkpointer把暂停前后的状态都保存下来。核心用法是在节点里调用interrupt当前执行会暂停外部拿到控制权确认之后用同一个thread_id继续执行代码片段中interrupt调用位置的上下文会被保留。from langgraph.types import interrupt def confirm_refund(state: ServiceAgentState) - dict: # 第一次执行到这里时会抛出中断信号等待外部确认 user_confirm interrupt({action: confirm_refund, amount: state[amount]}) # 恢复执行时这里的返回值就是外部传入的确认结果 if user_confirm and user_confirm.get(confirmed): return {messages: [(system, 退款已执行)]} return {messages: [(system, 退款已取消)]}这种模式的要求是外部必须维护同一个thread_id。中断时用graph.get_state(config)拿到当前快照用graph.invoke(Command(resume...), configconfig)传回确认结果。刚开始接触这个概念有点绕我的理解捷径是中断不是终止而是把执行器挂起把决策权借给外部。5.2 用数据库Checkpointer实现持久化记忆InMemorySaver适合Demo和测试进程重启后记忆就丢了。实际项目里要用持久化数据库方案。LangGraph官方提供了对Postgres、Redis等存储的支持不同场景选择不同存储方案特点适用场景InMemorySaver快、无持久化本地开发、测试PostgresSaver持久化、可多实例共享生产环境多副本部署RedisSaver低延迟、支持过期机制高并发会话场景接入PostgresSaver需要额外安装依赖并初始化数据表。官方通常提供setup命令或者要求手动执行建表SQL。数据库方案不影响业务代码逻辑只需要在compile时替换checkpointer其他代码可以保持不变这也是LangGraph接口设计比较友好的一点。5.3 流的三种状态与开发调试技巧每一个执行到中途或被暂停的图都可以通过快照查看状态。调试时最有用的命令是state_snapshot graph.get_state(config) for task in state_snapshot.next: print(task)state_snapshot.next表示接下来待执行的节点集合。如果列表为空说明图已经执行完毕如果不为空说明执行中断在那个节点上。我调试条件路由不按预期走时经常打印这个字段来确认实际路径。LangGraph还支持流式返回graph.stream会逐节点产出状态变化适合观察长流程的中间变化。形式上比invoke复杂一点但对于理解图的执行顺序帮助很大。6. 常见问题与排坑实录6.1 状态字段被覆盖而不是合并这是最常遇到的问题。现象是多轮对话中上一轮的记录被这一轮覆盖了。原因基本是字段没有声明Reducer。记住一个规律普通值字段比如count、intent默认覆盖符合直觉。列表累积字段比如messages、history必须加Annotated[list, operator.add]。我在做模拟项目X时第一次写多轮客服工作流就漏了这个排查了一下午。现在基本养成习惯定义State时先想清楚每个字段是“唯一值”还是“累积列表”。6.2 invoke返回的不是最终回答不少新手期待invoke直接返回模型的最终文本实际返回的是整个State字典。你需要自己从字典里取。如果想要简化可以在最后一个节点里把要暴露的字段整理成{response: ...}外部读取result[response]这样外部接口层就不用关心内部State结构。6.3 条件边映射表写错导致KeyErroradd_conditional_edges的第三个参数是一个映射字典键是路由函数可能的返回值值是节点名。路由返回了字典里不存在的键时会直接报KeyError。防止这个问题的做法是在路由函数里严格限定返回值并且做一个兜底def route_by_intent(state: ServiceAgentState) - str: return { order: query_order, complaint: submit_complaint, }.get(state[intent], generate_answer)6.4 thread_id没有传导致记忆不生效Checkpointer编译进图中后如果你invoke时忘记传config每次运行就会是一个全新的会话前面的记忆全都没了。而且不会报错只表现为“模型不记得之前说过什么”排查难度挺高。我的经验是只要compile时带了checkpointer调用时务必带上config {configurable: {thread_id: ...}}。6.5 版本差异导致的API不兼容LangGraph从早期版本到现在API变化包括MemorySaver改名为InMemorySaver一些内置类型从langgraph.prebuilt挪到了别的位置部分远端调用方式也变了。遇到网上旧教程时多看官方文档的changelog。最好的判断标准是在本地写个小示例跑通再复制别人的代码。7. 一点经验心得LangGraph这套框架的使用体验和之前用的普通容器编排很不一样。它逼你把业务流程先想清楚再把每个环节拆成独立节点。刚开始我觉得这一步很麻烦但实际项目越复杂这种前置设计的价值就越大。相比在普通代码里穿插大量分支和状态变量图结构的可观测性和调试便利性显然高一个档次。如果只看一个建议我会说把State的定义当成系统接口设计来做。字段怎么命名、哪些允许累积、哪些上下层节点共享这些在动手写节点之前想清楚后面能少走很多弯路。至于什么节点内部用规则逻辑还是LLM那是后话图的结构定了替换只是改一个节点内部实现的事。最后分享一个小技巧做图设计时先在纸上画出你预期的工作流流转图标出每个节点会读哪些字段、写哪些字段再开始写代码。这一步可能省掉你最痛苦的调试阶段。后续如果要做横向扩展比如接入更复杂的工具集、加多模态输入也只需要在图上加节点和边原有代码的稳定性基本不受影响。