在实际 AI 应用开发中，构建一个能自主决策、调用工具并管理复杂流程的智能体（Agent）是许多开发者的目标。LangChain 提供了基础的 Agent 构建模块，但当任务流程变得复杂、需要状态管理或涉及多步骤循环时，仅靠 LangChain 的 Agent 接口会显得力不从心。这时，LangGraph 作为一个基于有向图（Graph）的工作流编排框架，其价值就凸显出来了。它允许你将 Agent 的逻辑清晰地定义为节点和边，从而构建出具备长期记忆、条件分支和循环的动态工作流。

本文面向已经了解 LangChain 基础概念，希望将智能体能力从简单的“一问一答”升级到复杂、可编排流程的开发者。我们将通过一个完整的实战案例，带你理解 LangGraph 的核心概念，并一步步构建一个具备动态决策能力的智能体工作流。你将学会如何定义状态、创建节点、编排流程，并最终运行一个可以处理多轮交互、根据结果调整策略的智能体系统。

1. 理解 LangGraph：从链到图的智能体进化

在深入代码之前，必须厘清 LangGraph 与 LangChain 的关系及其解决的问题。LangChain 是一个用于构建由语言模型驱动的应用程序的框架，其Agent和AgentExecutor是早期实现智能体逻辑的核心。它们通常遵循“思考-行动-观察”的循环，但这个循环是隐式且相对固定的，状态管理较为简单。

而 LangGraph 是 LangChain 生态系统中的一个库，它引入了“图”这一抽象。在图论中，图由节点（Node）和边（Edge）组成。在 LangGraph 的语境下：

• 节点（Node）：通常是一个可调用的函数或工具，负责执行一项具体任务，例如调用语言模型、执行搜索、更新数据库。
• 边（Edge）：定义了节点之间的流转条件。它决定了在当前节点执行完毕后，下一步应该前往哪个节点。边可以是固定的，也可以是基于当前状态的动态条件。
• 状态（State）：一个贯穿整个工作流的共享数据结构。每个节点读取并修改这个状态，状态的变化驱动着工作流的走向。

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

1. 显式流程控制：工作流的每一步都清晰可见，不再是黑盒循环，便于调试和理解。
2. 复杂逻辑编排：轻松实现条件分支（if-else）、循环（while）、并行等复杂逻辑。
3. 持久化状态（长期记忆）：图的状态可以被持久化到数据库，从而实现跨会话的记忆，这是构建具有“长期记忆”智能体的基础。
4. 人类介入点：可以在特定节点设置“中断”，等待人工审核或输入，这对于关键业务流程至关重要。

简单来说，如果你需要构建的智能体只是回答一个问题并调用一次工具，LangChain Agent 可能就够了。但如果你需要构建一个能处理多轮对话、根据中间结果选择不同路径、甚至需要记住之前交互内容的智能体，LangGraph 是更强大的工具。它不是一个替代品，而是一个在复杂场景下的增强和演进。

2. 环境准备与核心依赖配置

开始构建前，我们需要一个干净的 Python 环境。建议使用 Python 3.9 或更高版本。

2.1 创建虚拟环境与安装依赖

首先，创建一个新的项目目录并初始化虚拟环境。

mkdir langgraph-agent-demo cd langgraph-agent-demo python -m venv venv

激活虚拟环境：

• Windows:venv\Scripts\activate
• macOS/Linux:source venv/bin/activate

接下来，安装核心依赖。我们将使用 LangChain 和 LangGraph 的最新稳定版，并选择 OpenAI 作为语言模型后端。同时，为了示例需要，我们安装一个用于网页内容提取的工具langchain-community。

pip install langgraph langchain langchain-openai langchain-community

注意：langchain是一个元包，通常安装它和特定版本的集成包（如langchain-openai）是标准做法。langgraph是独立包，专门用于图工作流。

2.2 配置 API 密钥

本示例使用 OpenAI 的 GPT 模型。你需要在环境变量中设置你的 API 密钥。

