LangGraph实战：构建有状态多步骤AI工作流与智能体应用

1. 先搞清楚 LangGraph 到底解决了什么核心问题

如果你正在接触 LangChain，并且发现用 Chain 和 Agent 处理稍微复杂一点的业务流程时，代码会变得又长又乱，状态流转全靠自己手动维护，那么 LangGraph 就是你一直在找的那个东西。它不是一个全新的框架，而是 LangChain 生态里专门用来编排复杂、有状态、多步骤工作流的库。

简单来说，LangGraph 把智能体（Agent）或者任何多步骤的 LLM 应用，抽象成一个有向图（Graph）。图中的每个节点（Node）是一个执行单元（比如调用一次 LLM、执行一个工具、查询数据库），边（Edge）定义了节点之间的流转条件。这样一来，你就不用写一堆if-else和全局变量来管理“上一步做了什么”、“下一步该去哪”，而是用声明式的方式画出一个工作流蓝图，让框架帮你驱动执行。

它最核心的价值就两点：

1. 状态管理：内置了一个State对象，在整个图执行过程中流转，所有节点都能读写它。你不用再操心怎么把上一步的输出塞给下一步的输入。
2. 循环与条件分支：这是它比普通 Chain 强大的地方。你可以轻松实现“如果 LLM 返回的结果不满足要求，就循环回去重新处理”，或者“根据工具执行结果，决定走 A 分支还是 B 分支”。这让构建真正自主的、能应对复杂场景的智能体成为可能。

所以，这篇文章不是泛泛而谈概念，而是会直接带你从环境搭建开始，一步步构建一个能实际运行的智能体，并拆解其中每个关键参数和踩坑点。无论你是想用 LangGraph 做客服对话、数据分析流水线，还是复杂决策系统，这个从零到一的实战流程都是通用的。

2. 环境准备与核心概念对齐

在开始写代码之前，先把环境和核心概念对齐，能避免后面很多“为什么跑不起来”的问题。

2.1 基础环境搭建

LangGraph 是 Python 库，建议使用 Python 3.8 及以上版本。首先创建一个干净的虚拟环境，这是管理依赖的好习惯。

# 创建并激活虚拟环境（以 conda 为例） conda create -n langgraph-demo python=3.10 conda activate langgraph-demo # 安装核心库 pip install langgraph langchain-openai

这里注意几个关键点：

• langgraph是核心框架。
• langchain-openai是 LangChain 官方维护的 OpenAI 集成包，比旧的openai包兼容性更好。我们用它来调用大模型。
• 你很可能还需要其他工具包，比如langchain-community（包含很多社区工具和加载器）、tavily-python（网络搜索工具）等，可以根据项目需要后续安装。一开始保持环境简洁。

接下来是配置 API Key。你需要一个 OpenAI 的 API Key（或其他 LangGraph 支持模型的 Key）。永远不要把 Key 硬编码在代码里提交到版本库。

# 在终端中设置环境变量（Linux/macOS） export OPENAI_API_KEY='your-api-key-here' # 在 Windows PowerShell 中 $env:OPENAI_API_KEY='your-api-key-here'

更推荐的做法是使用.env文件配合python-dotenv管理。

2.2 理解 LangGraph 三要素

在动手前，脑子里要对下面三个核心概念有清晰的认识，后面写代码就是把这些概念实例化：

1. State（状态）：

   • 这是一个贯穿整个工作流的共享数据存储，类型通常是TypedDict或Pydantic BaseModel。
   • 它定义了工作流中需要传递的所有数据字段，比如messages（对话历史），intermediate_steps（工具调用结果），question（用户问题）等。
   • 关键：每个节点（Node）的函数，接收一个State对象，修改它，然后返回更新后的State。框架负责传递。

2. Node（节点）：

   • 工作流中的一个步骤，可以是一个普通函数。这个函数必须接收一个State对象作为参数，并返回一个更新后的State对象（或包含State的字典）。
   • 节点里可以干任何事：调用 LLM、执行工具、计算、查询数据库等。

3. Edge（边）：

   • 决定工作流下一步该执行哪个节点。分为两种：
     • 条件边（Conditional Edge）：根据State中的某个值，动态决定下一个节点。这是实现循环和分支的关键。
     • 普通边（Normal Edge）：无条件地指向下一个节点。
   • LangGraph 预定义了END和START两个特殊节点，分别代表结束和开始。

一个常见的误区：以为 LangGraph 是 LangChain 的替代品。其实不是。LangChain 是一个庞大的生态，提供了模型 I/O、检索、工具、链等基础组件。LangGraph 是 LangChain 生态中专门负责编排和有状态执行的那一部分。你可以只用 LangGraph 的图编排能力，配合其他模型 SDK；但通常，结合 LangChain 提供的丰富工具和链来构建节点，效率最高。

