在实际 AI 应用开发中构建一个能处理复杂、多步骤任务的智能体Agent是核心挑战。传统的线性调用链难以应对需要状态管理、条件分支和循环迭代的场景。LangGraph 作为 LangChain 框架中用于构建有状态、多参与者工作流的核心库正是为了解决这一问题而生。它借鉴了图计算的思想将 Agent 的执行过程建模为一张图Graph节点代表执行单元边代表控制流从而能够清晰地描述和运行包含循环、分支和并行等复杂逻辑的 Agent。本文面向希望从 LangChain 基础使用进阶到复杂 Agent 开发的工程师。我们将从 LangGraph 的核心概念入手通过一个完整的、可运行的案例带你理解其设计哲学、State 管理、节点与边的关系并最终构建一个具备“长期记忆”和决策能力的对话 Agent。你将掌握如何将业务逻辑转化为 LangGraph 图结构并了解在生产环境中部署此类应用的关键考量。1. 理解 LangGraph从链Chain到图Graph的思维跃迁在深入代码之前必须厘清 LangGraph 要解决的根本问题以及它与 LangChain 其他组件的区别。这决定了你是否能正确使用它而非仅仅套用模板。1.1 LangChain 与 LangGraph 的角色分工LangChain 是一个用于开发由语言模型驱动的应用程序的框架。它提供了大量的“链”Chains将模型调用、工具使用、记忆、提示词模板等环节串联起来形成线性的、确定性的执行流程。例如一个典型的检索问答链RetrievalQA会按照“接收问题 - 检索文档 - 组合提示词 - 调用模型 - 返回答案”的顺序执行。然而许多现实任务并非线性。例如一个客服 Agent 可能需要1理解用户意图2根据意图决定查询知识库还是调用订单 API3如果信息不足主动反问用户4将多轮对话的历史考虑在内。这个过程包含条件判断分支和等待用户输入循环。LangGraph 就是为了描述和控制这种非线性、有状态的工作流而设计的。你可以把它看作是 LangChain 生态系统中的“工作流引擎”或“状态机控制器”。它不替代 LangChain 的模型调用、工具集成等基础能力而是为这些能力提供更强大的编排方式。简单对比LangChain Chain 像一条装配线物料从起点到终点经过固定的工序。LangGraph Graph 像一张交通路线图你可以根据当前所在位置状态和路标条件选择不同的道路边前往下一个地点节点。1.2 LangGraph 的核心构建块State、Node、Edge理解 LangGraph 的三大核心概念是设计和实现工作流的关键。1. State状态State 是一个字典或 Pydantic 模型它代表了工作流在任意时刻的“全部记忆”。所有节点都读取和更新这个共享的 State。典型的状态字段包括messages: 对话消息列表是 LangChain 生态中最常见的状态用于保存聊天历史。question: 用户当前的问题。context: 检索到的相关文档内容。next: 指示下一步应该执行哪个节点的标志。任何自定义的业务数据如intermediate_steps工具调用记录、user_id等。State 的设计是 LangGraph 应用的基石它决定了信息如何在节点间流动。2. Node节点节点是一个普通的 Python 函数或可调用对象它接收当前的 State 作为输入执行一些操作如调用 LLM、使用工具、处理数据并返回一个更新后的 State 字典或包含更新字段的字典。 节点的核心职责是处理业务逻辑。例如一个“调用模型”的节点会从 State 中取出messages发送给 LLM然后将 LLM 的回复追加到messages中并返回。3. Edge边边定义了控制流。它决定了在当前节点执行完毕后接下来应该执行哪个或哪些节点。边可以是条件边Conditional Edge: 根据 State 中的某个值如 LLM 的输出内容动态决定下一个节点。这是实现分支逻辑的关键。普通边: 无条件地指向下一个节点。起始边Entry Point: 指定工作流的入口节点。通过组合节点和边你就绘制出了一张完整的工作流蓝图。1.3 “长期记忆”与普通记忆的区别在相关热词中“langgraph 长期记忆”被频繁提及。这里的“长期记忆”并非指 LangGraph 内置了某种特殊的向量数据库而是指其State 的持久化能力。普通记忆如ConversationBufferMemory 通常只存在于单次链式调用中或者通过简单的窗口机制保留最近几轮对话。其生命周期和状态管理相对简单。LangGraph 的“长期记忆” 得益于其明确的状态管理模型你可以轻松地将整个 State 对象序列化后存储到数据库如 Redis、PostgreSQL或文件中。当同一个会话Session再次被触发时你可以加载之前保存的 State让工作流从上次中断的地方继续执行。这使得构建跨对话回合、记住用户偏好和历史操作的复杂 Agent 成为可能。2. 环境准备与项目初始化我们将构建一个具备工具调用能力和多轮对话记忆的智能客服 Agent。这个 Agent 能回答关于天气和公司产品的问题并在无法回答时礼貌地告知。2.1 环境与依赖配置首先确保你的 Python 环境版本在 3.8 以上。然后安装必要的依赖包。我们使用 OpenAI 的 GPT 模型作为 LLM 引擎因此需要其 SDK。同时安装 LangChain 和 LangGraph 的核心库。# 创建并进入项目目录 mkdir langgraph-agent-demo cd langgraph-agent-demo # 创建虚拟环境可选但推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 安装核心依赖 pip install langchain langgraph langchain-openai # 安装用于结构化输出的库可选用于更精确的工具调用 pip install langchain-community注意langchain是一个元包通常会安装核心组件。明确指定langgraph和langchain-openai以确保版本兼容。生产环境中务必使用requirements.txt或pyproject.toml固定版本。接下来你需要一个 OpenAI 的 API 密钥。将其设置为环境变量不要在代码中硬编码。# 在终端中设置临时 export OPENAI_API_KEYyour-api-key-here # 在Windows CMD中设置 # set OPENAI_API_KEYyour-api-key-here2.2 项目结构设计一个清晰的项目结构有助于管理复杂度。建议按以下方式组织langgraph-agent-demo/ ├── agents/ │ ├── __init__.py │ └── customer_service_agent.py # Agent 图定义 ├── tools/ │ ├── __init__.py │ └── weather_tool.py # 自定义工具 ├── config/ │ └── settings.py # 配置管理如API密钥 ├── run_agent.py # 主运行脚本 └── requirements.txt本文为简化演示我们将核心代码放在一个文件中但会注明各部分对应的模块。3. 构建第一个 LangGraph Agent智能客服我们将分步构建一个包含工具调用和条件路由的客服 Agent。3.1 定义工具Tools工具是 Agent 扩展能力的手臂。我们先定义一个模拟的天气查询工具。# tools/weather_tool.py from langchain.tools import tool from typing import Optional tool def get_weather(location: str, date: Optional[str] None) - str: 查询指定地点和日期的天气情况。 Args: location: 城市名例如“北京”、“上海”。 date: 日期格式为 YYYY-MM-DD。如果为 None则查询当前天气。 Returns: 天气情况的字符串描述。 # 这是一个模拟工具。真实场景应调用如 OpenWeatherMap 的 API。 if date: return f{location}在{date}的天气是晴朗温度25°C。 else: return f{location}当前天气多云温度22°C。使用tool装饰器可以自动生成符合 OpenAI Function Calling 格式的工具描述这对于后续的 Agent 调用至关重要。3.2 设计 State 并初始化 LLMState 是工作流的共享内存。我们使用 TypedDict 或 Pydantic 来定义其结构以获得更好的类型提示。这里我们使用 LangGraph 提供的StateGraph和MessagesState来简化。# agents/customer_service_agent.py from typing import TypedDict, Annotated, Sequence from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage from langchain_openai import ChatOpenAI from langchain.tools import Tool from tools.weather_tool import get_weather # 1. 定义状态结构 class AgentState(TypedDict): # 对话消息历史add_messages 是一个归约函数用于智能地追加消息 messages: Annotated[Sequence[BaseMessage], add_messages] # 一个可选字段用于指示下一个要执行的动作由模型决定 next: str # 2. 初始化 LLM 和工具 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 将工具包装成列表 tools [get_weather] # 让 LLM 绑定这些工具使其具备调用能力 llm_with_tools llm.bind_tools(tools)这里的关键是Annotated[Sequence[BaseMessage], add_messages]。add_messages是一个“归约函数”它告诉 LangGraph 如何合并当前节点返回的新消息与 State 中已有的消息。它会自动处理消息去重、排序等逻辑是管理对话历史的推荐方式。3.3 创建节点Nodes节点是执行具体任务的函数。我们的 Agent 至少需要两个节点一个负责调用模型并决定行动call_model另一个负责执行工具execute_tools。# agents/customer_service_agent.py (续) from langchain_core.messages import ToolMessage import json def call_model(state: AgentState): 模型调用节点。 读取历史消息调用绑定了工具的LLM并将LLM的响应可能是消息或工具调用请求添加到状态中。 messages state[messages] # 调用模型 response llm_with_tools.invoke(messages) # 返回一个字典更新状态中的 messages 字段。 # add_messages 函数会处理如何合并。 return {messages: [response]} def execute_tools(state: AgentState): 工具执行节点。 处理上一步模型返回的工具调用请求并行执行所有工具并将结果封装为 ToolMessage。 messages state[messages] last_message messages[-1] # 最新的一条消息应该是模型的请求其中包含工具调用 tool_calls last_message.tool_calls if not tool_calls: raise ValueError(fNo tool calls found in last message: {last_message}) tool_messages [] for tool_call in tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 根据工具名找到对应的工具函数 tool_to_use next(tool for tool in tools if tool.name tool_name) # 执行工具 result tool_to_use.invoke(tool_args) # 创建工具执行结果消息 tool_message ToolMessage( contentjson.dumps(result) if isinstance(result, dict) else str(result), tool_call_idtool_call[id], nametool_name ) tool_messages.append(tool_message) # 返回工具执行结果这些结果将被添加到消息历史中 return {messages: tool_messages}3.4 定义边Edges与条件路由边控制流程。我们需要决定在call_model节点之后如果模型返回了工具调用请求就前往execute_tools节点如果模型直接给出了最终答案则结束流程。在execute_tools节点之后流程应该回到call_model节点让模型根据工具执行结果生成最终回答。# agents/customer_service_agent.py (续) def should_continue(state: AgentState) - str: 条件路由函数。 根据最新消息判断下一步是执行工具还是结束。 messages state[messages] last_message messages[-1] # 如果最新消息包含工具调用则前往工具执行节点 if last_message.tool_calls: return execute_tools # 否则工作流结束 return END # 3. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(agent, call_model) # “agent”节点调用模型 workflow.add_node(tools, execute_tools) # “tools”节点执行工具 # 设置入口点 workflow.set_entry_point(agent) # 添加条件边从 agent 节点出来根据 should_continue 函数决定去向 workflow.add_conditional_edges( agent, should_continue, { execute_tools: tools, # 如果返回 “execute_tools”则前往 tools 节点 END: END # 如果返回 END则直接结束 } ) # 添加普通边从 tools 节点出来无条件回到 agent 节点让模型处理工具结果 workflow.add_edge(tools, agent) # 编译图生成可执行对象 app workflow.compile()至此一个具备基础工具调用和循环能力的 LangGraph Agent 就构建完成了。图的结构是agent - (条件判断) - tools - agent - ... - END。3.5 运行与验证 Agent创建一个主脚本来运行我们构建的 Agent。# run_agent.py from agents.customer_service_agent import app from langchain_core.messages import HumanMessage def run_conversation(): # 初始化状态包含用户的初始消息 initial_state {messages: [HumanMessage(content北京今天的天气怎么样)]} print(用户北京今天的天气怎么样) # 流式输出执行步骤可选便于调试 for event in app.stream(initial_state, stream_modevalues): event_messages event[messages] last_message event_messages[-1] # 打印模型响应或工具调用信息 if hasattr(last_message, tool_calls) and last_message.tool_calls: print(fAgent 决定调用工具: {last_message.tool_calls}) elif isinstance(last_message, AIMessage) and not last_message.tool_calls: print(fAgent{last_message.content}) elif isinstance(last_message, ToolMessage): print(f工具执行结果: {last_message.content}) if __name__ __main__: run_conversation()运行这个脚本python run_agent.py预期你会看到类似以下的输出这清晰地展示了 Agent 的思考与执行步骤用户北京今天的天气怎么样 Agent 决定调用工具: [{name: get_weather, args: {location: 北京}, id: call_...}] 工具执行结果: 北京当前天气多云温度22°C。 Agent根据查询结果北京当前天气为多云气温22摄氏度。这个流程展示了完整的“用户提问 - 模型决定调用工具 - 执行工具 - 模型根据结果生成回答”的循环。4. 进阶实现“长期记忆”与复杂状态管理上述示例中State 仅在单次app.invoke()调用内有效。要实现跨会话的“长期记忆”我们需要持久化 State。4.1 状态持久化方案LangGraph 的StateGraph本身不关心存储这给了我们灵活性。核心思路是在每次图执行前后对 State 进行序列化/反序列化并与一个唯一的session_id关联存储。# persistence.py import json import redis # 需要 pip install redis from typing import Any, Dict class StatePersistence: def __init__(self, redis_clientNone): # 使用 Redis 作为存储后端也可替换为数据库或文件 self.client redis_client or redis.Redis(hostlocalhost, port6379, db0) def save_state(self, session_id: str, state: Dict[str, Any]): 将状态序列化后保存 serialized json.dumps(state, defaultstr) # 处理不可序列化的对象 self.client.setex(flanggraph:state:{session_id}, 3600, serialized) # 设置1小时过期 def load_state(self, session_id: str) - Dict[str, Any]: 加载并反序列化状态 serialized self.client.get(flanggraph:state:{session_id}) if serialized: return json.loads(serialized) return None # 或返回初始状态 def clear_state(self, session_id: str): 清除某个会话的状态 self.client.delete(flanggraph:state:{session_id})4.2 集成持久化层的 Agent 运行器修改主运行脚本使其支持多轮对话和状态恢复。# run_agent_persistent.py from agents.customer_service_agent import app, AgentState from persistence import StatePersistence from langchain_core.messages import HumanMessage import uuid persistence StatePersistence() def run_agent_with_memory(session_id: str None, user_input: str None): if not session_id: session_id str(uuid.uuid4()) # 为新会话生成ID print(f新会话 ID: {session_id}) current_state {messages: []} else: # 尝试加载历史状态 loaded persistence.load_state(session_id) current_state loaded if loaded else {messages: []} print(f恢复会话: {session_id}) if user_input: # 将用户输入添加到状态中 current_state[messages].append(HumanMessage(contentuser_input)) print(f用户{user_input}) # 执行图 final_state None for event in app.stream(current_state, stream_modevalues): event_messages event[messages] last_message event_messages[-1] if isinstance(last_message, AIMessage) and not last_message.tool_calls: print(fAgent{last_message.content}) final_state event # 记录最终状态 # 执行结束后保存状态 if final_state: persistence.save_state(session_id, final_state) return session_id, final_state if __name__ __main__: # 模拟一个多轮对话 session_id None queries [上海明天天气如何, 你们公司的主打产品是什么, 再问一下北京的天气] for query in queries: session_id, _ run_agent_with_memory(session_id, query) print(- * 30)在这个例子中同一个session_id下的所有对话历史都会被保存在 Redis 中。当用户再次发起对话时Agent 能“记得”之前聊过的内容从而实现上下文连贯的交互。这就是 LangGraph 实现“长期记忆”的典型方式。5. 生产环境关键考量与最佳实践将 LangGraph Agent 从演示环境推向生产需要关注稳定性、性能和可观测性。5.1 状态设计的权衡State 并非越大越好。在设计 State 时需考虑序列化成本 过大的 State如包含大量向量嵌入会显著增加存储和网络开销。考虑只存储必要的引用 ID 或摘要。隐私与安全 避免在 State 中存储敏感信息如密码、个人身份信息。如需存储必须加密。结构稳定性 一旦工作流上线State 的结构键名、类型应尽量保持稳定。变更时需考虑数据迁移和兼容性。5.2 错误处理与鲁棒性生产环境的 Agent 必须能妥善处理失败。工具调用超时与失败 在execute_tools节点中增加重试逻辑和超时控制并返回清晰的错误信息给模型。模型调用异常 使用try...except包裹 LLM 调用处理网络错误、速率限制、上下文超长等问题并设置合理的回退策略。图执行中断 考虑使用checkpointerLangGraph 企业版功能或自定义实现来保存中间状态以便在崩溃后可以从断点恢复。5.3 性能优化异步执行 LangGraph 支持异步节点。如果工具调用或模型调用是 I/O 密集型使用async def定义节点函数并用ainvoke进行调用可以大幅提升吞吐量。缓存 对昂贵的操作如重复的向量检索、固定的 API 查询结果进行缓存避免重复计算。精简消息历史 对于超长对话可以使用ConversationSummaryBufferMemory或ConversationTokenBufferMemory来压缩历史消息再放入 State以节省令牌数和上下文窗口。5.4 可观测性与调试复杂的图难以调试。必须建立完善的监控。日志记录 在每个节点的入口和出口记录 State 的关键快照、耗时和结果。使用结构化日志如 JSON 格式。可视化 LangGraph 提供了workflow.get_graph().draw_mermaid()功能可以生成图的可视化表示帮助理解流程。链路追踪 集成 OpenTelemetry 等追踪工具为每次用户请求分配一个 Trace ID贯穿所有节点和外部调用便于排查问题。5.5 常见问题排查表问题现象可能原因检查点解决方案图编译失败提示 State 字段错误State 的 TypedDict 或 Pydantic 模型定义与节点返回值不匹配检查节点函数返回的字典键名和类型是否与 State 定义一致统一 State 定义和节点返回结构工具调用未被触发模型直接回复1. LLM 未正确绑定工具。2. 提示词未引导模型使用工具。3. 工具描述不够清晰。1. 检查llm.bind_tools(tools)是否成功。2. 检查传入模型的messages确保系统提示词鼓励工具使用。3. 检查工具的description和参数定义。1. 确认工具列表正确。2. 优化系统提示词。3. 完善工具文档字符串。出现KeyError找不到消息add_messages归约函数使用不当或手动修改了messages列表导致结构混乱。检查节点函数是返回{messages: [new_message]}还是直接修改了state[messages]。始终通过返回字典来更新 State让 LangGraph 处理合并逻辑。多轮对话后响应变慢或出错1.messages历史过长超出模型上下文。2. State 过大序列化/反序列化耗时。1. 检查messages列表长度。2. 监控 Redis 或数据库的读写延迟。1. 实现历史消息摘要或滑动窗口。2. 优化 State只存必要数据考虑更快的存储后端。条件路由 (should_continue) 逻辑错误路由函数返回的字符串与add_conditional_edges中定义的映射不匹配。打印should_continue函数的返回值并与边映射的键进行比对。确保路由函数的所有可能返回值都在边映射中有明确定义的目标节点或END。6. 扩展方向与学习路径掌握基础 LangGraph 工作流后你可以向更高级的应用场景探索多智能体协作 创建多个具备不同能力的 Agent 节点如“检索专家”、“分析专家”、“写作专家”通过 LangGraph 编排它们之间的协作与对话解决更复杂的任务。人工介入Human-in-the-loop 在图中设置“人工审核”节点当模型置信度低或任务关键时暂停自动化流程等待人工输入后再继续。与 LangChain Expression Language (LCEL) 结合 LCEL 用于构建声明式、可组合的链。你可以将 LCEL 链作为一个节点嵌入到 LangGraph 中结合两者的优势。集成外部工作流引擎 对于超大规模、需要严格事务和调度的场景可以探索将 LangGraph 作为决策核心与 Apache Airflow、Prefect 等工作流引擎集成。深入研究官方示例 LangGraph 官方仓库提供了大量示例包括 ReAct Agent、Multi-Agent Collaboration、Hierarchical Agents 等是最佳的学习资料。学习路径建议从理解 State、Node、Edge 这三个核心概念开始亲手实现一个带工具调用的简单循环 Agent。然后尝试加入持久化层实现记忆功能。接着设计一个包含并行执行和条件分支的复杂工作流。最后研究如何将现有的 LangChain Chain 或自定义函数优雅地集成到图中。在整个过程中始终将可观测性和错误处理作为设计的一部分这是区分原型与生产系统的关键。