LangGraph状态图驱动的Agent开发

可视化演示

LangGraph状态图框架总览与设计理念

LangGraph是LangChain团队推出的状态图驱动Agent开发框架，通过将Agent逻辑建模为有向图（Directed Graph），让开发者可以精确控制节点执行顺序、状态流转和分支跳转。它解决了传统链式Agent难以处理复杂状态管理和循环逻辑的问题，是构建生产级Agent应用的核心框架。

状态图模型的核心思想是将Agent的执行流程抽象为图结构：节点（Node）代表执行具体逻辑的函数，边（Edge）定义节点间的跳转关系。与传统链式调用不同，状态图支持条件分支、循环和并行执行，能够表达更复杂的Agent行为模式。

LangGraph与LangChain的关系是互补而非替代。LangChain提供LLM调用、工具集成、检索增强等基础能力，LangGraph在此基础上提供状态管理和流程编排能力。可以将LangChain看作Agent的「器官」（感知、思考、行动），LangGraph则是Agent的「神经系统」（协调、调度、控制）。

状态图驱动Agent相比传统链式Agent的核心优势在于：精确的状态管理（每个节点看到的状态是明确定义的）、灵活的流程控制（条件边实现动态路由）、可靠的执行保障（检查点机制支持回滚恢复）、以及可观测的执行过程（状态变化可追踪和调试）。

TypedDict状态定义与流转机制

LangGraph使用Python的TypedDict来定义状态结构，这是一种带有类型注解的字典类型。通过TypedDict，开发者可以明确定义状态包含哪些字段、每个字段的类型是什么，获得IDE的类型提示和编译时检查能力。

状态累加器（Reducer）是LangGraph状态管理的核心机制。默认情况下，节点返回的状态更新会覆盖对应字段的旧值。但开发者可以为特定字段定义自定义的合并策略，例如使用列表追加（operator.add）而非覆盖，使得多个节点的输出可以累积到同一字段中。

状态在节点间的传递遵循明确的规则：每个节点函数接收当前完整状态作为参数，返回需要更新的字段子集。LangGraph框架负责将节点返回的更新与当前状态按照累加器策略合并，生成新的状态传递给下一个节点。这种设计使得节点之间通过状态解耦，无需直接调用。

需要注意的是，LangGraph的状态不是简单的全局变量传递。每个状态字段可以有不同的合并策略：覆盖（默认）、追加（列表累加）、自定义函数（如取最大值、去重合并等）。这种精确可控的状态更新机制是LangGraph在多节点并发场景下保证状态一致性的关键。

from typing import TypedDict, Annotatedimport operator class AgentState(TypedDict):    messages: Annotated[list, operator.add]    current_step: str    results: Annotated[list, operator.add]    final_answer: str def process_node(state: AgentState):    result = f"Processed: {state['current_step']}"    return {        'messages': [{'role': 'assistant', 'content': result}],        'results': [result],        'current_step': 'done'    }

节点与边的编排执行设计

LangGraph的图由节点和边组成。节点是执行具体逻辑的函数，每个节点接收当前状态、执行操作、返回状态更新。节点函数可以是同步或异步的，可以调用LLM、执行工具、进行数据处理等任意操作。

边定义节点间的跳转关系，分为三种类型：普通边（固定跳转到指定节点）、条件边（根据状态动态决定下一节点）、入口边（定义图的起始节点）。条件边是实现复杂分支控制的关键，它接收当前状态作为输入，返回下一个节点的名称。

StateGraph是LangGraph的核心构建类。通过add_node添加节点、add_edge添加普通边、add_conditional_edges添加条件边，最后调用compile方法编译生成可执行的图。编译过程会验证图的连通性和边的合法性，确保不存在孤立节点或死循环。

节点编排支持并行执行：当多个节点没有依赖关系时，LangGraph可以同时执行它们。通过Send API可以将不同的输入分发到同一节点的多个并行实例，实现fan-out/fan-in的并行处理模式。并行执行的结果通过状态累加器合并到统一的状态中。

from langgraph.graph import StateGraph, START, END def route_by_status(state):    if state.get('approved'):        return 'archive'    return 'revise' graph = StateGraph(AgentState)graph.add_node('extract', extract_node)graph.add_node('review', review_node)graph.add_node('archive', archive_node)graph.add_node('revise', revise_node)graph.add_edge(START, 'extract')graph.add_edge('extract', 'review')graph.add_conditional_edges('review', route_by_status)graph.add_edge('archive', END)graph.add_edge('revise', 'extract')app = graph.compile()

人机交互节点与审批流程设计

LangGraph原生支持Human-in-the-loop（人机交互）机制，允许Agent在执行过程中暂停等待人工审批或确认。这种机制对于高风险操作（如发送邮件、修改数据库、执行金融交易）至关重要，确保关键决策有人类监督。

人机交互的实现基于interrupt_before和interrupt_after参数。在compile时指定哪些节点需要在执行前或执行后中断，当图执行到这些节点时会暂停并将当前状态保存到Checkpointer。外部系统可以通过get_state获取暂停时的状态，展示给人类用户进行审查。

恢复执行通过update_state方法实现。人类用户审查状态后，可以修改状态中的某些字段（如设置approved=True），然后调用update_state更新状态并恢复执行。图会从中断点继续执行下一个节点，整个过程无需重新开始。