# 在命令行中临时设置（仅当前会话有效） export OPENAI_API_KEY='your-api-key-here'

或者在 Python 代码中直接设置（不推荐用于生产环境）：

import os os.environ["OPENAI_API_KEY"] = "your-api-key-here"

2.3 验证安装与基础导入

创建一个名为demo_setup.py的脚本，验证环境是否就绪。

# demo_setup.py from langchain_openai import ChatOpenAI from langgraph.graph import StateGraph, END print("LangChain OpenAI 客户端导入成功。") print("LangGraph StateGraph 导入成功。") # 简单测试 LLM 连接（可选，会消耗 token） try: llm = ChatOpenAI(model="gpt-3.5-turbo") # 不实际调用，仅检查初始化 print(f"LLM 模型初始化成功: {llm.model_name}") except Exception as e: print(f"初始化失败: {e}")

运行该脚本，确保没有报错：

python demo_setup.py

3. 构建第一个 LangGraph 智能体：天气查询工作流

我们将构建一个简单的智能体，它接收用户关于天气的查询，决定是否需要查询具体城市，然后模拟调用一个天气工具并返回结果。这个例子包含了状态定义、节点、边和条件路由。

3.1 定义工作流状态

状态是 LangGraph 的核心，它是一个 TypedDict，定义了工作流中所有需要传递和修改的数据。我们使用typing中的TypedDict和Annotated，以及langgraph.graph的add_messages来实现消息的累加。

# weather_agent.py from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage import operator # 1. 定义状态结构 class AgentState(TypedDict): # 用户输入的问题 input: str # 从输入中解析出的城市（可能为空） city: str # 累积的对话消息历史，用于给LLM提供上下文 messages: Annotated[list, operator.add] # 工作流的最终输出 output: str

这里的关键是messages字段。Annotated[list, operator.add]是一个 LangGraph 的注解，它告诉框架在节点间传递状态时，对于messages列表，采用“追加”（add）的方式合并，而不是覆盖。这完美契合了对话历史累积的需求。

3.2 创建节点函数

节点是普通的 Python 函数，它接收当前状态，执行操作，并返回一个包含更新后状态字段的字典。

我们先创建第一个节点：解析节点。它的职责是分析用户输入，判断是否包含城市信息。

# weather_agent.py (续) llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) def parse_input(state: AgentState) -> dict: """解析用户输入，提取城市信息。""" user_input = state["input"] messages = [HumanMessage(content=f"请从以下用户问题中提取城市名。如果问题不涉及具体城市或无法提取，请回答‘无’。问题：{user_input}")] response = llm.invoke(messages) city = response.content.strip() # 更新状态 return {"city": city, "messages": state["messages"] + [HumanMessage(content=user_input), response]}

接着创建第二个节点：查询天气节点。它模拟调用一个天气 API。

# weather_agent.py (续) def query_weather(state: AgentState) -> dict: """模拟查询天气。""" city = state["city"] # 这里模拟一个固定的API响应，真实场景会调用如OpenWeatherMap的API weather_info = f"{city}的天气：晴，温度 22°C，湿度 65%。" ai_msg = AIMessage(content=weather_info) return {"output": weather_info, "messages": state["messages"] + [ai_msg]}

然后创建第三个节点：请求澄清节点。当城市信息缺失时，请求用户澄清。

# weather_agent.py (续) def ask_for_city(state: AgentState) -> dict: """向用户请求城市信息。""" clarification = "您想查询哪个城市的天气呢？" ai_msg = AIMessage(content=clarification) # 注意：在真实交互中，这里可能需要暂停工作流等待用户输入。 # 本例中我们简化处理，直接输出提示。 return {"output": clarification, "messages": state["messages"] + [ai_msg]}

3.3 定义条件路由边

我们需要一个路由函数，根据parse_input节点的结果，决定下一步是去查询天气还是请求澄清。

