1. 项目概述为什么需要 LangGraph如果你已经接触过 LangChain搭建过一些简单的 AI 应用链可能会遇到一个瓶颈当业务流程变得复杂不再是简单的“输入-处理-输出”直线时代码会迅速变得难以维护。比如你需要根据 AI 模型的输出结果来决定下一步是调用工具、查询数据库还是直接返回给用户甚至可能需要循环执行某些步骤直到满足条件。这种“有状态”的、带分支和循环的工作流用传统的链式调用写起来就像是在用面条代码控制一个状态机既混乱又容易出错。这就是 LangGraph 要解决的核心问题。它不是一个替代 LangChain 的工具而是一个基于 LangChain 构建的、专门用于创建有状态、多参与者Agent工作流的框架。你可以把它想象成给 AI 应用开发提供了一个可视化的流程图编辑器。在 LangGraph 里每个节点Node是一个独立的处理单元可以是一个 LLM 调用、一个工具、一段自定义逻辑边Edge定义了节点之间的流转条件。整个图的执行由一个持久化的“状态State”对象来驱动状态在节点间传递和更新决定了流程的走向。所以当你的项目标题是“LangGraph 入门到精通0x02基础 API (一)”时我们瞄准的就是那些已经会用LLMChain或AgentExecutor但渴望构建更复杂、更健壮 AI 智能体Agent或工作流的开发者。本文将彻底拆解 LangGraph 最核心的 API让你理解其设计哲学并能亲手搭建第一个可运行的工作流图。我们会避开那些高阶抽象直接从构建图的基石开始。2. 核心概念与设计哲学拆解在深入代码之前必须理解 LangGraph 的三个核心概念状态State、节点Node和边Edge。这是它区别于简单链Chain的关键。2.1 状态State工作流的记忆与上下文在 LangChain 的普通链中输入和输出通常是独立的字典。但在一个复杂工作流中我们需要在不同步骤间共享和修改信息。LangGraph 引入了“状态”的概念它是一个贯穿整个图执行周期的、可变的共享数据存储。状态通常是一个 TypedDict类型化字典或 Pydantic BaseModel。这为状态提供了类型提示和结构验证。一个典型的状态可能包含input: 用户原始输入。messages: 对话历史记录一个消息列表。intermediate_steps: 工具调用的中间结果。next: 指示下一步该执行哪个节点。状态的魔力在于每个节点都是一个函数它接收当前完整的状态作为输入然后返回一个对该状态的更新。LangGraph 内部会负责将这些更新合并到总状态中。这种设计使得节点间的数据传递变得清晰且类型安全。2.2 节点Node原子化的处理单元节点是工作流中的基本执行单元。在 LangGraph 中一个节点就是一个可调用对象Callable最常见的就是一个函数。这个函数必须遵循一个固定签名function_name(state: State) - PartialState。它接收当前状态执行逻辑如调用 LLM、运行工具然后返回一个字典这个字典包含了要对全局状态进行的修改。例如一个“调用 LLM”的节点函数可能会读取状态中的messages调用聊天模型然后将模型返回的新消息append到messages列表中最后以{messages: [new_message]}的形式返回。LangGraph 会自动将这个新消息合并到状态的messages列表里。2.3 边Edge流程的逻辑控制器边决定了执行完一个节点后下一步该去哪里。这是实现分支、循环等复杂逻辑的关键。边分为两种普通边Linear Edge直接连接两个节点表示无条件跳转到下一个节点。条件边Conditional Edge根据当前状态的值动态决定下一个要执行的节点。这通常通过一个路由函数Router Function来实现。条件边是实现智能体核心逻辑如“如果工具调用结果不完整则继续循环思考”的基石。通过组合节点和条件边你可以轻松构建出复杂的、图灵完备的工作流。设计哲学总结LangGraph 将复杂的工作流抽象为“基于状态流转的有向图”。状态是血液节点是器官边是血管和神经。这种模型极其贴合多轮对话、工具调用、循环校验等真实 AI 应用场景让开发者能从更高的维度去设计和思考问题而不是陷在回调地狱里。3. 基础 API 深度解析与实战理解了核心概念后我们开始动手。我们将从零开始构建一个最简单的 LangGraph一个包含“生成”和“校验”两个节点的线性工作流。这个例子虽然简单但能完整展示所有基础 API 的用法。3.1 定义状态State首先我们需要定义工作流的状态结构。这里我们使用TypedDict因为它轻量且直观。from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END # 1. 定义状态结构 class GraphState(TypedDict): # 用户的问题 question: str # 生成的答案 draft_answer: str # 校验反馈 feedback: str # 最终确认的答案 final_answer: str这里我们定义了一个包含四个字段的状态。Annotated和operator.add是 LangGraph 的“语法糖”用于声明某个字段的合并策略。例如feedback字段我们可能希望后续节点能覆盖它而不是追加所以这里先不用Annotated。更复杂的合并策略如追加到列表我们会在后续章节详解。3.2 创建节点Node函数接下来创建两个节点函数一个用于生成答案草案一个用于校验这个草案。# 2. 创建节点函数 def generate_node(state: GraphState) - dict: 模拟LLM生成答案的节点 question state[question] # 模拟一个简单的生成逻辑。真实场景中这里会调用 LLM。 draft f关于{question}的初步回答这是一个模拟生成的答案草案。 print(f[生成节点] 基于问题 {question} 生成了草案。) # 返回对状态的更新设置 draft_answer 字段 return {draft_answer: draft} def validate_node(state: GraphState) - dict: 模拟校验答案的节点 draft state.get(draft_answer, ) # 模拟一个简单的校验逻辑。真实场景可能调用另一个LLM进行事实性或安全性检查。 if 模拟 in draft: feedback 草案中包含模拟内容需要更具体的回答。 next_action revise # 指示需要修订 else: feedback 草案内容良好可以确认。 next_action confirm # 指示可以确认 print(f[校验节点] 校验草案{draft[:50]}...。反馈{feedback}) # 更新 feedback 字段并添加一个决定下一步的字段 return {feedback: feedback, next_step: next_action}每个节点函数都接收完整的GraphState并返回一个字典。这个字典的键必须是GraphState中定义的字段名值就是要更新到该字段的内容。LangGraph 的运行时会将这个更新字典应用到全局状态上。3.3 构建图Graph并添加边现在我们创建图结构将节点和边组装起来。# 3. 构建图 # 初始化一个状态图并指定状态的结构类 workflow StateGraph(GraphState) # 添加节点。第一个参数是节点名称第二个是节点函数。 workflow.add_node(generate, generate_node) workflow.add_node(validate, validate_node) # 设置入口点工作流从哪个节点开始 workflow.set_entry_point(generate) # 添加普通边从 generate 节点无条件指向 validate 节点 workflow.add_edge(generate, validate) # 添加条件边从 validate 节点出发根据状态决定下一步 def decide_next_step(state: GraphState) - str: 路由函数根据状态决定下一个节点 next_step state.get(next_step, end) if next_step revise: return generate # 返回修订跳回 generate 节点 elif next_step confirm: return finalize # 确认跳转到最终化节点 else: return END # 结束流程 workflow.add_conditional_edges( validate, # 源节点 decide_next_step, # 路由函数 # 路由函数可能返回的目标节点映射可选但建议提供用于清晰度和验证 { generate: generate, finalize: finalize, END: END } ) # 添加一个最终化节点和边 def finalize_node(state: GraphState) - dict: 最终化节点准备最终输出 final_answer f最终答案基于草案{state[draft_answer]}和反馈{state[feedback]}生成 print(f[最终化节点] 生成最终答案。) return {final_answer: final_answer} workflow.add_node(finalize, finalize_node) workflow.add_edge(finalize, END) # 最终化后直接结束 # 编译图得到可执行对象 app workflow.compile()这段代码是 LangGraph 的核心StateGraph图构建器。add_node注册节点。set_entry_point设置开始节点。add_edge添加无条件转移边。add_conditional_edges添加条件边。这是最强大的部分。它需要一个路由函数该函数检查当前状态返回下一个要执行的节点名称字符串。END是一个特殊的标记表示图执行结束。compile()将图定义编译成可执行的应用。编译过程会进行一系列检查和优化。3.4 执行图与查看结果图编译好后就可以像调用函数一样执行它了。输入是一个初始状态字典。# 4. 执行图 # 定义初始状态 initial_state: GraphState { question: LangGraph 是什么, draft_answer: , feedback: , final_answer: } print( 开始执行工作流 ) # 使用 stream 方法可以逐步查看执行过程和状态变化 for event in app.stream(initial_state): # event 是一个元组 (node_name, state_update) node_name, state_update event print(f节点 {node_name} 执行完毕。状态更新: {state_update}) print(- * 40) print(\n 最终状态 ) # 获取最终完整状态 final_state app.invoke(initial_state) # invoke 会执行到底并返回最终状态 print(final_state)执行上述代码你会看到类似以下的输出清晰地展示了状态随着节点执行而流转的过程 开始执行工作流 [生成节点] 基于问题 LangGraph 是什么 生成了草案。 节点 generate 执行完毕。状态更新: {draft_answer: 关于LangGraph 是什么的初步回答这是一个模拟生成的答案草案。} ---------------------------------------- [校验节点] 校验草案关于LangGraph 是什么的初步回答这是一个模拟...。反馈草案中包含模拟内容需要更具体的回答。 节点 validate 执行完毕。状态更新: {feedback: 草案中包含模拟内容需要更具体的回答。, next_step: revise} ---------------------------------------- [生成节点] 基于问题 LangGraph 是什么 生成了草案。 节点 generate 执行完毕。状态更新: {draft_answer: 关于LangGraph 是什么的初步回答这是一个模拟生成的答案草案。} ---------------------------------------- ... (可能会陷入循环因为我们的逻辑简单)这里暴露了我们第一个流程设计缺陷校验节点发现草案有问题后路由函数指示跳回generate节点但generate节点是“无记忆”的它再次基于原始问题生成了完全一样的草案导致死循环。在实际应用中我们需要让generate节点能考虑到之前的feedback。这引出了状态管理的下一个关键点。4. 状态State的合并策略与进阶管理在上面的例子中我们遇到了状态更新的一个核心问题如何合并当多个节点都可能更新同一个字段时以谁为准LangGraph 通过Annotated类型提示来定义字段的合并策略。4.1 使用 Annotated 定义合并策略让我们修改状态定义让messages消息列表和feedback历史能更好地被管理。from typing import TypedDict, List, Annotated import operator class ImprovedGraphState(TypedDict): question: str # 关键使用 Annotated 声明 messages 字段的合并策略为“追加” messages: Annotated[List[str], operator.add] # feedbacks 也采用追加策略记录所有校验反馈 feedbacks: Annotated[List[str], operator.add] draft_answer: str final_answer: strAnnotated[List[str], operator.add]告诉 LangGraph当多个节点返回的更新中都包含messages时应该使用operator.add即列表的操作来合并它们也就是将新的列表追加到旧的列表后面。这是处理对话历史、工具调用序列等场景的标配。4.2 改造节点函数以利用合并策略现在我们改造节点函数让它们能利用这个特性实现带反馈的循环生成。def generate_with_feedback_node(state: ImprovedGraphState) - dict: 生成节点能读取历史反馈 question state[question] # 获取最新的反馈如果有的话 all_feedbacks state.get(feedbacks, []) last_feedback all_feedbacks[-1] if all_feedbacks else 无 # 模拟一个更智能的生成结合问题和上次反馈 prompt f问题{question}。上一轮反馈{last_feedback}。请生成改进后的答案。 # 模拟LLM调用 new_draft f改进草案基于反馈‘{last_feedback}’这是更具体的回答。 # 更新消息历史记录本次生成 new_message f生成器基于反馈‘{last_feedback}’创建了草案。 print(f[智能生成节点] 提示{prompt}) return { draft_answer: new_draft, messages: [new_message] # 这里返回的列表会被追加到全局的 messages 中 } def validate_and_route_node(state: ImprovedGraphState) - dict: 校验节点并给出路由决策 draft state.get(draft_answer, ) # 模拟校验逻辑 if 具体 in draft and 模拟 not in draft: feedback 草案质量合格通过校验。 is_approved True else: feedback 草案仍需改进请补充具体细节。 is_approved False print(f[校验路由节点] 草案{draft[:30]}... 反馈{feedback}) # 更新反馈历史 update { feedbacks: [feedback], messages: [f校验器给出反馈{feedback}] } # 将路由决策也放入状态供条件边读取 if is_approved: update[next_step] finalize else: # 可以加入循环次数限制防止无限循环 if len(state.get(feedbacks, [])) 3: # 假设最多循环3次 update[next_step] finalize update[feedbacks].append(已达到最大修订次数强制结束。) else: update[next_step] generate return update关键改进generate_with_feedback_node会读取feedbacks历史使生成过程具备“记忆”。两个节点都返回了messages和feedbacks的更新。由于我们在状态定义中为这两个字段指定了operator.add合并策略LangGraph 会自动将每次节点返回的新列表追加到全局列表的末尾。validate_and_route_node加入了简单的循环限制逻辑这是一个非常重要的防错实践。在构建任何可能循环的图时都必须考虑退出条件避免因逻辑错误或模型输出不稳定导致的无限循环。4.3 编译并执行改进后的图重新构建并执行这个改进后的图你将看到行为更加智能和可控# 重建图 improved_workflow StateGraph(ImprovedGraphState) improved_workflow.add_node(generate, generate_with_feedback_node) improved_workflow.add_node(validate, validate_and_route_node) improved_workflow.add_node(finalize, finalize_node) # 复用之前的finalize_node需稍作调整适配新状态 improved_workflow.set_entry_point(generate) improved_workflow.add_edge(generate, validate) def improved_router(state: ImprovedGraphState) - str: return state.get(next_step, END) improved_workflow.add_conditional_edges(validate, improved_router, { generate: generate, finalize: finalize, END: END }) improved_workflow.add_edge(finalize, END) improved_app improved_workflow.compile() # 执行 initial_state ImprovedGraphState(question如何学习深度学习, messages[], feedbacks[], draft_answer, final_answer) print(执行改进后的工作流最多循环3次:) for event in improved_app.stream(initial_state, subgraphsTrue): # subgraphs 选项可展示更详细结构 node, update event print(f{node}: {update.get(draft_answer, update.get(feedbacks, update))}) final_state improved_app.invoke(initial_state) print(f\n最终答案: {final_state[final_answer]}) print(f消息历史: {final_state[messages]}) print(f反馈历史: {final_state[feedbacks]})通过这个例子你应该深刻理解了Annotated和合并策略的重要性。它使得状态管理从“覆盖”变为“演进”非常适合多轮交互场景。5. 常见问题、调试技巧与性能考量在实际使用 LangGraph 的基础 API 时你会遇到一些典型问题。这里记录了我的踩坑实录和解决方案。5.1 状态更新不生效或报错问题现象节点返回了更新字典但最终状态里没变化或者抛出KeyError或类型错误。排查清单字段名拼写确保节点返回的字典键名与TypedDict或Pydantic Model中定义的字段名完全一致。Python 是大小写敏感的。合并策略冲突如果一个字段在状态定义中声明为Annotated[List, operator.add]但某个节点返回的更新值不是列表比如返回了一个字符串合并时会出错。确保节点返回值的类型与字段声明的类型及合并策略兼容。未声明的字段如果你尝试更新一个没有在状态类中定义的字段默认情况下 LangGraph 会忽略它除非你使用了特定的配置。检查状态类定义是否包含了所有需要用到的字段。5.2 条件边Conditional Edge路由错误问题现象流程没有按预期分支或者报错ValueError: Invalid target node。排查步骤检查路由函数返回值路由函数必须返回一个字符串该字符串必须是已添加到图中的节点名称或者是预定义的END常量。使用print或日志在路由函数内部输出返回值确认其符合预期。验证目标映射在add_conditional_edges方法中提供path_map参数即我们示例中的字典是个好习惯。这个字典不仅是为了文档化LangGraph 在编译时会用它来验证路由函数所有可能的返回值是否都是有效的目标节点。如果路由函数可能返回一个不在path_map中的值编译就会失败帮助你在运行前发现问题。状态访问确保路由函数读取的状态字段在此时已被正确赋值。有时因为节点执行顺序问题某个字段在路由时可能还是初始值。5.3 无限循环与流程控制问题现象如图陷入死循环无法退出。解决方案强制退出条件在任何可能形成循环的路由逻辑中必须基于状态设置一个硬性退出条件。就像我们在validate_and_route_node里做的检查feedbacks的长度。最大步数限制app.invoke()和app.stream()方法支持config参数其中可以设置recursion_limit。这是一个全局安全网。# 设置递归限制为50步 result app.invoke(initial_state, config{recursion_limit: 50})使用interrupt机制LangGraph 支持更高级的中断Interrupt和人为接管Human-in-the-loop这属于进阶 API但在设计复杂工作流时非常有用可以从外部暂停或改变流程。5.4 调试与可视化打印与日志在每个节点函数开始和结束时打印状态快照是最直接的调试方式。LangGraph 可视化这是 LangGraph 的王牌功能之一。编译后的图对象可以直接输出为图片。# 需要安装 graphviz 和 pygraphviz from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法生成图片可以输出文本形式的 Mermaid 图 print(app.get_graph().draw_mermaid())可视化图表能让你一眼看清所有节点和边的结构对于检查条件边的逻辑是否正确、是否有意外循环至关重要。5.5 性能考量与最佳实践状态尽量轻量状态对象会在每个节点间传递和序列化如果使用持久化。避免在状态中存储大型对象如完整的文档内容。应该存储引用或摘要。节点函数保持纯净节点函数应尽量是纯函数其输出只依赖于输入的状态。这有利于测试、调试和并发执行。避免在节点内部修改全局变量或产生其他副作用。合理划分节点粒度一个节点应该完成一个逻辑上相对独立的任务。不要把所有代码塞进一个节点也不要拆得过细导致图过于复杂。平衡可读性和复用性。善用stream方法进行开发调试app.stream()会返回一个生成器逐步产出每个节点执行后的状态更新。这在开发阶段观察流程走向、定位问题节点非常高效。掌握了这些基础 API 和核心概念你已经具备了用 LangGraph 构建复杂 AI 工作流的能力。从定义一个清晰的状态结构开始到实现每个节点函数再到通过边将它们有机连接起来这个过程就像在绘制一幅智能应用的蓝图。记住好的图设计始于对业务逻辑的清晰拆解。在下一篇文章中我们将深入 LangGraph 更强大的特性持久化状态Checkpoint和多智能体Multi-Agent协作这将让你的应用真正具备长期记忆和分工协作的能力。
