LangGraph实战：构建有状态多步骤AI智能体的完整指南-平芜编程栈

最近在尝试构建复杂的AI应用时，你是否遇到过这样的困境：单个大模型调用无法处理多步骤任务，不同工具和模型之间的状态流转混乱不堪，代码里充满了难以维护的if-else逻辑？这正是传统AI应用开发中普遍存在的痛点。LangGraph的出现，就是为了解决这类问题而生。它不是一个新的大模型，而是一个基于LangChain的框架，专门用于构建有状态、多步骤的AI智能体（Agent）。本文将带你从零开始，彻底掌握LangGraph的核心概念、设计模式与实战应用，无论你是想入门AI应用开发，还是希望将现有单点AI能力升级为自动化工作流，都能在这里找到完整的答案。

1. LangGraph是什么？为什么你需要它？

在深入代码之前，我们必须先理解LangGraph要解决的核心问题。简单来说，LangGraph是一个用于构建有状态、多步骤工作流的库。它将你的AI应用建模为一个“图”（Graph），图中的节点（Node）代表一个执行单元（如调用一次大模型、执行一个工具），边（Edge）则定义了节点之间的流转逻辑。

1.1 从LangChain到LangGraph：思维的转变

你可能已经熟悉LangChain，它通过Chain将多个组件（如LLM、Prompt、Output Parser）链接起来。但Chain本质上是线性的、无状态的。对于“根据用户问题搜索网络，然后分析结果，最后生成报告并保存”这样的复杂场景，线性Chain就显得力不从心。

LangGraph引入了“图”和“状态”的概念：

图（Graph）：描述了应用的整体流程和决策路径。
状态（State）：在整个工作流执行过程中持久化并传递的数据。这是LangGraph的灵魂，它使得工作流可以记住之前步骤的结果，并根据这些结果决定下一步走向。

1.2 核心应用场景：AI智能体（Agent）

LangGraph最典型的应用就是构建智能体。一个智能体通常包含：

规划（Planning）：理解目标，拆解步骤。
执行（Action）：调用工具（如搜索、计算、查数据库）或大模型。
观察（Observation）：获取工具执行结果。
循环（Loop）：根据观察结果，决定是继续执行还是结束。

这个经典的“规划-执行-观察”循环，用LangGraph可以非常优雅地实现。此外，它还非常适合构建需要分支、循环、并行或人工审核的复杂业务流程。

2. 环境准备与核心概念

在动手之前，我们需要搭建好开发环境，并厘清几个最关键的概念。

2.1 环境搭建

本文示例使用Python环境。请确保已安装Python（建议3.8+版本）。

首先，安装必要的库。LangGraph通常与LangChain配合使用，并需要一个LLM提供商（如OpenAI）。

pip install langgraph langchain langchain-openai

如果你使用其他LLM（如通义千问、DeepSeek等），请安装对应的LangChain集成包。

2.2 理解核心抽象：State, Node, Edge

这是学习LangGraph必须跨越的三个门槛。

State（状态）：
- 这是一个贯穿整个图执行过程的字典。
- 你需要预先定义这个字典的“结构”（Schema），即有哪些字段，分别是什么类型。
- 每个节点都可以读取和修改State中的字段。
- 例如，一个聊天机器人的State可能包含：messages（对话历史），user_query（当前问题），search_results（中间搜索结果）。
Node（节点）：
- 图中的一个功能单元。它本质上是一个函数，接收当前的State，执行一些操作（如调用LLM），然后返回一个包含更新后State字段的字典。
- 例如，一个“调用搜索工具”的节点，会读取State中的user_query，调用搜索引擎API，然后将结果写入State的search_results字段。
Edge（边）：
- 决定执行完一个节点后，下一步该去哪个节点。
- 分为两种：
  - 条件边（Conditional Edge）：根据State中的某个值（通常是LLM的判断）来决定下一个节点。这是实现分支逻辑的关键。
  - 普通边（Normal Edge）：无条件地指向下一个节点。