# weather_agent.py (续) def route_after_parse(state: AgentState) -> Literal["query_weather", "ask_for_city"]: """根据解析出的城市信息决定路由。""" city = state.get("city", "") # 如果城市信息有效且不是“无”，则去查询天气 if city and city != "无" and len(city) < 20: # 简单长度过滤，防止LLM胡言乱语 return "query_weather" else: return "ask_for_city"

3.4 组装工作流图

现在，我们将节点和边组合成一个完整的工作流。

# weather_agent.py (续) # 1. 创建图构建器 workflow = StateGraph(AgentState) # 2. 添加节点 workflow.add_node("parse_input", parse_input) workflow.add_node("query_weather", query_weather) workflow.add_node("ask_for_city", ask_for_city) # 3. 设置入口点 workflow.set_entry_point("parse_input") # 4. 添加条件边 workflow.add_conditional_edges( "parse_input", route_after_parse, # 路由判断函数 { "query_weather": "query_weather", "ask_for_city": "ask_for_city" } ) # 5. 添加普通边（从查询天气或请求澄清到结束） workflow.add_edge("query_weather", END) workflow.add_edge("ask_for_city", END) # 6. 编译图 app = workflow.compile()

3.5 运行与验证工作流

编译后的app就是一个可执行的工作流。我们创建两个测试用例来验证其行为。

# weather_agent.py (续) if __name__ == "__main__": # 测试用例1：包含城市信息 print("=== 测试1：查询‘北京天气怎么样？’ ===") initial_state1 = {"input": "北京天气怎么样？", "city": "", "messages": [], "output": ""} result1 = app.invoke(initial_state1) print(f"最终输出: {result1['output']}") print(f"最终城市: {result1['city']}") print("---\n") # 测试用例2：不包含城市信息 print("=== 测试2：查询‘今天会下雨吗？’ ===") initial_state2 = {"input": "今天会下雨吗？", "city": "", "messages": [], "output": ""} result2 = app.invoke(initial_state2) print(f"最终输出: {result2['output']}") print(f"最终城市: {result2['city']}") # 可选：查看完整的状态演变 # import pprint # pprint.pprint(result2)

保存并运行这个文件：

python weather_agent.py

你应该看到类似以下的输出：

=== 测试1：查询‘北京天气怎么样？’ === 最终输出: 北京的天气：晴，温度 22°C，湿度 65%。 最终城市: 北京 --- === 测试2：查询‘今天会下雨吗？’ === 最终输出: 您想查询哪个城市的天气呢？ 最终城市: 无

这个简单的例子展示了 LangGraph 的核心流程：定义状态、创建节点、通过条件边控制流程。parse_input节点后的条件路由，实现了智能体的首次决策。

4. 进阶实战：构建具备工具调用与循环的 Research Agent

现在我们来构建一个更复杂的智能体，它模拟一个研究助手，能够根据用户主题，自动决定是否需要联网搜索，并整合信息生成报告。这个例子将引入工具调用、循环（直到满足条件才结束）以及更复杂的状态管理。

4.1 定义进阶状态与工具

我们首先定义一个更丰富的状态，并创建一个模拟的网页搜索工具。