3. 实战：构建一个具备网络搜索能力的问答智能体

我们现在来构建一个经典的智能体：它能理解用户问题，自主决定是否需要联网搜索来获取最新信息，然后结合搜索到的信息给出最终回答。这个流程完美契合了 LangGraph 的“条件分支”和“多步骤”特性。

3.1 定义状态与工具

首先，定义智能体工作流中需要流转的状态。我们使用TypedDict。

from typing import TypedDict, List, Annotated import operator from langchain_core.messages import BaseMessage # 定义状态结构 class AgentState(TypedDict): # 对话消息历史，这是与LLM交互的核心 messages: Annotated[List[BaseMessage], operator.add] # 用户提出的原始问题 question: str # 是否需要执行搜索的标志位 should_search: bool

Annotated[List[BaseMessage], operator.add]这个写法是 LangGraph 的魔法。它告诉框架，对于messages这个字段，当多个节点返回的新messages列表时，不要替换，而是用operator.add（即列表的+操作）将它们合并起来。这确保了对话历史能不断累积。

接下来，准备工具。这里我们使用一个模拟的搜索工具，真实项目中可以换成 Tavily、SerperAPI 或自定义工具。

from langchain.tools import tool @tool def search_web(query: str) -> str: """当需要获取最新、实时或未知信息时，使用此工具进行网络搜索。""" # 这里是模拟返回。真实情况请接入搜索API。 print(f"[搜索工具被调用] 查询词: {query}") mock_results = { "今天北京的天气": "北京今天晴转多云，气温15-25摄氏度，南风2-3级。", "LangGraph是什么": "LangGraph 是 LangChain 项目的一部分，用于构建有状态、多智能体工作流的库。", "特斯拉最新股价": "根据模拟数据，特斯拉（TSLA）最新股价为 $175.32，上涨 2.1%。" } return mock_results.get(query, f"未找到关于 '{query}' 的明确信息。")

3.2 构建工作流图：节点与边

这是最核心的部分。我们将工作流拆解成几个节点，并连接它们。

from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage # 初始化大模型 llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) # 1. 路由节点：判断是否需要搜索 def router_node(state: AgentState) -> AgentState: """根据对话历史和问题，判断是否需要调用搜索工具。""" system_prompt = """你是一个助手。你需要判断用户的问题是否需要联网搜索最新信息才能回答。 例如，问“今天天气”、“最新新闻”、“实时股价”等需要搜索。 而问“你是谁”、“怎么编程”等通用知识或无需实时信息的问题，则不需要搜索。 只回答“需要”或“不需要”。""" # 构建判断用的消息 judge_messages = [ SystemMessage(content=system_prompt), HumanMessage(content=f"用户问题：{state['question']}") ] response = llm.invoke(judge_messages) decision = response.content.strip() # 更新状态 state["should_search"] = decision == "需要" # 也可以把判断过程记录到消息历史，这里我们选择不记录，保持主对话历史干净 return state # 2. 搜索节点：执行搜索并记录结果 def search_node(state: AgentState) -> AgentState: """执行搜索，并将搜索结果以系统消息格式加入历史。""" if not state["should_search"]: return state search_query = state["question"] search_result = search_web.invoke(search_query) # 将搜索结果作为一条“系统”或“工具”消息加入历史，供后续节点使用 search_result_msg = SystemMessage(content=f"[网络搜索结果]：{search_result}") state["messages"].append(search_result_msg) return state # 3. 回答节点：生成最终答案 def answer_node(state: AgentState) -> AgentState: """综合对话历史和可能有的搜索结果，生成最终答案。""" # 最终回答时，我们基于全部 messages 历史来生成 response = llm.invoke(state["messages"]) # 将AI的回答加入消息历史 state["messages"].append(response) return state # 开始构建图 builder = StateGraph(AgentState) # 添加节点 builder.add_node("router", router_node) builder.add_node("search", search_node) builder.add_node("answer", answer_node) # 设置入口点 builder.set_entry_point("router") # 添加边：router节点之后，根据 `should_search` 决定分支 def decide_next_node(state: AgentState): # 根据 router 节点设置的状态标志决定 if state["should_search"]: return "search" # 需要搜索，去 search 节点 else: return "answer" # 不需要搜索，直接去 answer 节点 builder.add_conditional_edges( "router", # 源节点 decide_next_node, # 条件判断函数 # 条件函数返回的字符串必须对应已添加的节点名或 `END` { "search": "search", "answer": "answer" } ) # 添加普通边：search 节点之后，必然去 answer 节点 builder.add_edge("search", "answer") # answer 节点之后，工作流结束 builder.add_edge("answer", END) # 编译图，得到可执行对象 graph = builder.compile()

3.3 运行与调试

现在，我们可以运行这个智能体了。