理解了这三个概念，LangGraph的整个设计思路就清晰了：定义好数据容器（State），安排好处理单元（Node），并用逻辑线（Edge）把它们连接起来。

3. 第一个LangGraph应用：智能对话助手

让我们通过一个最简单的例子，将上述概念具象化。我们将构建一个能根据用户问题决定是否要联网搜索的对话助手。

3.1 定义状态（State）

首先，我们定义工作流需要维护哪些数据。这里使用TypedDict来获得更好的类型提示。

from typing import TypedDict, List, Annotated import operator from langchain_core.messages import HumanMessage, AIMessage class AgentState(TypedDict): # 对话消息历史 messages: Annotated[List[HumanMessage | AIMessage], operator.add] # 是否需要搜索（由LLM判断） needs_search: bool # 搜索结果 search_result: str

messages：使用Annotated和operator.add是LangGraph的一个约定，表示这个字段是“追加”模式。每次节点返回的messages列表都会自动追加到历史中，而不是覆盖。
needs_search：一个布尔标志。
search_result：存储搜索到的文本。

3.2 创建节点（Nodes）

我们需要三个节点：router（路由判断）、search_node（搜索）、general_chat_node（普通聊天）。

from langchain_openai import ChatOpenAI from langgraph.graph import END, StateGraph # 初始化LLM llm = ChatOpenAI(model="gpt-3.5-turbo") def router(state: AgentState) -> dict: """判断用户问题是否需要联网搜索""" last_message = state[“messages”][-1].content # 构建一个Prompt让LLM判断 prompt = f”用户的问题是：‘{last_message}’。这个问题需要实时或最新的网络信息才能准确回答吗？只需要回答‘是’或‘否’。” response = llm.invoke(prompt) decision = response.content.strip() needs_search = decision == “是” return {“needs_search”: needs_search} def search_node(state: AgentState) -> dict: """模拟搜索节点（实际项目中替换为真实搜索API调用）""" query = state[“messages”][-1].content # 这里模拟搜索过程 print(f”[搜索节点] 正在搜索: {query}”) # 模拟返回搜索结果 mock_result = f”关于‘{query}’的模拟搜索结果：当前最新版本是v2.0，发布于2023年底。” return {“search_result”: mock_result, “messages”: [AIMessage(content=f”我已搜索到信息：{mock_result}”)]} def general_chat_node(state: AgentState) -> dict: """普通聊天节点，直接回答""" last_message = state[“messages”][-1] response = llm.invoke([last_message]) return {“messages”: [response]}

3.3 构建图（Graph）并连接边（Edges）

现在，我们把节点组装起来，并定义它们之间的流转逻辑。

from langgraph.graph import StateGraph, END # 1. 创建图构建器，并指定状态结构 workflow = StateGraph(AgentState) # 2. 添加节点 workflow.add_node(“router”, router) workflow.add_node(“search”, search_node) workflow.add_node(“general_chat”, general_chat_node) # 3. 设置入口点：所有流程都从`router`开始 workflow.set_entry_point(“router”) # 4. 添加条件边：根据`router`的结果决定下一步 def decide_next_step(state: AgentState) -> str: if state.get(“needs_search”): return “search” # 需要搜索，则前往`search`节点 else: return “general_chat” # 否则，直接聊天 workflow.add_conditional_edges( “router”, # 从哪个节点出发 decide_next_step, # 决定下一个节点的函数 {“search”: “search”, “general_chat”: “general_chat”} # 可能的目的地映射 ) # 5. 添加普通边：`search`节点执行完后，应进入`general_chat`节点来总结回答 workflow.add_edge(“search”, “general_chat”) # 6. 添加普通边：`general_chat`节点执行完后，工作流结束 workflow.add_edge(“general_chat”, END) # 7. 编译图，得到可执行对象 app = workflow.compile()

3.4 运行与测试

现在，我们可以运行这个图了。

# 初始化状态 initial_state = AgentState( messages=[HumanMessage(content=“LangGraph的最新版本是什么？”)], needs_search=False, search_result=“” ) # 执行图 final_state = app.invoke(initial_state) # 查看最终结果 for msg in final_state[“messages”]: print(f”{msg.type}: {msg.content}”)

