LangGraph：构建复杂AI工作流与有状态智能体的图计算框架

1. 项目概述：从LangChain到LangGraph的范式演进

如果你在过去一年里深度参与过基于大语言模型的应用开发，那么“LangChain”这个名字对你来说一定不陌生。它几乎成了构建AI应用的事实标准框架，通过其强大的链式编排能力，让我们能够将大模型、工具、记忆和外部数据源串联起来，创造出功能丰富的智能体。然而，随着我们构建的应用越来越复杂，从简单的问答机器人到需要多轮决策、状态管理和分支协作的复杂工作流，单纯依赖“链”的线性思维开始显得捉襟见肘。这正是LangGraph诞生的背景。

LangGraph并非要取代LangChain，而是对其核心范式的一次重要升级和补充。你可以把它理解为LangChain生态系统中的“工作流引擎”或“有状态编排层”。如果说LangChain的Chain是定义了一条从A到B的固定流水线，那么LangGraph则允许你绘制一张完整的流程图，其中包含条件判断、循环、并行执行以及持久化的状态。它引入了“图”的计算模型，将应用中的每个步骤（节点）和步骤之间的流转逻辑（边）清晰地定义出来，特别适合构建具有复杂逻辑、长期记忆和人类参与回路的智能体系统。

简单来说，当你需要构建一个能处理多轮对话、根据中间结果动态调整执行路径、或者协调多个“子智能体”协作完成任务的AI应用时，LangGraph就是你一直在寻找的工具。它让智能体的行为更像一个拥有明确规划和状态感知的“智能体”，而不仅仅是一个反应式的工具调用链。

2. 核心设计理念与架构拆解

2.1 有状态图计算：智能体的“大脑”模型

LangGraph最核心的设计理念是有状态图计算。这听起来有点学术，但理解它对用好这个框架至关重要。

在传统的LangChain链中，数据（或者说“上下文”）通常是以字典的形式在链的各个环节间传递。虽然方便，但对于需要记住之前发生了什么、并根据历史做决策的复杂场景，管理起来就很麻烦。LangGraph将整个系统的运行抽象为一个对“状态”的持续读写和更新过程。

这个“状态”是一个定义良好的字典（State），它包含了应用运行所需的所有信息，例如：用户的输入、大模型的回复、工具调用的结果、历史对话记录、乃至任何你自定义的变量。图中的每个节点（Node）都是一个函数，它接收当前的完整状态，执行一些操作（比如调用大模型、运行工具），然后返回一个更新后的状态字典。LangGraph的运行时引擎负责将更新后的状态传递给下一个节点。

这种设计带来了几个关键优势：

1. 状态集中管理：所有中间数据和历史都存储在同一个状态对象里，查询和追溯变得非常清晰。
2. 自然的循环与分支：基于状态的判断来决定下一步走向哪个节点，是实现循环（比如“直到满足条件才退出”）和条件分支（比如“如果工具调用失败，则转向人工审核节点”）的天然方式。
3. 可持久化与可恢复：由于整个系统的快照就是当前的状态，你可以轻松地将状态保存到数据库，并在之后某个时间点精确地恢复执行，这对于需要长时间运行或支持异步交互的智能体来说至关重要。

2.2 核心组件详解：State、Node、Edge

要构建一个LangGraph应用，你需要理解三个核心组件：状态（State）、节点（Node）和边（Edge）。

状态（State）：通常通过定义一个TypedDict或Pydantic模型来声明。这相当于为你的智能体定义了一张数据表格的 schema。例如，一个客服智能体的状态可能包含user_input,chat_history,agent_scratchpad（智能体思考过程）,knowledge_base_results等字段。LangGraph使用注解来定义每个字段的更新方式，比如是覆盖、追加还是其他自定义合并逻辑。

from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class AgentState(TypedDict): # 用户的最新问题 input: str # 对话历史，使用add_messages注解实现消息列表的追加 messages: Annotated[List, add_messages] # 从知识库检索到的文档片段 context: List[str] # 智能体决定要采取的动作 next_action: str

节点（Node）：节点是执行具体工作的函数。它接收一个状态字典，返回一个更新后的状态字典。一个节点可以做任何事情：调用LLM、执行代码、查询数据库、调用API。关键在于，节点只关心它需要读写状态的哪些部分。例如，一个“检索”节点可能只读取state[“input”]，然后调用向量数据库，最后将结果写入state[“context”]。