# 初始化一个状态 initial_state: AgentState = { "messages": [SystemMessage(content="你是一个有用的助手。")], # 初始系统提示 "question": "今天北京的天气怎么样？", "should_search": False # 初始值不重要，会被 router 节点覆盖 } # 执行图 final_state = graph.invoke(initial_state) # 查看最终的消息历史 for msg in final_state["messages"]: print(f"{msg.type}: {msg.content}")

运行后，你会看到类似这样的输出：

system: 你是一个有用的助手。 [搜索工具被调用] 查询词: 今天北京的天气怎么样？ system: [网络搜索结果]：北京今天晴转多云，气温15-25摄氏度，南风2-3级。 ai: 根据最新的网络搜索信息，北京今天（指您提问的当天）的天气是晴转多云，气温在15到25摄氏度之间，风力为南风2-3级。天气不错，适合外出活动。

关键调试点：

1. 检查状态流转：在router_node里打印decision变量，确认条件判断是否符合预期。
2. 检查工具调用：确保search_web工具被正确调用，并且返回的格式是字符串。
3. 检查消息历史：最终state[‘messages’]列表里消息的顺序和内容是否正确。错误的顺序可能导致 LLM 理解混乱。

4. 进阶：实现长期记忆与复杂循环

上面的例子是一个简单的二分支流程。但智能体常常需要更复杂的交互，比如多轮工具调用直到满足条件为止。这就需要用上 LangGraph 的循环能力。我们改造一下智能体，让它能处理“持续搜索直到信息足够”的场景。

4.1 改造状态与工具

我们引入一个search_count计数器和一个is_satisfied满意度标志。

class AdvancedAgentState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] question: str # 存储最近一次工具调用的结果 last_tool_result: str # 记录搜索次数 search_count: int # AI 判断信息是否已足够 is_satisfied: bool # 一个更“笨”的搜索工具，每次只返回一点信息 @tool def incremental_search(query: str) -> str: """模拟一个每次只返回部分信息的搜索。""" print(f"[增量搜索] 查询: {query}") # 模拟第一次和第二次搜索返回不同信息 if not hasattr(incremental_search, “call_count“): incremental_search.call_count = 0 incremental_search.call_count += 1 if incremental_search.call_count == 1: return “信息片段A：北京今日白天晴。“ elif incremental_search.call_count == 2: return “信息片段B：最高气温25度。“ else: return “信息片段C：无明显补充信息。“

4.2 构建带循环的工作流

这个工作流的核心思想是：LLM -> 判断是否满意 -> 不满意 -> 再搜索 -> 再判断，形成一个循环。

from langgraph.graph import StateGraph, END from langchain_core.messages import ToolMessage def llm_node(state: AdvancedAgentState) -> AdvancedAgentState: """LLM节点：分析当前信息，并决定下一步行动。""" # 构建给LLM的提示，要求它判断信息是否足够并可能提出新的搜索指令 prompt = f"""你正在回答这个问题：{state[‘question‘]} 目前你掌握的信息如下（来自之前的搜索）： {state.get(‘last_tool_result‘, ‘暂无信息‘)} 请判断：当前信息是否足够你给出一个完整、准确的回答？ 如果足够，请直接生成最终答案。 如果不够，请明确指出你还缺少哪部分信息。你的输出将驱动下一次搜索。""" human_msg = HumanMessage(content=prompt) state[“messages”].append(human_msg) response = llm.invoke(state[“messages”]) state[“messages”].append(response) # 记录LLM的思考 # 简单判断：如果回答中包含“足够”或答案看起来完整，则标记满意 # 更复杂的实现可以让LLM输出一个结构化字段（如 `need_more: bool`) answer_text = response.content if “足够” in answer_text or len(answer_text) > 50: # 简单启发式规则 state[“is_satisfied”] = True else: state[“is_satisfied”] = False # 可以把LLM指出的缺失信息存下来，作为下一次搜索的query state[“question”] = f“{state[‘question‘]} - 补充：{answer_text}” return state def tool_node(state: AdvancedAgentState) -> AdvancedAgentState: """工具节点：执行搜索。""" result = incremental_search.invoke(state[“question”]) state[“last_tool_result”] = result state[“search_count”] += 1 # 将工具执行结果以 ToolMessage 格式加入历史，这是标准做法 tool_msg = ToolMessage(content=result, tool_call_id=“mock_id”) state[“messages”].append(tool_msg) return state def final_answer_node(state: AdvancedAgentState) -> AdvancedAgentState: """最终回答节点（循环结束后调用）。""" # 基于完整的对话历史生成最终答案 final_prompt = “请基于所有对话历史，给出最终答案。” state[“messages”].append(HumanMessage(content=final_prompt)) final_response = llm.invoke(state[“messages”]) state[“messages”].append(final_response) return state # 构建图 builder = StateGraph(AdvancedAgentState) builder.add_node(“llm”, llm_node) builder.add_node(“tool”, tool_node) builder.add_node(“final”, final_answer_node) builder.set_entry_point(“llm”) # 关键：定义条件边，实现 LLM -> (工具 or 最终答案) 的循环 def decide_after_llm(state: AdvancedAgentState): if state[“is_satisfied”]: return “final” # 满意，去最终回答 else: if state[“search_count”] < 3: # 防止无限循环，设置上限 return “tool” # 不满意且未超限，去搜索 else: return “final” # 搜索次数太多，强制结束 builder.add_conditional_edges( “llm”, decide_after_llm, {“tool”: “tool”, “final”: “final”} ) # 工具执行完后，无条件返回 LLM 节点进行下一轮判断 builder.add_edge(“tool”, “llm”) # 最终回答后结束 builder.add_edge(“final”, END) advanced_graph = builder.compile()