预期输出：

[搜索节点] 正在搜索: LangGraph的最新版本是什么？ ai: 我已搜索到信息：关于‘LangGraph的最新版本是什么？’的模拟搜索结果：当前最新版本是v2.0，发布于2023年底。 ai: （这里会是general_chat节点基于搜索结果生成的最终友好回答）

这个简单的例子展示了LangGraph的核心流程：状态流转、节点执行和条件分支。router节点修改了needs_search状态，条件边根据这个状态值决定路径，最终所有结果都汇聚到messages中。

4. 构建高级智能体：ReAct模式实战

ReAct（Reasoning + Acting）是智能体的经典模式，它让LLM循环地进行“思考-行动-观察”。用LangGraph实现ReAct智能体非常直观。

4.1 设计状态与工具

我们将创建一个能使用计算器和搜索工具的智能体。

from typing import TypedDict, Annotated, List, Union import operator from langchain_core.messages import BaseMessage from langchain.tools import tool from langchain_community.tools.tavily_search import TavilySearchResults # 定义工具 @tool def calculator(expression: str) -> str: “”“计算一个数学表达式。例如：‘3 * 7 + 5’“”“ try: # 警告：实际项目中请使用更安全的eval替代方案，如ast.literal_eval或数学库 result = eval(expression) return f”计算结果: {result}” except Exception as e: return f”计算错误: {e}” # 假设我们有一个搜索工具（需要API Key） # search_tool = TavilySearchResults() # 为演示，我们创建一个模拟工具 @tool def web_search(query: str) -> str: “”“在网络上搜索信息。”“” return f”关于‘{query}’的模拟搜索结果：这是一个模拟的搜索结果摘要。” # 可用工具列表 tools = [calculator, web_search] # 定义状态 class ReActState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 用于记录LLM的“思考”过程 scratchpad: Annotated[List[BaseMessage], operator.add] # 当前轮次，防止无限循环 step: int

4.2 创建核心节点：Agent和Tools

在ReAct图中，通常有两个主要节点：agent（负责思考并决定使用哪个工具）和tools（执行工具）。

from langchain.agents import create_react_agent from langchain_core.prompts import ChatPromptTemplate # 1. 创建Agent节点 def agent_node(state: ReActState): # 构建Prompt，包含指令、工具描述、历史（scratchpad）和当前问题 prompt = ChatPromptTemplate.from_messages([ (“system”, “你是一个有帮助的助手，可以使用工具。请严格按以下格式回应：\n思考：<你的推理过程>\n行动：<要使用的工具名>\n行动输入：<工具的输入>”), (“placeholder”, “{chat_history}”), # 这里会填入scratchpad (“human”, “{input}”) ]) # 将工具绑定到LLM llm_with_tools = llm.bind_tools(tools) chain = prompt | llm_with_tools # 调用LLM agent_response = chain.invoke({ “chat_history”: state[“scratchpad”], “input”: state[“messages”][-1].content }) # 将LLM的响应记录到scratchpad中 return { “scratchpad”: [agent_response], “step”: state[“step”] + 1 } # 2. 创建Tools节点 def tools_node(state: ReActState): # 从scratchpad中取出最新的LLM响应 last_message = state[“scratchpad”][-1] # 解析LLM响应，提取“行动”和“行动输入” # 这里简化处理，实际应使用更健壮的解析器 content = last_message.content action_line = [l for l in content.split(‘\n’) if l.startswith(‘行动：’)][0] action_input_line = [l for l in content.split(‘\n’) if l.startswith(‘行动输入：’)][0] tool_name = action_line.replace(‘行动：’, ‘’).strip() tool_input = action_input_line.replace(‘行动输入：’, ‘’).strip() # 找到对应的工具并执行 tool_to_use = next((t for t in tools if t.name == tool_name), None) if tool_to_use: tool_result = tool_to_use.invoke(tool_input) result_msg = AIMessage(content=f”观察：{tool_result}”) else: result_msg = AIMessage(content=f”观察：工具‘{tool_name}’未找到。”) # 将工具执行结果（观察）记录到scratchpad return {“scratchpad”: [result_msg]}