def retrieve_node(state: AgentState): # 从状态中获取用户问题 question = state[“input”] # 调用检索工具（例如向量数据库） docs = retriever.invoke(question) # 返回更新后的状态（仅更新context字段） return {“context”: docs}

边（Edge）：边决定了流程的走向。这是LangGraph实现复杂逻辑的魔法所在。边分为两种：

• 条件边（Conditional Edge）：根据当前状态的某个值，决定下一个节点是谁。这通常通过一个路由函数实现。
• 普通边（Regular Edge）：无条件地指向下一个节点。

通过组合节点和边，你就能构建出包含if-else、while循环等复杂逻辑的工作流图。

2.3 与LangChain的共生关系

理解LangGraph和LangChain的关系能帮助你更好地在项目中定位它们。LangGraph深度集成在langchain包中（from langgraph import …），它复用并强化了LangChain的许多核心概念：

• LCEL（LangChain Expression Language）：你依然可以使用LCEL来优雅地组合链，并将这些链作为LangGraph图中的一个节点。这让你能复用大量现有的LangChain生态组件。
• Runnable 协议：LangGraph的节点函数本质上是符合Runnable协议的，这意味着它们可以无缝嵌入到现有的LangChain生态中。
• 智能体（Agent）的重新定义：在LangGraph的范式下，智能体不再是一个特殊的AgentExecutor对象，而是一个由“LLM调用节点”、“工具执行节点”和“路由逻辑边”构成的图。这提供了对智能体决策过程前所未有的透明度和控制力。

简而言之，LangChain提供了丰富的“乐高积木”（模型、工具、检索器），而LangGraph提供了搭建复杂动态“建筑结构”的蓝图和粘合剂。

3. 从零构建一个LangGraph智能体：以研究助手为例

理论说得再多，不如亲手构建一个。让我们来实现一个相对复杂的“研究助手”智能体。它的工作流程是：接收一个研究主题，先联网搜索最新信息，然后根据搜索内容撰写一份初步报告，最后自动检查报告中的事实性陈述，并为有疑问的部分添加引用备注。这个过程涉及条件判断和多个步骤的循环。

3.1 定义状态与工具

首先，定义智能体的状态。这个状态需要贯穿整个研究流程。

from typing import TypedDict, Annotated, List, Optional from langchain_core.messages import BaseMessage, HumanMessage from langgraph.graph.message import add_messages class ResearchState(TypedDict): # 用户提出的研究主题 topic: str # 对话/指令历史 messages: Annotated[List[BaseMessage], add_messages] # 网络搜索得到的内容 search_results: List[str] # 撰写的报告草稿 draft: str # 需要核实的事实列表（事实陈述， 来源） facts_to_check: List[tuple[str, str]] # 最终带引用的报告 final_report: Optional[str]

接下来，装备智能体所需的工具。我们需要一个搜索工具和一个用于检查事实/引用的工具（这里用一个大模型调用模拟）。

from langchain_community.tools import TavilySearchResults from langchain_openai import ChatOpenAI # 初始化工具和模型 search_tool = TavilySearchResults(max_results=3) llm = ChatOpenAI(model=“gpt-4-turbo-preview”) def fact_check_tool(claim: str, context: str) -> str: “”“模拟一个事实检查工具。在实际应用中，这里可以接入专业的Fact-Check API。”“” prompt = f”请判断以下陈述在提供的上下文中是否得到完全支持，并指出支持它的原文依据（如果有的话）：\n陈述：{claim}\n上下文：{context}” response = llm.invoke(prompt) return response.content

3.2 构建图节点：搜索、撰写、检查

我们将工作流分解为四个主要节点。

节点1：搜索节点（search_node）这个节点负责执行网络搜索。

def search_node(state: ResearchState): topic = state[“topic”] print(f”[搜索节点] 正在搜索主题：{topic}”) # 调用搜索工具 results = search_tool.invoke({“query”: topic}) # 提取搜索结果的文本内容 search_contents = [result[“content”] for result in results] return {“search_results”: search_contents}

节点2：撰写报告节点（draft_node）这个节点基于搜索结果为研究主题撰写初稿。

