1. 为什么我最终选择了LangGraph 1.0来搭智能体工作流先说说我自己的经历。早先做智能体应用我试过多种方案要么是纯靠代码手写状态机把每一步转移逻辑硬编码在if/else里要么用一些轻量级编排框架但一旦流程复杂状态维护就成了一场灾难。直到LangGraph升级到1.0我把一个多轮对话、工具调用、人工审核混合的流程迁移过来之后才真正感受到用图来编排智能体这件事的价值。LangGraph 1.0核心就三样东西状态State、节点Node、边Edge。思路并不复杂把一次智能体任务拆成几步每一步是一个节点节点之间通过边连接而整个运行过程中的数据存在一个全局状态里节点读写这个状态边决定下一步走向哪个节点。你可以把它理解为一张流程图但这是一张活的流程图——状态里的数据会驱动流程的走向节点可以在运行过程中动态改变执行逻辑。这篇文章适合谁看如果你已经写过一些基于大模型的应用比如调过API做过简单的工具调用但对如何搭建多个节点协作、带条件分支、可人工介入的完整工作流还比较模糊那这篇内容正好帮你把骨架搭起来。我会从代码层面讲清楚状态怎么定义、节点怎么编写、边怎么连接最后落一个能跑的Demo。2. 状态、节点、边LangGraph工作流的三个基本组件2.1 状态是智能体的便签本在LangGraph里状态不是某个节点的私有变量而是整个图上所有节点共享的数据容器。每个节点执行完都可以往状态里写入内容其他节点也能读到。这个设计让节点之间的数据交换变得很自然——你不需要显式地把A节点的输出传给B节点只需要各自读写同一个状态。我习惯把状态理解成一张便签本每个节点干活之前看一眼便签本干完活再往上添几笔。LangGraph会对这张便签本做版本化管理每一步的中间结果都可以被追踪和回放这给后面调试带来了很大的便利。定义一个状态很简单在1.0版本里通常用TypedDict配合注解from typing import TypedDict, Annotated from langgraph.graph import StateGraph class AgentState(TypedDict): user_input: str intent: str # 节点写入的中间结果 tool_name: str tool_result: str final_answer: str需要注意一个细节LangGraph对状态字段有一套默认的更新规则——后写的覆盖先写的。但在某些场景下你希望字段是追加而非覆盖比如对话历史就需要定义一个reducer去指定合并逻辑。这个后面单独展开讲。2.2 节点是一个个干活的小工节点就是执行单位。你可以把节点理解为带输入输出函数的函数。在LangGraph 1.0里一个节点接收当前状态作为输入返回一个字典作为对状态的更新。一个典型的节点长这样from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini) def parse_intent_node(state: AgentState) - dict: 把用户输入翻译成明确的意图 prompt f请判断以下用户请求的意图只返回一个词查询、计算、还是闲聊。\n输入{state[user_input]} resp llm.invoke(prompt) intent resp.content.strip() return {intent: intent}这里有一个关键点节点返回的字典里只会更新你写的那几个字段其他字段保持不变。这个设计让每个节点都尽量小、只干一件事——你不需要考虑会不会动到别人的数据只需声明自己负责的字段即可。2.3 边描述了节点之间的转移条件边定义了节点执行的先后顺序。最简单的是普通边normal edge表示执行完A下一步走B。还有一种条件边conditional edge它像一个路口的分岔标记——根据当前状态里的某个字段决定下一步到底走哪个节点。在LangGraph 1.0里边不仅规定了执行的顺序还给整张图引入了结构化分支的能力。很多智能体应用之所以需要在图里跑而不是纯写代码就是因为代价分支逻辑往往会随着业务规则变化而变化散落在代码里的if/else很难维护而图上几条边就能把规则清晰勾勒出来。2.4 图的运行时从哪里开始到哪里结束一张图至少要有一个入口节点和一个结束点。入口节点就是整个工作流开始时第一个执行的节点结束时让某个节点返回一个特殊标记或连接到一个结束边。在LangGraph 1.0中比较推荐的做法是为工作流定义合理的入口状态和结束条件。如果你不确定某个流程链条该怎么拆一个简单的方法是把用户的每一次触发→处理→响应拆成一个节点再把那些需要决策的分岔点做成条件边。3. 从0到1实现第一个LangGraph工作流3.1 安装和最小运行环境我建议直接用虚拟环境安装避免污染全局环境python -m venv .venv source .venv/bin/activate pip install langgraph langchain-openai现在LangGraph 1.0已经发布依赖包相对精简。注意langgraph和langchain的版本要匹配避免遇到API签名不一致的问题。我这里测试时用的版本组合是langgraph1.0、langchain-openai0.2跑起来没有冲突。3.2 一张最小可执行的图我们来做一个场景用户提了一个问题系统先判断是否和天气查询相关是的话走天气查询节点否则走普通问答节点。from typing import TypedDict, Literal from langgraph.graph import StateGraph class WeatherState(TypedDict): user_input: str need_weather: bool response: str def classify_node(state: WeatherState) - dict: # 实践中这里会调LLM这里用简单关键词示意 text state[user_input] need_weather 天气 in text or 下雨 in text or 温度 in text return {need_weather: need_weather} def weather_node(state: WeatherState) - dict: return {response: f抱歉我还无法实时获取天气。你问的是{state[user_input]}} def chat_node(state: WeatherState) - dict: return {response: f这是一个普通问答节点你输入的是{state[user_input]}} # 条件路由函数 def route_after_classify(state: WeatherState) - Literal[weather_node, chat_node]: if state[need_weather]: return weather_node return chat_node builder StateGraph(WeatherState) builder.add_node(classify_node, classify_node) builder.add_node(weather_node, weather_node) builder.add_node(chat_node, chat_node) builder.set_entry_point(classify_node) builder.add_conditional_edges(classify_node, route_after_classify) builder.add_edge(weather_node, __end__) builder.add_edge(chat_node, __end__) graph builder.compile()运行起来非常直接result graph.invoke({user_input: 明天会下雨吗, need_weather: False, response: }) print(result)你在结果里能看到整个状态的最终快照。这就是最简单的LangGraph工作流一个判断节点两条条件分支加上两个终态节点。3.3 在LangGraph 1.0里如何跑图在1.0版本里invoke是同步执行ainvoke是异步执行。如果你的节点里有异步IO比如调用大模型API的异步版本就用ainvoke。另一个差异是流式输出。对大模型应用来说用户往往希望字一个个蹦出来而不是等很久一次性返回。LangGraph 1.0提供了stream方法可以逐节点流式输出状态更新也可以配合大模型的stream模式输出token级的结果。我实际项目中的经验优先用astream配合stream_modemessages这样实时性最好。普通非流式场景下直接用invoke即可。4. 理解状态属性与Reducer机制从覆盖到合并4.1 为什么需要Reducer默认情况下一个节点返回的字段值会覆盖状态里的旧值。这在很多场景下没问题但有些字段天然是累积型的。比如多轮对话的消息列表每轮结束都要往同一个字段追加新消息而不是把旧的冲掉。LangGraph 1.0提供了一个Annotated类型机制配合reducer函数来定义字段的合并方式。下面的代码展示如何把messages字段设为追加模式from typing import Annotated, TypedDict from langgraph.graph import MessagesState from langchain_core.messages import AnyMessage def add_messages(left, right): # 实践中直接复用LangChain的实现这里示意 return left right class MultiTurnState(TypedDict): messages: Annotated[list[AnyMessage], add_messages]你可能会问为什么不直接在节点里自己写state[messages] state[messages] [new_msg]因为Reducer的价值在于多个节点并行或循环执行时仍能合并你不需要在每个节点里关心合并细节声明一次就够了。4.2 状态字段的默认覆盖与显式追加来看一个实际对比场景默认行为使用Reducer后的行为user_input字段节点返回新值则覆盖一般不需要合并messages历史记录只保留最后一次返回的列表追加新消息tool_result中间结果覆盖按需保留或拼接写状态时有一条朴素的原则作为流程控制依据的字段用覆盖即可作为上下文积累的字段用Reducer追加。搞混了会出现典型的历史消息丢失问题——排查起来费时费力。4.3 状态初始化和跨图传递每次调用invoke时传入的字典就作为初始状态。如果图比较大建议在入口处预留一个默认值工厂函数把缺省字段补全而不是在每个节点里去if xxx not in state这样判断。def default_state(user_input: str) - dict: return { user_input: user_input, intent: , tool_name: , tool_result: , final_answer: , }这样所有节点拿到的状态结构永远完整节点内就不需要写多余的防御性判断了。5. 边与路由控制智能体的决策核心5.1 普通边 vs 条件边怎么选普通边适合顺序执行条件边适合分岔路由。判断标准很朴素执行完这个节点下一步是不是看情况而定如果是就走条件边。我在实际项目里把条件边归类成三类内容路由比如根据LLM输出的意图标签分发到对应处理节点。路由到工具当用户请求需要调用外部API时先走工具选择节点再由条件边决定调用哪个工具节点。循环与退回比如生成的答案质量不过关条件边可以把流程拉回重新生成节点或者执行完工具后回到主对话节点让LLM基于工具结果继续推理。5.2 条件边函数的定义和注册方式看一个更完整的条件边写法def router(state: AgentState) - str: intent state[intent] if intent 查询: return query_node elif intent 计算: return calculator_node return fallback_node builder.add_conditional_edges(parse_intent_node, router, { query_node: query_node, calculator_node: calculator_node, fallback_node: fallback_node, })在LangGraph 1.0里add_conditional_edges接收三个参数源节点、路由函数、路由映射字典。路由函数的返回值必须是映射字典里的key之一这样才能保证图的结构是确定的。需要特别提醒一个坑路由函数里尽量不要做大模型调用。如果你让路由逻辑也依赖大模型输出整体延迟会叠加而且不容易调试。我通常把大模型调用放在数据解析节点里让路由只读解析好的字段这样所有分支判断都变成了纯规则稳定且快。5.3 循环边与递归限制LangGraph天然支持循环这是它比很多编排框架强的地方。比如一个反思再回答的流程生成回答 → 检查质量 → 不合格则回到生成节点 → 重新生成。builder.add_node(generate_node, generate_answer) builder.add_node(check_node, check_quality) builder.add_node(regenerate_node, regenerate_answer) builder.add_edge(generate_node, check_node) builder.add_conditional_edges(check_node, route_after_check) builder.add_edge(regenerate_node, generate_node)但注意无限循环会卡死整个工作流所以LangGraph在编译时可以设置递归上限graph builder.compile(recursion_limit10)一旦执行步数超过限制就会抛出异常。我在项目里会让check_node在第三或第四次循环时强制放行保证用户体验。6. 实战搭建一个多步骤智能体工作流6.1 场景设定与流程拆解下面我们做一个更接近真实需求的例子一个商品推荐助手。用户输入需求系统先判断是否需要查商品库如果需要先调用检索工具拿到商品列表再做分析推荐如果不需要直接回答。整个流程拆成四个节点parse_intent_node解析用户需求判断意图和关键实体。search_products_node根据实体去模拟商品库检索。recommend_node基于检索结果生成推荐语。fallback_node处理非推荐类闲聊。6.2 关键代码实现from typing import TypedDict, Literal from langgraph.graph import StateGraph class RecState(TypedDict): user_input: str need_search: bool keyword: str product_list: list output: str def parse_intent_node(state: RecState) - dict: # 简单规则代替LLM text state[user_input] if (买 in text or 推荐 in text) and (手机 in text or 电脑 in text or 耳机 in text): return {need_search: True, keyword: text} return {need_search: False, keyword: } def search_products_node(state: RecState) - dict: keyword state[keyword] # 实际项目里这里会调用向量检索或数据库 mock_products [ {name: 智能降噪耳机, price: 599}, {name: 轻薄商务本, price: 5999}, {name: 旗舰手机, price: 3999}, ] if 耳机 in keyword: products [mock_products[0]] elif 电脑 in keyword: products [mock_products[1]] elif 手机 in keyword: products [mock_products[2]] else: products mock_products return {product_list: products} def recommend_node(state: RecState) - dict: items 、.join([p[name] for p in state[product_list]]) return {output: f根据你的需求为你推荐{items}。需要更详细信息可以继续追问。} def fallback_node(state: RecState) - dict: return {output: 我现在主要负责商品推荐你可以问我手机、电脑、耳机等产品。} def route_after_parse(state: RecState) - Literal[search_products_node, fallback_node]: return search_products_node if state[need_search] else fallback_node builder StateGraph(RecState) builder.add_node(parse_intent_node, parse_intent_node) builder.add_node(search_products_node, search_products_node) builder.add_node(recommend_node, recommend_node) builder.add_node(fallback_node, fallback_node) builder.set_entry_point(parse_intent_node) builder.add_conditional_edges(parse_intent_node, route_after_parse) builder.add_edge(search_products_node, recommend_node) builder.add_edge(recommend_node, __end__) builder.add_edge(fallback_node, __end__) rec_graph builder.compile()调用测试result rec_graph.invoke({ user_input: 想买一个降噪耳机有推荐吗, need_search: False, keyword: , product_list: [], output: , }) print(result[output])6.3 这个架构能怎么扩展这套骨架最大的好处是可以无痛加节点。比如后续想加比价功能只需要新增一个compare_price_node在search_products_node和recommend_node之间加一条条件边决定是否先比价再推荐因为节点之间完全通过状态解耦新增节点时不需要改动已存在的节点函数改图结构即可。这就是用图编排相比硬编码状态机最明显的高级感。7. 调试与可观测性LangGraph的隐藏优势7.1 使用内置绘制功能检查图结构LangGraph 1.0内置了图的可视化能力。编译后的图对象可以直接生成Mermaid或PNG这个功能在刚开始搭流程时特别好用——你不需要在脑内推演节点之间的连接直接看图画就一目了然。一个典型做法是# 生成并保存图结构 with open(graph.png, wb) as f: f.write(rec_graph.get_graph().draw_mermaid_png())当然具体API可能随版本略有变化但思路是一致的编译后的graph对象提供了查看和导出图结构的方法。我每次调新流程第一件事就是把图画出来看一眼常常能发现一些自以为连对了但实际连错的边。7.2 运行时状态追踪调用invoke时加一个config参数就能开启详细的调试信息from langchain_core.runnables import RunnableConfig config RunnableConfig(recursion_limit10, tags[demo]) result rec_graph.invoke(input_state, configconfig)结合LangSmith之类的追踪工具也可以用LangGraph自带的回调机制每一步节点的输入输出、耗时都能被记录下来。排查问题时我最常用的是回溯最近几步的状态快照——先看看某一步状态里的字段是不是和预期一致再判断是节点问题还是边路由问题。7.3 常见的失败模式与排查链路在实际使用中我遇到的失败大多集中在以下几类一是节点返回了状态中不存在的字段。LangGraph一般不会马上报错但这个字段会被忽略掉最终结果里就是你想要的字段没更新。排查方法是把每个节点的返回值打出来挨个字段对照。二是条件边路由函数返回了映射里没有的key。这个会直接抛KeyError报错信息里会把路由函数名和返回的key都显示出来定位很快。三是循环递归超限。这种问题发生在节点之间形成了死循环比如某个校验节点永远判定为不合格流程就会不断循环直到触发recursion_limit。解决思路是在校验节点里增加已重试次数字段达到一定次数后强制走出口。8. 从固定流程到自适应智能体下一步可以这样扩展8.1 让图自己决定下一个工具现在很多智能体应用不满足于固定的流程顺序而是希望大模型根据用户输入动态决定要调用哪些工具。LangGraph里可以实现一个工具选择节点让LLM先输出所需工具列表再通过条件边逐一执行。def tool_selector_node(state: AgentState) - dict: prompt 根据用户输入选择需要调用的工具可选search_web, calculate, get_time。只返回工具名多个工具用逗号分隔。 resp llm.invoke(prompt) tools [t.strip() for t in resp.content.split(,)] return {selected_tools: tools}然后后续的路由就读取selected_tools里的工具名逐一dispatch到对应的工具节点。这个模式在1.0里非常顺滑因为状态天然支持列表类型的字段工具执行结果也能累积在状态里。8.2 人工审批与暂停恢复很多业务流程需要人工参与比如AI先生成内容 → 人工审核 → 通过则发送不通过则退回重写。LangGraph 1.0支持中断和恢复机制。你可以把需要人工审核的节点设置为一个中断点图执行到这里会暂停等待外部传入新的状态后再继续。这样实际落地时运营人员可以介入每个关键环节而不是完全交给模型自动跑。8.3 多智能体协作的前景最后提一个方向多个LangGraph图之间可以通过作为节点的方式嵌套组合。也就是说一个小图可以作为一个大图里的节点存在从而实现多智能体分层协作。比如一个客服主管图里用户提问先分诊再分别调用售后子图和技术支持子图。我个人的感受是LangGraph 1.0最大的价值在于把流程和逻辑解耦了。以前写复杂的智能体最痛苦的是维护那一堆隐性的流转关系。现在图结构本身就承载了流程状态管理交给框架代码里剩下的全是业务逻辑。对一个开发者来说这种体验确实值得专门写一篇文章来分享。