最近在尝试将大模型能力集成到业务流程中发现单纯依靠提示词调用API的方式很难处理复杂的、多步骤的决策任务。比如一个简单的客服场景可能需要先理解用户意图、查询知识库、生成回复、再根据用户反馈决定是否转人工这个过程涉及状态流转和条件判断。这时一个能够编排和控制任务流的框架就变得至关重要。LangGraph 正是为解决这类问题而生它让构建具备“思考”和“行动”循环的智能体Agent变得像搭积木一样清晰。本文将从零开始带你深入 LangGraph 的核心不仅会剖析其“图”与“状态”的架构思想还会通过一个完整的、可运行的代码实战项目演示如何构建一个具备工具调用和记忆能力的智能体。无论你是想入门智能体开发还是希望将 LangGraph 应用于实际项目这篇文章都将提供一条清晰的路径。我们将覆盖环境搭建、核心组件详解、代码实战以及避坑指南确保你能跟着一步步实现。1. LangGraph 是什么为什么需要它在深入代码之前我们有必要先理解 LangGraph 要解决的根本问题以及它的设计哲学。这能帮助我们在后续开发中做出更合理的设计选择。1.1 智能体与任务编排的挑战传统的单次大模型调用Completion适用于问答、翻译、摘要等一次性任务。但当任务变复杂时比如“分析这份财报并生成一份投资建议报告”可能需要拆解为提取关键数据、进行行业对比、评估风险、生成结构化报告等多个步骤。这些步骤之间存在依赖关系并且可能需要根据中间结果动态决定下一步做什么。早期我们可能用if-else或while循环来硬编码这种逻辑但这样代码会迅速变得臃肿且难以维护。智能体框架的目标就是将这种“决策-执行-再决策”的循环抽象出来让开发者更关注于定义“节点”做什么和“边”怎么流转而不是控制流细节。1.2 LangGraph 的核心有状态的工作流图LangGraph 是 LangChain 生态系统中的一个库它建立在 LangChain 丰富的组件模型、工具、记忆等之上专门用于构建有状态、多环节的工作流。你可以把它想象成一个功能强大的流程图引擎。它的核心思想是将一个智能体的执行过程建模为一张图Graph。这张图由两种基本元素构成节点Nodes代表一个可执行的操作单元。比如“调用大模型”、“执行某个工具”、“更新内存”。边Edges定义了节点之间的流转条件。通常基于当前“状态”来决定下一步走到哪个节点。而状态State是贯穿整个图执行过程的上下文容器它是一个字典或Pydantic模型记录了所有节点共享的信息例如用户输入、模型回复、工具执行结果、历史对话等。1.3 与 LangChain 的关系与区别这是初学者最容易混淆的点。简单来说LangChain是一个用于构建大模型应用的工具箱。它提供了连接模型、向量数据库、工具链等各种组件的标准化接口其AgentExecutor也可以运行简单的智能体。LangGraph是 LangChain 家族中专门用于构建复杂、有状态工作流的框架。它提供了更强大、更灵活的任务编排能力特别适合需要循环、分支、并行或持久化状态的场景。可以类比为LangChain 提供了螺丝刀、扳手等工具而 LangGraph 提供了一套设计精良的夹具和流水线让你能更高效、更可靠地组装复杂机器。1.4 典型应用场景了解适用场景能帮你判断何时该引入 LangGraph多步骤任务助手如数据分析、报告生成、复杂查询需要先搜索、再总结、再验证。对话式智能体需要维护对话历史、用户偏好并能根据上下文调用不同工具的聊天机器人。自动化业务流程例如自动处理客服工单分类、检索知识库、生成回复、升级处理。模拟与游戏构建具有记忆和决策能力的NPC。复杂决策系统需要反复调用工具、验证结果、并最终给出结论的系统。接下来我们就开始动手搭建环境并构建我们的第一个智能体。2. 环境准备与项目初始化为了确保示例的通用性和可复现性我们将使用 OpenAI 的 GPT 模型作为大模型引擎并使用虚拟工具进行演示。你可以根据自身情况替换为其他模型如 DeepSeek、通义千问等。2.1 环境与依赖操作系统Windows/macOS/Linux 均可。Python版本建议使用 Python 3.10 或 3.11这是当前多数AI库兼容性最好的版本。首先创建一个新的项目目录并初始化虚拟环境这是管理项目依赖的最佳实践。# 创建项目目录 mkdir langgraph-agent-tutorial cd langgraph-agent-tutorial # 创建并激活虚拟环境 (以 macOS/Linux 为例) python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate # 升级 pip pip install --upgrade pip接下来安装核心依赖。我们将安装langgraph,langchain-openai用于接入OpenAI以及langchain-community其中包含一些社区工具和组件。pip install langgraph langchain-openai langchain-community重要提示LangGraph 和 LangChain 生态版本更新较快如果遇到兼容性问题可以尝试指定稍早的稳定版本例如pip install langgraph0.0.40。本文示例基于当前主流稳定版本编写。2.2 设置 API 密钥我们需要一个 OpenAI API 密钥。如果你没有可以前往 OpenAI 平台注册获取。请注意保管你的密钥不要将其提交到代码仓库中。在项目中创建一个.env文件来存储密钥# .env 文件内容 OPENAI_API_KEY你的实际api密钥然后在 Python 代码中我们可以使用dotenv包来加载它。先安装这个包pip install python-dotenv2.3 项目结构规划一个清晰的目录结构有助于管理复杂的智能体项目。建议如下langgraph-agent-tutorial/ ├── .env # 环境变量API密钥等 ├── requirements.txt # 项目依赖列表 ├── main.py # 主入口文件智能体定义与运行 ├── tools/ # 自定义工具目录 │ └── custom_tools.py ├── graphs/ # LangGraph 图定义目录 │ └── basic_agent.py └── utils/ # 工具函数 └── state_models.py # 状态模型定义我们先创建requirements.txt文件记录依赖langgraph0.0.40 langchain-openai0.0.5 langchain-community0.0.10 python-dotenv1.0.0 openai1.0.0 # langchain-openai 的底层依赖现在基础环境已经就绪。让我们进入最核心的部分——理解 LangGraph 的组件。3. LangGraph 核心组件深度剖析要玩转 LangGraph必须吃透三个核心概念状态State、节点Node和边Edge。它们共同构成了智能体的“骨架”与“灵魂”。3.1 状态State智能体的记忆体状态是一个在所有节点间共享和修改的数据结构。在 LangGraph 中我们通常使用 Pydantic 的BaseModel来定义它这能提供良好的类型提示和数据验证。让我们定义一个最简单的智能体状态它需要记录对话和最新的思考。# utils/state_models.py from typing import List, Dict, Any, Optional from typing_extensions import TypedDict from langgraph.graph.message import add_messages from pydantic import BaseModel # 方式一使用 TypedDict灵活兼容性好 class AgentState(TypedDict): 智能体的状态定义 messages: List[Dict[str, Any]] # 对话消息历史 current_step: str # 当前执行到了哪一步 tool_outputs: Optional[List[str]] # 工具调用结果 # 方式二使用 Pydantic BaseModel推荐有类型验证和默认值 class AgentStatePydantic(BaseModel): 使用Pydantic定义的智能体状态 messages: List[Dict[str, Any]] [] # 默认空列表 current_step: str start tool_outputs: Optional[List[str]] None user_input: Optional[str] None # 用户最新输入 final_answer: Optional[str] None # 最终答案 class Config: arbitrary_types_allowed True关键点解析messages这是最关键的字段通常用于存储与LLM交互的对话历史。LangGraph 提供了add_messages函数来方便地遵循特定消息格式。其他字段如current_step,tool_outputs是自定义的用于控制流程和传递数据。使用 Pydantic 可以设置默认值并在运行时进行类型检查避免难以调试的错误。3.2 节点Node执行单元节点是一个函数它接收当前State执行一些操作如调用LLM、运行工具然后返回一个更新后的State或对State的修改。一个典型的“调用模型”节点如下# graphs/basic_agent.py 片段 from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage import os from dotenv import load_dotenv load_dotenv() # 加载 .env 中的 API_KEY # 初始化模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyos.getenv(OPENAI_API_KEY)) def call_model(state: AgentStatePydantic): 节点函数调用大模型生成回复 # 1. 从状态中获取消息历史 conversation_history state.messages # 2. 可以在此处添加系统提示词 system_prompt SystemMessage(content你是一个有帮助的助手。请根据用户请求和已有信息进行回复。) messages_for_llm [system_prompt] conversation_history[-6:] # 限制历史长度防止token超限 # 3. 调用模型 response llm.invoke(messages_for_llm) # 4. 将模型的回复添加到消息历史中 new_messages state.messages [{role: assistant, content: response.content}] # 5. 返回更新后的状态只更新需要修改的字段 return {messages: new_messages, current_step: model_called}节点设计原则单一职责一个节点最好只做一件事如只调用模型或只执行一个工具。纯函数思想节点不应有副作用如直接修改全局变量应通过返回的字典来声明对状态的更改。错误处理在实际项目中节点内应有try-except来处理模型调用失败或工具异常。3.3 边Edge与条件路由控制流边决定了执行完一个节点后下一步该去哪个节点。最简单的边是“总是转到下一个节点”。但智能体的强大之处在于条件路由。在 LangGraph 中我们通过定义一个“路由函数”来实现条件边。这个函数接收State返回下一个要执行的节点名称字符串。# graphs/basic_agent.py 片段 def should_use_tool(state: AgentStatePydantic) - str: 路由函数根据模型回复判断是否需要调用工具。 这是一个简单的规则如果回复中包含‘[调用工具]’字样则去工具节点。 last_message state.messages[-1] last_content last_message.get(content, ) if isinstance(last_message, dict) else last_message.content if [调用工具] in last_content: return tool_node # 前往工具节点 else: return end # 前往结束节点 def check_tool_result(state: AgentStatePydantic) - str: 路由函数检查工具调用结果后决定下一步。 例如如果工具执行成功则继续让模型总结如果失败则直接结束或重试。 if state.tool_outputs and len(state.tool_outputs) 0: # 假设工具执行成功返回结果不为空 if state.tool_outputs[-1] ! ERROR: return call_model # 返回模型节点进行总结 return end # 否则结束条件路由的威力通过这种方式你可以构建出复杂的决策树例如模型生成 → 判断是否需要搜索 → 是则调用搜索工具 → 根据搜索结果判断是否足够 → 不足则再次搜索或询问用户 → 足够则生成最终答案。3.4 图的组装StateGraph这是将节点和边组合成可执行工作流的地方。StateGraph是 LangGraph 的核心类。# graphs/basic_agent.py 继续 from langgraph.graph import StateGraph, END # 1. 创建图并指定状态的结构使用我们定义的Pydantic模型 workflow StateGraph(AgentStatePydantic) # 2. 添加节点 workflow.add_node(call_model, call_model) # 添加模型节点 workflow.add_node(tool_node, execute_tool) # 添加工具节点execute_tool函数需提前定义 # 3. 设置入口点 workflow.set_entry_point(call_model) # 4. 添加普通边无条件流转 workflow.add_edge(tool_node, call_model) # 工具执行完后回到模型节点 # 5. 添加条件边 workflow.add_conditional_edges( call_model, # 从哪个节点出发 should_use_tool, # 路由判断函数 { tool_node: tool_node, # 如果函数返回tool_node则前往tool_node节点 end: END # 如果函数返回end则直接结束图执行 } ) workflow.add_conditional_edges( tool_node, check_tool_result, { call_model: call_model, end: END } ) # 6. 编译图得到可执行对象 app workflow.compile()至此一个具备基本“思考-行动”循环的智能体框架就搭建好了。app对象就是我们的智能体我们可以向它传入初始状态来运行。4. 完整实战构建一个天气查询智能体理论已经足够现在我们来构建一个实用的智能体它能够理解用户关于天气的询问调用一个模拟的天气查询工具并将结果以友好的方式回复给用户。4.1 第一步定义工具工具是智能体与外界交互的“手”和“脚”。我们先定义一个模拟的天气查询工具。# tools/custom_tools.py from langchain.tools import tool from typing import Optional tool def get_weather(location: str, date: Optional[str] None) - str: 查询指定地点和日期的天气情况。 参数: location: 城市名例如“北京”、“上海”。 date: 日期格式为‘YYYY-MM-DD’。如果为None则查询今天天气。 返回: 天气情况的字符串描述。 # 这是一个模拟工具实际项目中应调用真实的天气API weather_data { 北京: {today: 晴朗气温 5~15°C微风, tomorrow: 多云气温 7~18°C}, 上海: {today: 小雨气温 10~18°C东风3级, tomorrow: 阴天气温 12~20°C}, 广州: {today: 晴朗气温 20~28°C, tomorrow: 晴朗气温 22~30°C}, } city_info weather_data.get(location) if not city_info: return f抱歉未找到 {location} 的天气信息。 if date is None or date today: return f{location}今天的天气是{city_info[today]} else: # 简单模拟明天返回明天的数据 return f{location}{date}的天气预计是{city_info.get(tomorrow, 信息暂缺)}4.2 第二步增强状态与工具节点我们需要更新状态模型以更好地支持工具调用。同时编写一个通用的工具执行节点。# utils/state_models.py (更新) from pydantic import BaseModel, Field from typing import List, Dict, Any, Optional from langchain_core.messages import BaseMessage class AgentState(BaseModel): 增强的智能体状态 messages: List[BaseMessage] Field(default_factorylist) # 使用LangChain消息对象 current_step: str start # 用于存储模型决定要调用的工具信息 tool_calls: Optional[List[Dict]] None # 用于存储工具执行后的原始结果 tool_outputs: Optional[List[str]] None # 最终给用户的答案 final_answer: Optional[str] None class Config: arbitrary_types_allowed True# graphs/weather_agent.py from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor from tools.custom_tools import get_weather import os from dotenv import load_dotenv from utils.state_models import AgentState load_dotenv() llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 将工具包装成 ToolExecutor方便调用 tools [get_weather] tool_executor ToolExecutor(tools) # 1. 工具调用节点 def execute_tools(state: AgentState): 执行模型请求调用的所有工具 messages state.messages last_message messages[-1] tool_calls last_message.tool_calls if hasattr(last_message, tool_calls) else [] if not tool_calls: # 如果没有工具调用直接返回原状态 return {tool_outputs: [], current_step: no_tool_called} outputs [] for tool_call in tool_calls: # 执行工具 result tool_executor.invoke(tool_call) # 将结果格式化为 ToolMessage outputs.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) # 将工具执行结果添加到消息历史中 new_messages messages outputs return {messages: new_messages, tool_outputs: outputs, current_step: tools_executed} # 2. 模型调用节点绑定工具 # 关键将工具定义绑定到模型这样模型才知道可以调用什么工具 llm_with_tools llm.bind_tools(tools) def call_model(state: AgentState): 调用已绑定工具的模型 messages state.messages # 调用模型模型现在知道有 get_weather 工具可用 response llm_with_tools.invoke(messages) # 将模型的响应可能包含工具调用请求添加到历史 new_messages messages [response] return {messages: new_messages, current_step: model_called}4.3 第三步定义智能体的决策逻辑路由这是智能体的“大脑”决定在什么情况下该做什么。# graphs/weather_agent.py (继续) def route_model(state: AgentState) - str: 核心路由逻辑根据模型的最新回复决定下一步。 如果模型请求调用工具则去执行工具否则工作流结束。 messages state.messages last_message messages[-1] # 检查最后一条消息是否是AIMessage且包含工具调用 if hasattr(last_message, tool_calls) and last_message.tool_calls: return execute_tools # 前往工具执行节点 else: # 模型直接给出了最终答案流程结束 if isinstance(last_message, AIMessage): return {final_answer: last_message.content} return end # 或者直接结束4.4 第四步组装并运行智能体现在我们把所有部件组装起来并运行一个完整的对话。# graphs/weather_agent.py (继续) def main(): # 创建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(call_model, call_model) workflow.add_node(execute_tools, execute_tools) # 设置入口点 workflow.set_entry_point(call_model) # 添加条件边 workflow.add_conditional_edges( call_model, route_model, # 路由函数 { execute_tools: execute_tools, # 需要调用工具 end: END # 直接结束 } ) # 工具执行完后无条件回到模型节点让模型总结工具结果 workflow.add_edge(execute_tools, call_model) # 编译图 app workflow.compile() # 运行智能体 print( 天气查询智能体启动 ) user_input input(请输入您的问题 (例如北京今天天气怎么样): ) # 初始化状态 initial_state AgentState( messages[HumanMessage(contentuser_input)] ) # 执行图 final_state app.invoke(initial_state) # 输出最终结果 print(\n 智能体回复 ) for msg in final_state[messages]: if isinstance(msg, AIMessage) and not msg.tool_calls: print(f助手: {msg.content}) if isinstance(msg, ToolMessage): print(f[工具调用结果]: {msg.content}) # 或者直接获取最终答案字段 if final_state.get(final_answer): print(f\n最终答案: {final_state[final_answer]}) if __name__ __main__: main()4.5 运行与结果分析保存所有文件后在项目根目录运行python graphs/weather_agent.py你会看到类似以下的交互过程 天气查询智能体启动 请输入您的问题 (例如北京今天天气怎么样): 上海明天天气如何 智能体回复 [工具调用结果]: 上海tomorrow的天气预计是阴天气温 12~20°C 助手: 根据查询结果上海明天的天气预计是阴天气温在12到20摄氏度之间。发生了什么你输入了“上海明天天气如何”初始状态被设置为包含这条用户消息。图从call_model节点开始。模型收到消息识别出需要查询天气于是它没有直接回复而是生成一个工具调用请求tool_calls指定调用get_weather工具参数为location“上海”。路由函数route_model检测到模型请求调用工具于是将执行流转到execute_tools节点。工具节点执行模拟的天气查询并将结果格式化为ToolMessage放回状态。工具节点执行完毕后通过无条件边回到call_model节点。模型再次被调用这次它的输入历史中包含了用户的问题和工具返回的结果。于是它生成了一段友好的、整合了工具结果的最终回复。路由函数发现这次模型的回复不包含工具调用于是流程结束输出最终答案。这个“模型思考 → 决定调用工具 → 执行工具 → 模型总结”的循环就是智能体最核心的工作模式。通过 LangGraph我们清晰地定义了这个循环的每一步。5. 常见问题与排查思路FAQ在实际开发中你肯定会遇到各种问题。下面是一些高频问题及其解决方案。问题现象可能原因排查思路与解决方案报错OpenAI API连接失败或 4011. API_KEY 未设置或错误。2. 网络问题或代理配置。3. OpenAI 服务暂时不可用。1. 检查.env文件是否正确并在代码中打印os.getenv(“OPENAI_API_KEY”)的前几位确认。2. 检查网络在中国大陆可能需要配置网络环境。3. 访问 OpenAI status 页面查看服务状态。错误Model‘s maximum context length is ...消息历史state.messages过长超过了模型的上下文窗口。1. 在call_model节点中限制传入模型的历史消息条数例如messages[-10:]。2. 使用 LangChain 的ConversationSummaryMemory或ConversationBufferWindowMemory来管理长对话。3. 升级到上下文更长的模型如gpt-4-turbo。智能体陷入死循环不停调用工具路由逻辑有缺陷或者工具返回的结果无法让模型满足。1.添加循环限制在状态中增加iteration_count字段在路由函数中检查超过一定次数如10次则强制跳转到END。2.优化工具和提示词确保工具返回的信息是模型能理解的。给模型更明确的系统提示例如“如果工具返回‘未找到信息’请直接告知用户不要重复调用”。3. 使用 LangGraph 的interrupt机制或Send功能进行更精细的控制。报错... is not JSON serializable状态State中存储了无法被 JSON 序列化的对象如自定义类实例。1.状态尽量使用基本类型字符串、数字、列表、字典。2. 如果必须存储复杂对象确保其实现了__dict__方法或自定义序列化逻辑。3. 使用 Pydantic 模型定义状态它能更好地处理序列化。工具绑定后模型仍不调用工具1. 模型绑定工具的代码未生效。2. 用户问题描述不够清晰模型认为无需调用工具。3. 工具描述tool下的 docstring不够准确。1. 确认llm.bind_tools(tools)被正确调用并且这个绑定后的模型llm_with_tools被传入节点函数使用。2. 在系统提示词中明确要求模型使用工具例如“当你需要查询实时信息时请务必使用提供的工具”。3. 优化工具的函数名和文档字符串使其意图更明显。如何保存和加载智能体的状态记忆默认状态下图执行完状态就消失了。1.短期记忆使用state.messages保存对话历史。2.长期记忆/持久化在图执行 (app.invoke) 前后将state字典序列化如用json.dumps存储到数据库或文件。下次运行时反序列化并作为初始状态传入。LangGraph 也支持与外部存储集成。报错Thethinking_budgetparameter must be a positive integer你使用的可能是 Anthropic Claude 等模型并传入了不支持的参数thinking_budget。1. 检查代码中初始化模型时是否错误添加了thinking_budget参数该参数是特定模型如claude-3-7-sonnet用于“思考”功能的。2. 如果不需要此功能移除该参数。如果需要请确认你使用的模型是否支持它并查阅对应模型的 API 文档。6. 进阶技巧与最佳实践掌握了基础之后下面这些实践能让你的智能体更健壮、更强大。6.1 状态设计的艺术最小化状态只存储必要的数据。状态越复杂图的调试和推理越困难。优先使用messages字段传递主要信息。使用 Pydantic强烈推荐使用 PydanticBaseModel定义状态。它能提供自动类型验证、IDE 自动补全和清晰的架构定义。分离配置与运行时状态像模型 API 密钥、工具列表这类配置信息不应放在状态里。它们应该在图编译前就确定好。6.2 高效的错误处理与重试智能体在调用外部 API 或工具时很容易失败。节点内部捕获在每个可能出错的节点尤其是工具调用和模型调用内部使用try-except。状态反馈将错误信息记录到状态如state[“last_error”] str(e)让路由函数能根据错误类型决定下一步如重试、换工具、直接向用户道歉。设置重试节点可以专门设计一个retry_node当工具调用失败时路由到此节点该节点可以修改参数后再次尝试或记录失败次数后放弃。6.3 利用 LangGraph 预构建组件LangGraph 提供了一些“开箱即用”的高级组件能极大简化开发。ToolNode 一个预构建的节点专门用于执行工具调用列表。你可以直接用ToolNode(tools[...])来替代我们手写的execute_tools函数。MessagesState 一个预定义的状态类型专门为基于消息的对话设计简化了消息列表的管理。State 一个更通用的状态基类配合annotation使用非常灵活。6.4 与 MCPModel Context Protocol集成MCP 是一种新兴协议旨在标准化大模型与外部工具/数据源之间的连接。虽然 LangGraph 本身不直接依赖 MCP但它们的理念相通。你可以将 MCP Server 提供的功能封装成 LangChain Tool然后集成到你的 LangGraph 智能体中。例如如果你有一个通过 MCP 连接数据库的 Server你可以创建一个 LangChain Tool 来调用它这样你的智能体就能通过 LangGraph 的工作流来查询数据库了。# 伪代码示例 from langchain.tools import tool import requests tool def query_database_via_mcp(query: str) - str: 通过MCP Server查询数据库 # 假设你的MCP Server运行在本地 8000 端口 response requests.post(http://localhost:8000/query, json{sql: query}) return response.json().get(result, 查询失败)6.5 可视化与调试LangGraph 支持将编译好的图导出为图片这对于理解复杂工作流和调试至关重要。# 在编译图之后 app workflow.compile() # 导出为PNG图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法显示图片可以打印文本结构 print(app.get_graph().print_ascii())在 Jupyter Notebook 或支持图形显示的环境中这能生成一张清晰的流程图展示所有节点和边。7. 总结从入门到精通的路径通过本文我们完成了从 LangGraph 核心概念理解到构建一个完整、可运行智能体的全过程。我们拆解了状态、节点、边这三个基石并通过一个天气查询Agent实战演示了“模型-工具”协作的经典循环。关键收获LangGraph 是一个工作流引擎它通过“图”来编排复杂的、有状态的智能体任务远超简单提示链的能力。状态管理是核心设计一个好的状态模型是构建稳定智能体的第一步。条件路由是智能的体现通过add_conditional_edges你可以让智能体根据中间结果动态改变执行路径。工具是能力的延伸将 LangChain Tools 集成进来智能体就能操作外部世界。下一步可以探索的方向复杂工作流尝试构建包含并行执行add_edge多个目标、子图嵌套工作流的智能体。长期记忆集成向量数据库如Chroma, Pinecone让智能体能够记住之前的对话并从知识库中检索。多智能体协作用 LangGraph 构建多个智能体让它们通过共享状态或消息进行协作完成更宏大的任务。生产化部署研究如何将 LangGraph 应用打包为 API 服务使用 FastAPI并加入监控、日志和稳定性保障。智能体开发是一个充满挑战和乐趣的领域而 LangGraph 提供了一个极其强大且优雅的范式。希望这篇教程能成为你探索之旅的一块坚实垫脚石。最好的学习方式就是动手不妨基于今天的代码尝试改造一个你工作中遇到的重复性任务看看智能体能否帮你自动化。如果在实践中遇到问题欢迎在评论区交流探讨。 从零构建LangGraph智能体：基于状态与图的工作流编排实战
最近在尝试将大模型能力集成到业务流程中发现单纯依靠提示词调用API的方式很难处理复杂的、多步骤的决策任务。比如一个简单的客服场景可能需要先理解用户意图、查询知识库、生成回复、再根据用户反馈决定是否转人工这个过程涉及状态流转和条件判断。这时一个能够编排和控制任务流的框架就变得至关重要。LangGraph 正是为解决这类问题而生它让构建具备“思考”和“行动”循环的智能体Agent变得像搭积木一样清晰。本文将从零开始带你深入 LangGraph 的核心不仅会剖析其“图”与“状态”的架构思想还会通过一个完整的、可运行的代码实战项目演示如何构建一个具备工具调用和记忆能力的智能体。无论你是想入门智能体开发还是希望将 LangGraph 应用于实际项目这篇文章都将提供一条清晰的路径。我们将覆盖环境搭建、核心组件详解、代码实战以及避坑指南确保你能跟着一步步实现。1. LangGraph 是什么为什么需要它在深入代码之前我们有必要先理解 LangGraph 要解决的根本问题以及它的设计哲学。这能帮助我们在后续开发中做出更合理的设计选择。1.1 智能体与任务编排的挑战传统的单次大模型调用Completion适用于问答、翻译、摘要等一次性任务。但当任务变复杂时比如“分析这份财报并生成一份投资建议报告”可能需要拆解为提取关键数据、进行行业对比、评估风险、生成结构化报告等多个步骤。这些步骤之间存在依赖关系并且可能需要根据中间结果动态决定下一步做什么。早期我们可能用if-else或while循环来硬编码这种逻辑但这样代码会迅速变得臃肿且难以维护。智能体框架的目标就是将这种“决策-执行-再决策”的循环抽象出来让开发者更关注于定义“节点”做什么和“边”怎么流转而不是控制流细节。1.2 LangGraph 的核心有状态的工作流图LangGraph 是 LangChain 生态系统中的一个库它建立在 LangChain 丰富的组件模型、工具、记忆等之上专门用于构建有状态、多环节的工作流。你可以把它想象成一个功能强大的流程图引擎。它的核心思想是将一个智能体的执行过程建模为一张图Graph。这张图由两种基本元素构成节点Nodes代表一个可执行的操作单元。比如“调用大模型”、“执行某个工具”、“更新内存”。边Edges定义了节点之间的流转条件。通常基于当前“状态”来决定下一步走到哪个节点。而状态State是贯穿整个图执行过程的上下文容器它是一个字典或Pydantic模型记录了所有节点共享的信息例如用户输入、模型回复、工具执行结果、历史对话等。1.3 与 LangChain 的关系与区别这是初学者最容易混淆的点。简单来说LangChain是一个用于构建大模型应用的工具箱。它提供了连接模型、向量数据库、工具链等各种组件的标准化接口其AgentExecutor也可以运行简单的智能体。LangGraph是 LangChain 家族中专门用于构建复杂、有状态工作流的框架。它提供了更强大、更灵活的任务编排能力特别适合需要循环、分支、并行或持久化状态的场景。可以类比为LangChain 提供了螺丝刀、扳手等工具而 LangGraph 提供了一套设计精良的夹具和流水线让你能更高效、更可靠地组装复杂机器。1.4 典型应用场景了解适用场景能帮你判断何时该引入 LangGraph多步骤任务助手如数据分析、报告生成、复杂查询需要先搜索、再总结、再验证。对话式智能体需要维护对话历史、用户偏好并能根据上下文调用不同工具的聊天机器人。自动化业务流程例如自动处理客服工单分类、检索知识库、生成回复、升级处理。模拟与游戏构建具有记忆和决策能力的NPC。复杂决策系统需要反复调用工具、验证结果、并最终给出结论的系统。接下来我们就开始动手搭建环境并构建我们的第一个智能体。2. 环境准备与项目初始化为了确保示例的通用性和可复现性我们将使用 OpenAI 的 GPT 模型作为大模型引擎并使用虚拟工具进行演示。你可以根据自身情况替换为其他模型如 DeepSeek、通义千问等。2.1 环境与依赖操作系统Windows/macOS/Linux 均可。Python版本建议使用 Python 3.10 或 3.11这是当前多数AI库兼容性最好的版本。首先创建一个新的项目目录并初始化虚拟环境这是管理项目依赖的最佳实践。# 创建项目目录 mkdir langgraph-agent-tutorial cd langgraph-agent-tutorial # 创建并激活虚拟环境 (以 macOS/Linux 为例) python -m venv venv source venv/bin/activate # Windows 下使用 venv\Scripts\activate # 升级 pip pip install --upgrade pip接下来安装核心依赖。我们将安装langgraph,langchain-openai用于接入OpenAI以及langchain-community其中包含一些社区工具和组件。pip install langgraph langchain-openai langchain-community重要提示LangGraph 和 LangChain 生态版本更新较快如果遇到兼容性问题可以尝试指定稍早的稳定版本例如pip install langgraph0.0.40。本文示例基于当前主流稳定版本编写。2.2 设置 API 密钥我们需要一个 OpenAI API 密钥。如果你没有可以前往 OpenAI 平台注册获取。请注意保管你的密钥不要将其提交到代码仓库中。在项目中创建一个.env文件来存储密钥# .env 文件内容 OPENAI_API_KEY你的实际api密钥然后在 Python 代码中我们可以使用dotenv包来加载它。先安装这个包pip install python-dotenv2.3 项目结构规划一个清晰的目录结构有助于管理复杂的智能体项目。建议如下langgraph-agent-tutorial/ ├── .env # 环境变量API密钥等 ├── requirements.txt # 项目依赖列表 ├── main.py # 主入口文件智能体定义与运行 ├── tools/ # 自定义工具目录 │ └── custom_tools.py ├── graphs/ # LangGraph 图定义目录 │ └── basic_agent.py └── utils/ # 工具函数 └── state_models.py # 状态模型定义我们先创建requirements.txt文件记录依赖langgraph0.0.40 langchain-openai0.0.5 langchain-community0.0.10 python-dotenv1.0.0 openai1.0.0 # langchain-openai 的底层依赖现在基础环境已经就绪。让我们进入最核心的部分——理解 LangGraph 的组件。3. LangGraph 核心组件深度剖析要玩转 LangGraph必须吃透三个核心概念状态State、节点Node和边Edge。它们共同构成了智能体的“骨架”与“灵魂”。3.1 状态State智能体的记忆体状态是一个在所有节点间共享和修改的数据结构。在 LangGraph 中我们通常使用 Pydantic 的BaseModel来定义它这能提供良好的类型提示和数据验证。让我们定义一个最简单的智能体状态它需要记录对话和最新的思考。# utils/state_models.py from typing import List, Dict, Any, Optional from typing_extensions import TypedDict from langgraph.graph.message import add_messages from pydantic import BaseModel # 方式一使用 TypedDict灵活兼容性好 class AgentState(TypedDict): 智能体的状态定义 messages: List[Dict[str, Any]] # 对话消息历史 current_step: str # 当前执行到了哪一步 tool_outputs: Optional[List[str]] # 工具调用结果 # 方式二使用 Pydantic BaseModel推荐有类型验证和默认值 class AgentStatePydantic(BaseModel): 使用Pydantic定义的智能体状态 messages: List[Dict[str, Any]] [] # 默认空列表 current_step: str start tool_outputs: Optional[List[str]] None user_input: Optional[str] None # 用户最新输入 final_answer: Optional[str] None # 最终答案 class Config: arbitrary_types_allowed True关键点解析messages这是最关键的字段通常用于存储与LLM交互的对话历史。LangGraph 提供了add_messages函数来方便地遵循特定消息格式。其他字段如current_step,tool_outputs是自定义的用于控制流程和传递数据。使用 Pydantic 可以设置默认值并在运行时进行类型检查避免难以调试的错误。3.2 节点Node执行单元节点是一个函数它接收当前State执行一些操作如调用LLM、运行工具然后返回一个更新后的State或对State的修改。一个典型的“调用模型”节点如下# graphs/basic_agent.py 片段 from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage import os from dotenv import load_dotenv load_dotenv() # 加载 .env 中的 API_KEY # 初始化模型 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyos.getenv(OPENAI_API_KEY)) def call_model(state: AgentStatePydantic): 节点函数调用大模型生成回复 # 1. 从状态中获取消息历史 conversation_history state.messages # 2. 可以在此处添加系统提示词 system_prompt SystemMessage(content你是一个有帮助的助手。请根据用户请求和已有信息进行回复。) messages_for_llm [system_prompt] conversation_history[-6:] # 限制历史长度防止token超限 # 3. 调用模型 response llm.invoke(messages_for_llm) # 4. 将模型的回复添加到消息历史中 new_messages state.messages [{role: assistant, content: response.content}] # 5. 返回更新后的状态只更新需要修改的字段 return {messages: new_messages, current_step: model_called}节点设计原则单一职责一个节点最好只做一件事如只调用模型或只执行一个工具。纯函数思想节点不应有副作用如直接修改全局变量应通过返回的字典来声明对状态的更改。错误处理在实际项目中节点内应有try-except来处理模型调用失败或工具异常。3.3 边Edge与条件路由控制流边决定了执行完一个节点后下一步该去哪个节点。最简单的边是“总是转到下一个节点”。但智能体的强大之处在于条件路由。在 LangGraph 中我们通过定义一个“路由函数”来实现条件边。这个函数接收State返回下一个要执行的节点名称字符串。# graphs/basic_agent.py 片段 def should_use_tool(state: AgentStatePydantic) - str: 路由函数根据模型回复判断是否需要调用工具。 这是一个简单的规则如果回复中包含‘[调用工具]’字样则去工具节点。 last_message state.messages[-1] last_content last_message.get(content, ) if isinstance(last_message, dict) else last_message.content if [调用工具] in last_content: return tool_node # 前往工具节点 else: return end # 前往结束节点 def check_tool_result(state: AgentStatePydantic) - str: 路由函数检查工具调用结果后决定下一步。 例如如果工具执行成功则继续让模型总结如果失败则直接结束或重试。 if state.tool_outputs and len(state.tool_outputs) 0: # 假设工具执行成功返回结果不为空 if state.tool_outputs[-1] ! ERROR: return call_model # 返回模型节点进行总结 return end # 否则结束条件路由的威力通过这种方式你可以构建出复杂的决策树例如模型生成 → 判断是否需要搜索 → 是则调用搜索工具 → 根据搜索结果判断是否足够 → 不足则再次搜索或询问用户 → 足够则生成最终答案。3.4 图的组装StateGraph这是将节点和边组合成可执行工作流的地方。StateGraph是 LangGraph 的核心类。# graphs/basic_agent.py 继续 from langgraph.graph import StateGraph, END # 1. 创建图并指定状态的结构使用我们定义的Pydantic模型 workflow StateGraph(AgentStatePydantic) # 2. 添加节点 workflow.add_node(call_model, call_model) # 添加模型节点 workflow.add_node(tool_node, execute_tool) # 添加工具节点execute_tool函数需提前定义 # 3. 设置入口点 workflow.set_entry_point(call_model) # 4. 添加普通边无条件流转 workflow.add_edge(tool_node, call_model) # 工具执行完后回到模型节点 # 5. 添加条件边 workflow.add_conditional_edges( call_model, # 从哪个节点出发 should_use_tool, # 路由判断函数 { tool_node: tool_node, # 如果函数返回tool_node则前往tool_node节点 end: END # 如果函数返回end则直接结束图执行 } ) workflow.add_conditional_edges( tool_node, check_tool_result, { call_model: call_model, end: END } ) # 6. 编译图得到可执行对象 app workflow.compile()至此一个具备基本“思考-行动”循环的智能体框架就搭建好了。app对象就是我们的智能体我们可以向它传入初始状态来运行。4. 完整实战构建一个天气查询智能体理论已经足够现在我们来构建一个实用的智能体它能够理解用户关于天气的询问调用一个模拟的天气查询工具并将结果以友好的方式回复给用户。4.1 第一步定义工具工具是智能体与外界交互的“手”和“脚”。我们先定义一个模拟的天气查询工具。# tools/custom_tools.py from langchain.tools import tool from typing import Optional tool def get_weather(location: str, date: Optional[str] None) - str: 查询指定地点和日期的天气情况。 参数: location: 城市名例如“北京”、“上海”。 date: 日期格式为‘YYYY-MM-DD’。如果为None则查询今天天气。 返回: 天气情况的字符串描述。 # 这是一个模拟工具实际项目中应调用真实的天气API weather_data { 北京: {today: 晴朗气温 5~15°C微风, tomorrow: 多云气温 7~18°C}, 上海: {today: 小雨气温 10~18°C东风3级, tomorrow: 阴天气温 12~20°C}, 广州: {today: 晴朗气温 20~28°C, tomorrow: 晴朗气温 22~30°C}, } city_info weather_data.get(location) if not city_info: return f抱歉未找到 {location} 的天气信息。 if date is None or date today: return f{location}今天的天气是{city_info[today]} else: # 简单模拟明天返回明天的数据 return f{location}{date}的天气预计是{city_info.get(tomorrow, 信息暂缺)}4.2 第二步增强状态与工具节点我们需要更新状态模型以更好地支持工具调用。同时编写一个通用的工具执行节点。# utils/state_models.py (更新) from pydantic import BaseModel, Field from typing import List, Dict, Any, Optional from langchain_core.messages import BaseMessage class AgentState(BaseModel): 增强的智能体状态 messages: List[BaseMessage] Field(default_factorylist) # 使用LangChain消息对象 current_step: str start # 用于存储模型决定要调用的工具信息 tool_calls: Optional[List[Dict]] None # 用于存储工具执行后的原始结果 tool_outputs: Optional[List[str]] None # 最终给用户的答案 final_answer: Optional[str] None class Config: arbitrary_types_allowed True# graphs/weather_agent.py from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor from tools.custom_tools import get_weather import os from dotenv import load_dotenv from utils.state_models import AgentState load_dotenv() llm ChatOpenAI(modelgpt-3.5-turbo, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 将工具包装成 ToolExecutor方便调用 tools [get_weather] tool_executor ToolExecutor(tools) # 1. 工具调用节点 def execute_tools(state: AgentState): 执行模型请求调用的所有工具 messages state.messages last_message messages[-1] tool_calls last_message.tool_calls if hasattr(last_message, tool_calls) else [] if not tool_calls: # 如果没有工具调用直接返回原状态 return {tool_outputs: [], current_step: no_tool_called} outputs [] for tool_call in tool_calls: # 执行工具 result tool_executor.invoke(tool_call) # 将结果格式化为 ToolMessage outputs.append(ToolMessage(contentstr(result), tool_call_idtool_call[id])) # 将工具执行结果添加到消息历史中 new_messages messages outputs return {messages: new_messages, tool_outputs: outputs, current_step: tools_executed} # 2. 模型调用节点绑定工具 # 关键将工具定义绑定到模型这样模型才知道可以调用什么工具 llm_with_tools llm.bind_tools(tools) def call_model(state: AgentState): 调用已绑定工具的模型 messages state.messages # 调用模型模型现在知道有 get_weather 工具可用 response llm_with_tools.invoke(messages) # 将模型的响应可能包含工具调用请求添加到历史 new_messages messages [response] return {messages: new_messages, current_step: model_called}4.3 第三步定义智能体的决策逻辑路由这是智能体的“大脑”决定在什么情况下该做什么。# graphs/weather_agent.py (继续) def route_model(state: AgentState) - str: 核心路由逻辑根据模型的最新回复决定下一步。 如果模型请求调用工具则去执行工具否则工作流结束。 messages state.messages last_message messages[-1] # 检查最后一条消息是否是AIMessage且包含工具调用 if hasattr(last_message, tool_calls) and last_message.tool_calls: return execute_tools # 前往工具执行节点 else: # 模型直接给出了最终答案流程结束 if isinstance(last_message, AIMessage): return {final_answer: last_message.content} return end # 或者直接结束4.4 第四步组装并运行智能体现在我们把所有部件组装起来并运行一个完整的对话。# graphs/weather_agent.py (继续) def main(): # 创建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(call_model, call_model) workflow.add_node(execute_tools, execute_tools) # 设置入口点 workflow.set_entry_point(call_model) # 添加条件边 workflow.add_conditional_edges( call_model, route_model, # 路由函数 { execute_tools: execute_tools, # 需要调用工具 end: END # 直接结束 } ) # 工具执行完后无条件回到模型节点让模型总结工具结果 workflow.add_edge(execute_tools, call_model) # 编译图 app workflow.compile() # 运行智能体 print( 天气查询智能体启动 ) user_input input(请输入您的问题 (例如北京今天天气怎么样): ) # 初始化状态 initial_state AgentState( messages[HumanMessage(contentuser_input)] ) # 执行图 final_state app.invoke(initial_state) # 输出最终结果 print(\n 智能体回复 ) for msg in final_state[messages]: if isinstance(msg, AIMessage) and not msg.tool_calls: print(f助手: {msg.content}) if isinstance(msg, ToolMessage): print(f[工具调用结果]: {msg.content}) # 或者直接获取最终答案字段 if final_state.get(final_answer): print(f\n最终答案: {final_state[final_answer]}) if __name__ __main__: main()4.5 运行与结果分析保存所有文件后在项目根目录运行python graphs/weather_agent.py你会看到类似以下的交互过程 天气查询智能体启动 请输入您的问题 (例如北京今天天气怎么样): 上海明天天气如何 智能体回复 [工具调用结果]: 上海tomorrow的天气预计是阴天气温 12~20°C 助手: 根据查询结果上海明天的天气预计是阴天气温在12到20摄氏度之间。发生了什么你输入了“上海明天天气如何”初始状态被设置为包含这条用户消息。图从call_model节点开始。模型收到消息识别出需要查询天气于是它没有直接回复而是生成一个工具调用请求tool_calls指定调用get_weather工具参数为location“上海”。路由函数route_model检测到模型请求调用工具于是将执行流转到execute_tools节点。工具节点执行模拟的天气查询并将结果格式化为ToolMessage放回状态。工具节点执行完毕后通过无条件边回到call_model节点。模型再次被调用这次它的输入历史中包含了用户的问题和工具返回的结果。于是它生成了一段友好的、整合了工具结果的最终回复。路由函数发现这次模型的回复不包含工具调用于是流程结束输出最终答案。这个“模型思考 → 决定调用工具 → 执行工具 → 模型总结”的循环就是智能体最核心的工作模式。通过 LangGraph我们清晰地定义了这个循环的每一步。5. 常见问题与排查思路FAQ在实际开发中你肯定会遇到各种问题。下面是一些高频问题及其解决方案。问题现象可能原因排查思路与解决方案报错OpenAI API连接失败或 4011. API_KEY 未设置或错误。2. 网络问题或代理配置。3. OpenAI 服务暂时不可用。1. 检查.env文件是否正确并在代码中打印os.getenv(“OPENAI_API_KEY”)的前几位确认。2. 检查网络在中国大陆可能需要配置网络环境。3. 访问 OpenAI status 页面查看服务状态。错误Model‘s maximum context length is ...消息历史state.messages过长超过了模型的上下文窗口。1. 在call_model节点中限制传入模型的历史消息条数例如messages[-10:]。2. 使用 LangChain 的ConversationSummaryMemory或ConversationBufferWindowMemory来管理长对话。3. 升级到上下文更长的模型如gpt-4-turbo。智能体陷入死循环不停调用工具路由逻辑有缺陷或者工具返回的结果无法让模型满足。1.添加循环限制在状态中增加iteration_count字段在路由函数中检查超过一定次数如10次则强制跳转到END。2.优化工具和提示词确保工具返回的信息是模型能理解的。给模型更明确的系统提示例如“如果工具返回‘未找到信息’请直接告知用户不要重复调用”。3. 使用 LangGraph 的interrupt机制或Send功能进行更精细的控制。报错... is not JSON serializable状态State中存储了无法被 JSON 序列化的对象如自定义类实例。1.状态尽量使用基本类型字符串、数字、列表、字典。2. 如果必须存储复杂对象确保其实现了__dict__方法或自定义序列化逻辑。3. 使用 Pydantic 模型定义状态它能更好地处理序列化。工具绑定后模型仍不调用工具1. 模型绑定工具的代码未生效。2. 用户问题描述不够清晰模型认为无需调用工具。3. 工具描述tool下的 docstring不够准确。1. 确认llm.bind_tools(tools)被正确调用并且这个绑定后的模型llm_with_tools被传入节点函数使用。2. 在系统提示词中明确要求模型使用工具例如“当你需要查询实时信息时请务必使用提供的工具”。3. 优化工具的函数名和文档字符串使其意图更明显。如何保存和加载智能体的状态记忆默认状态下图执行完状态就消失了。1.短期记忆使用state.messages保存对话历史。2.长期记忆/持久化在图执行 (app.invoke) 前后将state字典序列化如用json.dumps存储到数据库或文件。下次运行时反序列化并作为初始状态传入。LangGraph 也支持与外部存储集成。报错Thethinking_budgetparameter must be a positive integer你使用的可能是 Anthropic Claude 等模型并传入了不支持的参数thinking_budget。1. 检查代码中初始化模型时是否错误添加了thinking_budget参数该参数是特定模型如claude-3-7-sonnet用于“思考”功能的。2. 如果不需要此功能移除该参数。如果需要请确认你使用的模型是否支持它并查阅对应模型的 API 文档。6. 进阶技巧与最佳实践掌握了基础之后下面这些实践能让你的智能体更健壮、更强大。6.1 状态设计的艺术最小化状态只存储必要的数据。状态越复杂图的调试和推理越困难。优先使用messages字段传递主要信息。使用 Pydantic强烈推荐使用 PydanticBaseModel定义状态。它能提供自动类型验证、IDE 自动补全和清晰的架构定义。分离配置与运行时状态像模型 API 密钥、工具列表这类配置信息不应放在状态里。它们应该在图编译前就确定好。6.2 高效的错误处理与重试智能体在调用外部 API 或工具时很容易失败。节点内部捕获在每个可能出错的节点尤其是工具调用和模型调用内部使用try-except。状态反馈将错误信息记录到状态如state[“last_error”] str(e)让路由函数能根据错误类型决定下一步如重试、换工具、直接向用户道歉。设置重试节点可以专门设计一个retry_node当工具调用失败时路由到此节点该节点可以修改参数后再次尝试或记录失败次数后放弃。6.3 利用 LangGraph 预构建组件LangGraph 提供了一些“开箱即用”的高级组件能极大简化开发。ToolNode 一个预构建的节点专门用于执行工具调用列表。你可以直接用ToolNode(tools[...])来替代我们手写的execute_tools函数。MessagesState 一个预定义的状态类型专门为基于消息的对话设计简化了消息列表的管理。State 一个更通用的状态基类配合annotation使用非常灵活。6.4 与 MCPModel Context Protocol集成MCP 是一种新兴协议旨在标准化大模型与外部工具/数据源之间的连接。虽然 LangGraph 本身不直接依赖 MCP但它们的理念相通。你可以将 MCP Server 提供的功能封装成 LangChain Tool然后集成到你的 LangGraph 智能体中。例如如果你有一个通过 MCP 连接数据库的 Server你可以创建一个 LangChain Tool 来调用它这样你的智能体就能通过 LangGraph 的工作流来查询数据库了。# 伪代码示例 from langchain.tools import tool import requests tool def query_database_via_mcp(query: str) - str: 通过MCP Server查询数据库 # 假设你的MCP Server运行在本地 8000 端口 response requests.post(http://localhost:8000/query, json{sql: query}) return response.json().get(result, 查询失败)6.5 可视化与调试LangGraph 支持将编译好的图导出为图片这对于理解复杂工作流和调试至关重要。# 在编译图之后 app workflow.compile() # 导出为PNG图片 from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法显示图片可以打印文本结构 print(app.get_graph().print_ascii())在 Jupyter Notebook 或支持图形显示的环境中这能生成一张清晰的流程图展示所有节点和边。7. 总结从入门到精通的路径通过本文我们完成了从 LangGraph 核心概念理解到构建一个完整、可运行智能体的全过程。我们拆解了状态、节点、边这三个基石并通过一个天气查询Agent实战演示了“模型-工具”协作的经典循环。关键收获LangGraph 是一个工作流引擎它通过“图”来编排复杂的、有状态的智能体任务远超简单提示链的能力。状态管理是核心设计一个好的状态模型是构建稳定智能体的第一步。条件路由是智能的体现通过add_conditional_edges你可以让智能体根据中间结果动态改变执行路径。工具是能力的延伸将 LangChain Tools 集成进来智能体就能操作外部世界。下一步可以探索的方向复杂工作流尝试构建包含并行执行add_edge多个目标、子图嵌套工作流的智能体。长期记忆集成向量数据库如Chroma, Pinecone让智能体能够记住之前的对话并从知识库中检索。多智能体协作用 LangGraph 构建多个智能体让它们通过共享状态或消息进行协作完成更宏大的任务。生产化部署研究如何将 LangGraph 应用打包为 API 服务使用 FastAPI并加入监控、日志和稳定性保障。智能体开发是一个充满挑战和乐趣的领域而 LangGraph 提供了一个极其强大且优雅的范式。希望这篇教程能成为你探索之旅的一块坚实垫脚石。最好的学习方式就是动手不妨基于今天的代码尝试改造一个你工作中遇到的重复性任务看看智能体能否帮你自动化。如果在实践中遇到问题欢迎在评论区交流探讨。