def draft_node(state: ResearchState): topic = state[“topic”] search_content = “\n\n”.join(state[“search_results”]) print(f”[撰写节点] 基于搜索内容撰写‘{topic}’的初稿...”) prompt = f”你是一个专业的研究员。请根据以下关于‘{topic}’的网络搜索信息，撰写一份结构清晰、内容全面的研究报告草稿。报告应包含引言、核心发现和总结。\n\n搜索信息：\n{search_content}” response = llm.invoke(prompt) draft = response.content # 一个简单的启发式方法：从草稿中提取可能的事实陈述（以句号分割，过滤短句） sentences = [s.strip() for s in draft.split(‘.’) if len(s.strip()) > 20] # 这里简化处理，将前3个长句作为待检查事实，并关联其来源（用搜索结果的第一个作为示例来源） facts = [(sent, state[“search_results”][0]) for sent in sentences[:3]] if state[“search_results”] else [] return {“draft”: draft, “facts_to_check”: facts}

节点3：事实检查节点（fact_check_node）这个节点对草稿中的存疑事实进行核查。

def fact_check_node(state: ResearchState): facts = state[“facts_to_check”] checked_notes = [] print(f”[检查节点] 正在核查 {len(facts)} 项事实陈述...”) for claim, source_context in facts: result = fact_check_tool(claim, source_context) checked_notes.append(f”- 陈述：{claim}\n 检查结果：{result}”) notes_str = “\n”.join(checked_notes) return {“facts_to_check”: []} # 清空待检查列表，表示已处理

节点4：生成最终报告节点（finalize_node）这个节点综合草稿和事实检查备注，生成最终报告。

def finalize_node(state: ResearchState): draft = state[“draft”] # 在实际中，这里会整合fact_check_node的输出。本例中我们直接添加一个说明部分。 print(“[终稿节点] 生成最终报告...”) final_report = f”{draft}\n\n---\n**备注**：报告中的部分陈述已通过初步信息源核对，建议对关键数据进一步溯源。” return {“final_report”: final_report}

3.3 编排图结构与条件路由

现在，我们用StateGraph把这些节点和逻辑连接起来。

from langgraph.graph import StateGraph, END # 1. 创建图，并指定状态类型 workflow = StateGraph(ResearchState) # 2. 添加节点 workflow.add_node(“search”, search_node) workflow.add_node(“draft”, draft_node) workflow.add_node(“fact_check”, fact_check_node) workflow.add_node(“finalize”, finalize_node) # 3. 设置入口点 workflow.set_entry_point(“search”) # 4. 添加边，定义流程 workflow.add_edge(“search”, “draft”) # 搜索完后直接撰写 workflow.add_edge(“finalize”, END) # 生成最终报告后结束 # 5. 关键：添加条件边。在撰写草稿后，根据是否有待检查的事实来决定下一步。 def should_check_facts(state: ResearchState) -> str: “”“路由函数：判断是否需要进入事实检查环节。”“” if state[“facts_to_check”] and len(state[“facts_to_check”]) > 0: return “fact_check” else: return “finalize” workflow.add_conditional_edges( “draft”, # 源节点 should_check_facts, # 路由判断函数 {“fact_check”: “fact_check”, “finalize”: “finalize”} # 目标映射 ) workflow.add_edge(“fact_check”, “finalize”) # 检查完后去生成终稿 # 6. 编译图 app = workflow.compile()

3.4 运行与可视化

现在，我们可以运行这个研究助手了。

# 准备初始状态 initial_state = {“topic”: “2024年人工智能在医疗诊断领域的最新进展”, “messages”: []} # 运行图 final_state = app.invoke(initial_state, config={“recursion_limit”: 50}) print(“\n=== 研究完成 ===”) print(f”最终报告长度：{len(final_state[‘final_report’])} 字符”) print(“报告预览：”, final_state[‘final_report’][:500], “...”)

LangGraph还有一个非常强大的功能，就是可以将你定义的图可视化出来，这对于调试和理解复杂工作流至关重要。

# 将图导出为PNG图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法直接显示，可以保存到文件 app.get_graph().draw_mermaid_png(output_file_path=“research_workflow.png”)

生成的流程图会清晰地展示出从search到draft，然后根据条件分支到fact_check或finalize的完整路径，让你对智能体的决策流程一目了然。