# research_agent.py from typing import TypedDict, Annotated, Literal, List from langgraph.graph import StateGraph, END, START from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_core.tools import tool import operator import json # 1. 定义进阶状态 class ResearchState(TypedDict): # 用户原始查询 query: str # 累积的消息历史 messages: Annotated[List, operator.add] # 已收集的研究资料 research_materials: List[str] # 当前轮次（防止无限循环） iteration: int # 最终报告 report: str # 标记是否需要继续搜索 should_continue: bool # 2. 创建一个模拟的搜索工具 @tool def web_search_tool(query: str) -> str: """ 模拟网页搜索工具。根据查询返回模拟的搜索结果。 在实际项目中，这里应替换为真实的 SerperAPI、Tavily 或 Google Search API 调用。 """ # 这是一个模拟响应库 mock_responses = { "LangGraph 是什么": "LangGraph 是 LangChain 的一个库，用于构建基于有向图的工作流，特别适合多步骤、有状态的智能体应用。它提供了状态管理和条件路由。", "LangGraph 长期记忆": "LangGraph 通过将图的状态持久化到数据库（如 PostgreSQL、Redis）来实现长期记忆，使得智能体可以跨会话记住上下文。", "智能体开发框架对比": "主流智能体框架包括 LangChain Agent、LangGraph、AutoGen、CrewAI。LangGraph 在复杂工作流编排和状态管理上优势明显。", "天气": "这是一个通用词汇，没有具体信息。" } # 返回匹配的模拟结果，若无匹配则返回默认信息 for key, value in mock_responses.items(): if key.lower() in query.lower(): return f"模拟搜索到信息：{value}" return f"模拟搜索到信息：关于‘{query}’，未找到高度相关的特定资料。"

4.2 创建决策与执行节点

这个工作流将包含三个核心节点：

1. 规划节点：分析当前查询和已有材料，决定下一步是搜索还是生成报告。
2. 搜索节点：调用搜索工具收集信息。
3. 报告节点：整合所有材料，生成最终报告。

# research_agent.py (续) llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0) tools = [web_search_tool] llm_with_tools = llm.bind_tools(tools) def planner_node(state: ResearchState) -> dict: """规划节点：决定下一步行动。""" messages = state["messages"] # 构建给LLM的提示，包含历史和研究材料 prompt = f""" 你是一个研究助手。当前用户查询是：{state['query']} 目前已收集的研究材料有：{state['research_materials']} 当前是第 {state['iteration']} 轮分析。 请决定下一步行动： 1. 如果已有材料足以回答用户查询，或者迭代次数超过3次，则选择 `generate_report`。 2. 如果还需要更多信息，则选择 `search_web`。 只返回 `generate_report` 或 `search_web` 中的一个词。 """ decision_msg = HumanMessage(content=prompt) response = llm.invoke([decision_msg]) decision = response.content.strip().lower() should_continue = decision == "search_web" and state["iteration"] < 3 update = { "messages": state["messages"] + [decision_msg, response], "should_continue": should_continue, "iteration": state["iteration"] + 1 } # 将决策也存入状态，供后续边使用 update["_next_step"] = decision return update def search_node(state: ResearchState) -> dict: """搜索节点：调用工具获取信息。""" # 基于查询和历史，让LLM决定搜索关键词 prompt = f"根据以下查询和历史，生成一个简洁的网页搜索关键词。查询：{state['query']}" search_query_msg = HumanMessage(content=prompt) response = llm.invoke([search_query_msg]) search_keyword = response.content.strip() # 调用工具 tool_result = web_search_tool.invoke(search_keyword) # 更新材料列表 new_materials = state["research_materials"] + [tool_result] # 构造工具调用和结果消息，以便记录到历史 tool_call_id = f"call_{state['iteration']}" ai_msg_with_tool = AIMessage(content="", tool_calls=[{ "name": "web_search_tool", "args": {"query": search_keyword}, "id": tool_call_id }]) tool_msg = ToolMessage(content=tool_result, tool_call_id=tool_call_id) return { "research_materials": new_materials, "messages": state["messages"] + [search_query_msg, response, ai_msg_with_tool, tool_msg] } def report_node(state: ResearchState) -> dict: """报告节点：生成最终答案。""" prompt = f""" 基于以下用户查询和收集到的研究材料，生成一份简洁、准确的研究报告。 用户查询：{state['query']} 研究材料： {chr(10).join(state['research_materials'])} 报告： """ report_msg = HumanMessage(content=prompt) response = llm.invoke([report_msg]) return { "report": response.content, "messages": state["messages"] + [report_msg, response], "should_continue": False # 生成报告后停止循环 }

4.3 组装带循环的工作流