4.3 构建并运行ReAct图

现在构建一个循环，让Agent和Tools交替执行，直到LLM决定结束。

from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver # 创建图 workflow = StateGraph(ReActState) workflow.add_node(“agent”, agent_node) workflow.add_node(“tools”, tools_node) workflow.set_entry_point(“agent”) # 定义边逻辑：Agent之后总是去Tools workflow.add_edge(“agent”, “tools”) # 关键：从Tools节点出来的条件边，决定是继续循环还是结束 def should_continue(state: ReActState) -> str: last_message = state[“scratchpad”][-1] # 如果LLM在思考中说了“最终答案：”，或者步骤超过5步，则结束 if “最终答案：” in last_message.content or state[“step”] > 5: return “end” else: return “continue” workflow.add_conditional_edges( “tools”, should_continue, {“continue”: “agent”, “end”: END} # 继续循环回Agent，或结束 ) # 编译图 app = workflow.compile()

4.4 测试ReAct智能体

# 使用检查点存储器，便于调试和观察中间状态 memory = MemorySaver() app_with_memory = workflow.compile(checkpointer=memory) config = {“configurable”: {“thread_id”: “test_react_1”}} initial_state = ReActState( messages=[HumanMessage(content=“请先计算15的平方，然后搜索一下LangGraph的作者信息。”)], scratchpad=[], step=0 ) # 以流式方式执行，可以看到每一步的输出 for event in app_with_memory.stream(initial_state, config, stream_mode=“values”): event[“scratchpad”][-1].pretty_print()

这个ReAct智能体会先调用计算器工具得到225，然后调用搜索工具获取信息，最后LLM综合两者给出最终答案。通过scratchpad，我们可以清晰地看到LLM“思考-行动-观察”的完整链条。

5. 高级特性与工程化实践

掌握了基础构建后，我们来看一些让LangGraph应用更健壮、更强大的高级特性。

5.1 持久化与检查点（Checkpoints）

对于长时间运行或需要中断恢复的智能体，检查点功能至关重要。它能在每个节点执行后自动保存完整状态。

from langgraph.checkpoint.sqlite import SqliteSaver import tempfile import os # 使用SQLite持久化检查点 db_file = tempfile.NamedTemporaryFile(delete=False).name checkpointer = SqliteSaver.from_conn_string(f”sqlite:///{db_file}”) # 在编译图时传入checkpointer persistent_app = workflow.compile(checkpointer=checkpointer) config = {“configurable”: {“thread_id”: “user_session_123”}} try: # 第一次执行 state1 = persistent_app.invoke(initial_state, config) print(“第一次执行完成。”) # 模拟应用重启后，可以从检查点恢复，继续执行 # 通过指定`config`中的相同thread_id，可以加载历史状态 # 这对于实现多轮对话、异步任务恢复非常有用 finally: os.unlink(db_file) # 清理临时文件

5.2 人工干预节点（Human-in-the-Loop）

在某些关键业务流程中，需要引入人工审核。LangGraph可以轻松地将“等待人工输入”作为一个节点。

def human_review_node(state: AgentState): """模拟人工审核节点。在实际应用中，这里会连接到一个Web界面或通知系统。""" task_description = state[“messages”][-1].content print(f”\n=== 需要人工审核 ===\n任务：{task_description}\n==================”) # 在实际系统中，这里会阻塞等待用户输入。 # 为演示，我们模拟用户输入“批准” human_decision = “批准” # 模拟输入 return {“human_decision”: human_decision} # 在图中加入人工审核节点，并设置条件边，只有审核通过才进入下一步。

5.3 子图（Subgraphs）与模块化

复杂的应用可以拆分成多个子图，提高可维护性和复用性。子图本身也是一个StateGraph，可以被主图当作一个节点来调用。