4. 高级模式与实战技巧

4.1 人工监督与中断：Interrupt和Human-in-the-Loop

对于高风险或关键业务流程，让人类参与审核是刚需。LangGraph通过interrupt机制优雅地支持了这一点。

你可以在图中任何一个节点之后设置中断，将执行暂停，等待外部输入（比如用户在UI上点击“批准”或修改某些内容），然后再继续执行。

from langgraph.graph import StateGraph, END from langgraph.checkpoint import MemorySaver from langgraph.prebuilt import ToolNode, tools_condition # 使用检查点存储器，这是支持中断和恢复的基础 memory = MemorySaver() workflow = StateGraph(…, checkpointer=memory) # … 添加节点和边 … # 在某个节点后配置中断 workflow.add_node(“human_review”, human_review_node) # 假设这是一个等待人工输入的节点 workflow.add_edge(“some_node”, “human_review”) workflow.add_edge(“human_review”, “next_node”) # 当调用到`human_review`节点时，图的执行会暂停，并返回一个包含`waiting_for_input`标记的结果。 # 你的前端应用可以据此展示一个审核界面。当人工操作完成后，你再通过`app.update_state()`传入新的状态，并继续执行。

注意：中断功能强烈依赖于检查点（Checkpointer）。检查点不仅记录了当前状态，还记录了图的执行位置，使得暂停和恢复成为可能。在生产环境中，你需要将MemorySaver替换为如PostgresSaver或RedisSaver这样的持久化存储。

4.2 多智能体协作：MessagesState与子图

LangGraph非常适合构建多智能体系统，其中不同的智能体扮演不同角色，通过消息传递进行协作。MessagesState是一个预定义的状态类型，专门为基于消息的对话场景设计。

你可以为每个智能体定义一个子图（Subgraph），然后将这些子图作为主图的节点。主图负责在不同智能体之间路由消息。

from langgraph.graph import StateGraph, MessagesState from langgraph.prebuilt import create_react_agent # 定义状态，使用预建的MessagesState，它自动处理消息列表的追加 class MultiAgentState(MessagesState): current_agent: str # 记录当前该谁发言 # 创建两个不同专长的智能体 analyst_agent = create_react_agent(llm, tools=[data_analysis_tool]) writer_agent = create_react_agent(llm, tools=[writing_assistant_tool]) # 创建主图 workflow = StateGraph(MultiAgentState) # 将智能体子图添加为主图的节点 workflow.add_node(“analyst”, analyst_agent) workflow.add_node(“writer”, writer_agent) # 定义一个路由函数，根据消息内容或规则决定下一个发言的智能体 def router(state: MultiAgentState) -> str: last_message = state[“messages”][-1] if “数据分析” in last_message.content: return “analyst” else: return “writer” workflow.add_conditional_edges(“analyst”, router) workflow.add_conditional_edges(“writer”, router) workflow.set_entry_point(“analyst”) # 从分析师开始 app = workflow.compile()

在这个模式下，analyst和writer节点各自都是一个完整的、拥有推理和工具调用能力的智能体子图。主图像一个调度员，根据对话内容决定将任务交给谁，实现了智能体间的协同工作。

4.3 性能优化与检查点策略

当你的图变得复杂或需要长时间运行时，性能和资源管理就变得重要。

1. 选择性状态更新：在节点函数中，只返回你真正修改了的状态字段。这可以减少序列化和传递的数据量。LangGraph的状态合并机制非常高效，但明确返回{“updated_field”: new_value}仍然是最佳实践。

