直接切入正题。这两年做大模型应用尤其是Agent类项目最绕不开的一个痛点就是逻辑稍微复杂一点代码就开始失控。今天聊的这个LangGraph就是专门来治这个病的。LangGraph是LangChain团队推出的一个框架核心思路一句话就能讲明白把Agent的工作流建模成一张图节点是你要执行的步骤边是步骤之间的流转规则状态则是贯穿全程的数据载体。它能解决的问题很具体——当你需要让一个Agent具备多步推理、循环反思、条件分支、甚至人工介入确认这些能力时普通链式调用会写得越来越拧巴而图结构天生就是干这个的。这篇文章的目标受众很明确已经了解LangChain基础用法开始写Agent但觉得不顺手想找一个更可控、更适合生产落地的编排方案的人。市面上聊LangGraph的教程不少但多数是翻译文档的复读机讲一堆概念却不说人话。我这篇尽量反过来先讲清楚它设计的逻辑再给一套能直接上手的代码和踩坑记录。1. LangGraph到底解决了什么问题1.1 传统Agent编排的三个典型困境很多人第一次写Agent用的是LangChain的AgentExecutor。说实话入门确实快写个ReAct模式的Agent几行代码就完事。但一到真实场景你会撞见三堵墙。第一堵墙是状态管理混乱。Agent运行过程中中间结果散落在各个局部变量里你想拿到某一步的推理记录、某些中间输出要么靠回调钩子硬捞要么自己缝一个全局对象到处传。代码越多传递链路越长出bug的时候根本不知道是哪一步弄脏了数据。第二堵墙是循环控制靠prompt碰运气。比如你希望Agent在生成代码之后先自测一遍发现报错就回到上一步修改最多重试3次。用传统方式实现你得在prompt里写“如果遇到错误就重新调用工具”模型不一定听你的流程控制完全不deterministic。第三堵墙是无法暂停和恢复。生产环境里Agent常需要等人工审批后才能继续或者跑了一半挂了下次要从断点续跑。AgentExecutor对这些场景的支持约等于零你得自己再造一套持久化机制。这三堵墙本质上指向同一个根源Agent不是一个简单的text-in-text-out的链它是一个有状态、有分支、可能反复循环的状态机。用链式思维去建模天花板就摆在那。1.2 图结构如何重塑工作流LangGraph把问题换个角度解你不需要在代码层通过函数调用来隐式控制流程而是显式地定义一张有向图。这张图里**Node节点**是一个Python异步或同步函数负责执行某个具体逻辑比如“调用大模型”“调用搜索工具”“判断是否需要重试”**Edge边**定义了执行顺序从A节点走完接下来进入B节点State状态是一个共享的数据对象每个节点都可以读取和更新它另外还有条件边Conditional Edge执行完一个节点后根据状态里的某个字段决定下一步走哪条路。用这套抽象你可以画出这样的流程大模型生成回答 → 如果回答需要调用工具则进入工具节点 → 工具返回结果后回到大模型节点 → 如果达到最大迭代次数则强制结束。整张图就是Agent的完整执行逻辑每一步修改都很直观控制权完全在你手里不用再赌模型的自觉性。这套设计有它天然的工程优势——图和执行器是分离的。你可以把一张图编译成一个可执行对象反复调用也可以把图的结构序列化存起来做可视化甚至能引用同一个图的不同示例并行跑不同的输入。这对想上生产的团队来说吸引力非常大。1.3 和LangChain的关系以及何时不必用它LangGraph并不是要取代LangChain它和LangChain的模型接口、工具封装完美兼容。你可以直接用ChatOpenAI、Tool、create_react_agent这类现成组件再把它嵌进LangGraph的节点里。也就是说LangChain负责提供积木块LangGraph负责把积木搭成可管控的流水线。但我也要泼盆冷水不是所有项目都需要上LangGraph。如果你的Agent只是简单的“问一句答一句”偶尔调一个工具用常规链或者AgentExecutor反而更轻快。图结构的建模需要设计状态、拆节点是有初始成本的。我的判断标准就一条当流程里出现明确的循环、分支、记忆或人工干预需求时再上LangGraph不迟。2. 核心概念拆解状态、节点与边2.1 先理解状态Agent的数据中枢State是LangGraph里最基础也最容易误解的概念。它不是简单的dict而是一个有schema约束的对象。框架支持TypedDict、Pydantic或dataclass来定义建议用TypedDict最轻。一个常见的设计长这样from typing_extensions import TypedDict from langgraph.graph import StateGraph class AgentState(TypedDict): messages: list current_step: str retry_count: int final_answer: str关键在于每个节点函数都接收当前State作为参数返回一个partial state字典框架会把返回值合并更新到全局状态上。这个“返回部分更新”的设计很巧妙节点只需声明自己改了什么字段不必把整个状态原样传回去。需要特别注意列表字段的更新策略。默认情况下如果你在节点里返回{messages: [...]}框架是按覆盖处理不是追加。想实现追加得用Annotated配合operator.add做reducerfrom typing import Annotated from typing_extensions import TypedDict import operator class AgentState(TypedDict): messages: Annotated[list, operator.add]这里用operator.add作为reducer含义就是每次节点返回的messages会和原始列表执行加法合并。这个坑我一开始不知道结果节点里怎么处理消息都不对排查了好久才发现是Reducer机制的问题。2.2 节点函数的正确写法节点本质就是个普通函数但写法上有讲究。基础的签名是(state) - dict。def call_model(state: AgentState) - dict: response llm.invoke(state[messages]) return {messages: [response], current_step: model_done}如果要支持异步改成async def即可。还有一点很容易忽略如果你想在节点内部使用工具工具的结果要合并到状态里不然下一步模型拿不到工具输出。框架还支持带config参数的节点函数用于读取运行时配置比如thread_id。签名长这样def node_with_config(state: AgentState, config: dict) - dict: thread_id config[configurable][thread_id] ...2.3 边的类型与选择时机边有两种普通边和条件边。普通边就是无脑顺序执行——A执行完永远走B。条件边则是一个函数接收state返回下一步要去的节点名字符串或节点名列表。def should_continue(state: AgentState): if state[retry_count] 3: return end return call_model这里有个设计建议条件判断函数尽量简洁不要在里面做大逻辑处理它只负责“看状态、选路”真正的处理逻辑放在节点里。这样图的结构一目了然调试的时候只需关心每个路口怎么分叉。2.4 入口与终点的设定StateGraph创建后需要用set_entry_point指定入口节点然后通过add_edge或add_conditional_edges连接节点。图编译时需要明确哪些节点是终结点框架允许一个“end”的特殊节点名实际编译后也可以不用专门定义。注意一点入口节点可以只有一个但可以存在多个终结点。比如“正常返回”和“达到重试上限强制结束”都算终点这是合理设计。没有终点的图编译不过去这是个明显的报错点。3. 从零搭建你的第一个LangGraph应用3.1 环境准备与最小可运行图安装很简单pip install langgraph langchain-openai记一下当前文档多数示例基于LangGraph 0.2版本API和早期0.1相比有变化比如StateGraph的导入路径、编译方式都有调整。我在末尾的常见问题区专门列一个版本坑。然后直接写最小图from typing_extensions import TypedDict from langgraph.graph import StateGraph class MyState(TypedDict): value: int def double(state: MyState) - dict: return {value: state[value] * 2} def add_one(state: MyState) - dict: return {value: state[value] 1} builder StateGraph(MyState) builder.set_entry_point(double) builder.add_node(double, double) builder.add_node(add_one, add_one) builder.add_edge(double, add_one) builder.add_edge(add_one, end) graph builder.compile() result graph.invoke({value: 1}) print(result) # {value: 3}这套流程跑通之后LangGraph的地基就算打好了。注意add_edge(add_one, end)里的end是框架内建的终结点标记不需要为它定义节点。3.2 带条件分支的Agent构建实例下面进入正题搭一个带搜索工具、支持重试循环的Agent。先定义工具from langchain_core.tools import tool from langchain_openai import ChatOpenAI tool def search_web(query: str) - str: 搜索公开信息 return f这是{query}的模拟搜索结果。 llm ChatOpenAI(modelgpt-4o-mini, temperature0) tools [search_web] llm_with_tools llm.bind_tools(tools)然后构建状态机class AgentState(TypedDict): messages: Annotated[list, operator.add] retry_count: int def call_model(state: AgentState) - dict: response llm_with_tools.invoke(state[messages]) return {messages: [response]} def execute_tool(state: AgentState) - dict: last_message state[messages][-1] tool_outputs [] for tool_call in last_message.tool_calls: output {role: tool, content: f工具返回{tool_call[name]}} tool_outputs.append(output) return {messages: tool_outputs} def should_continue(state: AgentState) - str: last_message state[messages][-1] if not last_message.tool_calls: return end return exec_tool把图串起来builder StateGraph(AgentState) builder.add_node(model, call_model) builder.add_node(exec_tool, execute_tool) builder.set_entry_point(model) builder.add_edge(model, exec_tool, conditionshould_continue)等一下这段有问题——我没有把retry_count的更新逻辑放进去先保留这个思路后面补上。还有一处要注意如果模型发出的tool_calls有多个execute_tool里需要循环处理每个调用逐个解析name、args并做异常保护否则一个工具报错会拖垮整个Agent。我在代码里写了模拟返回实际项目里你会在这里调用具体的搜索、计算或数据库工具。3.3 编译、调用与Streaming输出调用图使用graph.invoke或graph.stream。Streaming是生产环境最常用的模式它能把节点的执行过程实时推送出来体验接近流式大模型。实测发现graph.stream返回的chunk结构很有规律——每个事件流是一个dict键为node_name或__interrupt__等值里含messages列表需要自己写一个解析函数把消息抽出来。调用的完整示例for event in graph.stream({messages: [{role: user, content: 帮我查一下今天的新闻}]}, config{configurable: {thread_id: thread-1}}): for key, value in event.items(): print(f节点 {key} 输出: {value})加上config里的thread_id这行代码是LangGraph做记忆管理的关键入口后面细说。4. 状态记忆、持久化与人工介入4.1 Checkpoint与Thread设计原理LangGraph的持久化机制核心是Checkpoint检查点。每当一个节点执行完成框架会把当前状态写进checkpoint存到指定的存储器里。这样即使程序崩溃或者会话暂停随时可以根据checkpoint恢复状态。而thread_id就是会话的唯一标识。同一个thread_id下的多次invoke共享状态不同thread_id之间互相隔离。这个设计对应到业务场景就是一个用户开一个会话thread_id存数据库Agent随时可以续聊、回溯。想要启用持久化只要给checkpoint传一个BaseSaver实例。官方内置了几种saver最简单不过MemorySaver内存存储重启全丢生产环境通常用langgraph-checkpoint-postgres或langgraph-checkpoint-sqlite。from langgraph.checkpoint.memory import MemorySaver memory MemorySaver() graph builder.compile(checkpointermemory)再次invoke时只要传入相同的thread_id状态就会自动续接graph.invoke({messages: [question]}, config{configurable: {thread_id: user_123}})4.2 记忆只靠消息追加是远远不够的很多人以为把消息列表攒起来就是记忆了实践下来发现两层问题。第一层大模型的上下文窗口有限。对话轮数一多消息列表超长要么截断要么摘要。我试过在节点里加一个判断当len(state[messages]) 20时调用一个总结模型把早期消息压成摘要再用摘要替换旧消息。这个方案需要你自己维护消息列表的“摘要替换”逻辑LangGraph本身不替你管。第二层除了对话内容跨会话的状态也可能需要记忆。比如用户偏好、关键业务参数这些不适合混在消息列表里。正确做法是额外定义状态字段比如user_profile在节点里读写它。这属于Agent记忆设计的高级话题但理解thread和checkpoint之后实现起来并不复杂。4.3 interrupt等人工审批的关键机制LangGraph有个很有价值的原语叫interrupt()。在节点执行到某一步时调用它图会暂停执行把控制权还给外部程序。之后你可以通过graph.invoke传入新的数据比如人工审批结果图会从暂停的地方继续跑。我做一个实操场景Agent生成订单申请后需要人工确认确认通过才发通知。节点里写from langgraph.types import interrupt def review_node(state: AgentState) - dict: decision interrupt({order: state[order_info]}) return {approved: decision}外部流程程序先invoke一次拿到interrupt返回的内容展示给用户用户点了确认后再invoke一次传入审批结果。这里要理解“同一个thread_id下第二次invoke不是在重跑而是在续跑”和普通调用完全不同。5. 常见的坑与排查记录5.1 State更新不生效八成是Reducer没写对我遇到的最多问题就是这个。在节点里返回{messages: new_msg}结果状态里还是原来的列表。原因就是Reducer默认是覆盖不是追加。解决办法给字段加Annotated[list, operator.add]。但要注意operator.add对空列表以外的合并行为是把新列表直接接上去。如果你的业务需要去重、排序这类定制合并逻辑得自己写函数做Reducer。5.2 条件边只匹配上了默认分支add_conditional_edges的第三个参数是条件函数它返回的字符串必须和已添加的节点名完全匹配。我踩过一次坑是返回值多打了一个空格或者大小写不齐框架也不报错就直接走了默认分支数据就乱了。调试方法很简单在条件函数里加print输出返回值。5.3 版本升级导致API不兼容LangGraph迭代非常快很多教程里的写法在0.2版本会报错。几个典型变化StateGraph导入路径从langgraph.graph到langgraph.graph没有大改但内部默认checkpointer从MemorySaver的用法有变化。add_conditional_edges的签名发生了调整早期版本传的是dict映射现在简化成直接返回目标节点名。用create_react_agent快速构建时要注意它返回的是一个预编译好的图不一定支持你自定义State。我的建议是装库时统一锁定版本把langgraph.__version__记录到项目的requirements里别追新追到跑不起来。5.4 生产环境用内存Checkpointer数据全丢开发测试用MemorySaver没问题一旦部署到生产服务进程重启意味着所有对话状态归零。生产环境的正确姿势是用Postgres Saver或者Redis Saver把检查点数据持久化到外部存储。刚开始我图省事UAT环境上线第二天服务重启用户就发现所有会话断了复盘才发现是内存型checkpointer的锅。6. 什么时候选LangGraph以及后续方向给一个实用的选型建议表基于我自己的实践场景推荐方案单轮问答、简单工具调用普通Chain或AgentExecutor多步固定流程含人工确认LangGraph图结构让节点一目了然需要循环反思、自动修正LangGraph配合条件边和计数器需要暂停恢复、断点续跑LangGraph配合Postgres Saver超大规模、跨语言服务可以考虑自研状态机或评估是否用LangGraph Server用LangGraph写Agent的最终收益不是因为它“时髦”而是它把Agent从黑盒变回了可控系统。每次状态变更都留下记录每个分支都看得见每步执行都能追踪——这对复杂度上来的项目是真正的救命稻草。我在实际项目中有一个体会不要一上来就试图把所有逻辑都塞进一张图。可以先把主流程画出来跑通后再把重试、反思、人类确认这些“异常分支”逐渐加进去。一次图变更就测一次这比脑内推演可靠得多。再分享一个实用小技巧调试条件边的时候在条件函数入口直接加一行print(state)你会发现自己对状态变化的感知比看任何文档都来得快。Graph的可视化能力够用了但真正定位问题靠的还是“看状态怎么流”。LangGraph的后续玩法还有很多比如多Agent协作、结合向量库做长期记忆。不过这些都是后话了——先把状态、节点、边这套基本功练扎实你会发现Agent工程的复杂度突然变得可控起来。