ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

LangGraph 状态机实战:构建可控 Agent 的分支与循环

LangGraph 状态机实战:构建可控 Agent 的分支与循环 1. 为什么单靠 LangChain 写 Agent 迟早会失控我最早接触 Agent 开发的时候用的是最直觉的写法把工具定义好塞给大模型然后写一个while循环让模型自己决定下一步调什么工具、什么时候停。这套东西在 demo 阶段跑得飞快二十行代码就能做出一个能查天气、能算数、能搜网页的助手发给同事看的时候大家都觉得挺唬人。但只要业务稍微复杂一点问题就全冒出来了。最典型的是三件事第一模型偶尔会陷入死循环同一个工具反复调用烧 token 烧到心疼第二中间某一步失败了整个流程只能从头再来前面已经拿到的结果全白费第三也是最要命的你根本说不清楚这个 Agent 到底应该怎么走因为流程完全藏在模型的脑子里出了问题只能靠打印日志一点点猜。这就是一段式状态机和多段式状态机的区别。一段式就是一个大状态里啥都干所有逻辑揉在一起靠条件判断硬撑而 LangGraph 走的是显式状态机的路子——把 Agent 的每一步拆成独立的节点节点之间用边连起来状态在节点之间流转。你不再问模型下一步想干嘛而是问当前状态满足什么条件该走哪条边。打个生活化的比方。一段式 Agent 像是让一个实习生自己看着办你只说把这事搞定他可能干得不错也可能跑偏LangGraph 像是给他画了一张流程图每一步该做什么、做完交给谁、什么情况下打回重做全都写在纸上。前者灵活但不可控后者可控且可调试。LangGraph 和 LangChain 的关系也得说清楚这是新手最容易绕晕的地方。LangChain 提供的是积木——模型封装、工具封装、提示词模板、检索器这些都是零件LangGraph 提供的是装配图纸和流水线——它管的是这些零件怎么按顺序、按条件、按循环组合成一个有状态的流程。你可以只用 LangChain 不用 LangGraph也可以只用 LangGraph 而工具自己手写两者不是替代关系是分层关系。很多人问LangChain 和 LangGraph 的区别一句话LangChain 管能做什么LangGraph 管按什么顺序做、做到哪一步了。那到底什么样的场景值得上 LangGraph我的判断标准很简单只要你的 Agent 流程里出现了分支或者循环就该考虑它。比如先判断用户意图是查订单就走订单流程是退款就走退款流程这是分支生成内容后自检不合格就重新生成最多重试三次这是循环。这两种结构用纯 LangChain 硬写代码会迅速变成意大利面条而用 LangGraph 就是画几个节点连几条边的事。2. LangGraph 的三个核心概念状态、节点、边在动手写代码之前必须把 LangGraph 的抽象模型吃透。它的世界观非常小小到只有三个东西State状态、Node节点、Edge边。理解了这三个剩下的都是排列组合。2.1 State整个流程共享的那块白板State 是整个图的共享内存。你可以把它想象成一块白板每个节点都能上来读上面的内容也能往上写东西。所有节点看到的是同一块白板这就是为什么 LangGraph 能做多步推理——上一步的结论写在白板上下一步直接读。State 的定义方式通常是一个TypedDict或者用 Pydantic 模型。字段就是你想在流程里传递的数据。比如一个客服 AgentState 里可能有user_query用户问题、intent识别出的意图、order_info订单信息、final_answer最终回复。这里有个新手必踩的坑State 的字段更新规则。默认情况下节点返回的字典会覆盖对应字段。但如果你希望某个字段是累加的比如消息历史messages每轮对话都往里追加而不是覆盖就得用Annotated配合 reducer 函数。最常见的写法是from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] intent: stradd_messages这个 reducer 的作用就是新消息追加到列表末尾而不是把整个列表替换掉。我第一次写的时候没加这个结果多轮对话永远只有最后一条消息排查了半天才发现是覆盖问题。这个细节官方文档里有但很容易被跳过。2.2 Node一个只做一件事的函数Node 就是一个普通的 Python 函数签名是接收 State返回要更新的字段字典。它不需要继承任何类不需要装饰器除非你要做特殊处理就是一个函数。def classify_intent(state: State): query state[messages][-1].content # 调用模型判断意图 intent llm.invoke(f判断这句话的意图{query}) return {intent: intent}Node 的设计哲学是单一职责。一个节点只干一件事要么调模型要么调工具要么做数据转换。不要把判断意图 查订单 生成回复塞进一个节点那样你就又回到一段式状态机了。拆得越细越容易调试越容易复用。我个人的经验是如果一个节点的函数体超过 30 行就该考虑拆了。拆节点的成本很低但调试一个巨型节点的成本极高。2.3 Edge决定下一步去哪Edge 分两种普通边和条件边。普通边就是干完 A 直接干 B用add_edge(A, B)声明。条件边是干完 A 之后根据 State 里的某个值决定去 B 还是 C用add_conditional_edges声明需要提供一个路由函数。def route_by_intent(state: State): if state[intent] 查订单: return query_order elif state[intent] 退款: return refund else: return fallback graph.add_conditional_edges(classify, route_by_intent)路由函数返回的是一个字符串这个字符串必须和某个节点的名字对应。这里有个隐蔽的坑返回的字符串拼错了不会报错而是直接抛异常说找不到节点。所以路由函数的返回值最好用常量或者枚举别手写字符串。还有一个特殊节点叫END表示流程结束。所有分支最终都要能走到END否则图会一直转下去。我见过有人忘了给某个分支连END结果那个分支跑完就卡住了日志也不报错非常难查。把这三个概念串起来看State 是数据Node 是计算Edge 是控制流。整个 LangGraph 程序就是定义 State → 写 Node → 连 Edge → 编译成图 → 调用。这个模型简单到有点朴素但正是这种朴素让它能表达任意复杂的流程。3. 从零搭一个带分支和循环的 Agent光讲概念没意思直接上一个能跑的完整例子。我选一个最经典的场景智能客服意图路由。用户说一句话Agent 先判断意图然后走不同分支其中退款分支还带一个自检循环。3.1 环境准备与依赖安装先说环境。LangGraph 对 Python 版本要求是 3.9 以上我建议直接用 3.11兼容性最好。依赖装这几个就够pip install langgraph langchain langchain-openai如果你用 conda 管理环境注意一个坑conda 默认源里的 langgraph 版本可能偏旧建议用 pip 装或者把 conda 的 pip 优先级调高。我遇到过 conda 装出来的版本缺少add_messages的情况换成 pip 就好了。模型这块任何兼容 OpenAI 接口的服务都能用把base_url和api_key配好即可。我下面用ChatOpenAI举例你换成自己的模型客户端就行。3.2 定义 State 和节点函数先定义 State。这个客服场景需要传递用户消息、识别出的意图、订单信息、退款草稿、重试次数。from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages class CustomerServiceState(TypedDict): messages: Annotated[list, add_messages] intent: str order_info: str refund_draft: str retry_count: int注意retry_count是普通字段默认覆盖我们在节点里手动加一。messages用add_messages累加。然后写节点。第一个节点是意图分类def classify_intent(state: CustomerServiceState): last_msg state[messages][-1].content prompt f判断下面这句话的意图只返回一个词 查订单 / 退款 / 其他 用户说{last_msg} result llm.invoke(prompt).content.strip() return {intent: result}第二个节点是查订单def query_order(state: CustomerServiceState): # 实际项目里这里查数据库 order_info 订单号 12345状态已发货预计明天送达 return {order_info: order_info}第三个节点是生成退款草稿def draft_refund(state: CustomerServiceState): draft llm.invoke( f根据用户诉求生成一段退款说明{state[messages][-1].content} ).content return {refund_draft: draft, retry_count: state.get(retry_count, 0) 1}第四个节点是自检def check_refund(state: CustomerServiceState): draft state[refund_draft] result llm.invoke( f这段退款说明是否礼貌且完整只回答 合格 或 不合格\n{draft} ).content.strip() return {intent: result} # 借用 intent 字段传自检结果这里我偷了个懒用intent字段传自检结果实际项目里应该单独开一个字段别学我。3.3 用条件边实现意图路由现在把节点连起来。先建图graph StateGraph(CustomerServiceState) graph.add_node(classify, classify_intent) graph.add_node(query_order, query_order) graph.add_node(draft_refund, draft_refund) graph.add_node(check_refund, check_refund) graph.add_node(fallback, lambda s: {order_info: 抱歉我没理解您的意思})设置入口点graph.set_entry_point(classify)然后加条件边从classify出发根据意图分流def route_intent(state: CustomerServiceState): if state[intent] 查订单: return query_order elif state[intent] 退款: return draft_refund else: return fallback graph.add_conditional_edges(classify, route_intent)query_order和fallback都是终点直接连ENDgraph.add_edge(query_order, END) graph.add_edge(fallback, END)3.4 用循环边实现退款自检重试退款分支要复杂一点生成草稿 → 自检 → 合格就结束不合格就回去重新生成最多三次。graph.add_edge(draft_refund, check_refund) def route_check(state: CustomerServiceState): if state[intent] 合格: return end elif state.get(retry_count, 0) 3: return end # 超过三次强制结束 else: return retry graph.add_conditional_edges( check_refund, route_check, {end: END, retry: draft_refund} )注意add_conditional_edges的第三个参数是一个映射字典把路由函数的返回值映射到实际节点。这样路由函数返回end和retry这种语义化字符串映射到END和draft_refund可读性好很多。最后编译并调用app graph.compile() result app.invoke({ messages: [(user, 我要退款)], retry_count: 0 }) print(result[refund_draft])跑通之后你会发现整个流程的每一步都是可见的、可断点的、可重放的。这就是显式状态机的价值。4. 调试、持久化与踩坑实录代码能跑通只是第一步真正让 LangGraph 在生产里站住脚的是它的调试能力和持久化机制。这一块我踩的坑最多单独拎出来讲。4.1 用 stream 看清每一步到底发生了什么app.invoke()只给你最终结果中间过程是黑盒。调试的时候一定要用app.stream()for event in app.stream({messages: [(user, 我要退款)], retry_count: 0}): print(event)它会每执行完一个节点就 yield 一次你能清楚看到classify输出了什么、draft_refund生成了什么、check_refund判定结果是什么。我第一次调意图路由的时候模型把我要退款识别成了其他就是因为提示词里没给例子加上 stream 一看就发现了。还有一个更细粒度的astream_events能拿到 token 级别的流式输出做前端打字机效果的时候用得上。不过它的事件格式比较复杂新手先用stream就够了。4.2 Checkpointer让流程能暂停、能恢复、能重放LangGraph 有个杀手锏叫Checkpointer检查点。它会在每个节点执行后自动保存一次 State 快照。这意味着什么意味着你的 Agent 可以中途暂停等人类审批后再继续意味着程序崩了可以从上一个检查点恢复不用从头跑意味着你可以回放整个执行历史看每一步的状态变化。开启方式很简单编译的时候传一个 checkpointerfrom langgraph.checkpoint.memory import MemorySaver memory MemorySaver() app graph.compile(checkpointermemory)调用的时候必须传config里面带thread_idconfig {configurable: {thread_id: user-001}} app.invoke({messages: [(user, 我要退款)]}, config)thread_id就是会话 ID同一个 ID 的多次调用会共享状态历史。这个设计对多轮对话特别友好——你不需要自己维护对话历史checkpointer 帮你存着。MemorySaver是内存版重启就没了。生产环境要换成SqliteSaver或者 Postgres 版把检查点落到数据库。我建议开发阶段就用 Sqlite方便查看历史比内存版好用太多。这里有个坑用了 checkpointer 之后State 里不能有不可序列化的对象。比如你往 State 里塞了一个数据库连接或者模型客户端保存检查点的时候会直接报错。解决办法是把这些重资源放在节点函数外部作为全局变量State 里只放纯数据。4.3 人类介入interrupt 与审批流Checkpointer 带来的另一个能力是interrupt也就是在某个节点前暂停等人类输入后再继续。这在需要审批的场景里非常有用比如退款金额超过一定阈值必须人工确认。from langgraph.types import interrupt def human_approval(state: CustomerServiceState): decision interrupt(请审批这笔退款) return {intent: decision}调用时如果走到这个节点invoke会返回一个中断信号你拿到后做人工审批再用Command(resume...)恢复执行。这套机制让AI 自动处理 人类兜底的混合流程变得非常自然。我实测下来interrupt 的恢复逻辑有个细节要注意恢复时传入的值会作为 interrupt 函数的返回值而不是覆盖 State。所以你在节点里要显式地把这个返回值写回 State否则人工审批的结果就丢了。4.4 几个高频踩坑与排查思路把我在实际项目里踩过的坑整理成表方便对照排查现象根因解决多轮对话只有最后一条消息State 字段没加 reducer被覆盖用Annotated[list, add_messages]路由报找不到节点路由函数返回的字符串和节点名不匹配用常量或枚举别手写字符串图跑完卡住不结束某个分支没连到 END检查所有分支的出口保存检查点报序列化错误State 里有不可序列化对象重资源放节点外State 只放纯数据循环停不下来缺少最大重试次数保护在路由函数里加计数判断条件边不生效路由函数返回值不在映射字典里检查映射字典的 key 是否齐全这些坑的共同点是报错信息往往不直接指向根因。比如找不到节点其实是路由函数拼写问题序列化错误其实是 State 设计问题。所以调试 LangGraph 的核心方法是用 stream 看每一步的输入输出而不是盯着报错信息猜。5. 什么场景该上 LangGraph什么场景别硬上聊完技术细节说点更实际的选型判断。我见过太多人一上来就用 LangGraph 写一个本来十行代码能搞定的东西结果图比业务逻辑还复杂。工具是好工具但用错地方就是负担。5.1 适合 LangGraph 的四类场景第一类是多分支路由。用户输入进来先分类再走不同处理链路这是 LangGraph 最舒服的场景。意图识别、工单分类、内容审核分流都属于这一类。第二类是带自检和重试的生成流程。生成 → 检查 → 不合格重生成这种循环用 LangGraph 表达非常自然而且重试次数、退出条件都是显式的不会失控。第三类是需要人类介入的审批流。Checkpointer interrupt 这套组合让AI 处理到一半停下来等人确认变得很简单这是纯 LangChain 很难优雅实现的。第四类是长流程、需要断点续跑的任务。比如一个要跑十几步的数据处理流水线中间某步失败了你希望从失败点恢复而不是从头来。Checkpointer 就是为这个设计的。5.2 不适合硬上的三类场景反过来这几种情况我建议别用 LangGraph单步问答。用户问一句模型答一句没有任何分支和循环。这种用 LangChain 的chain或者直接调模型就行上 LangGraph 是杀鸡用牛刀。纯工具调用型 Agent。如果流程就是模型自己决定调哪个工具没有固定的分支结构那用 LangChain 的AgentExecutor或者更轻量的方案反而更合适。LangGraph 的价值在于流程可控如果流程本来就是让模型自由发挥那它的优势发挥不出来。原型验证阶段。如果你还在探索这个 Agent 到底该怎么做流程还没定型那先用最简单的写法快速试错等流程稳定了再迁移到 LangGraph。过早引入状态机会让你的探索速度变慢。5.3 和其他工作流工具的定位差异市面上工作流工具很多Dify、Coze、n8n 这些可视化平台和 LangGraph 的定位其实不一样。可视化平台适合非程序员快速搭流程拖拖拽拽就能出东西但灵活性和可编程性受限LangGraph 是代码优先的适合需要精细控制、需要嵌入到现有 Python 项目里的场景。我个人的判断是如果流程固定、参与者是非技术人员用可视化平台如果流程复杂、需要和代码深度集成、需要自定义状态逻辑用 LangGraph。两者不是竞争关系是不同层次的选择。还有一个常见困惑是LangGraph 和普通状态机比如 C 语言里手写的状态机有什么区别。本质区别在于传统状态机是确定性的状态转移完全由代码逻辑决定LangGraph 的状态转移可以由模型决定条件边的路由函数里可以调模型。这就把确定性流程和模型自主决策结合起来了既有流程的可靠性又有模型的灵活性。6. 把状态机思维迁移到其他 Agent 项目LangGraph 教会我的最重要的一件事其实不是某个 API 怎么用而是状态机思维。这套思维一旦建立你会发现它能迁移到很多地方。6.1 状态机思维的核心把隐式流程变成显式结构大多数 Agent 项目出问题根源都是流程藏在代码的 if-else 里或者藏在模型的脑子里。状态机思维要求你把流程画出来有哪几个状态、每个状态做什么、什么条件下转移到下一个状态、哪些状态是终态。这个画出来的动作本身就是价值。我现在的习惯是动手写代码前先用纸笔画出状态图节点是圆圈边是箭头条件写在箭头旁边。画完再写代码几乎不会出现写到一半发现流程漏了的情况。这套思维不限于 LangGraph。你用任何框架写 Agent甚至不用框架都可以先画状态图。图清楚了代码只是翻译。6.2 从一段式到多段式的渐进式重构如果你手上已经有一个一段式的 Agent想迁移到状态机不用推倒重来。我的建议是渐进式重构第一步先把现有流程里的关键决策点找出来这些点就是未来的条件边。第二步把每个决策点之间的处理逻辑抽成独立函数这些就是未来的节点。第三步用 LangGraph 把这些节点和边连起来先跑通主流程。第四步再逐步加上 checkpointer、interrupt 这些高级能力。这样重构风险最小每一步都能验证。我迁移过一个跑了半年的客服 Agent整个过程分了三周每周迁移一部分线上一直没出问题。6.3 状态设计的几条经验法则最后分享几条我在 State 设计上总结的经验都是踩坑换来的State 字段宁少勿多。只放真正需要在节点间传递的数据临时变量放在节点函数内部。State 越干净checkpointer 序列化越省事调试时看状态也越清晰。给每个字段想清楚更新语义。是覆盖还是累加覆盖用普通字段累加用 reducer。想不清楚就会出 bug。重资源绝不进 State。模型客户端、数据库连接、文件句柄这些放节点函数外部。State 里只放纯数据这是铁律。给循环加硬性上限。任何可能循环的边都要有最大次数保护。模型有时候会钻牛角尖没有上限的循环就是烧钱机器。路由函数保持纯粹。路由函数只做判断不做副作用操作。它应该是读 State返回字符串不调模型、不写数据库。这样路由逻辑才能被单独测试。这套东西说起来简单但真正在项目里坚持下来Agent 的可靠性会有质的提升。我从一段式写到多段式最大的感受就是可控性比灵活性更重要。一个你能完全掌控的 Agent哪怕能力弱一点也比一个能力强但随时可能失控的 Agent 有价值。LangGraph 给的就是这份掌控感。
返回列表