2. 检查点配置：检查点虽然强大，但频繁保存会影响性能。你可以在编译图时配置检查点策略。

   from langgraph.checkpoint import MemorySaver checkpointer = MemorySaver( serde=“json”, # 序列化方式 # at=… 可以配置在哪些节点之后自动保存检查点 ) app = workflow.compile(checkpointer=checkpointer)

   对于非常长的运行链，可以考虑只在关键的决策点或外部调用（如工具调用）之后保存检查点，而不是每一步都保存。

3. 异步支持：LangGraph原生支持异步节点函数。如果你的节点涉及大量I/O操作（如网络请求、数据库查询），使用async def定义节点函数，并用ainvoke调用图，可以显著提升吞吐量。

   async def async_search_node(state: State): result = await async_http_client.get(…) return {“data”: result} # 异步调用 final_state = await app.ainvoke(initial_state)

5. 常见问题与调试指南

在实际使用LangGraph的过程中，你可能会遇到一些典型问题。以下是一些排查思路和解决方案。

5.1 状态更新不符合预期

问题：节点修改了状态，但下游节点读取到的值还是旧的。排查：

• 首先确认你的状态字段注解是否正确。例如，对于列表追加，必须使用Annotated[List, add_messages]或operator.add。如果错误地使用了默认的覆盖方式，更新就会丢失。
• 在节点函数中多使用print或日志输出当前状态和返回的状态，确认更新逻辑。
• 检查是否有多个节点在并发修改同一个字段（虽然LangGraph默认是顺序执行，但在复杂路由下需理清顺序）。

5.2 图陷入无限循环

问题：智能体在一个循环里出不来，直到达到递归限制。排查：

• 检查条件边（Conditional Edge）的路由逻辑：这是最常见的循环来源。确保你的路由函数在所有可能的状态下都能最终导向END或一个不指向自身的节点。可以打印路由函数的输入和返回值来调试。
• 设置递归限制：在app.invoke()时传入config={“recursion_limit”: N}，这是一个安全网，防止无限循环耗尽资源。
• 可视化你的图：使用app.get_graph().draw_mermaid()生成图表，直观检查是否存在循环路径。确保每个循环都有退出条件。

5.3 工具调用失败或LLM响应格式错误

问题：在智能体节点中，工具调用抛出异常，或者LLM返回的内容无法被解析为工具调用。排查：

• 隔离测试工具：首先确保你的工具函数在LangGraph上下文之外能正常工作。
• 包装工具调用：在节点函数内部使用try…except包裹工具调用和LLM调用，并将错误信息捕获后写入状态，以便后续节点进行错误处理或重试。
• 优化Prompt：对于ReAct模式的智能体，LLM输出格式错误通常与Prompt指令不清晰有关。确保你的系统消息明确要求以Action:和Action Input:的格式输出。使用LangChain的ToolsAgent或create_react_agent等预建组件可以减少这类问题。

5.4 检查点恢复失败

问题：保存的检查点无法加载，或者恢复后状态不对。排查：

• 序列化兼容性：确保状态中所有对象都是可序列化的（如JSON或Pickle）。避免在状态中存储数据库连接、文件句柄等不可序列化的对象。复杂对象应存储其引用ID，在节点中重新初始化。
• 检查点键冲突：每个运行图实例的线程ID（thread_id）必须是唯一的。如果你在恢复时使用了重复的thread_id，可能会加载到错误的状态。确保你的thread_id管理逻辑正确。
• 存储后端一致性：如果你使用了外部数据库（如Postgres）作为检查点存储，确保图编译时使用的检查点序列化/反序列化配置与存储时一致。

5.5 调试与日志记录实战技巧

1. 使用print进行快速调试：在每个节点的开始和结束打印状态的关键字段，这是最直接的方法。
2. 利用langsmith进行追踪：如果你使用LangSmith，LangGraph的每一步执行都会自动生成详细的追踪记录，包括每个节点的输入输出、耗时、工具调用详情等，这是生产环境调试的利器。
3. 状态快照：在关键节点后，手动将状态字典保存到文件或日志系统，便于事后分析。

   import json def some_node(state): # … 业务逻辑 … with open(f”state_snapshot_{node_name}.json”, “w”) as f: json.dump(state, f, indent=2, default=str) # default=str处理不可序列化对象 return new_state

4. 图结构验证：在编译图之前，仔细检查边的连接，特别是条件边的目标映射字典，确保所有引用的节点名都已正确添加。

LangGraph将AI应用的开发从“链式思维”提升到了“图式思维”，为你处理复杂、有状态、多分支的工作流提供了强大而优雅的抽象。初学时会觉得概念稍多，但一旦理解了State、Node、Edge和Checkpoint这几个核心支柱，构建复杂智能体就会变得像画流程图一样直观。从简单的线性链升级到LangGraph，就像是给你的AI应用装上了规划和决策的“大脑”，让它能真正胜任那些需要记忆、思考和判断的复杂任务。