如果你最近在搭Agent,应该已经刷到过不少次“LangGraph”这个名字。我第一次看到它时没太当回事,心想这不就是LangChain又套了个壳吗?直到我自己的Agent项目开始失控——任务链一长,代码就往“状态字典 + while True + if 分支”里滑,每加一个判断条件函数就胖一圈,中途出错还得从头再来——我才认认真真把它读完。读完之后最大的感受是:它改的不是API,而是你组织Agent流程的思维方式。
这篇是LangGraph系列的第一篇,我把它的定位、五个核心概念、真正该用的场景,以及一个不用调用大模型也能跑通的Demo骨架完整讲一遍。适合已经写过一点LangChain、但还没摸清Graph思路的开发者;如果你纯新手,前面的基础概念部分也能帮你补上。读完你至少能回答两个问题:这东西到底解决什么问题?我手头的项目值不值得迁移过去?
1. 为什么Agent项目越写越痛苦:LangGraph要解决的到底是什么
1.1 从链式到循环:失控感从哪里来
用LangChain写过一点业务的人,应该都对LCEL很熟悉:prompt | model | parser,像流水线一样从上往下串。这种写法在固定流程下非常舒服,可一旦你开始做真正的Agent,痛苦就来了。Agent不是一条直线,它是“干一步、看结果、再决定下一步”的循环,比如先规划、再执行、检查结果不对就重试、重试还不行就换工具,最后才输出。
LCEL想表达这种循环,只能硬拗。常见做法是写一个超级大的函数,里面塞满中间变量,再套一个while循环,循环里用if判断是否要退出。刚开始还行,两个工具三个条件的时候代码还能看;等到工具一多、分支一多,状态就分不清了:这个变量是哪个环节写的?这一步的结果在哪个字段里?程序跑到一半崩了,前面花了多少钱调用的token全白费。
我印象最深的一次,是给一个内部工具加“人工确认后再执行”的需求。原来的实现是:Agent先生成计划,然后调用一个工具直接执行。加确认就意味着流程要在半路上停下来等人点按钮,完事后还得从停住的位置接着跑。用LCEL链加回调硬写,代码绕到我自己都看不下去。LangGraph真正让我觉得“对了”的,就是它把这种半路停、接着跑、跑错了回溯,直接做成了框架的第一等公民。
1.2 “有状态、有循环、可中断”意味着什么
LangGraph的定位是一套低层编排引擎,专门用来构建有状态、多参与者、图结构的Agent应用。这话有点官方,拆开看其实就是三个词。
有状态:整个Agent运行过程中,所有节点共享同一个State对象。它有点像一块贴在墙上的大白板,每个环节都能往上写东西,也都能看到别人写过什么,谁也不会把数据揣在自己兜里。
有循环:节点不是只能按顺序走一遍,条件边可以让你从后面的节点绕回前面的节点。这意味着“检查不通过就重写”“工具调用失败就换一个”这类逻辑,不再需要你手写循环,而是直接在图上画一条回边。
可中断:执行到某个节点之前可以停下来,把当前所有状态存好,等你处理完人工审批再继续;也可以从历史某个节点重新跑。这对长流程和需要审计的场景价值极大。
这三个能力,恰好就是LCEL和手写while循环最别扭的地方。所以LangGraph不是比LangChain“更高级”,而是补上了LangChain不擅长的那块拼图。
1.3 谁最应该现在动手学
我不是劝所有人都立刻迁移。但如果你的项目出现下面这些信号,LangGraph大概率能帮你省事:需要多个工具协作、同一个任务要反复迭代多轮、流程中有需要人类确认的环节、任务跑一次时间很长且希望断点续跑、或者你希望把Agent每一步的路径都记录下来方便排查。反过来,如果你只是做一个单轮的检索问答,输入问题返回答案,那真的不需要上Graph,普通函数就够了。
系列后面几篇我会往深了拆,但这一篇的核心目标,是先帮你把图化思维立起来。因为后面所有进阶功能——多Agent协作、人机交互、流式渲染,全都是建立在这套核心概念之上的。
2. LangGraph的五个核心概念,以及它们为什么这样设计
2.1 State:节点之间传递的不只是字典
State在LangGraph里通常就是一个TypedDict,用来描述整个流程共享的数据结构。我一开始以为它只是个普通的全局字典,用多了才发现它的设计是“全量共享 + 增量更新”。每个节点读的时候拿到的是当前完整的State,写的时候只需要返回自己修改的那几个字段,LangGraph负责把这些返回值合并回全局State。
举个例子:
from typing import TypedDict class WritingState(TypedDict): topic: str title: str draft: str review_result: str revision_count: int默认情况下,节点返回的字段会直接覆盖State里同名字段。如果你希望某个字段是“累加”而不是“覆盖”,比如要把多轮对话消息不断追加到一个列表里,可以用Annotated加一个reducer:
from typing import TypedDict, Annotated import operator class ChatState(TypedDict): messages: Annotated[list, operator.add] current_plan: str这个operator.add就是告诉LangGraph:每次节点返回messages字段,不要替换,而是和旧列表拼接起来。这个机制一度被我忽略,直到做多轮Agent才发现它是控制状态更新方式的钥匙。
把State类比成贴便签的共享白板最贴切:每个人都可以看全貌,但只往上面贴自己负责的那张便签,而不是把整块白板擦掉重写。
2.2 Node:每个节点都是普通函数
Node是图上一个个处理步骤,在代码里就是一个普通Python函数:接收当前State,返回需要更新的字段。不需要继承什么基类,不需要实现什么接口,这让我在迁移老代码时非常轻松——把原来大函数里的一步拆出来,包成一个函数就行。
def write_node(state: WritingState): draft = f"标题:{state['title']}\n正文:围绕{state['topic']}展开的初稿……" return {"draft": draft}为什么设计成普通函数而不是类?我自己的理解是,普通函数最好测试、最好组合,而且能和LangChain的模型、工具、解析器无缝混用。你在节点里调model.invoke()也好、调外部API也好、跑一段本地计算也好,它不关心,只要最终返回一个字典就行。
有一个经验值得分享:节点里的逻辑越薄,图的价值越大。一开始我习惯把“生成方案 + 执行 + 校验”全写进一个节点,后来发现这样没法单独回放、也没法精确插入人工审批。把每一步拆成独立节点后,整个流程就像拆开的积木,哪里出问题一眼就能看到。
2.3 Edge、条件边与图的组合方式
节点是积木,Edge就是积木之间的连接线。普通边表示确定性的流转,比如“写完初稿之后必须评审”。LangGraph里用add_edge声明,恒定的执行顺序一目了然。真正的精髓是条件边add_conditional_edges,它允许你写一个路由函数,根据当前State的内容决定下一步去哪个节点。
def route_after_review(state: WritingState) -> str: if state["review_result"] == "pass": return "publish" return "write_node"这个路由函数就是Agent“自主决策”的落点。它本身不调用模型,只是读取State、返回一个节点名字符串,然后图框架负责把执行流引导到对应节点或终止点END。条件边的存在,让图不再只是一条流水线,而是一张真正可以拐弯、可以绕回头的路线图。
把普通边想象成地铁的固定线路,条件边则是换乘闸机:你刷票的时候系统会根据你的目的地自动引导你走哪条线。Agent里每个关键判断点都应该是一个“换乘闸机”。
2.4 Checkpointer:让执行可中断、可回放
如果说State、Node、Edge是LangGraph的地基,那Checkpointer就是把它从玩具变成生产工具的关键组件。它负责把每一步执行完之后的完整State保存下来,用thread_id区分不同的会话。同一个会话里,你可以在中断后继续执行,也可以回溯到历史节点重新跑。
from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() app = graph_builder.compile(checkpointer=checkpointer) config = {"configurable": {"thread_id": "demo-session-1"}} result = app.invoke({"topic": "LangGraph入门", "revision_count": 0}, config)“时间旅行”是这个词最直观的体现:只要保存了checkpoint,你就能看到这个会话每一步的State快照,甚至可以回到某一步之前换条路再走一次。做调试、做审计、做人工干预,都依赖这个能力。后面第6章我会专门讲生产环境该选哪种checkpointer,这里先记住概念。
下面这张表可以帮你快速回顾五个概念:
| 概念 | 作用 | 生活类比 |
|---|---|---|
| State | 所有节点共享的数据结构 | 共享白板 |
| Node | 一个处理步骤,普通函数 | 工位上的操作员 |
| Edge | 确定性的流转关系 | 流水线传送带 |
| Conditional Edge | 根据状态决定下一步 | 智能换乘闸机 |
| Checkpointer | 持久化状态,支持中断与回放 | 行车记录仪 |
3. 先别急着上手:什么场景才真正需要LangGraph
3.1 这些场景用了会真香
我在自己的项目里验证过几个典型场景,体验确实比手写循环好太多。
第一种是“规划-执行-检查”循环。Agent先根据用户目标生成一份计划,然后逐步执行,每执行一步检查结果是否符合预期,不符合就回退到执行节点重新来。这种流程在LCEL里要写递归或者复杂回调,在LangGraph里就是一条从“检查节点”指回“执行节点”的条件边。
第二种是需要人工审批的流程。比如Agent生成一段代码、一份合同,不能直接执行,要先停在一个interrupt_before节点前等人类确认。LangGraph原生支持这种暂停,而且暂停期间State完整保存在checkpoint里,恢复执行时不会丢上下文。
第三种是长流程任务的断点续跑和数据回溯。跑一个多阶段数据分析任务动辄十几分钟,中途挂了不用从头开始,直接定位到失败节点,修复后从该节点续跑;出了线上问题,还能翻历史State看看到底是哪一步决策导致的结果。
3.2 这些场景暂时用不上
说实话,有不少项目是不需要上LangGraph的。如果你的需求是:单轮问答、固定顺序的检索增强生成、或者简单的文本处理流水线,那LCEL甚至直接写Python函数都更轻量。强行上Graph,反而要为“图结构”付出额外的心智负担。
我还见过一种误区,觉得“用了LangGraph就能让Agent变强”。它不是模型能力的放大器,只是流程编排工具。模型该迷糊还是迷糊,该调用错工具还是调用错,Graph真正改善的是你对流程的控制力,不是单点智能。把它当银弹,预期一定会落空。
3.3 使用图的真实成本
作为一篇介绍文章,我觉得有必要把成本说透。LangGraph的学习曲线比LCEL陡,核心倒不是概念多,而是你要开始用图的方式思考问题——哪些是节点,哪些是边,节点的输入输出边界在哪,状态里该放什么不该放什么。这些设计判断需要一点时间适应。
调试成本也略高。图多了之后,一个节点的错误可能和上一个节点的输入耦合,排查时要结合checkpoint的回放功能。好在LangGraph提供了app.get_graph().draw_ascii()这类可视化方法,把整个图结构打印出来,调试的时候对照着看会直观很多。
我的建议是:不要为了用而用。当流程超过5个步骤、存在分支循环或需要人工介入时,图的价值会明显压过成本;如果只是两三个步骤的线性调用,忍忍就行。
4. 30分钟跑通一个带审阅循环的写作助手Demo
4.1 环境准备与最小依赖
这个Demo最大的好处是不用配大模型API也能跑,因为核心目的是让你理解图结构本身。等跑通了,再把“模拟生成”替换成真实的LLM调用就行。
先装包,Python 3.9以上环境。我这里用langgraph和langchain-core,如果你后面要接模型,再装对应的模型包,比如langchain-openai:
pip install langgraph langchain-core安装完成后,建议先跑一句import langgraph确认没问题。我遇到过装完版本冲突导致StateGraph导不出来的情况,通常把langchain-core升级到最新版就能解决。
4.2 状态、节点与条件边的完整实现
我们做一个最简单的“写作助手”:给定主题,先生成标题,再写初稿,然后评审,评审不通过就回到“写初稿”重来,通过就发布。整个流程天然包含一个循环,非常适合展示Graph的运作方式。
先定义State:
from typing import TypedDict from langgraph.graph import StateGraph, START, END class WritingState(TypedDict): topic: str title: str draft: str review_result: str revision_count: int然后定义四个节点函数。评审节点的逻辑我做了简化:前两轮都判定“需要修改”,第三轮才通过,这样你运行一次就能看到完整的两轮循环:
def brainstorm_node(state: WritingState): return {"title": f"《{state['topic']}完全指南》"} def write_node(state: WritingState): draft = f"标题:{state['title']}\n正文:围绕{state['topic']}展开的初稿内容。" return {"draft": draft} def review_node(state: WritingState): if state["revision_count"] >= 2: return {"review_result": "pass"} return {"review_result": "revise", "revision_count": state["revision_count"] + 1} def publish_node(state: WritingState): return {"final_text": state["draft"] + "\n——已发布。"}接下来是重头戏:把节点组装成图,并通过条件边把“评审”和“重写/发布”连接起来:
def route_after_review(state: WritingState) -> str: if state["review_result"] == "pass": return "publish" return "write_node" graph_builder = StateGraph(WritingState) graph_builder.add_node("brainstorm_node", brainstorm_node) graph_builder.add_node("write_node", write_node) graph_builder.add_node("review_node", review_node) graph_builder.add_node("publish_node", publish_node) graph_builder.add_edge(START, "brainstorm_node") graph_builder.add_edge("brainstorm_node", "write_node") graph_builder.add_edge("write_node", "review_node") graph_builder.add_conditional_edges( "review_node", route_after_review, {"publish": "publish_node", "write_node": "write_node"}, ) graph_builder.add_edge("publish_node", END) app = graph_builder.compile()注意add_conditional_edges的第三个参数是路由返回值到实际节点名的映射。如果你是直接把返回值写成节点名,可以省略映射,但显式写出来更清晰,也能防止路由返回值写错。
4.3 运行日志与输出解读
现在调用它。这里我建议用stream而不是invoke,因为stream能打印出每一步哪个节点执行了、返回了什么,非常适合第一次跑图时建立直观感受:
config = {"recursion_limit": 10} for event in app.stream({"topic": "LangGraph入门", "revision_count": 0}, config): print(event)你会看到类似这样的输出,每个元素代表一个节点执行后产生的状态更新:
{'brainstorm_node': {'title': '《LangGraph入门完全指南》'}} {'write_node': {'draft': '标题:《LangGraph入门完全指南》\n正文:围绕LangGraph入门展开的初稿内容。'}} {'review_node': {'review_result': 'revise', 'revision_count': 1}} {'write_node': {'draft': '...第二次修改...'}} {'review_node': {'review_result': 'revise', 'revision_count': 2}} {'write_node': {'draft': '...第三次修改...'}} {'review_node': {'review_result': 'pass'}} {'publish_node': {'final_text': '...——已发布。'}}看到这个序列,你就能理解“图上的循环”怎么走了:write_node和review_node之间那根条件边,让执行流自动绕了两圈。这比手写while循环清晰太多——哪个节点跑了多少次、每次返回了什么,全都有记录。
如果不用stream,直接invoke拿到的是最终完整状态,适合在生产代码里取最终结果:
result = app.invoke({"topic": "LangGraph入门", "revision_count": 0}) print(result["final_text"])4.4 把模拟生成替换成真实LLM调用
这个Demo里的文本生成都是写死的,替换成真实模型非常简单。以write_node为例,你只需要在里面调用一个模型客户端,把返回的content放进draft字段:
from langchain_openai import ChatOpenAI llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7) def write_node(state: WritingState): response = llm.invoke([ {"role": "user", "content": f"请围绕主题'{state['topic']}'写一篇短文,字数300字。"} ]) return {"draft": response.content}其他节点同理。brainstorm_node可以改成让模型拟标题,review_node可以改成让模型根据评分规则判断是否通过,并返回是否通过的结果。这样整个图就从一个模拟骨架变成了一个真实可用的Agent。
5. 条件边的进阶玩法:从分支到防死循环
5.1 路由函数的契约与映射
条件边的路由函数有一个硬性契约:接收完整的State,返回一个字符串,这个字符串要么直接是节点名,要么能被映射字典转换成节点名。它不能返回None,不能返回不存在的节点。我第一次写的时候就在路由里漏了一个分支场景,结果运行时框架直接报节点找不到,排查了好半天。
为了避免这类问题,我有三个小习惯:一是把节点名定义成模块级常量,不写魔法字符串;二是路由函数里先处理兜底分支,让所有状态都有去处;三是写完后用draw_ascii()画一遍图,肉眼确认每条边正确:
print(app.get_graph().draw_ascii())输出是一张用ASCII字符画的图,虽然简陋,但哪两个节点相连、条件边通向哪里,一目了然。这个技巧在图上节点超过五六个之后会非常救命。
5.2 循环不是bug,是Agent的基本形态
很多从LCEL过来的人,第一次看到图上有回边会不习惯,觉得这不是成环了吗,会不会死循环。实际上,Agent的很多核心能力都依赖这种循环:检索不满意就换个关键词再检、代码运行出错就读取错误信息重新生成、生成结果不符合格式要求就再修一轮。循环是Agent的基本形态,不是例外。
LangGraph对循环没有任何偏见,它甚至会老老实实地记录每一轮的状态变化。你需要做的,是想清楚退出条件。退出条件放在哪里?就是条件边的路由函数里。回到我们的写作助手Demo,退出条件是review_result == "pass",而review_node负责更新这个字段,责任边界很清晰。
这种“循环控制逻辑由节点状态驱动”的写法,比在函数里写while True和break要规范得多。因为它把“本轮结果”和“退出判断”都变成了显式的State字段,可以打印、可以存储、可以在线排查。
5.3 分支多的项目怎么组织路由逻辑
真实项目里,条件边不会只有两三个分支。一个复杂Agent可能有“继续执行”“请求人工介入”“换工具重试”“直接结束”等多个出口。组织不好就会变成一坨if-else。
我目前的实践是:单独维护一张路由表,把条件和目标节点集中定义。比如用一个类或一组常量名词管理所有节点名,路由函数只负责提炼“当前意图”,映射字典负责把意图翻译成节点。这样增删一个分支,只需要改映射,不需要翻整个图的代码。路由函数保持纯净也很重要——它里面不做任何副作用操作,只读State返回字符串,这样测试时可以单独喂各种State去验证它会不会返回预期节点。
6. 从Demo到生产:持久化、流式与人工介入的实操经验
6.1 Checkpointer怎么选
Demo阶段用MemorySaver没问题,进程还在,状态就在;但进程一重启,所有会话历史全部消失。生产环境必须持久化。LangGraph自带好几种checkpointer,我整理了一张表供你参考:
| Checkpointer | 存储后端 | 适用场景 |
|---|---|---|
| MemorySaver | 内存 | 单进程Demo、测试 |
| SqliteSaver | SQLite文件 | 单机部署、轻量持久化 |
| PostgresSaver | PostgreSQL | 多实例生产、需要共享状态 |
多实例部署时一定要选PostgresSaver这类共享存储,否则两个实例各存各的,同一会话的状态对不上。这一条我是吃了亏才记住的:之前用MemorySaver跑多进程服务,前端拿不到跨进程的会话状态,排查半天才发现是checkpoint没有共享。
6.2 invoke与stream的使用区别
很多新手分不清invoke和stream。简单说,invoke是等图全部跑完,一次性返回最终State;stream是边跑边返回,每次节点执行完就吐出一个事件。交互式前端、逐字展示Agent思考过程,用stream;后台批处理任务,用invoke就够了。
还有一个容易忽略的点:stream默认返回的是每个节点的状态更新,而不是全量State。如果你想拿到完整状态快照,可以指定stream_mode="values"。我自己做调试日志时经常用这个模式,能清楚看到State每一步变成了什么样。
for chunk in app.stream(input_data, config, stream_mode="values"): print(chunk)6.3 人工介入:让机器停在需要审批的地方
人工介入是Agent落地时常被忽视、但极为重要的能力。LangGraph的处理方式是在编译时指定interrupt_before,让图执行到某个节点前自动暂停,并等待继续执行的指令:
graph_builder = StateGraph(WritingState) # ... 添加节点和边 ... app = graph_builder.compile( checkpointer=checkpointer, interrupt_before=["review_node"], )这样每次运行到review_node之前,执行流都会停住,State会被完整保存。人类可以在外部审核当前State,点通过之后再用同一个thread_id调用一次,图会从暂停的位置继续跑。这种“机器跑一半、人来拍板”的模式,特别适合内容审核、代码执行确认、大额操作授权等场景。
6.4 我从真实项目中踩过的几个坑
有些坑是文档里不大会写的,遇到了很容易卡人。我把自己踩过的问题列在这里,希望能帮你省点时间。
第一个坑是条件边返回了未注册的节点名。路由函数里如果漏掉一个分支,运行时会直接抛错,而且错误信息不太直观。解决方式就是前面说的:节点名用常量、映射显式写全、兜底分支一定要有。
第二个坑是TypedDict字段缺失导致的KeyError。节点直接访问state["xxx"],而该字段在一次调用中还没被任何节点写入过,就会崩。别赌执行顺序一定先写后读,用state.get("xxx", 默认值)更稳,或者在一开始的输入里把全字段初始化好。
第三个坑是状态更新方式不符合预期。默认是覆盖,想累加字段必须配Annotated[list, operator.add]。我见过有人在State里定义列表字段,节点返回时直接把整张历史表“覆盖”掉了,排查半天才发现是没配reducer。
第四个坑是异步环境。LangGraph有配套的异步接口,比如ainvoke、astream。在异步代码里忘记加await,表现出来就是程序卡住或者返回协程对象,一查一个准。
第五个坑也是最容易忽视的:MemorySaver在进程重启后状态消失。如果你只是本地调通,没问题;一旦要部署成长期服务,开机第一件事就是把checkpointer换成数据库版本。
这篇先到这里。整体梳理下来你会发现,LangGraph真正改变的是你组织Agent的方式:从“一堆循环和临时变量”变成“一张可以命名、可以保存、可以回放的图”。我自己的体会是,它不是让Agent变聪明,而是让Agent变可控,这个可控性在真实业务里比模型本身的聪明程度更稀缺。下一篇我会专门拆一个真实的多工具调度Agent,把工具注册、条件边组合、断点续跑串起来讲,如果你已经把这里的最小Demo跑通了,那篇你会跟得很轻松。