之前在处理一个内部知识库问答项目时我最初用 LangChain 的Chain把流程串起来结果越写越痛苦分支判断要手写if-else步骤多了状态不好维护想实现“先检索再决定是否追问”这种逻辑时控制流几乎写成了意大利面。后来换成 LangGraph 重构整个流程瞬间清晰了很多。这也是我坚持推荐 LangGraph 给身边开发者的原因它把 Agent 和复杂任务流程真正当成“图”来建模而不是一条写死的链。这篇教程会从零开始结合可运行的代码示例把 LangGraph 的核心概念、条件路由、循环控制、子图和并行分支一次讲透。内容按“概念 → 环境 → 原理 → 实战 → 排错 → 最佳实践”的顺序展开适合刚接触 LangGraph 的初学者也适合用过 LangChain 但还没系统掌握 LangGraph 的开发者。文中代码都基于常见版本编写版本差异我会在对应位置指出。1. 背景与核心概念1.1 LangGraph 是什么LangGraph 是一个用于构建有状态、可编排的 AI 应用和 Agent 的框架。它由 LangChain 团队维护但和 LangChain 的定位不同LangChain 提供各种大模型、工具、记忆组件的封装而 LangGraph 更关注流程本身的编排与控制。如果用一句话概括LangGraph 让开发者以“图”的方式描述 AI 应用的执行流程。节点Node代表一个处理步骤边Edge代表步骤之间的流转关系。图可以包含分支、循环、并行和子图天然适合实现复杂的 Agent 行为。很多人第一次听到 LangGraph 会问LangChain 已经很流行了为什么还要一个 LangGraph实际上两者不是替代关系。LangGraph 内部可以调用 LangChain 的模型、工具和组件也可以完全脱离 LangChain 单独运行。它的核心价值在于控制流control flow而不是模型封装。1.2 LangGraph 解决什么问题传统 LangChain 的Chain是线性结构适合“先 A 再 B 再 C”的固定流程。但真实业务中AI 应用的流程常常是这样的用户输入问题后先判断是否需要查数据库。如果不需要直接让模型回答。如果需要先查询知识库再根据检索结果决定是直接回答还是补充提问。某些步骤可能要反复执行直到结果满足条件。这种场景用if-else也能写但代码会越来越难维护。状态变量散落各处退出条件不清晰调试困难更不用说实现并行分支、人工审批、循环修正这类高级能力。LangGraph 把这些控制流抽象成了图结构状态State全局共享所有节点都可以读写。节点Node一个普通函数或可调用对象。边Edge定义节点之间的流转关系包括普通边和条件边。条件边conditional_edge根据当前状态决定跳转到哪个节点。这种设计的直接收益是流程可视化很直观调试时可以定位到具体节点状态集中管理不容易出现隐式传参复杂控制流循环、并行、分支都有对应的实现方式。1.3 常见应用场景LangGraph 的应用场景很多以下几类在工程中最为常见Agent 工具调用模型决定调用哪些工具调用后根据结果决定是否继续调用或直接结束。知识库问答流程检索增强生成RAG流程中根据问题路由到不同检索器再对结果进行合并和评分。多步骤任务处理例如“先规划再执行再检查”的 Plan-and-Execute 模式。人工审核流程某些高风险操作需要等待人工确认LangGraph 的interrupt机制可以暂停图执行。数据清洗流水线按条件过滤、分支处理、并行计算后汇总结果。掌握 LangGraph本质上就掌握了一种描述复杂 AI 流程的通用语言。2. 环境准备与版本说明2.1 运行环境LangGraph 是 Python 库本文示例在以下环境演示操作系统Windows 10/11、macOS 或主流 Linux 发行版Python3.9 及以上LangGraph以 0.1.x 或 0.2.x 常见版本为例LangChain Core可选依赖用于使用 LangChain 的模型封装由于 LangGraph 更新较快不同小版本的 API 会有差异。建议以官方文档和你实际安装的版本身为准。安装时不要固定写死我下面给的版本号可以先安装最新版遇到 API 变化时查阅对应文档。2.2 安装依赖建议在虚拟环境中操作。使用venv创建隔离环境python -m venv langgraph-demo激活环境Windowslanggraph-demo\Scripts\activatemacOS / Linuxsource langgraph-demo/bin/activate然后安装 LangGraphpip install langgraph如果希望配合 LangChain 使用模型可以一并安装pip install langchain-core langchain-openai安装完成后验证版本python -c import langgraph; print(langgraph.__version__)如果提示找不到模块说明安装没有成功检查虚拟环境是否激活以及 pip 是否指向当前环境。2.3 示例项目结构本文实战部分会创建类似下面结构的项目langgraph-tutorial/ ├── main.py # 主流程构建并运行图 ├── agent.py # Agent 节点调用模型 ├── routes.py # 路由函数条件判断 ├── state.py # State 定义 └── subgraph.py # 子图示例先不用急着创建文件下面的内容会按顺序逐个实现。3. 核心原理拆解3.1 State图中的全局状态State 是 LangGraph 中最重要的概念之一。它本质上是一个数据结构在图的整个执行过程中传递。每个节点读取 State 中的部分数据处理后再把结果写回 State下一个节点就能看到更新。用 Python 的TypedDict定义 State 最简单from typing_extensions import TypedDict class MyState(TypedDict): query: str documents: list answer: strLangGraph 内部会对 State 的更新做合并。默认情况下如果节点返回的字段在 State 中已经存在新的值会覆盖旧的值如果字段是列表可以使用add_markdown这类 reducer 控制追加行为这个问题后面会专门讲。3.2 Node图中的处理单元Node 就是一个普通函数。函数接收 State 作为参数返回一个字典字典里的键值对会被更新到 State 中。def retrieve_node(state: MyState): query state[query] documents [文档1, 文档2] # 模拟检索结果 return {documents: documents}注意函数签名参数必须是state返回值必须是字典。这个约定很简单但也是初学者最容易忽略的。3.3 Edge确定流程方向的连接Edge 定义节点之间的连接。普通边表示无条件跳转A 节点执行完一定进入 B 节点。LangGraph 中内置了START和END两个特殊节点。START是图的入口END是图的终点END节点不执行任何逻辑只表示流程结束。构建图的基本流程是创建StateGraph。添加节点。添加边。编译图。执行图。下面是最小示例from langgraph.graph import StateGraph, START, END from typing_extensions import TypedDict class State(TypedDict): value: str def node_a(state: State): print(执行 node_a) return {value: A 的结果} def node_b(state: State): print(执行 node_b接收到 state[value]) return {} graph StateGraph(State) graph.add_node(a, node_a) graph.add_node(b, node_b) graph.add_edge(START, a) graph.add_edge(a, b) graph.add_edge(b, END) app graph.compile() result app.invoke({value: })这里add_edge(START, a)表示从入口进入a节点add_edge(a, b)表示执行完a后进入b最后add_edge(b, END)表示b执行完流程结束。invoke方法接收初始 State 字典返回最终 State。3.4 conditional_edge让流程自己“做选择”conditional_edge是 LangGraph 最核心的 API它让流程执行过程中可以根据当前 State 动态决定下一个节点。先看方法签名不同版本略有差异graph.add_conditional_edges( source_node, routing_function, path_map )source_node从哪个节点出发做判断。routing_function一个函数接收 State返回一个路由键字符串。path_map一个字典把路由键映射到目标节点名。一个经典用法是根据用户输入决定是否查数据库def route_after_query(state: State): if state.get(need_search, False): return database return direct_answer graph.add_conditional_edges( query_node, route_after_query, { database: search_node, direct_answer: answer_node, } )route_after_query返回的字符串就是字典的键LangGraph 根据这个键跳转到对应的节点。如果路由函数返回了 path_map 中不存在的键运行时会报错所以路由函数和 path_map 必须严格保持一致。3.5 如何在节点函数中修改 State 状态值这是 LangGraph 新手最常问的问题之一。答案并不复杂在节点函数中直接 return 一个字典字典中的键就是要修改的 State 字段。def update_answer_node(state: State): source state.get(source, 未知) return { answer: f根据 {source} 生成最终回答, status: completed }执行后State 中的answer和status字段会被覆盖更新。这里有一个核心机制需要理解LangGraph 的 State 不是直接在原对象上修改而是通过 reducer 合并返回值。默认情况下如果返回的字段已存在就会覆盖如果字段不存在就会新增。对于列表字段默认行为也是整体覆盖而不是追加。如果希望列表字段“追加”而不是“覆盖”需要用到Annotated和 reducerfrom typing import Annotated from typing_extensions import TypedDict def merge_list(left: list, right: list): 将新列表追加到旧列表后面 return left right class State(TypedDict): messages: Annotated[list, merge_list]这样每次节点返回{messages: [新消息]}时LangGraph 会调用merge_list把旧列表和新列表拼接起来而不是直接替换。这个点非常关键。很多人写多轮对话时发现消息总是被覆盖通常就是没有配置 reducer。3.6 循环LangGraph 中的循环是如何实现的有些任务需要反复执行某个节点直到满足退出条件。传统循环用while或for实现而 LangGraph 的循环是通过“条件边指回自身”来实现的。下面是一个典型的“重试直到成功”模型def process_node(state: State): # 模拟处理达到一定条件才成功 current_attempt state.get(attempt, 0) 1 success current_attempt 3 return { attempt: current_attempt, success: success, message: f第 {current_attempt} 次尝试 } def route_after_process(state: State): if state.get(success): return end return retry graph.add_node(process, process_node) graph.add_edge(START, process) graph.add_conditional_edges( process, route_after_process, { end: END, retry: process, # 关键返回自己 } )当route_after_process返回retry且 path_map 指向process时流程会再次执行process_node。这就实现了循环而且退出条件清晰可见。这里需要注意防止死循环。LangGraph 默认对递归深度有一定限制但在自己的代码里最好加上迭代次数上限def route_after_process(state: State): if state.get(success) or state.get(attempt, 0) 5: return end return retry3.7 子图流程嵌套与复用子图Subgraph就是把一个已经编译好的图作为另一个图的节点。这个机制非常适合流程复用。比如你已经实现了一个“检索子图”多个父图都可以直接引用它。from langgraph.graph import StateGraph # 先定义子图 def subgraph_node(state: State): return {value: state[value] -经过子图} sub_builder StateGraph(State) sub_builder.add_node(sub_node, subgraph_node) sub_builder.add_edge(START, sub_node) sub_builder.add_edge(sub_node, END) sub_graph sub_builder.compile() # 父图引用子图 def main_node(state: State): return {} parent_builder StateGraph(State) parent_builder.add_node(main, main_node) parent_builder.add_node(sub, sub_graph) # 子图作为节点 parent_builder.add_edge(START, main) parent_builder.add_edge(main, sub) parent_builder.add_edge(sub, END) app parent_builder.compile()使用子图时需要注意子图编译后的类型仍然是CompiledStateGraph可以像普通节点一样添加到父图中。子图的 State 类型不一定需要和父图完全一致但要保证相互传递的字段能够匹配。如果需要更复杂的映射可以注册为subgraph节点并传入input和output映射函数。3.8 并行分支同时执行多个节点LangGraph 真正强大的地方之一是并行分支。当某个节点执行完它可以同时进入多个后续节点最后再汇总。def branch_a(state: State): return {branch_result: [A]} def branch_b(state: State): return {branch_result: [B]} def aggregate_node(state: State): results state.get(branch_result, []) return {aggregated: f汇总 {results}} builder StateGraph(State) builder.add_node(a, branch_a) builder.add_node(b, branch_b) builder.add_node(aggregate, aggregate_node) builder.add_edge(START, a) builder.add_edge(START, b) builder.add_edge(a, aggregate) builder.add_edge(b, aggregate) app builder.compile()这里START同时连接到a和b两个节点。LangGraph 会并发执行两个节点默认使用线程池两个节点执行完后再进入aggregate节点。这种模式非常适用于“多个来源分别检索再合并结果”的场景。需要提醒的是并行节点如果同时修改同一个 State 字段可能会产生覆盖。设计 State 时要让不同分支写不同字段或者在汇总节点中统一处理。4. 完整实战案例搭建一个带分支和循环的知识问答 Agent下面我们完成一个更完整的实战案例。这个 Agent 的功能是接收用户问题。判断是否需要检索知识库。如果需要进入子图执行检索。检索结果不满足条件时重新尝试检索。生成最终回答。这个案例会综合用到状态修改、条件路由、循环和子图是前面概念的完整串联。4.1 创建项目结构我们先创建项目目录mkdir langgraph-tutorial cd langgraph-tutorial按照前面的设计创建state.py、routes.py、agent.py、subgraph.py和main.py五个文件。4.2 定义 State 数据结构文件路径state.pyfrom typing import Annotated from typing_extensions import TypedDict def merge_results(left: list, right: list) - list: 将新检索结果追加到已有结果列表的末尾 return left right class AgentState(TypedDict): # 用户原始问题 query: str # 是否需要检索 need_search: bool # 检索结果使用 reducer 实现追加而非覆盖 documents: Annotated[list, merge_results] # 当前尝试次数 attempt: int # 检索结果是否充分 is_sufficient: bool # 最终回答 answer: str在AgentState中documents使用了Annotated[list, merge_results]这样多个节点返回的documents会自动拼接不会互相覆盖。4.3 定义路由函数文件路径routes.pyfrom state import AgentState def should_route_to_search(state: AgentState) - str: 根据 need_search 字段决定进入检索还是直接回答 if state.get(need_search): return to_search return direct_answer def after_search_route(state: AgentState) - str: 检索后判断结果充分则回答不充分则重新检索或失败结束 if state.get(is_sufficient): return answer if state.get(attempt, 0) 3: return give_up return retry这里两个路由函数分别对应入口判断和检索后的循环判断。after_search_route同时承担了循环退出条件的职责当尝试次数达到上限时跳转到give_up节点结束流程避免死循环。4.4 定义子图检索子流程文件路径subgraph.pyfrom langgraph.graph import StateGraph, START, END from state import AgentState def retrieve_node(state: AgentState): 模拟从知识库检索文档 query state[query] attempt state.get(attempt, 0) 1 # 模拟检索前几次不充分第三次成功 if attempt 3: documents [f关于 {query} 的权威文档] is_sufficient True else: documents [f关于 {query} 的相关片段第 {attempt} 次] is_sufficient False print(f[检索] 第 {attempt} 次尝试query{query}) return { attempt: attempt, documents: documents, is_sufficient: is_sufficient, } def build_search_subgraph(): 构建并返回检索子图 sub_builder StateGraph(AgentState) sub_builder.add_node(retrieve, retrieve_node) sub_builder.add_edge(START, retrieve) sub_builder.add_edge(retrieve, END) return sub_builder.compile()子图内部只有一个retrieve节点。这里为了演示假设前两次检索结果不充分第三次才成功。真实项目中可以把retrieve_node换成向量检索、数据库查询等逻辑。4.5 定义主图节点文件路径agent.pyfrom langgraph.graph import StateGraph, START, END from routes import should_route_to_search, after_search_route from state import AgentState from subgraph import build_search_subgraph def query_node(state: AgentState): 接收用户问题并模拟判断是否需要检索 query state[query] # 这里用简单规则演示包含“知识库”就触发检索 need_search 知识库 in query print(f[入口] 接收问题{query}是否需要检索{need_search}) return { need_search: need_search, documents: [], attempt: 0, is_sufficient: False, } def direct_answer_node(state: AgentState): 不经过检索直接回答 return { answer: f无需检索直接回答{state[query]} } def aggregate_answer_node(state: AgentState): 汇总检索结果生成最终回答 documents state.get(documents, []) all_text \n.join(documents) return { answer: f基于以下资料回答{state[query]}\n参考资料\n{all_text} } def give_up_node(state: AgentState): 检索多次后仍不充分给出兜底回答 return { answer: 抱歉多次检索后仍未找到足够的信息请尝试简化问题。 } def build_main_graph(): 构建主图 builder StateGraph(AgentState) # 添加子图作为节点 search_subgraph build_search_subgraph() builder.add_node(query, query_node) builder.add_node(search_subgraph, search_subgraph) builder.add_node(direct_answer, direct_answer_node) builder.add_node(aggregate_answer, aggregate_answer_node) builder.add_node(give_up, give_up_node) # 入口边 builder.add_edge(START, query) # 条件路由是否检索 builder.add_conditional_edges( query, should_route_to_search, { to_search: search_subgraph, direct_answer: direct_answer, }, ) # 检索完成后的条件路由可能回答、重试、放弃 builder.add_conditional_edges( search_subgraph, after_search_route, { answer: aggregate_answer, retry: search_subgraph, give_up: give_up, }, ) builder.add_edge(direct_answer, END) builder.add_edge(aggregate_answer, END) builder.add_edge(give_up, END) return builder.compile()这段代码把前面讲的 API 全部串起来了。注意builder.add_node(search_subgraph, search_subgraph)把编译好的子图当作普通节点添加。这样主图的结构非常清晰入口query节点判断是否需要检索。需要检索时进入search_subgraph。检索后根据条件决定是汇总回答、重试检索还是放弃。4.6 编写入口运行文件文件路径main.pyfrom agent import build_main_graph def main(): app build_main_graph() # 第一次运行问题中包含“知识库”会触发检索 result1 app.invoke({query: 知识库中如何配置权限}) print(\n最终回答) print(result1[answer]) print( * 50) # 第二次运行问题不包含“知识库”直接回答 result2 app.invoke({query: 今天天气怎么样}) print(\n最终回答) print(result2[answer]) if __name__ __main__: main()4.7 运行与验证在项目目录下运行python main.py预期输出类似[入口] 接收问题知识库中如何配置权限是否需要检索True [检索] 第 1 次尝试query知识库中如何配置权限 [检索] 第 2 次尝试query知识库中如何配置权限 [检索] 第 3 次尝试query知识库中如何配置权限 最终回答 基于以下资料回答知识库中如何配置权限 参考资料 关于 知识库中如何配置权限 的相关片段第 1 次 关于 知识库中如何配置权限 的相关片段第 2 次 关于 知识库中如何配置权限 的权威文档 [入口] 接收问题今天天气怎么样是否需要检索False 最终回答 无需检索直接回答今天天气怎么样从输出可以看到第一次问题触发了检索子图并且因为前两次结果不充分循环重试了三次第三次进入aggregate_answer_node。第二次问题被路由到direct_answer_node直接结束。这验证了条件路由、循环和子图的组合工作正常。4.8 在节点函数中修改 State 状态的实际表现最后通过一个小实验感受 State 修改机制。在main.py中添加调试代码def inspect_state(state): print(\n当前 State 中 documents 数量:, len(state.get(documents, []))) print(当前 attempt:, state.get(attempt))在retrieve_node中调用它会看到每次检索后documents数量都在增加而attempt被覆盖为最新值。这正好体现了 reducer 的作用列表字段追加普通字段覆盖。5. 常见问题与排查思路5.1 常见报错现象与解决问题现象常见原因解决思路ValueError: Expected ... to be an instance of ...State 类型定义与节点返回值类型不匹配检查 TypedDict 字段类型确保返回值符合 State 定义路由函数返回了 path_map 不存在的键path_map 键与路由函数返回值不一致检查add_conditional_edges的第三个字典参数节点函数返回了 State 中未定义的字段类型检查或合并不符合预期在 TypedDict 中补充该字段或删除返回值列表字段每次都被覆盖没有为列表字段配置 reducer使用Annotated[list, merge_function]循环不退出或递归过深退出条件不满足或没有设置尝试次数上限在路由函数中增加最大迭代次数判断子图与父图字段不匹配子图 State 定义和父图不同统一 State 定义或使用子图输入输出映射5.2 排查清单如果你运行 LangGraph 程序报错可以按下面的顺序排查确认安装版本python -c import langgraph; print(langgraph.__version__)。确认 State 类型定义所有节点返回的键是否都能在 State 中找到对应字段。确认节点函数签名参数是否为state返回值是否为字典。确认路由函数返回值与path_map中的键完全一致。确认边连接所有节点是否都有进入和离开的边特别是END是否正确连接。确认编译正常调用compile()后检查是否有报错。对于并发分支确认不同分支是否写了同一个 State 字段。5.3 LangGraph 和 LangChain 的区别这个问题被问得最多。简单总结LangChain 是一套完整的开发框架包含模型封装、提示词模板、工具、记忆、向量存储等能力。它更偏向“组件库”。LangGraph 专注于流程编排核心是图和状态控制。它可以脱离 LangChain 使用也可以配合 LangChain 的组件使用。在 LangChain 时代业务流程是一条条 Chain 串起来在 LangGraph 时代业务流程是一张图支持分支和循环。实际项目中经常用 LangChain 负责模型和工具封装用 LangGraph 负责把这些步骤组织成可控流程。6. 最佳实践与工程建议6.1 State 设计建议State 是 LangGraph 应用的核心数据结构设计时要注意字段不要过多按业务边界拆分。列表字段尽量配置 reducer避免覆盖。内部状态和对外输出分开例如用internal_status和final_answer两类字段。不要把所有临时变量都放进 State可以用节点内的局部变量处理。6.2 路由函数命名与维护路由函数只负责路由判断不要在路由函数中做复杂计算。路由键使用有语义的字符串例如to_search、direct_answer不要用1、2。path_map和路由函数放在一起维护避免修改路由函数后忘记改映射。6.3 循环与递归控制所有循环必须有明确的退出条件。建议在 State 中维护attempt或类似字段设置最大次数。日志中记录每次循环的attempt值方便定位死循环。如果需要长时间运行的 Agent考虑使用invoke的超时参数或后台任务机制。6.4 子图拆分与复用当一个子流程超过 3 个节点时优先考虑拆成子图。子图应该只依赖 State 中定义良好的接口不要隐式依赖外部全局变量。子图之间尽量解耦便于测试和复用。6.5 安全与生产环境注意如果流程中涉及数据库、文件或外部服务操作必须明确权限边界。子图内部执行的检索、写入操作要遵循最小权限原则。如果对接真实大模型注意 API Key 的保管建议使用环境变量或配置中心注入。对用户输入不要直接拼接成 SQL、命令或提示词先做合法性和长度校验。生产环境建议开启日志追踪在关键节点记录 State 变化方便回溯。6.6 测试方案LangGraph 应用的核心可测试性在于节点函数是普通函数。可以为每个节点写单元测试def test_retrieve_node(): from subgraph import retrieve_node state {query: 测试, attempt: 0, documents: [], is_sufficient: False} result retrieve_node(state) assert isinstance(result, dict) assert documents in result再配合一个“最小状态”和“边界状态”的测试列表就能覆盖大多数逻辑。7. 总结与下一步学习路线本文从 LangGraph 的背景和设计理念出发介绍了 State、Node、Edge、conditional_edge 等核心概念并深入讲解了在节点函数中修改 State 状态值的关键机制。随后通过一个包含条件路由、循环检测、子图和并行分支的完整实战案例演示了如何从零构建一个知识问答 Agent。你现在应该对 LangGraph 有了一个系统性的认识它本质上是一套状态驱动的图执行引擎理解State的读写规则和conditional_edge的路由机制就等于掌握了这个框架的钥匙。如果要继续深入学习建议按以下顺序推进把本文的示例自己敲一遍动手修改路由条件和循环次数观察 State 的变化。尝试接入真实的大模型和工具调用构建一个能自己决定调用哪些工具的 Agent。学习 LangGraph 的持久化特性理解如何实现跨会话的长期状态。了解人工介入interrupt机制这对于生产级别的审核流程非常重要。最后再深入阅读 LangGraph 相关源码和官方文档关注版本更新带来的 API 变化。LangGraph 作为一个快速演进的框架不同版本的 API 细节可能会变但“图 状态 路由”的核心思想是稳定的。把这套思想内化之后你就能在自己的项目中灵活设计各种复杂流程。如果本文对你有帮助可以收藏备用方便在后续开发中快速查阅。