LangGraph实战：从零构建具备长期记忆与工具调用能力的AI智能体

最近在尝试将大语言模型LLM从简单的问答工具升级为能自主执行复杂任务的智能体时你是否也遇到了这些困扰Agent 的逻辑流难以编排状态管理混乱记忆能力薄弱导致对话上下文丢失传统的 LangChain 在构建复杂工作流时代码往往变得冗长且难以维护。本文将为你系统性地拆解 LangGraph 这一新兴的 Agent 编排框架并结合 LangChain 生态手把手带你从零构建具备长期记忆和复杂推理能力的 AI 智能体。无论你是希望入门 AI 应用开发的新手还是寻求项目落地的进阶开发者都能从中获得一套完整、可复现的实战方案。1. 背景与核心概念为什么需要 LangGraph在深入代码之前我们首先要厘清几个核心概念以及它们之间的关系这有助于理解 LangGraph 要解决的根本问题。AI Agent智能体是什么简单来说它是一个能感知环境、进行决策并执行行动以实现目标的软件实体。在大模型语境下Agent 通常由一个大语言模型LLM作为“大脑”配合工具调用Tools、记忆Memory和任务规划Planning等模块构成。它不再是被动地回答单次提问而是能够主动使用搜索引擎、计算器、数据库等工具完成一系列连贯的任务比如“帮我分析上周的销售数据并写一份报告”。LangChain是一个用于开发由 LLM 驱动的应用程序的框架。它提供了丰富的模块如模型抽象、提示模板、链Chains、记忆存储和工具集成极大地简化了与 LLM 交互的复杂度。其核心抽象“链”可以将多个步骤串联起来。然而当任务流程不再是简单的线性链而是包含条件分支、循环、并行执行等复杂逻辑时传统的链式结构就会显得力不从心。代码会充斥着大量的if-else语句和状态管理可读性和可维护性急剧下降。这正是LangGraph登场的原因。LangGraph 是建立在 LangChain 之上的一个库它引入了图Graph的概念来编排 Agent 的工作流。你可以将工作流中的每个步骤节点和步骤之间的流转逻辑边可视化地定义出来从而清晰、灵活地构建出支持循环、分支和并行化的复杂 Agent。核心概念对比与关系LangChain 提供构建 LLM 应用的基础“砖块”模型、提示、工具、记忆。LangGraph 提供将这些“砖块”组装成复杂“机器”智能体的“蓝图”和“装配线”图结构。Agent 最终构建出的、能够自主工作的智能应用程序。Memory Agent 的“记忆”系统用于在多次调用或不同节点间持久化状态如对话历史、中间结果是构建连贯性智能体的关键。简单理解LangChain LangGraph 强大的、可编排的 AI Agent。2. 环境准备与版本说明在开始实战前我们需要搭建好开发环境。本文将使用 Python 作为开发语言并优先考虑本地化部署方案以方便调试和隐私保护。2.1 基础环境配置请确保你的系统已安装 Python推荐 3.8 及以上版本。我们将使用venv创建独立的虚拟环境避免包依赖冲突。# 1. 创建项目目录并进入 mkdir langgraph-agent-tutorial cd langgraph-agent-tutorial # 2. 创建虚拟环境Windows 用户使用 python -m venv venv python3 -m venv venv # 3. 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: # venv\Scripts\activate # 4. 激活后命令行提示符前应显示 (venv)2.2 依赖安装我们将安装 LangChain、LangGraph 的核心库并选用Ollama作为本地大模型运行工具它可以让您在本地无GPU或轻量GPU环境下运行如 Llama 3、Qwen 等开源模型。# 安装核心框架 pip install langchain langgraph langchain-community # 安装 Ollama 的 LangChain 集成包用于调用本地模型 pip install langchain-ollama # 安装用于结构化输出的可选包对 Agent 很有用 pip install langchain-openai # 即使不用 OpenAI它也包含一些有用的工具版本说明截至撰写时langchain: 0.1.0 (LangChain 已进入 0.x 时代API 与旧版有较大变化)langgraph: 0.0.20langchain-ollama: 0.1.0ollama软件需单独在 Ollama官网 下载安装并拉取模型。重要提示LangChain 生态版本迭代较快本文代码基于上述较新版本编写。若遇到 API 不兼容请参考官方文档或适当调整版本号例如pip install langchain0.1.0。2.3 本地模型准备Ollama前往 Ollama官网 下载并安装对应操作系统的软件。打开终端拉取一个合适的模型。这里我们选择轻量且性能不错的llama3.2:1b10亿参数作为示例对硬件要求极低。ollama pull llama3.2:1b运行模型服务。Ollama 默认会在11434端口启动一个本地 API 服务。ollama run llama3.2:1b # 在另一个终端窗口保持服务运行或直接让它在后台运行。至此你的开发环境已经就绪。3. LangGraph 核心原理解析LangGraph 的核心思想是将工作流抽象为一个有状态图Stateful Graph。理解以下几个关键组件是构建智能体的基础3.1 状态State状态是一个字典或 Pydantic 模型它随着工作流的执行而演变包含了所有节点需要共享和修改的信息。例如它可能包含messages: 对话消息列表来自用户和AI。intermediate_steps: 工具调用及其结果的记录。next: 指示下一步该执行哪个节点。任何你自定义的业务数据。3.2 节点Nodes节点是工作流中的基本执行单元。每个节点是一个函数它接收当前的状态作为输入执行一些操作如调用 LLM、运行工具然后返回一个更新后的状态或对状态的修改指令。3.3 边Edges边定义了节点之间的流转逻辑。分为两种条件边Conditional Edge根据当前状态的值例如LLM 的输出是继续还是结束决定下一个要执行的节点。这实现了分支和循环。普通边无条件地从一个节点指向下一个节点。3.4 图Graph与编译你将节点和边组合起来定义一个图结构。然后通过graph.compile()方法将其编译成一个可执行的、类似链Chain的对象。这个编译后的对象管理着状态的传递和节点的调度。工作流程简述初始化一个状态。将状态传入编译后的图。图根据当前状态和边逻辑决定执行哪个节点。节点执行更新状态。重复步骤 3-4直到到达终止节点。返回最终状态。4. 实战构建你的第一个 LangGraph Agent让我们从一个经典的“工具调用” Agent开始。这个 Agent 将学会使用一个计算器工具来回答数学问题。4.1 项目结构初始化在项目根目录下创建以下文件langgraph-agent-tutorial/ ├── basic_agent.py # 基础工具调用Agent ├── memory_agent.py # 带记忆的对话Agent └── requirements.txt # 依赖列表可由 pip freeze requirements.txt 生成4.2 构建基础工具调用 Agent (basic_agent.py)# basic_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_ollama import ChatOllama from langchain.tools import tool from langchain_core.messages import HumanMessage, AIMessage # 1. 定义状态结构 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 关键使用 operator.add 让消息列表自动累积 next: str # 用于决定下一个节点 # 2. 创建工具 # 定义一个简单的计算器工具 tool def calculator(expression: str) - str: 计算一个数学表达式。支持 , -, *, /, **。例如calculator(\2 3 * 4\) # 警告实际生产中应对表达式进行严格安全检查避免代码注入。 try: result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e} # 工具列表 tools [calculator] # 3. 绑定模型和工具 llm ChatOllama(modelllama3.2:1b, temperature0) # 为模型绑定工具使其知道可以调用哪些工具 llm_with_tools llm.bind_tools(tools) # 4. 定义节点函数 def call_model(state: AgentState): 调用大模型决定是回复还是调用工具 print(f\n[节点 - call_model] 当前消息历史: {state[messages]}) # 获取最新的用户消息 last_message state[messages][-1] # 调用绑定了工具的模型 response llm_with_tools.invoke([last_message]) # 检查模型的响应是普通消息还是工具调用请求 if response.tool_calls: # 模型要求调用工具 print(f 模型决定调用工具: {response.tool_calls}) # 将模型的响应包含工具调用信息添加到消息历史 new_messages state[messages] [response] # 设置下一个节点为 call_tool return {messages: new_messages, next: call_tool} else: # 模型直接给出最终回答 print(f 模型直接回复: {response.content}) new_messages state[messages] [response] # 设置下一个节点为 END结束流程 return {messages: new_messages, next: __end__} def call_tool(state: AgentState): 执行模型请求的工具调用 print(f\n[节点 - call_tool] 执行工具...) last_message state[messages][-1] new_messages state[messages].copy() # 遍历模型响应中的所有工具调用请求 for tool_call in last_message.tool_calls: # 根据工具名找到对应的工具函数 tool_to_use {t.name: t for t in tools}[tool_call[name]] # 执行工具传入参数 tool_output tool_to_use.invoke(tool_call[args]) print(f 执行工具 {tool_call[name]}参数 {tool_call[args]}结果: {tool_output}) # 将工具执行结果作为一条新消息添加到历史 new_messages.append(AIMessage(contenttool_output, tool_call_idtool_call[id])) # 工具执行后需要再次让模型根据结果进行总结或下一步决策 return {messages: new_messages, next: call_model} # 5. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(call_model, call_model) workflow.add_node(call_tool, call_tool) # 设置入口点 workflow.set_entry_point(call_model) # 添加边 workflow.add_conditional_edges( call_model, # 这是一个路由函数根据状态中的 next 字段决定下一个节点 lambda state: state[next], # 映射关系state[next] 的值 - 下一个节点名 { call_tool: call_tool, __end__: END, } ) workflow.add_edge(call_tool, call_model) # 工具执行完后无条件回到模型节点 # 编译图 app workflow.compile() # 6. 运行 Agent if __name__ __main__: print( 启动基础工具调用 Agent ) # 初始化状态用户输入一个问题 initial_state { messages: [HumanMessage(content请问 15 乘以 8 等于多少)], next: call_model } # 流式输出执行过程方便观察 print(\n--- 执行流程追踪 ---) for event in app.stream(initial_state, stream_modevalues): event[messages][-1].pretty_print() # 获取最终结果 final_state app.invoke(initial_state) print(\n--- 最终回答 ---) print(final_state[messages][-1].content)运行与验证在终端执行python basic_agent.py你应该能看到类似以下的输出清晰地展示了 Agent 的思考过程Reasoning和行动Action 启动基础工具调用 Agent --- 执行流程追踪 --- [节点 - call_model] 当前消息历史: [HumanMessage(content请问 15 乘以 8 等于多少)] 模型决定调用工具: [{name: calculator, args: {expression: 15 * 8}, id: ...}] [节点 - call_tool] 执行工具... 执行工具 calculator参数 {expression: 15 * 8}结果: 计算结果: 120 [节点 - call_model] 当前消息历史: [HumanMessage(...), AIMessage(...), AIMessage(content计算结果: 120, tool_call_id...)] 模型直接回复: 15 乘以 8 等于 120。 --- 最终回答 --- 15 乘以 8 等于 120。这个简单的 Agent 已经具备了“思考-行动-观察”的循环能力。模型首先决定调用计算器工具思考然后执行工具得到结果行动-观察最后根据结果生成面向用户的自然语言回复。5. 进阶为 Agent 注入长期记忆Memory没有记忆的 Agent 就像金鱼每次对话都是新的开始。LangGraph 通过状态管理天然支持记忆。我们将构建一个能记住对话历史的聊天 Agent。5.1 构建带记忆的对话 Agent (memory_agent.py)# memory_agent.py from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_ollama import ChatOllama from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, BaseMessage from langgraph.checkpoint.memory import MemorySaver from langgraph.graph.message import add_messages # 1. 定义增强的状态结构使用 add_messages 这个 LangGraph 内置的归约器 class ChatState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] # 自动合并消息列表 user_input: str # 当前轮的用户输入 # 2. 初始化模型和记忆存储 llm ChatOllama(modelllama3.2:1b, temperature0.7) # 提高 temperature 使回复更多样 # 创建一个内存检查点存储用于持久化对话状态此处为内存可替换为数据库 memory MemorySaver() # 3. 定义系统提示词赋予 Agent 角色和记忆指令 system_prompt SystemMessage(content你是一个友好且健谈的助手。请根据我们的对话历史进行自然、连贯的回复。如果用户提到之前聊过的内容请记得它。) def chat_node(state: ChatState): 聊天节点结合历史生成回复 print(f\n[Chat Node] 历史消息数: {len(state[messages])}) # 构建给模型的完整消息列表系统提示 历史消息 最新用户输入 messages_for_llm [system_prompt] state[messages] [HumanMessage(contentstate[user_input])] # 调用模型 response llm.invoke(messages_for_llm) print(f 助手回复: {response.content}) # 更新状态将用户输入和AI回复都添加到 messages 中 # add_messages 归约器会自动处理合并 return { messages: [HumanMessage(contentstate[user_input]), response], user_input: # 清空当前输入等待下一轮 } # 4. 构建图这次更简单是线性对话流 workflow StateGraph(ChatState) workflow.add_node(chat, chat_node) workflow.set_entry_point(chat) workflow.add_edge(chat, END) # 单轮对话结束 # 5. 编译图并注入记忆检查点功能 # checkpointer 使得每次调用都能基于特定的 config 中的线程ID来保存和加载状态 app workflow.compile(checkpointermemory) # 6. 运行多轮对话 if __name__ __main__: print( 启动带记忆的对话 Agent ) config {configurable: {thread_id: user_123}} # 线程ID标识一个独立的对话会话 # 第一轮对话 print(\n--- 第一轮自我介绍 ---) initial_state {messages: [], user_input: 你好我叫小明。} result app.invoke(initial_state, configconfig) print(f助手: {result[messages][-1].content}) # 第二轮对话测试记忆 print(\n--- 第二轮询问名字测试记忆---) # 注意我们不需要再传入 messages因为检查点会从 thread_id 恢复之前的状态。 # 我们只需要传入新的 user_input。 result app.invoke({user_input: 你还记得我叫什么名字吗}, configconfig) print(f助手: {result[messages][-1].content}) # 第三轮对话继续聊天 print(\n--- 第三轮新话题 ---) result app.invoke({user_input: 今天天气怎么样}, configconfig) print(f助手: {result[messages][-1].content}) # 我们可以检查存储的状态 print(f\n当前对话线程的完整消息历史) for msg in result[messages]: print(f {msg.type}: {msg.content[:50]}...)运行与验证python memory_agent.py输出将展示 Agent 如何记住上下文 启动带记忆的对话 Agent --- 第一轮自我介绍 --- [Chat Node] 历史消息数: 0 助手回复: 你好小明很高兴认识你。我叫Ollama是一个AI助手。有什么我可以帮你的吗 助手: 你好小明很高兴认识你。我叫Ollama是一个AI助手。有什么我可以帮你的吗 --- 第二轮询问名字测试记忆--- [Chat Node] 历史消息数: 2 # 注意这里历史消息数不再是0包含了第一轮的对话 助手回复: 当然记得你刚才告诉我你叫小明。很高兴再次和你聊天小明 助手: 当然记得你刚才告诉我你叫小明。很高兴再次和你聊天小明 --- 第三轮新话题 --- [Chat Node] 历史消息数: 4 助手回复: 作为一个AI我无法获取实时天气信息。不过你可以告诉我你所在的城市我可以根据一般情况给你一些穿衣或活动建议哦 助手: 作为一个AI我无法获取实时天气信息。不过你可以告诉我你所在的城市我可以根据一般情况给你一些穿衣或活动建议哦通过MemorySaver和thread_id我们轻松实现了跨多次invoke调用的长期对话记忆。这是构建实用聊天机器人和复杂会话式 Agent 的基石。6. 常见问题与排查思路在开发 LangGraph Agent 过程中你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案AttributeError: ‘str‘ object has no attribute ‘tool_calls‘1. 模型响应没有被正确解析为AIMessage对象。2. 使用的模型不支持或未正确配置工具调用。1. 确保使用llm.bind_tools(tools)绑定工具并用invoke调用。2. 检查模型是否支持function calling如 GPT-4, Claude, 较新的 Llama 3。Ollama 的llama3.2:1b支持基础工具调用。3. 打印response的类型和内容确认其结构。图编译或执行时报状态结构错误状态TypedDict中Annotated的归约器如operator.add,add_messages使用不当或与节点返回值不匹配。1.消息列表优先使用from langgraph.graph.message import add_messages。2.普通列表追加使用operator.add。3. 确保节点函数返回的字典中对应键的值是增量更新而不是完整替换。例如返回{messages: [new_message]}会被add_messages自动追加到历史。Agent 陷入无限循环条件边add_conditional_edges的逻辑有误或者节点没有正确设置状态中的next字段。1. 在节点函数中打印state[‘next‘]或关键决策变量。2. 检查条件边的路由函数确保所有可能的分支都映射到了有效的节点或END。3. 可以为图设置最大循环次数app workflow.compile(…, interrupt_before[“node_name”])并配合超时机制。Checkpoint相关错误1. 未正确传递config参数。2. 检查点存储如MemorySaver未正确初始化或注入。1. 使用记忆功能时每次invoke必须传入相同的config如{“configurable”: {“thread_id”: “xxx”}}以恢复状态。2. 确保编译时传入了checkpointer参数。3. 对于生产环境考虑使用RedisSaver或SqliteSaver替代MemorySaver。Ollama 连接错误或模型未加载1. Ollama 服务未启动。2. 模型名称拼写错误或未拉取。1. 在终端运行ollama serve确保服务运行。2. 运行ollama list确认模型存在。3. 在代码中检查ChatOllama的base_url参数默认http://localhost:11434。工具调用参数错误工具函数的参数定义Pydantic 模型或类型注解与模型生成的参数不匹配。1. 使用tool装饰器时确保函数有清晰的文档字符串docstring模型会据此生成参数。2. 打印tool_call[‘args‘]查看模型实际生成的参数。3. 考虑使用StructuredTool定义更严格的参数模式。7. 最佳实践与工程建议将 LangGraph Agent 从原型推向生产需要考虑以下方面7.1 状态设计最小化状态只将需要在节点间传递和持久化的数据放入状态。避免将整个应用上下文塞进去。使用 Pydantic 模型对于复杂状态使用pydantic.BaseModel替代TypedDict能获得更好的类型验证和序列化支持。清晰的归约器理解add_messages用于消息列表和operator.add用于普通列表的区别。自定义归约器可用于复杂合并逻辑。7.2 节点与图结构节点职责单一每个节点应只做一件事如“调用模型”、“调用工具”、“验证输入”。这提高了可测试性和复用性。利用子图对于复杂的、可复用的逻辑序列可以将其封装成一个子图然后在主图中作为一个节点引用。这有助于管理复杂度。可视化调试LangGraph 支持导出Mermaid图。在开发阶段使用workflow.get_graph().draw_mermaid()来可视化你的工作流检查逻辑是否正确。7.3 错误处理与鲁棒性节点级 Try-Catch在节点函数内部对可能失败的操作如网络调用、工具执行进行异常捕获并返回错误信息到状态中由专门的“错误处理节点”处理。设置超时与中断对于可能长时间运行或卡住的图在invoke时设置超时或利用interrupt_before/after在特定节点前/后设置断点。验证输入在第一个节点或专门的“验证节点”中对用户输入进行清洗和验证防止恶意输入或无效数据流入后续流程。7.4 生产环境部署持久化存储将MemorySaver替换为RedisSaver或SqliteSaver以便在服务重启后保留对话状态并支持多实例部署。异步支持LangGraph 天然支持异步。对于 I/O 密集型的节点如调用外部 API使用async def定义节点函数并用ainvoke、astream进行调用可以大幅提升并发性能。配置化管理将模型参数、工具列表、系统提示词等抽离到配置文件如config.yaml或环境变量中便于不同环境开发、测试、生产的切换。日志与监控为关键节点添加详细的日志记录记录状态变化、决策路径和耗时。这对于排查生产问题和理解 Agent 行为至关重要。7.5 测试策略单元测试节点单独测试每个节点函数模拟输入状态断言输出状态。集成测试全图针对关键用户旅程编写端到端测试验证从初始状态到最终输出的正确性。模拟外部依赖在测试中使用unittest.mock模拟 LLM 响应和工具调用使测试快速、稳定且不依赖外部服务。通过本文的讲解和实战你已经掌握了使用 LangGraph 和 LangChain 构建具备工具调用和长期记忆能力的 AI 智能体的核心技能。从理解图计算的基本原理到一步步实现基础 Agent 和记忆 Agent再到学习生产级的最佳实践这条路径为你深入探索更复杂的多智能体协作、人工反馈循环等高级主题打下了坚实的基础。建议你接下来尝试修改工具集、设计更复杂的图逻辑或将其集成到 Web 框架如 FastAPI中打造出真正实用的 AI 应用。