★ FEATURED ARTICLE 先说结论如果你还在用一层while循环加一堆if-else拼装大模型调用LangGraph 绝对是值得你花一个周末认真研究的东西。它不是又一个套壳框架而是把智能体这件事从写流程升级成了画状态图让复杂的多步推理、工具调用、人工介入、状态回溯都有了清晰的落点。这篇文章我不打算念文档而是从实际构建智能体的角度把 LangGraph 的核心设计、关键陷阱和一套能直接跑通的实践方案拆开讲清楚。我自己是在一个客服工单自动分类的项目里第一次大规模用上 LangGraph 的。之前用纯代码编排十来个工具函数、七八个判断分支代码已经乱到改一个逻辑要全局搜索三遍。换成 LangGraph 之后图的结构、状态的流动、每一步的输入输出都变得肉眼可见排查问题从猜变成了看。这篇文章适合两类人一类是刚接触 Agent 开发、想找个靠谱框架起步的另一类是已经被各种编排代码折磨过、想找个更优雅方案的。不管你是哪种跟着这篇文章走一遍你会有种原来 Agent 还能这么写的感觉。1. 这一框架解决的是写 Agent 最恶心的那部分1.1 手写 ReAct 循环的痛点我们先回忆一下不用框架时一个带工具调用的 Agent 长什么样。核心逻辑通常是这样把用户问题塞给大模型大模型可能返回一个工具调用请求代码解析这个请求、执行对应的工具函数、把结果拼回去再丢给大模型如此循环直到大模型说我不需要工具了直接回答。听起来很简单对吧但一旦工具数量超过五个分支逻辑开始变多这个循环就会失控。你需要手动维护一个messages列表每次迭代都要小心翼翼地追加消息你需要判断工具调用是参数格式错误还是业务逻辑错误你还需要处理工具执行超时了怎么办、同一轮调用里多个工具并行执行的结果怎么合并这类边角问题。更别提如果你还想支持用户中途打断、修改参数、然后继续这种交互手写代码的复杂度会直接爆炸。我用过一个很典型的方案用一个大字典保存整个对话状态每次循环结束后手动更新这个字典。前两周还好到了第三周这个字典里有十几个键每个键都有好几层嵌套结构我根本记不清哪一层需要在哪一步更新。出现 Bug 的时候调试器里看那个巨大的字典人都是懵的。1.2 StateGraph 和普通流程图的核心区别LangGraph 的核心思路不是给你一个更高级的循环而是让你用图来描述整个 Agent 的行为。一个图由节点Node和边Edge构成。节点是你希望执行的一步操作比如调用大模型、执行搜索工具、生成最终回答边定义了节点之间的流转关系。但这里有个关键区别普通流程图是数据流驱动数据从一个节点流到下一个节点LangGraph 的 StateGraph 是状态驱动。什么意思整个图共享一个状态对象这个对象会随着节点的执行不断更新。每个节点读取这个状态、做一些事情、然后把更新后的状态返回。图的下一个节点看到的是更新后的同一个状态而不是上一个节点的输出。这个设计带来的直接好处是节点之间不需要关心彼此的输入输出签名只要约定好状态里有什么字段就行。你可以随时插入一个新的节点只要它操作的状态字段和相邻节点兼容整个图依然能跑。这在迭代开发中太重要了。我经常是先画一个三步的图跑通然后再加一个意图判断节点插在最前面整个过程不需要改动原有节点的一行业务代码。如果打个比方普通代码编排就像一条流水线每个工位只处理自己的零件零件形态变了整条线就得改LangGraph 则像一块共享的白板每个角色节点都往白板上写自己的内容下一个角色只要看白板上有什么就能继续干活。白板本身状态是贯穿始终的。2. 核心概念逐层拆解节点、边、状态和 Reducer2.1 状态是你唯一需要操心的数据在 LangGraph 里状态State本质上就是一个数据结构。官方文档里通常用 TypedDict 来定义但实际项目中我个人更推荐用dataclass因为当你需要定义复杂的嵌套结构或者要给状态字段写默认值、写辅助方法时dataclass的灵活性高得多。一个典型的研究助手状态可能长这样from dataclasses import dataclass, field dataclass class ResearchState: topic: str # 当前研究主题 plan: list field(default_factorylist) # 研究计划步骤 search_results: dict field(default_factorydict) # 每个计划步骤对应的搜索结果 report: str # 最终研究报告 max_steps: int 5 # 最大搜索轮次 current_step: int 0 # 当前执行到第几步每个节点函数接收一个state参数返回一个字典。返回的字典就是你要更新的状态字段。要注意的是你只需要返回你想更新的字段不需要返回全部状态。LangGraph 会把你返回的部分更新和原有状态做合并。这个合并背后的逻辑就是 Reducer。2.2 Reducer多节点写入同一状态的合并规则这是 LangGraph 和普通 DAG 框架最不一样、也最容易踩坑的地方。默认情况下如果有两个节点都要往状态里同一个字段写东西后执行的节点会把先执行节点的结果整个覆盖掉。这是绝大多数新手的第一个 Bug 来源我以为节点 A 往search_results里写了三个关键词的结果节点 B 又往里写了一个关键词结果回头一看search_results里只剩节点 B 写的那一个。解决办法是给状态字段指定 Reducer。Reducer 是一个函数它在节点返回新值和当前状态里的旧值之间执行合并逻辑。最常用的是operator.add它的作用是如果字段的旧值和新值都是列表就把它们拼接起来。用dataclass定义带 Reducer 的状态需要用到Annotated类型from typing import Annotated, TypedDict import operator class ResearchState(TypedDict): topic: str search_results: Annotated[list, operator.add] # 每次写入自动追加而不是覆盖我实际项目里几乎给所有列表字段都配了operator.add因为绝大多数场景下你希望的是收集所有节点的产出而不是用最后一个节点的产出。如果你的合并逻辑更复杂比如要去重或者要按照某个字段聚合那就写一个自定义函数传进去。Reducer 函数签名是(old_value, new_value) - merged_value灵活度很高。2.3 条件边和 Command控制流的分岔口有了状态和节点接下来需要决定下一步去哪。LangGraph 有两种边普通边和条件边。普通边就是固定流转比如节点 A 结束后必然去节点 B。条件边则是在节点执行完后根据状态内容动态决定下一个节点是谁。条件边的核心是一个路由函数。这个函数接收当前状态返回一个字符串这个字符串必须是一个已注册的节点名或边的关键字。看个例子def should_continue(state: ResearchState): if state[current_step] state[max_steps]: return generate_report # 达到上限去生成报告 return search_tool # 没达到上限继续搜索返回的是下一步要去哪个节点的名字。LangGraph 会拿着这个返回值去匹配图中定义好的条件边映射。这里有一个进阶玩法值得了解Command对象。在一些复杂场景下你可能希望一个节点同时在更新状态和指定下一步跳转两件事上拥有一票否决权。用Command可以做到from langgraph.types import Command from typing import Literal def my_node(state: ResearchState) - Command[Literal[search, end]]: # 做点事情 if state[something]: return Command( update{important_flag: True}, # 先更新状态字段 gotosearch # 再强制指定下一步 ) return Command(update{important_flag: False}, gotoend)这种写法比条件边更直接适合那种路由决策和状态更新强耦合的场景。但我个人建议能不用就不用条件边直观、可调试性强Command 用得多了图的可读性会变差。3. 从零构建一个可用的研究助手智能体前面原理讲了不少这一节我们动手写一个能真正跑起来的东西。需求很明确用户输入一个主题智能体自动拆解研究问题、多次调用搜索工具收集信息、最后生成一份结构化的研究报告。3.1 先想清楚需求再画图动手写代码之前我建议你先在纸上画一下这个 Agent 的行为。我的设计是这样的入口节点plan_research接收用户主题调用大模型生成研究问题列表写入状态的plan字段。条件节点should_search检查current_step是否达到max_steps决定下一步是去搜索还是去写报告。执行节点search_tool根据当前步骤对应的研究问题调用搜索工具把结果存入search_results。汇总节点summarize把最新搜索结果和大模型已有消息拼起来让大模型做一轮信息提取更新消息历史。终止节点generate_report拿全部研究结果生成报告。这里面search_tool执行完后会通过Command主动把current_step加一然后无条件回到should_search形成循环。这个搜索 - 判断 - 搜索的环路就是研究型 Agent 的核心节律。3.2 节点函数怎么写才不会踩坑每个节点就是一个普通的 Python 函数签名统一是(state) - dict或(state) - Command。我直接贴一个实际可用的版本重点看节点之间怎么通过状态协作。from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage from langchain_community.tools import DuckDuckGoSearchRun llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) search DuckDuckGoSearchRun() PLAN_PROMPT 你是研究规划助手。请把用户的研究主题拆解为最多3个具体搜索问题每个问题单独一行。只输出问题本身。 def plan_research(state: ResearchState): resp llm.invoke([SystemMessage(contentPLAN_PROMPT), HumanMessage(contentstate[topic])]) questions [line.strip() for line in resp.content.splitlines() if line.strip()] return {plan: questions, current_step: 0} def search_tool(state: ResearchState) - Command[Literal[should_continue]]: step state[current_step] query state[plan][step] if step len(state[plan]) else state[topic] result search.run(query) return Command( update{ search_results: {query: result}, # 注意这里返回一个单元素字典 current_step: step 1 }, gotoshould_continue )这里有个细节search_results字段是一个Annotated[list, operator.add]类型的列表但我在search_tool里返回的是一个字典。这会不会类型冲突不会。LangGraph 的operator.add处理的是字段级别的追加合并如果字段值是字典那每次写入会整体替换答案是会。所以这里我建议把search_results定义成Annotated[list, operator.add]然后每次写入的是一个单元素列表而不是字典。正确写法class ResearchState(TypedDict): topic: str plan: Annotated[list, operator.add] search_results: Annotated[list, operator.add] # 存 {query: result} 字典的列表 report: str max_steps: int current_step: int然后def search_tool(state: ResearchState) - Command[Literal[should_continue]]: step state[current_step] query state[plan][step] if step len(state[plan]) else state[topic] result search.run(query) return Command( update{ search_results: [{query: query, result: result}], current_step: step 1 }, gotoshould_continue )summarize节点会读取search_results列表的最后一个元素把查询和结果包装成一条消息追加到messages列表里然后让大模型提取要点。3.3 编译运行图怎么变成可以调用的对象节点定义好之后就是构建图from langgraph.graph import StateGraph, START, END from typing import Literal def should_continue(state: ResearchState) - Literal[search_tool, generate_report]: if state[current_step] state[max_steps]: return generate_report return search_tool builder StateGraph(ResearchState) builder.add_node(plan_research, plan_research) builder.add_node(search_tool, search_tool) builder.add_node(summarize, summarize) builder.add_node(generate_report, generate_report) builder.add_edge(START, plan_research) builder.add_edge(plan_research, search_tool) builder.add_edge(search_tool, should_continue) # 条件节点作为边先注册 builder.add_conditional_edges(search_tool, should_continue) builder.add_edge(generate_report, END) graph builder.compile()注意add_conditional_edges的写法第一个参数是从哪个节点出发第二个参数是路由函数。意思是当search_tool节点执行完后调用should_continue函数根据返回值决定跳转到search_tool继续循环还是generate_report结束。调用也很简单result graph.invoke({topic: LangGraph 状态管理最佳实践}, config{recursion_limit: 20}) print(result[report])invoke的第一个参数是初始状态LangGraph 会自动补全其他字段。config里的recursion_limit是递归深度上限循环图里这个参数特别重要不设置的话跑几轮可能就报超出限制的错误。整个流程跑下来你会发现图内部每个节点的输入输出都清晰得像看动画状态一步步累积搜索结果一条条追加最终报告基于完整上下文生成。这种可观测性是纯代码编排很难做到的。4. 持久化、人工介入和部署中的实际经验4.1 Checkpointer让 Agent 真正有记忆LangGraph 有一个概念叫 Checkpointer作用是把每一步执行完的状态保存下来。有了它Agent 就有了两样能力断点续跑和对话历史管理。断点续跑的意思是如果图中某一步抛了异常或者你主动设置了中断点图执行到那里会暂停检查一下状态修完再继续。这在生产环境太重要了。我之前做客服工单分类偶尔会遇到工具 API 超时没有 Checkpointer 的话整个会话就得重新跑一遍用户等得暴躁。有了 Checkpointer超时之后我可以从失败节点重新执行前面已经干完的活不白干。对话历史管理更直白每次graph.invoke之后图的状态会存到 Checkpointer 里。下一次用户发消息你只需要带着同一个thread_id去invoke图就自动从上次结束的状态继续而不是从头开始。from langgraph.checkpoint.memory import InMemorySaver checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) config {configurable: {thread_id: user-123}} graph.invoke({topic: LangGraph}, configconfig) # 用户追加问题 graph.invoke({topic: 顺便说说 Checkpointer 怎么用}, configconfig)不过InMemorySaver只适合单进程本地调试。多实例部署时多个服务进程各自持有内存状态同一个用户的请求打到不同实例上就乱了。生产环境建议用数据库类型的 Checkpointer比如PostgresSaver或SqliteSaver。使用方式和内存版几乎一样只是初始化时多传一个连接参数。4.2 人机协同节点interrupt 的玩法研究助手还好但如果你构建的是一个需要人类审批或输入的 Agent比如帮我写邮件发送前让我看一下纯自动流转就不够了。LangGraph 提供了一个interrupt函数可以在节点内部暂停图的执行等待外部输入。from langgraph.types import interrupt def review_node(state): message f请确认是否发送邮件收件人{state[recipient]} decision interrupt({message: message}) if decision approve: return {approved: True} return {approved: False}图跑到interrupt时会抛出一个GraphInterrupt异常但状态已经保存在 Checkpointer 里。你的应用捕获这个异常后可以把interrupt返回的message发给前端等用户点了同意或拒绝之后再用Command(resume...)从暂停处继续。from langgraph.types import Command graph.invoke(Command(resumeapprove), configconfig)这个机制我一开始觉得挺绕后来发现它本质上就是把人的决策也当作一个外部工具调用和工具调用的流程天然统一。特别适合审批流、人工复核、敏感操作确认这类场景。4.3 部署时的常见坑部署 LangGraph 应用最典型的坑是 Checkpointer 类型选错。内测时候用InMemorySaver一切都好一上多实例就出现用户请求串线的诡异问题。排查了半天发现是没有跨实例共享状态。换成PostgresSaver后就好了。所以我的建议是项目第一天就按生产环境设计直接用数据库 Checkpointer省得后面返工。另一个坑是 LangGraph 的流式输出。它支持stream_mode参数可以是values每次状态变更都流式输出、updates只输出节点产生的更新或者自定义模式。前端要展示Agent 正在做什么用updates最合适。但注意一点stream_mode和interrupt配合时中断信息会出现在流的末尾格式还不太一样需要单独处理。5. 常见问题与排查技巧实录5.1 状态怎么改都不生效Reducer 写错了这是出现频率最高的问题没有之一。绝大多数为什么我更新了状态但下一个节点看到的是旧值的困惑都是因为字段没配 Reducer被后一个节点的返回值整个覆盖了。排查思路很固定先看这个字段是不是被多个节点写了。如果是就把字段类型改成Annotated[list, operator.add]或者自定义合并逻辑。还有一个常见的变体operator.add配在TypedDict字段上但字段初始值是字典或字符串节点返回的值也是字典或字符串那合并逻辑就是把两个字典做加法运算直接报TypeError。检查状态定义和节点返回类型的一致性是排查这类问题最快的路径。5.2 条件边返回了不存在的路由名这个问题我等会儿用一个速查表来汇总。条件边的路由函数返回值必须要能匹配一个已注册的节点名。如果你写了个拼写错误LangGraph 会抛出类似Node not found的异常。注意看报错信息它通常会明确指出找不到哪个节点。这种错基本是粗心造成的但有时也会因为重构改名后忘了更新条件边而出现。我的建议是给路由函数的返回值加上Literal类型标注让 IDE 帮你在写代码时就检查出来别等到运行时才暴露。5.3 循环死循环图停不下来循环图是 LangGraph 的强项也最容易出事。当你配置了递归限制 (recursion_limit)超限时LangGraph 会抛GraphRecursionError告诉你执行步数超出了限制值。解决死循环有几种思路一是设置recursion_limit这是兜底二是像我的current_step一样在状态里维护一个步数计数器到点自动退出三是更优雅一点在should_continue里判断最近几轮搜索结果是否有明显增益没有就提前收敛。实际项目中我把第三种方法用在了客服工单分类上如果最近两轮工具调用返回的内容和已有关键信息高度重合就认为信息够用了直接跳去生成最终回答。这个策略可以省掉很多不必要的 token 消耗。5.4 Checkpointer 相关的诡异问题用数据库 Checkpointer 时有时候会遇到corrupted state之类的报错。常见原因是状态对象里有不可序列化的字段比如某些自定义类的实例。LangGraph 在把状态存到数据库前会做序列化如果你往状态里塞了 Python 对象而这类没有实现序列化协议就会出现问题。解决办法是把状态里的复杂对象转成 JSON 兼容结构再存。我通常在状态里只放str、dict、list这类基础类型真正需要复杂对象时先把它们序列化成字符串存进状态用时再反序列化。丑是丑了点但绝对稳。还有一个容易忽视的点thread_id一定要保证唯一。我之前在测试环境随便生成 UUID结果一个用户开几个会话历史记录互相串场。下面把几种高频问题整理成一张速查表方便你排查时快速定位问题现象排查方向解决方案状态字段被后节点覆盖检查字段是否被多节点写入给字段配 Reducer如 operator.add条件边报节点不存在检查路由函数返回值加 Literal 类型标注核对节点名图执行步数超限检查循环退出条件设置 recursion_limit 添加步数计数状态存储报序列化错误检查状态里是否有复杂对象转成 JSON 兼容结构多实例下会话串线检查 Checkpointer 类型生产环境改用数据库 Checkpointerinterrupt 后无法恢复检查是否用了同一个 thread_id确保 config 里 thread_id 不变5.5 调试技巧画图之前先想状态最后一个建议也是最想强调的一点先用一个下午想清楚状态再动手写节点。LangGraph 的代码组织方式决定了你的状态设计就是整个应用的数据库设计。字段拆得越细后面加功能越容易字段定义越模糊后面到处都在为这个字段到底存了什么而吵架。我踩过比较大的坑是把中间过程和最终结果混在同一个字段里。比如用search_results既存搜索结果又存总结后的要点结果读取方根本分不清当前拿到的是原始内容还是加工过的内容。正确的做法是分成两个字段raw_results存原始搜索summarized_points存提炼过的要点。各节点职责清晰后面加缓存、加日志也都方便。写 Code 时每个节点函数开头和结尾各打印一下state的关键字段这个习惯在很多次排障里救过我。写在最后的一点个人体会LangGraph 真正让我觉得顺手的是图是数据、数据是图这种统一模型。状态贯穿整个流程节点只关心自己该读什么、写什么图结构就是可执行的架构文档。运行时你看一眼图的流转日志就知道 Agent 每一步在做什么、为什么这么走而不是面对一大坨调用链抓瞎。当然它也不是万能的。如果你的 Agent 只有两三个步骤、没有循环、不需要人机交互那你用手写代码反而更快。LangGraph 解决的是复杂的确定性流程编排问题没有这个复杂度用它就是杀鸡用牛刀。如果你打算深入建议从官方文档里那个ReAct agent示例开始跑通之后试着把里面的工具换成一个真实 API。等你体会到换个工具只需要加一个节点、不需要动主流程的时候你就会明白为什么一个图状编排框架能成为新一代智能体构建的主流选择。
阅读完成 · 觉得有帮助？
