不是标题党,也不打算让你看完就收藏吃灰。LangGraph这个框架,近两年在AI应用开发圈里热度一直很高,但很多人误以为它只是LangChain生态里的一个小工具,或者觉得“直接用LangChain的Agent不就够了,为什么要多学一个图框架”。直到你开始做真正复杂的多步骤任务、需要持久化记忆、或者要让AI在几个工具之间来回决策的时候,链式调用那套东西就会明显卡住。这篇把我自己从Chain迁移到Graph的完整思考、核心概念拆解、以及一个可直接改来用的实操示例全部摊开讲,适合正在做Agent开发、准备把AI能力落到真实业务里的程序员参考。
1. 为什么从Chain走到Graph:链式编排解决不了的三个场景
1.1 线性流程的天花板
很多人在LangChain里最早接触的是Chain,比如LLMChain、SequentialChain。它的思路很直观:把Prompt模板、模型调用、输出解析等组件串成一个流水线,上一个环节的输出就是下一个环节的输入。处理“给一段文本,先翻译再总结”这种固定顺序的任务时,链式模式完全够用。
但真实业务里没有那么多顺滑的直线。我举几个自己实际遇到过的场景:
第一,工具调用的循环。Agent需要判断用户问题是否需要调用查询工具,查完数据库后发现结果还不够,还需要再查一次,或者需要调用第二个工具。这种“查询—判断—再查询—再判断”的动作天然是一个环,而链式结构不支持回头路。用Chain硬写,只能靠一个Agent自己反复调用工具,流程不透明,中途想插入一个人工确认步骤更是无从下手。
第二,分支逻辑。用户输入的内容,先要判断是闲聊、查单、退换货还是投诉,不同意图走完全不同的处理流程。链式结构当然也能通过if-else在代码层面分叉,但分叉多了之后,业务逻辑散落在各个回调函数里,整条链路变得很难梳理,也很难可视化地排查“用户走到哪一步了”。
第三,状态持久化。聊天机器人需要记住对话历史,多轮任务中需要在多个节点之间共享中间结果。链式调用里每个环节都是独立的函数调用,状态要么塞在context里一层层传,要么自己维护外部存储,代码里到处是胶水逻辑。
这三个场景归纳起来就一句话:你需要的不再是一条线,而是一张可以决定走向的图。LangGraph的核心思路,就是让开发者用图的思维方式来编排AI工作流,节点表示动作,边表示流转关系,并且支持条件分支和循环。
1.2 图框架带来的思维转变
我最早看到LangGraph时,第一反应是“这不就是状态机吗”。想深一层之后发现,它确实是一个面向LLM应用的状态机,但比传统状态机多了一个关键能力:每个状态流转都可能触发一次模型调用,而模型的下一次输出又会决定往哪条边走。
这个思维转变很重要。以前写Agent,你的思维模式是“调用一个大模型,让它自己决定下一步”,LangGraph则是“把大模型的决策能力嵌入到流程图中,每一步做什么是开发者预先设计的,模型只在关键的判断节点上做选择”。前者是黑盒,后者是可观测、可控制、可断点续跑的白盒。
我自己切换到图模式之后,最大的感受是:调试从“猜Agent在想什么”变成了“看它走到了哪个节点,根据条件结果判断为什么走错了”。这种确定性,在开发阶段和线上运维阶段都是巨大的效率提升。
2. 核心概念拆解:看懂LangGraph的设计语言
2.1 State:整个图共享的状态对象
LangGraph里最核心的概念是State,它是一份贯穿整个图的数据结构,图里每个节点都能读取和修改它。你可以把State理解为一张在多个工人之间传递的“流转单”,每个工人在上面填自己的处理结果,下一个人继续在此基础上操作。
实际操作中,State通常用TypedDict类定义,比如:
from typing import TypedDict, Annotated, List class WorkflowState(TypedDict): question: str history: List[str] tool_results: List[str] intent: str final_answer: str我在项目里的经验是:State字段设计要克制,能用局部变量传的参数就不要放进State。因为所有写入State的数据最终都会被Checkpointer持久化保存,塞太多临时变量会让每次存储的数据量变大,恢复状态时也可能带上无用的脏数据。
2.2 Node:把动作封装成节点
Node是图里执行实际工作的单元,它本质上就是一个接收State、返回State更新的函数。下面是LangGraph最常见的节点定义方式:
from langgraph.graph import StateGraph def classify_node(state: WorkflowState) -> dict: # 调用模型判断意图 intent = call_llm_for_intent(state["question"]) # 只返回需要更新的字段 return {"intent": intent} def handle_order_node(state: WorkflowState) -> dict: # 处理订单查询工具调用 result = call_order_api(state["question"]) return {"tool_results": state["tool_results"] + [result]}需要特别注意的是:LangGraph节点返回的字典只会更新State中对应的字段。也就是上面classify_node只更新intent字段,其他字段原样保留。这个设计很实用,但也带来了一个常见坑——如果你把某个字段误写成覆盖,比如对历史列表用= new_list而不是+扩展,就会把之前节点写入的数据冲掉。
2.3 Edge:定义节点之间的流转关系
边用来定义节点之间的连接。LangGraph提供三种核心边类型:
- 普通边:固定从A到B。
- 条件边:根据一个路由函数的结果,决定从当前节点走向哪个后续节点。
- 入口边:定义整个图从哪里开始执行。
条件边是最常用的,因为它正好对应“让模型在关键岔路口做判断”的场景。一个典型的路由函数长这样:
from langgraph.graph import START, END def route_by_intent(state: WorkflowState) -> str: if state["intent"] == "order": return "handle_order" elif state["intent"] == "after_sale": return "handle_after_sale" else: return "small_talk" graph.add_conditional_edges("classify", route_by_intent, { "handle_order": "handle_order", "handle_after_sale": "handle_after_sale", "small_talk": "small_talk", })条件边让我可以在图里画出真实的业务判断逻辑,而不是把一堆判断塞进Agent的系统提示词里。AI负责理解用户意图,图的控制逻辑负责决策往哪走,责任边界清晰。
2.4 Checkpointer:给整个流程加上记忆
Checkpointer是LangGraph里我觉得价值被严重低估的一个特性。它的作用是把每一次节点执行后的State快照保存下来,保存位置可以选内存、文件系统或外部数据库。
有了Checkpointer,你至少获得三种能力:
- 断点续跑:流程执行到一半崩溃了,可以从最近一个保存点恢复,不用从头再来。
- 人工介入:在执行到某个敏感节点时暂停,等人审批确认后再继续。
- 多轮会话记忆:把同一用户的多次请求关联到同一个会话ID,每次新请求都从历史State恢复上下文。
实际使用中,不要每执行一步就持久化一次,那样开销太大。LangGraph允许你配置在执行到特定节点时save,我一般只在关键业务节点前设置持久化,把IO成本控制在合理范围内。
3. 从零搭建一个可复现的LangGraph Agent
3.1 定义一个够用但不过度设计的场景
我拿一个最常见的“智能客服流程”来做示例。场景要求是:收到用户提问后,先做意图识别;如果是查询类问题,调用一个模拟的查询工具;查询完成后再次判断是否还需要补充信息;整个流程支持多轮对话记忆。
这个场景覆盖了LangGraph最核心的四种能力:状态管理、节点封装、条件路由和记忆持久化。看懂这套结构,其他复杂Agent基本就是在这个骨架上加节点和边。
3.2 完整代码:一个带分支和记忆的查询Agent
import json from typing import TypedDict, Annotated, List from langgraph.graph import StateGraph, START, END from langgraph.checkpoint.memory import MemorySaver # 1. 定义全局状态 class AgentState(TypedDict): question: str history: Annotated[List[str], operator.add] # 使用add reducer避免覆盖 need_tool: bool tool_result: str answer: str # 2. 定义工具函数 def fake_order_query(order_id: str) -> str: # 模拟订单查询 return f"订单{order_id}状态为已发货,预计3天内送达" # 3. 定义各个节点 def intent_node(state: AgentState) -> dict: # 实际项目中这里会调用LLM判断意图 question = state["question"] need_tool = "订单" in question or "查" in question return {"need_tool": need_tool} def tool_node(state: AgentState) -> dict: # 从问题中简单提取一个订单号 order_id = "A12345" result = fake_order_query(order_id) return {"tool_result": result} def answer_node(state: AgentState) -> dict: if state.get("tool_result"): answer = f"查询结果:{state['tool_result']}" else: answer = "我现在可以帮你解答一般性问题,具体订单问题请提供订单号。" return {"answer": answer, "history": [f"Q: {state['question']}", f"A: {answer}"]} # 4. 条件路由函数 def should_use_tool(state: AgentState) -> str: return "tool" if state["need_tool"] else "answer" # 5. 构建图 graph = StateGraph(AgentState) graph.add_node("intent", intent_node) graph.add_node("tool", tool_node) graph.add_node("answer", answer_node) graph.add_edge(START, "intent") graph.add_conditional_edges( "intent", should_use_tool, {"tool": "tool", "answer": "answer"} ) graph.add_edge("tool", "answer") graph.add_edge("answer", END) # 6. 编译并启用内存检查点 checkpointer = MemorySaver() app = graph.compile(checkpointer=checkpointer) # 7. 执行 config = {"configurable": {"thread_id": "user-001"}} result = app.invoke( {"question": "帮我查一下订单A12345到哪里了", "history": []}, config=config ) print(result["answer"]) # 第二次问,带上thread_id后历史自动关联 result2 = app.invoke( {"question": "那这个订单什么时候发货的呢", "history": []}, config=config ) print(result2["answer"])3.3 这段代码里的几个设计要点
先说为什么用Annotated[List[str], operator.add]。这是LangGraph里一个容易忽略但很重要的机制。默认情况下,节点返回什么,State字段就会被覆盖为什么。如果history字段不允许追加,每次节点返回时都会用[new_item]替换掉之前的所有历史。加上operator.add这个reducer之后,LangGraph会自动把返回的列表元素追加到原有列表尾部,而不是替换整个列表。这就是状态更新策略的配置,理解了它就能避免很多“为什么运行几次之后历史只有最后一次数据”的诡异问题。
再说thread_id。它相当于对话会话的ID,同一个thread_id的多次invoke会被LangGraph认为是同一个会话。第一次调用时,State里的history是空的,执行完answer节点时history被写入了一条问答记录。第二次调用时,因为指定了相同的thread_id,LangGraph会先从checkpointer里恢复history字段,之后所有节点读到的state里就带上了上一轮的上下文。这个机制让我们不需要自己管理数据库里的对话记录,框架层就把活儿干了。
最后说条件路由的返回映射。should_use_tool返回的是逻辑分支名称"tool"或"answer",第三个参数里再定义分支名和实际节点的映射关系。这个映射是可选的,如果路由函数返回的名字和节点名一致,可以不写映射。
4. 工具选型:LangGraph、AutoGen、CrewAI到底怎么选
我经常被问到一个问题:现在做多Agent编排有好几个框架,LangGraph、AutoGen、CrewAI各有各的说法,到底学哪个。我的答案一向很简单:**如果你的应用核心是“可控的业务流程+LLM决策”,选LangGraph;如果你的应用核心是“多个Agent自由对话协作”,可以考虑AutoGen;如果你需要高度抽象的多Agent角色扮演,可以看看CrewAI。**但三个框架都学一遍没必要,真要上手干活,LangGraph带来的长期收益最大。
为什么会这么说?因为LangGraph本质上是图执行引擎,控制逻辑在开发者手里,模型的自由度被约束在节点内部。这种模式放到生产环境有巨大的优势:流程可观测、出错了可以看路径回放、中途可以暂停人工介入。这些都是企业级应用最关心的特性。
AutoGen的核心强调Agent之间的对话,让多个Agent互相发消息来完成任务。这种方式在做研究原型、探索式场景时很灵活,但对话过程天然不可控,Agent之间聊着聊着跑偏了是常有的事。你很难在AutoGen里精准地表达“必须先完成A步骤,然后才有条件地执行B步骤”。
CrewAI把Agent封装成带有角色和目标的“员工”,用团队协作的方式组织,概念很友好,上手也快。但封装的层次比较高,一旦需要细粒度控制某个流程节点,或者需要自定义状态持久化策略,你会发现自己被框架抽象限制住了,还得绕回底层去改。
我的建议是:学LangGraph,因为它是离“图的本质”最近的框架。图、节点、状态这些概念,本身就是软件工程的基础知识,学会了之后就算未来不用LangGraph,迁移到任何别的编排框架都很快。而其他框架的Agent抽象则不太通用,换个工具可能就要推翻重学。
5. 常见问题与踩坑实录
5.1 状态覆盖:Date 的隐性坑
我最早写LangGraph时,遇到过一类非常隐蔽的Bug:节点A返回了字段a和b,节点B返回了字段b和c,按理说最后State里应该同时有a、b、c三个字段。但实际运行时,a字段有时候存在有时候不存在。
排查后发现是节点返回的问题。LangGraph在节点执行结束后,会用节点返回的字典更新State,这个更新逻辑对每个字段单独判断。如果前一个节点返回了a和b,后一个节点只返回了b和c,那a是保留还是消失?答案是:**a还是会保留,因为更新的粒度是字段级,不是节点级。**但如果后一个节点返回的是{"a": None},那a字段就会被覆盖成None。
所以我的习惯是:节点返回值里只写本节点真正要改动的字段,其他字段一概不返回。宁可多写几个专门处理字段合并的小工具函数,也不要在节点里去手动操作整个State。
5.2 Checkpointer反序列化异常
使用持久化检查点时,不少人会遇到恢复State时类型不对的问题。比如之前State里存的是一个对象实例,恢复后变成了字典,或者字段缺失。这多半是因为State里存了非JSON序列化的数据,比如自定义类实例、函数对象、数据库连接。
经验法则:State里只放能序列化的数据。复杂对象放ID或引用,需要时在节点里重新加载。我见过有人把Pandas DataFrame直接放进State,看起来很方便,但恢复时会报错,最后只能改成存取文件路径。LangGraph的持久化设计偏向于让State保持轻量,你要顺着它的设计来,不要硬塞重对象。
5.3 并发更新同一个thread_id
LangGraph支持同一thread_id的多次invoke,但如果你在同一个thread_id上并发发起多个请求,可能会互相覆盖状态。比如用户双击提交按钮,两个请求同时从同一个State出发,后写入的那个会把先写入的覆盖掉。
解决方式有两个:一是业务层保证同一个会话串行处理,最简单的做法是加锁或队列;二是给每次请求用不同的thread_id,恢复时再用历史版本管理。实际项目中我用得最多的是第一种,因为用户操作天然是串行的,接口层做个防重就够了。
5.4 调试技巧:让图跑得明明白白
LangGraph提供了get_state和update_state这样的接口,可以在图执行过程中手动检查状态、甚至手动修改状态后再继续执行。我强烈建议在开发阶段把每一步的State都打印出来,路径调试比直接看最终输出直观得多:
# 模拟执行到中途,检查当前state current_state = app.get_state(config) print(current_state.values) # 手动修改state后继续 from langgraph.types import Command app.invoke(Command(resume={"need_tool": False}), config=config)这一段是调试利器,尤其是排查“为什么它走错分支”的时候,直接看关键字段在什么节点被改掉,比反反复复调Prompt高效得多。另外,LangGraph的draw_mermaid_png方法可以生成流程图,节点关系一眼就能看懂,强烈建议画出来贴在文档里,有人问流程时直接甩图。
6. 一些个人经验和最后一句话
如果让我给刚接触LangGraph的人一个上手路线,我会这么说:先别急着写Agent,去读一遍StateGraph的文档,然后用一个只有两个节点的图跑通invoke,再加入条件边,再加入checkpointer,最后再放工具调用。每一步都动手改代码、看状态变化,比看十篇教程都实在。
我自己在实际开发中的体会是:LangGraph的图模式不只是技术选择,更是一种思维方式。它逼着你在动手写代码之前,先把流程想清楚——哪些环节需要模型判断,哪些环节是固定逻辑,哪些环节允许人工介入。这个前置设计的过程,反而帮你节省了大量后期调试的时间。
最后分享一个实用小技巧:定义节点函数时,建议统一把函数第一行写成打印当前节点名和关键输入,这样不管最后逻辑多复杂,你都能通过日志快速定位问题。这个习惯我沿用到现在,每次排查线上问题时都受益。
如果你正在准备把LLM应用做成真正能被业务使用的系统,LangGraph值得花时间研究。它不是万能的,但至少把“可控”、“可恢复”、“可观测”这三个生产级需求,第一次这么清晰地统一到了一套API里。