★ FEATURED ARTICLE 如果你最近在折腾智能体开发大概率绕不开LangGraph这个名字。它1.0正式版出来之后把早期版本里那些东拼西凑的API和示例收拢了一遍整体思路清爽了不少。所谓速通指南其实就是把一条路从起点完整走到终点状态、节点、边这三个词你只要真正吃透了后面再写多复杂的编排心里都有底。这篇是系列第一讲我们先不碰那些高层封装只用最核心的StateGraph搭一个能跑起来的最小工作流同时把1.0里最容易踩的几个坑提前指出来。对于刚接触智能体编排的开发者这一篇看完可以直接上手复现如果你已经在用旧版LangChain那套链式写法也能借此把心法切换过来。1. 为什么是状态、节点、边先想一个很现实的问题普通调用大模型一次问答就结束了。可智能体不一样它要连续干活可能要调用工具、要自我修正、要多次追问甚至要根据中间结果动态决定下一步干什么。这种流程如果用普通函数一条条写下来跑几次没问题改起来就头疼了每个分支都靠if-else硬编码中间数据靠全局变量传根本没法维护也没法观察每一步发生了什么。LangGraph 的核心想法是把整个智能体工作流当成一张图来组织。图由三类东西组成状态State、节点Node、边Edge。打个比方把整个流程想象成一条流水线状态是流水线上流动的工件托盘里面装着半成品和加工参数节点是流水线旁的加工工位每个工位只负责一道工序边是工位之间的传送带决定加工完A之后下一个去哪个工位。这个比喻不算新颖但它确实能说清楚为什么这个抽象好用。有人可能会问线性列表难道不够吗还真不够。真实的智能体流程不是一根直线而是带分支、带回环的。比如一个“访谈-反思-再追问”的工作流它要不停地循环“提问-回答-判断是否继续”直到满足条件才进入总结阶段。这种结构用线性列表没法表达而图天然支持你可以把“回答”节点的出口通过一条条件边重新连回“提问”节点形成一个循环。这就是1.0里最值得掌握的一个能力不是只能做有向无环图而是可以跑真正的循环逻辑。LangGraph 1.0把这三者做得很规矩状态有明确的Schema决定了哪些字段允许存在、允许被谁改节点只是普通函数接收当前状态、返回增量更新不需要关心完整流程边用来声明静态关系和动态路由。这三个概念拆开看都不难但合在一起就足以描述绝大多数智能体的控制流。后面几节我逐个拆开讲并配合完整代码争取让你一次看明白。2. 核心概念逐个拆解2.1 状态工作流的“共享内存”状态是整张图中所有节点共享的数据容器。它不是一个黑盒而是你在代码里显式定义出来的一个Schema一般用TypedDict或 Pydantic 模型来声明。这个Schema决定了哪些key存在、每个key是什么类型。之所以要这么严谨是因为图里的节点不直接互相传参它们只和状态打交道每个节点读当前状态然后返回一个“更新建议”LangGraph负责把这些增量合并回全局状态。这里有个很容易忽略的点节点之间并不像普通函数那样node_a(result_of_b)传参而是通过状态间接沟通。好处是节点之间彻底解耦A节点根本不需要知道B节点做了什么只要读自己关心的字段就行代价是所有人必须遵守同一个状态Schema约定。你如果在一个节点里随手返回了一个Schema里没声明过的key运行时会直接报错这种报错刚开始会有点烦但习惯之后反而觉得安全——它能及时拦下拼写错误和数据流混乱。来看一个具体定义from typing import TypedDict, Annotated import operator class InterviewState(TypedDict): topic: str # 入参不会被修改 rounds: Annotated[int, operator.add] # 累计轮次 transcript: Annotated[list[str], operator.add] # 访谈记录 final_summary: str # 最终产出这才是LangGraph 1.0里面比较关键的一处设计reducer机制。默认情况下如果多个节点都想修改同一个key后一个节点的返回值会把前一个覆盖掉。这在很多场景下不对——比如两个节点同时往transcript里追加记录我明明想“拼接”结果被“覆盖”了。解决办法就是在定义状态时给这个key加上Annotated[类型, reducer函数]。上面代码里transcript加了operator.add那么LangGraph在合并这个字段时会把新返回的list和旧list拼起来而不是直接替换。rounds同理多个节点每次返回1会累加而不是被重置成1。这个机制理解到位之后很多诡异的bug都能想明白。比如你可能遇到过这种情况一个节点里返回return {count: state[count] 1}同时又给count声明了operator.add结果数据翻倍增长。因为节点函数内部算出了新值reducer又把你这个新值和旧值相加等于每走一轮都加了两次。所以在决定加不加reducer之前先想清楚“这个字段是只有一个节点写还是多个节点共同写”。一个节点写就默认覆盖别画蛇添足多个节点追加才需要operator.add或自定义合并函数。2.2 节点功能的封装单元节点是执行具体任务的单位。在LangGraph 1.0里节点本质上就是普通Python函数或可调用对象输入是当前状态对象输出是一个字典代表对状态的增量更新。你可以让它调用大模型、执行工具、读数据库、做字符串拼接或者只是做一个很简单的if判断。它不要求返回完整状态只要返回自己负责修改的那几个字段就行剩下的由运行时帮你补全。要特别强调一个容易误解的地方节点不等于大模型调用。很多新手会把“写智能体”等同于“在每个节点里塞一个LLM调用”其实完全没必要。节点可以是任何逻辑一个分类规则、一个超时判断、一个数据清洗函数都可以做成节点。我见过不少高效的工作流真正调用LLM的节点只有两三个其余全是普通Python逻辑用来路由、校验、缓存。把没必要花钱的步骤交给LLM既慢又贵这属于设计层面的误区。节点的命名也很重要。在add_node(interviewer, ask_question)里第一个参数“interviewer”是这个节点的全局唯一ID后面连边全靠这个名字。命名建议用清晰的动作名词比如“classifier”“summarizer”“tool_executor”别用node1、node2这种图一旦变大连你自己都记不住哪个是哪个。还有一个细节节点函数的返回值理论上可以是None。如果你想让某个节点只读状态、不做任何更新可以返回NoneLangGraph会认为是“无更新”。但这很少用一般节点总得干点事。比如一个“检查是否结束”的节点它读状态里的轮次和阈值然后直接返回None或返回路由值靠的是条件边做后续判断而不是靠更新状态。2.3 边控制流的组织者边负责把节点串起来有三种常见的连接方式。第一种是无条件边builder.add_edge(a, b)意思就是A跑完必跑B。第二种是条件边add_conditional_edges它不直接把两个节点绑死而是通过一个路由函数根据当前状态动态决定下一步去哪个节点。第三种是接特殊节点START是整个图的入口END是出口所有工作流至少要有一条从某个节点到END的路径否则图跑完没有终点运行时可能会一直在等。从START出发是最容易忽略的一步。很多人定义好节点和边之后直接compile()然后跑起来报错一看是压根没有“入口边”。正确的写法是builder.add_edge(START, interviewer)意思是图一启动第一个要执行的节点是interviewer。同样的一个图可以有多条进入END的边不同条件分支最后汇合到END是常见布局。条件边的写法在1.0里长这样builder.add_conditional_edges( expert, route_after_expert, { continue: interviewer, summarize: summarizer, }, )这里的route_after_expert是路由函数接收状态返回一个字符串键名字典{...}把返回的键名映射到目标节点名。如果返回值在字典里找不到运行时会抛异常这个我后面在第5节详细讲。正是这条条件边让“专家回答完之后判断是否需要继续追问”变成了现实需要就回到interviewer完成就切到summarizer。循环并不神秘就是一条回边加一个终止条件而已。3. 从零搭建第一个LangGraph工作流3.1 安装与版本确认前面讲了一堆概念现在落到实际操作。第一件事是安装。建议直接装1.0以上版本别再回头研究0.x的旧API了pip install langgraph1.0装完之后最好确认一下实际版本python -c import langgraph; print(langgraph.__version__)为什么要先确认版本因为这个框架版本迭代快网上搜到的很多示例还是0.6甚至更早的写法API差异很大。如果你照着旧代码写一边是langgraph.graph.MessageGraph一边是新的StateGraph很容易陷入“完全看不懂哪里报错”的困境。我自己的习惯是每次新建项目都先固定版本号比如langgraph1.0.0至少保证一周以内的示例不会错位。另外建议把 Python 升级到3.10以上原因主要是类型语法更舒服TypedDict、Annotated这些都用得顺手。太老的Python版本在类型标注上会有一堆兼容问题没必要自找麻烦。3.2 定义状态Schema我直接用一个“模拟专家访谈”的小例子把整张图串起来。这个场景很典型先提出话题请专家回答轮流几次之后生成总结。它的流程天然包含循环和汇聚非常适合练手。先定义状态Schemafrom typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, START, END class InterviewState(TypedDict): topic: str rounds: Annotated[int, operator.add] transcript: Annotated[list[str], operator.add] final_summary: str关于rounds和transcript上面的reducer说明一下rounds字段在“提问”“回答”两个节点里都会返回1我希望它累加成总轮次所以声明Annotated[int, operator.add]transcript也是同理每次提问和回答都追加一条内容最后我想要一个长长的完整记录而不是被最后一条覆盖。如果只是单一节点写这些字段我会顺手把Annotated去掉保持Schema简单。这里其实有一个比较隐蔽的坑如果rounds声明了相加红ucer节点函数内部再写state[rounds] 1那每一步会把旧值累加两次。所以写节点函数时要搞清楚“我返回的是增量”还是“我返回的是全量”。在声明了reducer的字段上你返回的是增量。3.3 编写节点函数接下来写三个节点函数。def ask_question(state: InterviewState): return {rounds: 1, transcript: [f提问能否谈谈{state[topic]}]} def get_answer(state: InterviewState): return { rounds: 1, transcript: [f回答关于{state[topic]}先说结论再说理由。], } def summarize(state: InterviewState): last_lines state[transcript][-4:] return {final_summary: 摘要 .join(last_lines)}ask_question和get_answer都只负责追加内容返回增量。summarize读取完整访谈记录提取最近几条拼成摘要。重点观察一下这三个函数都没有互相调用它们只读写状态。这就是节点的解耦性——每个函数都很短短到不用看其他节点的代码就能猜出它在干什么。真实项目里哪个节点逻辑复杂了你就知道该拆了。把节点注册进图builder StateGraph(InterviewState) builder.add_node(interviewer, ask_question) builder.add_node(expert, get_answer) builder.add_node(summarizer, summarize)注册顺序不影响执行。真正决定执行顺序的是边。3.4 连接节点与条件边现在把头和尾接上builder.add_edge(START, interviewer) builder.add_edge(interviewer, expert) def route_after_expert(state: InterviewState): if state[rounds] 3: return summarize return continue builder.add_conditional_edges( expert, route_after_expert, {continue: interviewer, summarize: summarizer}, ) builder.add_edge(summarizer, END)这个流程是这样的入口进interviewer提问然后无条件走到expert回答回答完之后route_after_expert看当前总轮次是否已经达到3次没到就回到interviewer再来一轮到了就去summarizer生成总结最后到END。这就是一个完整的多轮循环工作流。注意这里route_after_expert读的state[rounds]是一个已经被合并好的当前状态值。因为ask_question和get_answer都会返回rounds: 1所以每经历一次“提问-回答”循环rounds会累加1。当它累加到3时条件路由才会把它切走。这个计数逻辑要想明白不然很容易写出永远不结束的循环。我的建议是循环越复杂越好在路由函数里同时加一个绝对上限比如if state[rounds] 3 or state[rounds] 5:这种兜底逻辑防止某个节点更新出错时整个图跑死。3.5 编译与调用图组装完毕编译之后就可以用了graph builder.compile() result graph.invoke({ topic: LangGraph 1.0 快速上手, rounds: 0, transcript: [], final_summary: , }) print(result[rounds]) # 3 for line in result[transcript]: print(line) print(result[final_summary])这里有一个输入细节invoke传入的是完整状态字典。我一般会把所有字段都显式初始化包括rounds、transcript这些积累性字段不给运行时留猜测空间。有的字段确实可以缺省但显式写出来读代码的人一看就知道每条数据的默认形态。如果工作流跑得比较慢或者你想观察每一步发生了什么建议直接用流式接口for step in graph.stream( {topic: LangGraph 1.0 快速上手, rounds: 0, transcript: [], final_summary: }, stream_modevalues, ): if step[transcript]: print(step[transcript][-1])stream_modevalues会在每一步执行完后输出整个状态对象尤其适合调试循环和条件分支。stream_modeupdates则只输出每个节点返回的增量更快但信息量少。初次调试工作流用values模式打日志比直接看最终结果直观得多。4. 进阶实操带条件分支的“路由主控器”智能体循环解决了我们再解决分支。分支和循环在LangGraph里其实是一回事都靠条件边实现。下面给一个“智能问答路由”的场景用户抛一个任意问题进来系统先判断类型然后分流到技术问答、生活问答、闲聊三个链路之一最后统一汇总返回。这种结构在真实产品里非常常见相当于一个主控路由。路由节点很简单def classify(state: InterviewState): text state[topic] if 代码 in text or 部署 in text: return tech if 菜谱 in text or 旅行 in text: return life return chat对应的条件边映射builder.add_conditional_edges( classifier, classify, { tech: tech_agent, life: life_agent, chat: chat_node, }, )这里有个新手几乎必踩的坑条件边第三个参数是路由返回值到节点名的映射字典不是单纯的节点列表。返回值tech是“路由键”字典里必须有这个键并且它的value是add_node时的节点名。如果路由函数返回了tech但字典里写的是{tech_agent: tech_agent}运行时照样报错。我吃过这个亏现在写条件边时会不自觉地先检查一遍返回值集合和字典keys是否完全一致。分支链路最后如果要汇聚到同一个汇总节点LangGraph处理并发天然有一套逻辑当多个上游节点都指向同一节点时这个下游节点会等待所有上游完成后再执行。比如tech_agent和life_agent都指向final_responder那么两个分支跑完后final_responder才会启动。这保证了分支汇聚的正确性也是图模型相比手写线程调度省心的地方。并行 fan-out 也值得一提。如果一个节点后面接了多条普通出边这些下游节点会并行执行。比如你要同时调两个搜索工具就可以从一个search_router节点分别连两条边到search_web和search_db二者并发跑最后再汇到一个merge_result。实际项目中这一步能把总耗时从两次串行调用变成一次并发窗口收益非常直观。使用并行前确保这两个节点之间没有数据依赖——它们只能读同一个上游状态不能互相等对方的输出。5. 常见问题与排查技巧实录5.1 节点返回了状态Schema之外的key这是出现率最高的运行时报错之一大意是节点尝试修改一个不在状态Schema里的字段。常见原因包括字段名拼写不一致、大小写对不上、或者某个节点确实需要新加字段但忘了更新Schema定义。排查方法很简单打开报错信息里提到的节点把它返回的字典keys和Schema里的keys逐一对齐。我习惯把Schema定义放在图代码的最上方一旦节点增多这行定义就是所有人的契约。改节点前先改Schema改完Schema再检查所有旧节点是否还能兼容这个顺序能避免大半报错。5.2 条件边返回值没有命中path map报错信息通常会直接告诉你某个路由返回值在映射字典里不存在。这种问题大多是路由函数和映射字典各写各的没有对齐。我的做法是在路由函数里加一行临时日志def classify(state): # ... print( route:, route) return route跑一遍看输出再对照字典里的keys基本一眼就能找到差异。养成这个习惯之后你连调试面板都不太需要。5.3 reducer语义被误用导致数据翻倍或丢失前面反复提到过Annotated和reducer它作用于“多个节点返回该字段时的合并方式”不影响节点内部读状态。这个混淆是很多数据翻倍的来源。如果rounds声明了operator.add节点里再写state[rounds] 1并返回那自动叠加会把逻辑搞乱。建议是声明了累加reducer的字段节点返回增量声明了覆盖模式的字段节点才返回全量。如果你想要非常规的合并逻辑也可以自己写reducer函数def merge_list(left, right): return left right然后Annotated[list, merge_list]。这没什么神秘就是一个普通函数左侧是当前状态里的旧值右侧是节点返回的新值返回结果会写回状态。5.4 图陷入无限循环条件边回环本身不报错但如果终止条件没写对图就会一直跑下去。调试这种问题流式输出是你最好的朋友for step in graph.stream(input, stream_modevalues): print({k: step[k] for k in (rounds, transcript) if k in step})看到每一步的状态变化就能找到是哪条路让条件边反复走回同一节点。我处理这种问题的标准操作是把路由函数的判断条件打印出来对照状态值变化确认计数逻辑确实在递增。另一个更保险的做法是给编译后的对象传recursion_limitgraph builder.compile() try: graph.invoke(input, config{recursion_limit: 20}) except Exception as e: print(图可能进入循环或超过深度:, e)这相当于一个熔断器至少不会让进程一直卡死。5.5 老版本示例混用网上大量LangGraph教程还是旧写法比如StateGraph里直接传状态字典、老式的add_conditional_edges参数顺序、旧版 checkpointer 用法。这些混到1.0代码里轻则警告重则直接TypeError。我踩过几次之后养成了一个原则以官方最新文档的示例为基准其他博客只作为思路参考不直接复制。装完包之后可以先跑一遍仓库自带的示例确认当前版本API的手感这比看一百篇过时文章都管用。5.6 记忆与持久化checkpointer初见LangGraph 1.0里跨对话记忆靠的是检查点机制。把检查点挂到图上from langgraph.checkpoint.memory import InMemorySaver checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer)之后每次调用带上thread_id同一个线程的多次调用会自动共享状态历史config {configurable: {thread_id: session-123}} graph.invoke(new_input, configconfig)这套机制是后面做多轮真实对话的基础。第一讲先做到“知道有这回事”后续系列再展开讲存储选型和生产细节。6. 后续扩展思路聊到这里LangGraph的三大核心概念已经全部过了一遍你也拥有一个能跑循环、能走分支、能看流式输出的最小工作流。下一步自然是把它往真实项目上靠把示例里的“专家回答”替换成真实LLM调用把“总结节点”替换成结构化输出解析再加入工具调用节点、多智能体协作、持久化存储这些能力。我自己刚开始切换到这个框架时其实就是从普通链式调用改写过来的第一张图跑了整整一下午都在报错最后定位到的问题极其简单——状态Schema少了一个字段。这类问题排查多了你慢慢会对状态、节点、边形成肌肉记忆再复杂的图也不慌。如果你把这一讲的最小示例亲手跑通、改过一版自己的路由逻辑那这个系列接下来讲工具节点和多Agent编排时你会轻松很多。
阅读完成 · 觉得有帮助？