交互节点与自动节点可以混合编排在同一个图中。典型的模式是：自动节点处理数据提取和初步分析，在关键决策点设置人工审批节点，审批通过后继续自动执行后续步骤。这种混合模式既发挥了自动化的效率优势，又保证了关键决策的安全性。

状态持久化与检查点回溯机制

LangGraph通过Checkpointer机制实现状态持久化。在每个节点执行完成后，Checkpointer自动保存当前状态的完整快照。这些快照存储在持久化后端（如SQLite、PostgreSQL、Redis），即使应用重启也能恢复执行状态。

MemorySaver是最简单的Checkpointer实现，将状态保存在内存中，适合开发和测试。生产环境应使用SqliteSaver或PostgresSaver等持久化方案，确保状态数据不会因进程重启而丢失。Checkpointer为每个执行线程（thread_id）维护独立的状态历史。

状态回溯是Checkpointer的高级功能。通过get_state_history可以获取某个线程的所有历史状态快照，按照时间顺序排列。开发者可以选择任意历史版本恢复到当前状态，实现Agent执行的回滚功能。这对于调试错误、探索不同执行路径和审计追踪非常有价值。

检查点机制不仅用于错误恢复，还支持时间旅行调试、状态审计追踪、多分支探索等高级功能。时间旅行调试允许开发者回到Agent执行的任意历史节点，观察当时的状态并分析决策原因。多分支探索允许从同一历史状态创建多个分支，尝试不同的执行路径。

分支控制与流式输出实现

LangGraph支持多分支并行执行和结果聚合。通过Send API，可以将不同的输入数据分发到同一节点函数的多个并行实例中处理。每个实例独立执行并返回结果，所有结果通过状态累加器合并。这种fan-out/fan-in模式适合需要并行处理多个子任务的场景。

流式输出是LangGraph的重要特性，允许Agent执行过程中的中间状态和结果实时推送到前端。通过stream方法，开发者可以逐节点获取执行结果，而不是等待整个图执行完成。这对于长时间运行的Agent任务尤其重要，用户可以看到Agent的推理进度。

流式输出支持两种模式：values模式（每次推送完整的当前状态值）和updates模式（只推送状态的变化量）。values模式适合需要完整状态快照的场景，updates模式适合只需要增量更新的场景，减少网络传输量。

复杂工作流的分支控制需要综合考虑条件路由、并行执行和错误处理。条件边实现基于状态的动态路由，并行节点通过累加器合并结果，错误处理通过try-catch节点或回退边实现。合理组合这些机制可以构建出既灵活又可靠的Agent工作流。

AI 应用场景

Python 演示

from typing import TypedDict, Annotatedimport operator class AgentState(TypedDict):    messages: Annotated[list, operator.add]    current_step: str    approved: bool def extract_node(state: AgentState):    doc = "Sample document content"    return {        'messages': [{'role': 'assistant', 'content': f'Extracted: {doc}'}],        'current_step': 'extracted'    } def review_node(state: AgentState):    last_msg = state['messages'][-1]['content']    return {        'messages': [{'role': 'assistant', 'content': f'Reviewed: {last_msg}'}],        'current_step': 'reviewed',        'approved': True    } def archive_node(state: AgentState):    return {        'messages': [{'role': 'assistant', 'content': 'Archived successfully'}],        'current_step': 'archived'    } def revise_node(state: AgentState):    return {        'messages': [{'role': 'assistant', 'content': 'Needs revision'}],        'current_step': 'revising',        'approved': False    } def route_by_approval(state):    if state.get('approved'):        return 'archive'    return 'revise' from langgraph.graph import StateGraph, START, END graph = StateGraph(AgentState)graph.add_node('extract', extract_node)graph.add_node('review', review_node)graph.add_node('archive', archive_node)graph.add_node('revise', revise_node)graph.add_edge(START, 'extract')graph.add_edge('extract', 'review')graph.add_conditional_edges('review', route_by_approval)graph.add_edge('archive', END)graph.add_edge('revise', 'extract')app = graph.compile() result = app.invoke({'messages': [], 'current_step': '', 'approved': False})for msg in result['messages']:    print(msg['content'])print(f"Final step: {result['current_step']}")

1. class AgentState(TypedDict):使用TypedDict定义状态结构，明确状态包含messages、current_step和approved字段
2. messages: Annotated[list, operator.add]messages字段使用Annotated+operator.add定义累加器，新消息追加而非覆盖
3. def extract_node(state: AgentState):提取节点：接收状态，提取文档内容，返回状态更新
4. def review_node(state: AgentState):审核节点：读取最新消息进行审核，设置approved标志
5. def route_by_approval(state):条件路由函数：根据approved状态决定跳转到归档还是修改节点
6. graph = StateGraph(AgentState)创建状态图实例，传入状态类型定义
7. graph.add_node('extract', extract_node)添加节点：将函数注册为图中的执行节点
8. graph.add_conditional_edges('review', route_by_approval)添加条件边：从review节点出发，根据路由函数动态决定下一节点
9. graph.add_edge('revise', 'extract')添加回退边：修改节点执行后回到提取节点，形成循环
10. app = graph.compile()编译图：验证连通性，生成可执行的应用实例
11. result = app.invoke({'messages': [], 'current_step': '', 'approved': False})执行图：传入初始状态，运行整个流程直到END节点

常见问答

常见误区

参考来源

• webpageLangGraph DocumentationLangGraph官方文档
• webpageLangGraph GitHub RepositoryLangGraph源代码仓库
• webpageIntroducing LangGraphLangChain官方博客介绍LangGraph