这个工作流的关键在于引入循环：规划 -> (搜索 -> 规划) 循环，直到规划节点决定生成报告。

# research_agent.py (续) # 1. 创建图 workflow = StateGraph(ResearchState) # 2. 添加节点 workflow.add_node("planner", planner_node) workflow.add_node("search", search_node) workflow.add_node("generate_report", report_node) # 3. 设置入口点 workflow.set_entry_point("planner") # 4. 定义从规划节点出发的条件边 def decide_after_planning(state: ResearchState) -> Literal["search", "generate_report", "__end__"]: """根据规划节点的决策和循环条件路由。""" # 从 planner_node 存入的状态中获取决策 next_step = state.get("_next_step", "") should_continue = state.get("should_continue", False) if not should_continue or next_step == "generate_report": return "generate_report" elif next_step == "search_web" and should_continue: return "search" else: # 安全回退，结束流程 return "__end__" workflow.add_conditional_edges( "planner", decide_after_planning, { "search": "search", "generate_report": "generate_report", "__end__": END } ) # 5. 添加从搜索节点回到规划节点的边，形成循环 workflow.add_edge("search", "planner") # 6. 添加从报告节点到结束的边 workflow.add_edge("generate_report", END) # 7. 编译 app = workflow.compile()

4.4 运行与观察循环行为

现在我们可以运行这个研究助手，观察它如何循环决策。

# research_agent.py (续) if __name__ == "__main__": print("=== 启动研究助手 ===") initial_state = { "query": "请帮我研究一下 LangGraph 和它的长期记忆功能", "messages": [], "research_materials": [], "iteration": 0, "report": "", "should_continue": True } # 为了观察流程，我们可以使用流的模式，或者多次调用。 # 这里我们直接调用，并打印最终状态。 final_state = app.invoke(initial_state) print(f"\n用户查询: {final_state['query']}") print(f"\n最终迭代次数: {final_state['iteration']}") print(f"\n收集到的材料:") for i, mat in enumerate(final_state['research_materials']): print(f" {i+1}. {mat[:100]}...") # 截断显示 print(f"\n=== 生成的研究报告 ===") print(final_state['report']) print("="*50)

运行此脚本：

python research_agent.py

输出将展示智能体如何根据初始查询，可能先搜索“LangGraph 是什么”，将结果加入材料库，然后规划节点判断是否需要更多信息（例如关于“长期记忆”），从而发起第二次搜索，最后在材料足够或达到迭代上限时，生成一份整合报告。这个过程完美演示了 LangGraph 如何管理带有循环和状态依赖的多步骤工作流。

5. 关键配置、调试与生产环境考量

构建出可运行的工作流只是第一步。要让其在开发和生产环境中稳定可靠，还需要关注以下方面。

5.1 状态持久化与长期记忆

LangGraph 的State可以被持久化，这是实现“长期记忆”的关键。你需要一个Checkpointer来保存和加载图的状态。

from langgraph.checkpoint.sqlite import SqliteSaver from langgraph.graph import StateGraph, START, END import sqlite3 # 创建或连接到 SQLite 数据库 conn = sqlite3.connect("checkpoints.db") checkpointer = SqliteSaver(conn) # 在编译图时传入 checkpointer app = workflow.compile(checkpointer=checkpointer) # 调用时，需要指定一个线程ID（thread_id）来标识会话 config = {"configurable": {"thread_id": "user_session_123"}} initial_state = {...} # 第一次调用，状态会被保存 result1 = app.invoke(initial_state, config=config) # 一段时间后，可以基于同一个 thread_id 继续执行，状态会被恢复 new_state = {"input": "基于我们刚才的对话，..."} result2 = app.invoke(new_state, config=config) # 此时 result2 包含了之前的 messages 历史

生产环境中，你可能需要使用 PostgreSQL、Redis 等更强大的后端，LangGraph 提供了相应的Saver实现。

5.2 错误处理与超时控制

节点函数可能因网络、API限制或逻辑错误而失败。在生产工作流中必须加入错误处理。

