先思考一个常见场景你费了不少力气把大模型接进了业务系统却发现单次调用大模型根本解决不了真实问题。真实需求往往是多步骤的先判断意图再查资料然后决定是调用工具还是继续追问中间可能还要验证结果、重新尝试、多路并行处理。如果只用if-else硬写逻辑很快会变成一团乱麻如果只靠 LangChain 的 Chain 串行调用又很难处理分支、循环和状态回退。LangGraph 就是在这样的背景下出现的。这篇文章我会围绕 LangGraph 的核心概念、环境搭建、节点与状态设计、条件路由、循环控制、子图封装、并行分支、常见报错排查和工程落地建议展开。内容偏实战但概念部分也会讲透适合刚从 LangChain 过渡过来、或者想用 LangGraph 重构 Agent 工作流的读者。读完以后你应该能独立搭出一个带条件路由和循环检测的 LangGraph 应用也能理解为什么官方文档里反复强调 State、Node、Edge 这三个抽象。1. LangGraph 到底是什么为什么值得系统学1.1 从 LangChain 到 LangGraphAgent 编排思路的演进LangChain 早期给人的印象是“链式调用”把 Prompt、模型、输出解析器串成一条链。对简单的固定流程来说链式调用完全够用。但到了 Agent 场景问题就变了模型需要根据当前状态决定调哪个工具、要不要重试、什么时候该结束甚至要多个工具并行执行后再汇总结果。这种动态流程用固定链很难表达。LangGraph 把 LLM 应用建模成一张“图”。图里有节点、有边、有共享状态。节点是执行单元边描述了执行顺序状态是贯穿整个流程的数据结构。这个建模思路意味着你可以构建循环让流程在条件不满足时回到前面的节点重新执行也可以构建分支让不同输入走不同路径还可以把复杂流程拆成子图降低单张图的可维护性负担。翻译成通俗话就是LangChain 适合“流水线”LangGraph 适合“工作流”。如果业务逻辑里存在“模型判断下一步去哪”这类动态决策LangGraph 的表达能力会明显更强。1.2 LangGraph 和 LangChain 的区别很多同学刚开始会混淆 LangGraph 和 LangChain。这里做一个比较清晰的对照维度LangChainLangGraph核心抽象Chain、Tool、RetrieverStateGraph、Node、Edge、State流程结构偏线性串行图结构支持分支、循环、并行状态管理依赖外部传参链内状态比较弱全局 State节点更新后自动传递给后续节点适用场景固定流程、RAG、Prompt 管线Agent、多步决策、人工审核、循环校验流程学习曲线相对平缓需要理解图、状态、路由等概念这里需要强调一点LangGraph 并不是 LangChain 的替代品。LangChain 的模型封装、Prompt 模板、文档加载器、向量库集成这些能力依然有用。LangGraph 在编排层更擅长它本身也允许你在节点里继续使用 LangChain 的 ChatModel、Retriever、Tool 等组件。实际项目中两者经常搭配使用。1.3 适合谁学、学完能做什么LangGraph 适合以下几类读者已经用过 LangChain想在 Agent 场景中处理动态分支的开发者。想用大模型构建多步骤业务流的后端工程师例如工单自动分类、内容审核、需求拆解。在研究 ReAct、Plan-and-Execute、Self-Consistency 等 Agent 模式的算法工程师。对状态机、图流程编排感兴趣想找一种抽象方式管理复杂任务的人。学完本文后你应该能独立实现一个带条件路由的多节点工作流一个会“循环重试直到结果合格”的节点组合一个通过子图复用流程片的架构以及一个可以多个分支并行执行的流程骨架。2. 环境准备与版本说明2.1 安装 LangGraph 与依赖本文示例以 Python 环境为主。建议使用 Python 3.9 及以上版本并优先在虚拟环境中操作。安装命令如下pip install langgraph langchain-core如果你还需要调用大模型建议根据实际模型厂商安装对应依赖。例如 OpenAI 风格接口可以使用pip install langchain-openai这里要说明LangGraph 更新速度比较快API 在某些小版本中会有调整。本文示例基于langgraph较新的稳定用法但你在自己的环境中运行前最好先确认一下实际安装版本pip show langgraph或者直接在代码里打印版本信息import langgraph print(langgraph.__version__)如果版本差异较大优先参考官方文档中对应版本的迁移说明而不是死记本文代码。2.2 验证环境是否可用安装完成后可以写一个最小示例验证环境。先创建一个测试文件test_langgraph.pyfrom typing import TypedDict from langgraph.graph import StateGraph, START, END class State(TypedDict): message: str def hello_node(state: State) - State: return {message: state[message] LangGraph 环境正常} builder StateGraph(State) builder.add_node(hello, hello_node) builder.add_edge(START, hello) builder.add_edge(hello, END) graph builder.compile() result graph.invoke({message: 你好}) print(result)如果执行后输出类似下面的结果说明环境没有问题{message: 你好LangGraph 环境正常}2.3 示例项目结构实战阶段建议按下面的结构组织文件便于后续扩展langgraph_demo/ ├── .env ├── requirements.txt ├── main.py └── graph/ ├── __init__.py ├── state.py ├── nodes.py ├── edges.py └── workflow.py在实际项目中不要把所有节点和路由逻辑都塞进一个文件。按职责拆分会让你更容易定位问题也方便后续加入测试和日志。3. 核心概念拆解State、Node、Edge 与 GraphLangGraph 的核心抽象可以浓缩成一句话整个应用是一张图图里有节点节点之间靠边连接所有节点共享同一个 State 对象。下面逐个拆解。3.1 State全局共享的状态State 是 LangGraph 中贯穿整个流程的数据结构。它通常使用TypedDict定义也可以使用 Pydantic 的BaseModel。一个简单的状态定义如下from typing import TypedDict class State(TypedDict): messages: list current_step: str retry_count: int这里有几个关键点每个节点执行的输入是当前 State。每个节点返回的字典会覆盖或更新 State 中的字段。节点不需要返回全部字段只返回要修改的字段即可。如果节点返回了 State 中不存在的字段并且状态定义是 TypedDict可能会触发类型校验或运行时错误。想象一下每个节点拿到一个“共享笔记本”它只需要把自己负责的那一页写好后续节点就能看到最新内容。这种设计让节点之间的数据传递不再依赖函数签名而是统一通过 State 完成。3.2 Node状态转换函数Node 本质上是一个函数。它的输入是State输出是一个字典字典里的内容会被合并进 State。def analyze_node(state: State) - State: # 模拟分析逻辑 result f分析完成当前消息数量{len(state[messages])} return {analysis_result: result}这里需要注意函数签名。LangGraph 内部会把 State 作为关键字参数传给节点。如果你在节点里除了state还想接收额外配置可以通过config参数实现def analyze_node(state: State, config: dict) - State: # config 里通常包含 thread_id、metadata 等信息 print(当前线程 ID, config.get(configurable, {}).get(thread_id)) return {analysis_result: ok}不过大多数入门场景不需要 config可以暂不深入。3.3 Edge连接与条件路由Edge 描述节点之间的跳转关系。分为普通边和条件边。普通边写法builder.add_edge(node_a, node_b)表示node_a执行完成后立刻执行node_b。条件边写法builder.add_conditional_edges( node_a, route_function, {continue: node_b, end: END} )route_function接收当前的 State返回一个字符串键LangGraph 根据这个键找到对应的目标节点。条件路由看起来不复杂但它是 LangGraph 实现动态决策的关键。后面实战部分会重点演示。3.4 Graph 编译与执行把所有节点和边定义完成后调用compile()得到可执行的图app builder.compile() result app.invoke({messages: [用户消息]})invoke是最常见的同步执行入口。除了invokeLangGraph 还支持异步的ainvoke、流式输出的stream、以及带状态持久化的执行方式。入门阶段先掌握invoke即可。4. 完整实战案例一从零构建一个带条件路由的 Agent 工作流这一节我们做一个完整的小项目。需求是输入一个用户问题系统先判断问题类型然后决定走“普通回答”还是“工具查询”路径最后汇总输出。4.1 需求分析流程设计如下接收用户问题。用一个分类节点判断问题是否涉及实时数据。如果是实时数据类问题走工具查询路径。如果不是走普通回答路径。两条路径结束后都汇合到汇总输出节点。这个流程天然适合条件路由。4.2 定义状态模型创建state.pyfrom typing import TypedDict class WorkflowState(TypedDict): question: str question_type: str tool_result: str final_answer: str字段含义question用户原始问题。question_type分类节点的输出结果tool表示需要查询工具normal表示普通回答。tool_result工具查询结果。final_answer最终输出答案。4.3 编写节点函数创建nodes.pyfrom langchain_openai import ChatOpenAI from .state import WorkflowState def classify_node(state: WorkflowState) - WorkflowState: 判断问题类型。这里用一个简单的关键词规则避免引入模型调用。 question state[question] if any(keyword in question for keyword in [天气, 股票, 新闻, 汇率]): return {question_type: tool} return {question_type: normal} def normal_answer_node(state: WorkflowState) - WorkflowState: 普通回答节点直接让模型回答。 llm ChatOpenAI(modelgpt-4o-mini, temperature0) answer llm.invoke(f请用简洁的方式回答{state[question]}) return {final_answer: answer.content} def tool_query_node(state: WorkflowState) - WorkflowState: 模拟工具查询实际项目中可以接入天气 API、股票 API 等。 question state[question] # 这里先模拟真实环境替换成 HTTP 请求或数据库查询 if 天气 in question: tool_result 北京晴25 度东北风 2 级 else: tool_result f查询结果{question} 的实时数据暂未接入返回模拟数据 return {tool_result: tool_result} def final_answer_node(state: WorkflowState) - WorkflowState: 汇总输出。如果走了工具路径则把工具结果组织进最终答案。 if state.get(tool_result): return {final_answer: f根据工具查询{state[tool_result]}} return {final_answer: state.get(final_answer, 暂未生成答案)}为了演示方便上面的classify_node使用了关键词规则。真实项目中完全可以把分类节点改成调用大模型让模型输出一个question_type。这样 LangGraph 的图结构不用改只替换节点内部实现即可。4.4 构建条件路由图创建workflow.pyfrom langgraph.graph import StateGraph, START, END from .state import WorkflowState from .nodes import ( classify_node, normal_answer_node, tool_query_node, final_answer_node, ) def build_workflow(): builder StateGraph(WorkflowState) builder.add_node(classify, classify_node) builder.add_node(normal_answer, normal_answer_node) builder.add_node(tool_query, tool_query_node) builder.add_node(final_answer, final_answer_node) builder.add_edge(START, classify) # 条件路由分类节点之后根据 question_type 决定走哪条路径 builder.add_conditional_edges( classify, lambda state: state[question_type], { tool: tool_query, normal: normal_answer, }, ) builder.add_edge(tool_query, final_answer) builder.add_edge(normal_answer, final_answer) builder.add_edge(final_answer, END) return builder.compile()注意add_conditional_edges的第一个参数是源节点第二个参数是路由函数第三个参数是映射字典。路由函数返回的键必须在映射字典里存在。4.5 运行与验证创建main.pyfrom graph.workflow import build_workflow app build_workflow() if __name__ __main__: result app.invoke({question: 北京明天天气怎么样}) print(result) result2 app.invoke({question: 讲一个笑话}) print(result2)预期效果输入“北京明天天气怎么样”时question_type被识别为tool会走工具查询路径。输入“讲一个笑话”时question_type被识别为normal会走普通回答路径。如果你安装了langgraph的可视化插件还可以将图结构保存为图片方便检查from IPython.display import Image, display # 需要安装额外的绘图依赖 # pip install pygraphviz 或 使用 mermaid 输出 display(Image(app.get_graph().draw_mermaid_png()))这个可视化能力在排查复杂图结构时非常有用。5. 完整实战案例二循环、子图与并行分支第二个案例我们覆盖三个进阶能力循环控制、子图封装、并行分支。这三个能力是 LangGraph 在复杂工作流里真正拉开差距的地方。5.1 循环控制与循环检测在 LangGraph 中循环不是靠语法而是靠“条件边指回前面节点”实现的。假设我们有一个答案质量校验节点如果答案不合格就回到生成节点重新生成。逻辑如下class ReviewState(TypedDict): answer: str retry_count: int def generate_answer_node(state: ReviewState) - ReviewState: # 模拟生成答案实际项目中换成大模型调用 return {answer: f第 {state[retry_count] 1} 次生成的答案} def review_answer_node(state: ReviewState) - ReviewState: # 模拟评审前两次不合格第三次合格 if state[retry_count] 2: return {qualified: False} return {qualified: True} def route_after_review(state: ReviewState) - str: if state.get(qualified): return pass return retry构建图builder StateGraph(ReviewState) builder.add_node(generate, generate_answer_node) builder.add_node(review, review_answer_node) builder.add_edge(START, generate) builder.add_edge(generate, review) builder.add_conditional_edges( review, route_after_review, { pass: END, retry: generate, }, ) app builder.compile() result app.invoke({answer: , retry_count: 0})这里有几个容易踩的坑死循环风险如果review_answer_node永远不返回qualifiedTrue流程会一直重试。LangGraph 默认有递归限制超过限制会抛出GraphRecursionError。计数器更新循环重试时记得在某个节点更新retry_count否则无法跳出循环。上面的示例为了简洁没有在generate_answer_node里更新计数实际使用时建议在generate_answer_node返回retry_count 1。状态污染循环场景下每次迭代最好只修改本轮需要变更的字段避免旧数据残留。5.2 子图Subgraph封装子图的价值在于复用。当多个主流程都要走同一个复杂子流程时可以直接把子流程编译成一张子图然后作为节点加入主图。先定义一个子图# subgraph_demo.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class SubState(TypedDict): input_text: str sub_result: str def sub_process_node(state: SubState) - SubState: return {sub_result: f子图处理完成{state[input_text]}} sub_builder StateGraph(SubState) sub_builder.add_node(sub_process, sub_process_node) sub_builder.add_edge(START, sub_process) sub_builder.add_edge(sub_process, END) subgraph sub_builder.compile()然后在主图中直接把这个子图作为节点加入class MainState(TypedDict): input_text: str sub_result: str def main_node(state: MainState) - MainState: return {input_text: state[input_text] 经过主节点} main_builder StateGraph(MainState) main_builder.add_node(main_step, main_node) main_builder.add_node(call_subgraph, subgraph) main_builder.add_edge(START, main_step) main_builder.add_edge(main_step, call_subgraph) main_builder.add_edge(call_subgraph, END) main_app main_builder.compile() result main_app.invoke({input_text: 原始输入}) print(result)这里要注意子图内部的 State 定义和主图的 State 定义需要兼容。更稳妥的做法是让子图的 State 字段和主图对应字段保持一致这样状态传递不会因为字段缺失而报错。我个人的建议是子图不要过度设计。如果子流程只被调用一次直接用普通节点即可被复用 3 次以上再考虑封装成子图。封装过早会增加调试难度。5.3 并行分支LangGraph 支持多个节点从同一个源节点出发并行执行。最简单的方式就是添加多条普通边class ParallelState(TypedDict): data: str result_a: str result_b: str def node_a(state: ParallelState) - ParallelState: return {result_a: fA 处理{state[data]}} def node_b(state: ParallelState) - ParallelState: return {result_b: fB 处理{state[data]}} def merge_node(state: ParallelState) - ParallelState: return {data: f{state[result_a]}{state[result_b]}} builder StateGraph(ParallelState) builder.add_node(a, node_a) builder.add_node(b, node_b) builder.add_node(merge, merge_node) builder.add_edge(START, a) builder.add_edge(START, b) builder.add_edge(a, merge) builder.add_edge(b, merge) builder.add_edge(merge, END)执行时node_a和node_b会被并行执行。这里有一个细节LangGraph 的并行执行是“异步并发”的如果节点内部有阻塞的同步 IO 或 CPU 密集计算建议把节点函数改造成异步函数async def node_a(state: ParallelState) - ParallelState: await asyncio.sleep(1) return {result_a: A 完成}然后在主入口使用ainvokeresult await app.ainvoke({data: 测试数据})并行分支在使用时要注意一个常见坑多个并行节点可能同时修改 State 中的同一个字段造成覆盖。建议每个并行分支负责自己的独立字段合并时再统一处理。6. 常见问题与排查思路LangGraph 的报错信息有些比较直接有些比较抽象。下面整理几个高频问题以及我的排查顺序。问题现象常见原因解决思路ImportError: cannot import name StateGraph安装了错误的包或版本过旧执行pip install langgraph升级到新版本确认没有误装其他同名包节点函数返回了 State 中不存在的字段TypedDict 严格校验时报错或后续节点取不到值打开extra校验或调整 State 定义统一节点返回字段节点函数返回None导致状态没有更新节点函数没有返回值或返回了空字典节点必须返回字典或None如果不希望更新状态返回空字典{}而不是NoneGraphRecursionError或超过最大递归次数图中存在循环路径且没有正确的出口条件检查条件路由函数确认循环能跳出必要时在状态中增加retry_count限制调用invoke后结果里缺少部分字段有些节点没有按预期执行使用get_graph()可视化图结构在节点内部加日志输出异步节点与同步调用混用报错在同步入口invoke中使用async节点函数统一使用ainvoke或把异步节点改成普通函数6.1 节点返回None导致状态丢失很多初学者在写节点时会像这样写def my_node(state: State) - None: state[updated] True然后发现updated并没有生效。原因是 LangGraph 并不是靠“原地修改”来更新状态而是依赖节点函数的返回值。正确的写法是def my_node(state: State) - State: return {updated: True}如果你真的不需要更新状态也可以不写返回语句LangGraph 会把它当成不会修改状态的处理节点。但如果你返回值写成了None同时内部又修改了state这种隐式行为容易在调试时造成困惑。6.2 条件路由映射不到目标节点条件路由函数返回的键必须在映射字典中存在。比如builder.add_conditional_edges( node_a, route_func, {tool: node_b} )如果route_func返回了normal而映射字典里没有这个键LangGraph 会抛错。这里的建议是路由函数内做好兜底比如return normal if normal in mapping else ...或者直接在映射字典中覆盖所有可能返回值。6.3 编译报错Not all edges were specified如果你添加了一个节点但忘了给它连接边编译时可能报类似错误。排查顺序是用get_graph().draw_mermaid_png()看节点是否独立。检查每个节点是否都有入边和出边。检查START和END是否正确连接。7. 最佳实践与工程建议7.1 状态模型设计要克制State 设计会直接影响整个项目的可维护性。我的建议是State 字段不要太多能表达当前流程所需数据即可。把“临时中间结果”和“最终产物”分开命名。如果状态结构越来越复杂优先拆成多个子状态由不同子图管理。避免在 State 中存无法序列化的对象例如数据库连接、大模型实例。这类对象最好通过config或依赖注入传给节点。7.2 节点函数保持单一职责一个节点只做一件事。例如“生成回答”和“校验回答”分开“查询天气”和“汇总结果”分开。节点拆得越细条件路由和单元测试就越容易写。但这不代表节点越细越好太细会导致图结构碎片化同样不好维护。我个人的经验是以“是否能独立描述一个业务步骤”为判断标准。7.3 条件路由函数保持“纯函数”风格条件路由函数最好只根据 State 中的已有字段做判断不要在路由函数里调用外部 API 或大模型。判断逻辑越纯粹调试越简单。如果某个判断需要调用大模型可以把判断逻辑放进一个单独的“决策节点”把结果写入 State然后再用条件路由读取结果。7.4 持久化与可观测性LangGraph 支持持久化可以把图执行过程中的状态保存下来。如果生产环境需要人工审核、断点续跑建议启用持久化配置。但从入门角度先掌握以下两点更实际在关键节点加上日志比如print或logger.info输出当前节点名和关键 State 字段。使用 LangSmith 或类似可观测性工具追踪执行链路。如果团队没有这类基础设施先用本地日志把执行路径记录下来也能解决大部分排查问题。7.5 版本与依赖管理LangGraph 的 API 更新速度较快从早期版本到目前的版本部分 API 有过调整。上线前建议在requirements.txt中锁住langgraph和langchain-core的版本。升级前先看官方迁移指南不要盲目升级大版本。写一个最小冒烟测试验证核心图能正常编译和执行。使用国内镜像时注意 LangGraph 依赖较多安装失败时优先检查 pip 源和 Python 版本。8. 总结与学习路线这篇文章从 LangGraph 的核心抽象讲起覆盖了 State、Node、Edge、条件路由、循环控制、子图封装和并行分支并给出了两个完整示例。核心要记住的点有三个LangGraph 用图来描述流程流程状态全部挂在 State 上节点通过返回值更新状态。只要理解这三件事LangGraph 的大部分概念都可以自然延伸。如果你想继续深入我建议按这个顺序推进读懂官方文档的 Tutorial 部分把基本的 Chatbot 和 Agent 示例手动敲一遍。重点研究add_conditional_edges的各种用法尤其是多分支组合。尝试把现有项目中的一两个if-else流程改写成 LangGraph 图体会状态管理带来的变化。再深入持久化、流式输出、Checkpointer、Human-in-the-loop 这些进阶能力。最后才是研究如何把 LangGraph 服务化接入 FastAPI 或消息队列。你可以在动手实践时拿一个小需求反复重构例如“输入问题 → 判断是否需要联网检索 → 检索或直接回答 → 校验质量 → 重新生成或结束”。这个需求看起来简单但足够让你把条件路由、循环、状态传递全部练到。等你把这个流程吃透后面再学子图和并行分支都会轻松很多。如果本文对你有帮助可以收藏备用也欢迎在评论区交流你在使用 LangGraph 时遇到的具体问题。