# 定义一个处理“数据查询”的子图 subgraph_builder = StateGraph(AgentState) # ... 构建子图内部的节点和边 ... subgraph = subgraph_builder.compile() # 在主图中，可以将子图作为一个节点添加 main_workflow.add_node(“data_query_subgraph”, subgraph)

这种方式非常适合微服务架构，每个团队可以负责开发一个独立的子图。

6. 常见问题与调试技巧

在开发LangGraph应用时，你可能会遇到以下典型问题。

6.1 状态（State）更新不符合预期

这是最常见的问题。请牢记：

追加（Add） vs 覆盖（Replace）：使用Annotated[List[...], operator.add]的字段是追加，否则是覆盖。确保你的节点返回的字典键值与State定义匹配，且更新逻辑符合预期。
调试方法：在每个节点函数开始和结束时打印State。或者使用LangGraph的stream模式观察状态变化。

6.2 图陷入无限循环

智能体可能无法自行终止。解决方案：

设置最大步数：在State中设置一个step计数器，在条件边判断函数中检查是否超过阈值。
优化Prompt：明确指示LLM在何时输出“最终答案：”等终止标记。
使用超时机制：在图编译或调用时设置超时。

6.3 工具调用解析失败

LLM可能不按你指定的格式返回工具调用信息。

使用LangChain的bind_tools和with_structured_output：它们能强制LLM返回结构化的工具调用对象，比解析文本更可靠。
在Prompt中提供更清晰的示例：在系统提示词中给出2-3个完美的响应格式示例。

6.4 性能优化

对于复杂图，性能可能成为瓶颈。

异步执行：如果节点间没有严格的先后依赖，可以考虑使用add_node的异步版本，并利用asyncio。
缓存：对于纯函数节点或昂贵的LLM调用，可以考虑引入缓存机制。
精简State：只保留必要的数据在State中，避免传递过大的对象。

7. 最佳实践与项目架构建议

将LangGraph用于实际项目时，遵循以下实践能避免很多坑。

7.1 状态设计原则

最小化： State只存储工作流真正需要流转和共享的数据。中间计算结果尽量放在节点局部变量中。
明确类型：坚持使用TypedDict或Pydantic模型来定义State，这能极大提高代码可读性和减少运行时错误。
区分会话与流程：将会话ID、用户ID等元数据放在config中，而不是State里。State应专注于业务流程数据。

7.2 节点设计原则

单一职责：每个节点只做一件事。一个节点要么调用LLM，要么调用一个工具，要么处理数据。
幂等性：尽可能让节点函数是幂等的，即相同输入产生相同输出，这有利于调试和重试。
错误处理：节点内部应做好异常捕获，并选择是将错误信息放入State向下传递，还是直接抛出异常终止图。

7.3 图的版本管理与测试

版本化：当图的结构（节点、边）发生变化时，应视为一次版本升级。考虑如何平滑迁移正在运行中的旧图实例的状态。
单元测试：为每个节点函数编写单元测试，模拟输入State，验证输出State。
集成测试：为整个图编写集成测试，使用典型输入验证最终输出和状态变化。

7.4 生产环境部署

可观测性：在每个节点的入口和出口记录日志，包括State的快照。这对于排查生产问题至关重要。
监控与告警：监控图的执行时长、错误率、循环次数等指标。
配置化：将LLM的模型名称、API Key、工具开关等配置外置，便于不同环境（开发、测试、生产）的切换。

LangGraph代表了一种构建复杂AI应用的新范式，它将流程控制与AI能力解耦，通过声明式的“图”来管理复杂度。从简单的条件分支对话助手到拥有复杂推理循环的ReAct智能体，其核心思想始终是状态驱动和流程可视化。学习LangGraph的最佳方式，就是从一个具体需求开始，画出你的工作流程图，然后将其转化为节点和边。随着你对状态管理和条件流转越来越熟悉，你会发现构建强大、可靠且可维护的AI应用不再是一件令人畏惧的事情。

LangGraph实战：构建有状态多步骤AI智能体的完整指南