from tenacity import retry, stop_after_attempt, wait_exponential import asyncio @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10)) def reliable_llm_call(messages): """一个带重试机制的LLM调用函数。""" try: response = llm.invoke(messages) return response except Exception as e: print(f"LLM调用失败: {e}") raise # 触发重试 def robust_node(state: State): try: # ... 业务逻辑，使用 reliable_llm_call result = reliable_llm_call(state["messages"]) return {"output": result.content} except Exception as e: # 记录错误，并返回一个错误状态，让路由函数可以处理 return {"error": str(e), "should_stop": True}

在图定义中，你可以添加一个专门的error_handler节点，或者让路由函数检查状态中的error字段，并路由到降级处理或人工审核节点。

5.3 可视化与调试

LangGraph 支持将工作流图可视化，这对于理解和沟通复杂流程至关重要。

# 将图导出为 PNG 图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果环境不支持，可以输出 Mermaid 文本，在支持 Mermaid 的 Markdown 编辑器中查看 print(app.get_graph().draw_mermaid())

在调试时，可以启用更详细的日志，或者使用app.invoke(..., debug=True)来查看每一步的状态变化。

5.4 性能与成本优化

• 缓存：对 LLM 调用和工具调用（如搜索结果）实施缓存，避免重复计算。可以使用langchain.cache。
• 流式输出：对于报告生成等长文本节点，使用 LLM 的流式响应 (stream) 来提升用户体验。
• 限制循环次数：如我们例子中的iteration字段，必须设置硬性上限，防止因逻辑错误导致无限循环和 API 费用激增。
• 异步执行：如果节点间没有强依赖，可以考虑使用add_node的异步版本和asyncio来并行执行，提升整体吞吐量。

6. 常见问题排查清单

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

7. 生产环境最佳实践

将 LangGraph 智能体部署到生产环境，除了上述错误处理和性能优化，还需遵循以下实践：

1. 配置外部化：不要将 API 密钥、模型参数、数据库连接字符串等硬编码在代码中。使用环境变量或配置中心（如python-dotenv读取.env文件）。
2. 日志与监控：在每个节点函数的入口和出口添加结构化日志，记录输入、输出、耗时和错误。集成像 Prometheus 和 Grafana 这样的监控系统，跟踪工作流执行次数、成功率、延迟等指标。
3. 版本化管理图定义：工作流图（StateGraph的定义）是核心业务逻辑。应将其作为代码的一部分进行版本控制（Git）。考虑将复杂的图拆分成模块化组件，便于管理和测试。
4. 实现降级策略：对于关键节点（如 LLM 调用、外部 API 调用），设计降级方案。例如，当搜索工具失败时，可以转而从本地知识库中检索，或直接返回一个友好的提示信息，而不是让整个工作流崩溃。
5. 人工审核环节：对于涉及内容安全、重要决策或高成本操作的节点，可以设计边路由到一个“人工审核”节点，暂停工作流并通知审核人员，待人工确认后再继续执行。这可以通过 LangGraph 的interrupt机制实现。
6. 全面的测试：
   • 单元测试：单独测试每个节点函数。
   • 集成测试：测试整个工作流在不同输入下的行为。
   • 负载测试：模拟高并发场景，检查状态持久化后端和 LLM API 的承受能力。
7. 清晰的文档：使用app.get_graph().draw_mermaid()生成的图表作为技术文档的一部分，帮助团队成员理解智能体的决策流程。为每个节点和状态字段编写清晰的注释。

LangGraph 将智能体开发从线性的链式思维提升到了图式的工作流思维，为构建复杂、可靠、可维护的 AI 应用提供了强大的底层支持。从简单的条件路由到带有记忆和循环的复杂智能体，其核心始终在于对状态和流程的精确控制。开始你的项目时，建议从一个小而具体的工作流入手，逐步迭代增加节点和复杂性，并始终将可观测性和错误处理放在首位。