运行这个图，你会看到智能体在llm和tool节点间循环，直到is_satisfied为 True 或搜索达到3次上限，然后跳转到final节点给出答案。这就是 LangGraph 处理多步、循环任务的核心模式。

5. 生产环境部署与关键踩坑点

当你把 Demo 跑通，准备投入实际项目时，下面这些点需要特别注意。

5.1 状态设计的陷阱

• 状态字段的更新语义：这是最容易出错的地方。回顾我们用的Annotated[List[BaseMessage], operator.add]。如果你定义了一个字段是str类型，那么后一个节点的返回值会覆盖前一个节点的值。你需要想清楚每个字段是“累积”还是“替换”。
• 状态不可变：在节点函数内部，不要直接修改传入的state字典（虽然Python里可能行得通）。最佳实践是创建一份更新后的字典返回。LangGraph 内部会处理合并。
• 状态序列化：如果你需要持久化工作流状态（比如暂停后恢复），确保State中所有字段都是可序列化的（如基本类型、列表、字典）。避免放入数据库连接、文件句柄等不可序列化对象。

5.2 错误处理与持久化

• 节点错误：单个节点执行出错（如工具调用超时、LLM API 异常）会导致整个图执行失败。在生产环境中，你需要考虑：
  • 在节点函数内部使用try...except进行局部错误处理，并返回一个代表错误的状态。
  • 利用 LangGraph 的interrupt机制，在特定节点后暂停图执行，便于人工干预或异步回调。
  • 使用checkpointer持久化状态。这是 LangGraph 的高级特性，可以将图的状态保存到数据库（如Redis、PostgreSQL），实现故障恢复和长时运行任务。

  from langgraph.checkpoint.sqlite import SqliteSaver memory = SqliteSaver.from_conn_string(“:memory:”) # 示例用内存，生产用实际DB graph = builder.compile(checkpointer=memory) # 调用时传入 configurable 参数来支持断点续跑 config = {“configurable”: {“thread_id”: “user_123_session_1”}} graph.invoke(initial_state, config=config)

5.3 性能与成本考量

• 控制循环次数：像上面的例子一样，必须设置循环上限（search_count < 3），防止因LLM判断失误或工具失效导致无限循环，消耗大量 Token 和 API 费用。
• 优化提示词：节点中给 LLM 的提示词要精准。在router_node或decide_after_llm中的判断提示词，直接决定了循环逻辑的效率和准确性。让 LLM 输出结构化 JSON（如{“reasoning”: “…”, “next_step”: “search”}）比解析自然语言更可靠。
• 并发与流式：对于高并发场景，LangGraph 本身是框架，并发能力取决于你的部署方式（如 FastAPI 后端）。考虑使用异步节点函数。对于需要实时响应的场景，可以探索 LangGraph 的流式输出支持，逐步返回结果。

5.4 调试与监控

• 可视化：LangGraph 提供了graph.get_graph().draw_mermaid_png()方法（需要安装pygraphviz），可以将你的工作流图生成图片，直观查看节点和边的关系，对于复杂流程排查问题非常有用。
• 日志记录：在每个节点的关键步骤（如调用 LLM 前、工具执行后）添加详细的日志，记录输入输出和状态变化。这对于追踪生产环境中的诡异问题至关重要。
• 跟踪与评估：集成 LangSmith（LangChain 官方的跟踪平台）或类似的 LLM 应用监控工具。它可以记录每次图执行的所有步骤、耗时、Token 使用和中间状态，是进行性能分析和效果评估的利器。

最后，也是最实在的建议：不要一开始就设计极其复杂的图。从一个能跑通的最小可行图开始，逐步添加节点和分支。每加一个功能，就完整测试一遍状态流转。LangGraph 的强大在于其清晰的抽象，但复杂度的管理责任在开发者自己身上。先把单条路径跑稳，再考虑并发、持久化和错误恢复这些生产级特性。