LangGraph实战：构建具备长期记忆的本地AI智能体工作流

在实际 AI 应用开发中，构建一个能够自主规划、执行任务并记住上下文的智能体（Agent）是迈向复杂 AI 系统的关键一步。然而，仅凭单一的大语言模型（LLM）调用，往往难以处理多步骤、有状态的复杂任务。这时，我们需要一个框架来编排 LLM 调用、工具执行和状态流转，而 LangGraph 正是为此而生。它并非 LangChain 的替代品，而是其生态系统内一个专注于构建有状态、多参与者（Agent）工作流的强大库。

本文面向已经了解 LangChain 基础概念，希望将智能体从简单的“调用工具”升级为具备复杂决策和长期记忆能力的开发者。我们将从 LangGraph 的核心概念入手，通过一个完整的实战项目，带你构建一个本地运行的、具备记忆能力的 AI 智能体。你将学会如何定义状态、创建节点与边、管理工作流，并最终理解如何将 LangGraph 与 LangChain 的工具、模型等组件结合，打造出真正可用的智能应用。

1. 理解 LangGraph：从链（Chain）到图（Graph）的思维跃迁

在深入代码之前，必须厘清 LangGraph 要解决的核心问题以及它与 LangChain 的关系。这决定了你能否正确使用它，而非将其误用为一个复杂的链（Chain）。

1.1 LangChain 与 LangGraph 的定位差异

LangChain 是一个用于开发由语言模型驱动的应用程序的框架。它提供了模块化的组件，如模型 I/O、检索、记忆和代理（Agent），并通过“链”（Chain）将这些组件串联起来执行线性或略有分支的任务。传统的 LangChain Agent 本质是一个循环：思考 -> 选择工具 -> 执行 -> 观察 -> 再思考。

然而，当任务流程包含条件分支、循环、并行执行或多个协同工作的智能体时，简单的链或 Agent 循环会变得难以管理和维护。例如，一个客服机器人可能需要根据用户意图路由到不同的专业子智能体，子智能体处理完毕后需要将结果汇总。这种拓扑结构用“链”来表述非常笨拙。

LangGraph 的诞生就是为了描述和运行这种复杂的、有向的、可能带环的工作流图。它允许你将工作流中的每一步定义为一个“节点”（Node），节点之间的流转路径定义为“边”（Edge）。节点可以是 LLM 调用、工具执行、自定义函数，甚至是一个子图。边决定了基于当前状态，下一步应该执行哪个节点。

简单来说：

• LangChain提供了构建 AI 应用所需的砖块（模型、工具、记忆等）和粘合砖块的水泥（Chain）。
• LangGraph提供了设计并建造复杂建筑蓝图（有状态工作流图）的能力，并确保蓝图能按设计运行。

1.2 LangGraph 的核心概念：状态、节点、边与图

理解以下四个概念是使用 LangGraph 的基础：

1. 状态（State）：一个共享的、可变的字典，在工作流的所有节点间传递和更新。它定义了整个工作流的“记忆”和上下文。通常，你会定义一个 Pydantic 模型或 TypedDict 来明确状态的 schema，包含如messages（对话历史）、intermediate_steps（工具执行结果）、next（下一步指示）等字段。
2. 节点（Node）：工作流中的一个步骤。它是一个函数，接收当前状态作为输入，执行某些操作（如调用 LLM、运行工具），并返回一个更新后的状态字典（或包含更新字段的字典）。关键点：节点只负责修改状态中它关心的部分。
3. 边（Edge）：决定工作流控制流的条件逻辑。边连接节点，它根据当前状态的值，决定下一个应该执行的节点是谁。边可以是固定的（总是跳转到某个节点），也可以是条件式的（根据状态中的某个字段值，如next字段，决定路由）。
4. 图（Graph）：由节点和边构成的网络。在 LangGraph 中，你需要先创建一个StateGraph实例，然后添加节点和边，最后编译（compile）成一个可执行的计算图。

这种基于状态图的模型，使得实现循环、分支、并行和人工干预（Human-in-the-loop）等模式变得直观。

1.3 为什么需要“长期记忆”？

在热搜词中，“langgraph 长期记忆”被频繁提及。在传统对话中，LLM 通常只看到有限的上下文窗口。“记忆”机制的核心是突破这个窗口限制，让智能体能够记住更早的交互、学到的知识或做出的决策。

在 LangGraph 的上下文中，记忆主要通过两种方式实现：

1. 工作流状态（State）：作为短期或会话记忆，在单次工作流执行过程中持续存在并更新。
2. 外部记忆存储：通过 LangChain 的ChatMessageHistory等组件，将会话历史持久化到数据库（如 Redis、PostgreSQL）或向量库中，实现跨会话的长期记忆。LangGraph 的节点可以方便地读写这些外部存储。

接下来，我们将通过环境搭建和一个具体的智能体项目，将这些概念付诸实践。

2. 环境准备与项目初始化

我们将构建一个“本地研究助手”智能体。它的功能是：根据用户提出的复杂研究性问题（例如，“对比一下 LangChain 和 LangGraph 的技术特点”），能够自主规划步骤（如联网搜索、总结资料），并记住之前的对话历史，在后续问题中避免重复工作。

2.1 环境与依赖配置

首先，确保你的 Python 环境版本在 3.8 以上。建议使用虚拟环境。

# 创建并激活虚拟环境 (可选) python -m venv venv # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate

安装核心依赖。我们将使用langchain提供的基础模型和工具组件，使用langgraph构建工作流，使用langchain-community中的一些工具，并使用Ollama在本地运行开源大模型。

pip install langgraph langchain langchain-community

为了在本地运行模型，你需要安装并运行 Ollama 。Ollama 是一个本地大模型运行框架。安装后，拉取一个合适的模型，例如llama3.2或qwen2.5。

# 假设已安装 Ollama，拉取模型 ollama pull llama3.2:3b # 这里使用一个较小的版本用于演示，实际可根据硬件选择

2.2 项目结构规划

一个清晰的目录结构有助于管理复杂的智能体项目。建议如下：

local_research_agent/ ├── agent/ │ ├── __init__.py │ ├── graph.py # LangGraph 工作流定义 │ ├── state.py # 状态 Schema 定义 │ └── nodes.py # 各个节点函数实现 ├── tools/ │ ├── __init__.py │ └── web_search.py # 自定义工具示例 ├── memory/ │ └── __init__.py # 长期记忆存储相关 ├── config.py # 配置文件（模型、API Key等） ├── main.py # 主程序入口 └── requirements.txt

在requirements.txt中记录依赖：

langgraph>=0.0.40 langchain>=0.1.0 langchain-community>=0.0.10 ollama>=0.1.0 pydantic>=2.0.0 # 其他工具可能需要的库，如 duckduckgo-search duckduckgo-search>=4.1.0

3. 构建具备记忆的研究助手智能体

我们将分步骤构建智能体的核心组件：定义状态、创建工具、实现节点、编排图。

3.1 定义工作流状态（State）

在agent/state.py中，我们使用TypedDict来定义状态的结构。TypedDict比 Pydantic 模型更轻量，适合 LangGraph 的状态定义。

from typing import TypedDict, Annotated, List, Union, Optional from langchain_core.messages import BaseMessage import operator class AgentState(TypedDict): """智能体工作流的状态定义。""" # 消息列表：存储整个对话历史，是记忆的核心载体。 messages: Annotated[List[BaseMessage], operator.add] # 用户输入的最新问题 user_input: str # 存储工具调用及其结果，用于Agent的思考过程 intermediate_steps: Annotated[List[tuple], operator.add] # 控制流标志：决定下一步该执行哪个节点，例如 'continue', 'end', 'tool_call' next: str

关键解释：

• Annotated[List[BaseMessage], operator.add]：这是 LangGraph 的注解语法。它告诉框架，当多个节点都返回对messages字段的更新时，应该使用operator.add（即列表的+操作）来合并这些更新，而不是后一个覆盖前一个。这对于累积对话历史至关重要。
• messages：存储所有的HumanMessage,AIMessage,ToolMessage。这是实现对话记忆的基础。
• intermediate_steps：存储格式为(AgentAction, observation)的元组列表，是 ReAct 等 Agent 范式所需。
• next：一个简单的字符串路由键，我们的边（Edge）将根据它的值来决定流程走向。

3.2 创建工具（Tools）

智能体需要工具来与世界交互。我们先创建一个简单的网页搜索工具。在tools/web_search.py中：

from langchain_community.tools import DuckDuckGoSearchRun from langchain.tools import tool from typing import Optional # 使用 LangChain 社区集成的 DuckDuckGo 搜索工具 search = DuckDuckGoSearchRun() @tool def web_search_tool(query: str) -> str: """ 使用 DuckDuckGo 在互联网上搜索最新信息。 当用户的问题涉及实时、未知或需要最新资料的主题时，使用此工具。 Args: query: 搜索查询字符串。 Returns: 搜索结果的摘要文本。 """ # 在实际项目中，这里可以增加结果清洗、摘要提炼等逻辑 try: result = search.run(query) return result[:2000] # 限制返回长度，避免上下文过长 except Exception as e: return f"搜索过程中出现错误：{e}。请尝试简化查询词。"

你也可以创建其他工具，如计算器、数据库查询工具等。将所有工具放在一个列表中供智能体调用：

# 在 agent/nodes.py 或 config.py 中 from tools.web_search import web_search_tool research_agent_tools = [web_search_tool]

3.3 实现图节点（Nodes）

节点是工作流的执行单元。我们在agent/nodes.py中实现几个关键节点。

首先，初始化模型和工具链。我们使用 Ollama 本地模型和 LangChain 的 ReAct 代理框架来驱动单个“思考-行动”步骤。

from langchain_ollama import ChatOllama from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate from .state import AgentState import warnings warnings.filterwarnings('ignore') # 1. 初始化本地模型 llm = ChatOllama(model="llama3.2:3b", temperature=0) # 2. 定义 ReAct 风格的提示词模板 react_prompt_template = PromptTemplate.from_template(""" 你是一个专业的研究助手。请严格遵循以下格式回答问题： 问题：{input} 思考：你需要首先思考如何一步步解决问题。你可以使用以下工具： {tools} 如果你认为需要更多信息来回答，请使用工具。工具调用格式为： Action: 工具名称 Action Input: 工具的输入 工具会返回一个观察结果（Observation）。 当你拥有足够信息时，必须给出最终答案，格式为： Final Answer: 你的最终答案 历史对话： {chat_history} 开始！ 思考：{agent_scratchpad} """) # 3. 创建 ReAct 代理执行器 from langchain.agents import Tool # 假设 tools 列表已定义 from config import research_agent_tools agent = create_react_agent(llm, research_agent_tools, react_prompt_template) agent_executor = AgentExecutor(agent=agent, tools=research_agent_tools, verbose=False, handle_parsing_errors=True)

现在，实现具体的节点函数。节点函数接收state并返回一个包含状态更新字段的字典。

def route_question(state: AgentState) -> dict: """ 路由节点：分析用户输入，决定是直接回答还是需要研究。 这是一个简单的演示，实际可以使用一个更复杂的LLM调用或分类器。 """ user_input = state['user_input'].lower() # 简单规则：如果问题包含“搜索”、“最新”、“对比”等词，则进入研究流程 research_keywords = ['搜索', '查找', '最新', '对比', '研究', '什么是', 'how to', 'latest'] if any(keyword in user_input for keyword in research_keywords): next_step = 'research' else: # 简单问题，直接尝试回答 next_step = 'direct_answer' return {"next": next_step} def research_node(state: AgentState) -> dict: """ 研究节点：调用 ReAct 代理执行器来处理复杂问题，可能涉及多次工具调用。 """ # 从状态中提取用户输入和对话历史 user_input = state['user_input'] chat_history = state['messages'][:-1] # 最后一条是当前用户输入，历史是之前的 # 准备代理执行器的输入 agent_input = { "input": user_input, "chat_history": chat_history, # agent_scratchpad 由 executor 内部管理 } # 执行代理 result = agent_executor.invoke(agent_input) # 更新状态：将代理的最终回答添加到 messages 中 # result 中包含 'output' 和可能的 'intermediate_steps' from langchain_core.messages import AIMessage new_message = AIMessage(content=result['output']) # 注意：intermediate_steps 已经在 agent_executor 内部处理并添加到 result 中 # 我们需要将其同步到我们的 state 里 intermediate_steps_update = result.get('intermediate_steps', []) return { "messages": [new_message], "intermediate_steps": intermediate_steps_update, "next": "end" # 研究完成后，结束工作流 } def direct_answer_node(state: AgentState) -> dict: """ 直接回答节点：对于简单问题，直接调用LLM生成回答，不使用工具。 """ from langchain_core.messages import HumanMessage, AIMessage user_input = state['user_input'] chat_history = state['messages'][:-1] # 构建一个简单的提示 prompt = f""" 基于以下对话历史和当前问题，请直接给出简洁的回答。 如果你不知道答案，请如实说明，不要编造。 历史对话： {chat_history} 当前问题：{user_input} 回答： """ response = llm.invoke(prompt) new_message = AIMessage(content=response.content) return { "messages": [new_message], "next": "end" }

3.4 编排工作流图（Graph）

这是 LangGraph 的核心。在agent/graph.py中，我们将节点和边组装起来。

from langgraph.graph import StateGraph, END from .state import AgentState from .nodes import route_question, research_node, direct_answer_node def create_research_agent_graph(): """创建并编译研究助手智能体图""" # 1. 初始化一个状态图，指定状态类型 workflow = StateGraph(AgentState) # 2. 添加节点 workflow.add_node("route", route_question) # 路由决策节点 workflow.add_node("research", research_node) # 研究节点（使用工具） workflow.add_node("direct_answer", direct_answer_node) # 直接回答节点 # 3. 设置入口点 workflow.set_entry_point("route") # 4. 添加边（定义控制流） # 从 route 节点出发，根据其返回的 state['next'] 值路由 workflow.add_conditional_edges( "route", # 这是一个路由函数，它检查状态并返回下一个节点的名称 lambda state: state["next"], { "research": "research", "direct_answer": "direct_answer", } ) # 5. 从 research 和 direct_answer 节点到结束 workflow.add_edge("research", END) workflow.add_edge("direct_answer", END) # 6. 编译图 graph = workflow.compile() return graph # 创建图实例 research_agent_graph = create_research_agent_graph()

图结构解析：

1. 工作流从route节点开始。
2. route节点根据用户输入，将state[‘next’]设置为”research”或”direct_answer”。
3. add_conditional_edges方法会读取这个next值，并将工作流导向对应的节点。
4. research节点和direct_answer节点执行完毕后，都通过add_edge连接到END，表示工作流结束。
5. compile()方法将图结构编译成可执行对象。

3.5 主程序与交互循环

最后，在main.py中，我们创建一个简单的交互循环来使用这个智能体。

from agent.graph import research_agent_graph from agent.state import AgentState from langchain_core.messages import HumanMessage import json def main(): print("本地研究助手智能体已启动。输入‘退出’或‘quit’结束对话。") # 初始化一个空状态 current_state = { "messages": [], "user_input": "", "intermediate_steps": [], "next": "", } while True: try: user_input = input("\n用户: ").strip() if user_input.lower() in ['退出', 'quit', 'exit']: print("对话结束。") break if not user_input: continue # 1. 更新状态中的用户输入和消息 human_message = HumanMessage(content=user_input) current_state["user_input"] = user_input current_state["messages"] = current_state["messages"] + [human_message] # 2. 调用图执行工作流 print("智能体正在思考...") final_state = research_agent_graph.invoke(current_state) # 3. 获取并显示AI的回答 ai_messages = [msg for msg in final_state["messages"] if msg.type == "ai"] if ai_messages: latest_ai_msg = ai_messages[-1] print(f"\n助手: {latest_ai_msg.content}") else: print("\n助手: (未生成回答)") # 4. 为下一轮对话更新状态（保留历史消息） # 注意：invoke 返回的是最终状态，我们用它来更新 current_state，以保持记忆。 # 但需要清空 user_input 和 next，为下一次输入做准备。 current_state = final_state current_state["user_input"] = "" current_state["next"] = "" # 可选：打印当前状态，用于调试 # print(json.dumps({k: v for k, v in current_state.items() if k != 'messages'}, indent=2, default=str)) except KeyboardInterrupt: print("\n程序被中断。") break except Exception as e: print(f"\n处理过程中出现错误: {e}") # 可以选择重置状态或保留部分状态 # current_state["messages"].append(AIMessage(content=f"抱歉，处理时出错: {e}")) if __name__ == "__main__": main()

现在，运行python main.py，你就可以与你的本地研究助手对话了。它会根据问题复杂度决定是否调用搜索工具，并且能记住整个对话历史。

4. 关键配置、参数详解与运行验证

4.1 模型与工具配置详解

1. Ollama 模型选择：ChatOllama(model=”llama3.2:3b”, temperature=0)

   • model：必须与ollama list中显示的模型名称一致。对于研究任务，建议使用 7B 或更大参数量的模型以获得更好效果，但需要更多显存/内存。
   • temperature：控制生成随机性。0 表示确定性最强，适合需要准确答案的任务；更高的值（如 0.7-1.0）会使输出更有创造性。在工具调用场景，较低的温度有助于稳定性和准确性。

2. AgentExecutor 参数：

   • verbose=True：在开发时开启，可以打印详细的思考过程和工具调用日志，便于调试。
   • handle_parsing_errors=True：当 LLM 输出的内容无法被解析为有效的工具调用或答案时，这个选项可以防止整个程序崩溃，而是将错误信息返回给 LLM 让其重试。生产环境强烈建议开启。
   • max_iterations：限制代理的最大思考-行动循环次数，防止陷入无限循环。默认值通常为 15，对于复杂任务可以调高。

3. 工具定义：@tool装饰器会自动为函数生成描述，LLM 根据描述决定是否使用。描述 (docstring) 必须清晰准确，说明工具的用途、输入和输出。

4.2 运行验证与调试

启动程序后，进行多轮对话测试：

用户: 你好。 智能体正在思考... 助手: 你好！很高兴为你服务。我是一个AI助手，可以帮你解答问题或进行对话。有什么我可以帮助你的吗？ 用户: 搜索一下今天北京天气。 智能体正在思考... # 此处你会看到 verbose 模式下打印的思考、行动、观察过程 助手: 根据最新的搜索结果，今天北京天气晴朗，最高气温约25°C，最低气温约15°C，风力较小，适合户外活动。 用户: 上海呢？ 智能体正在思考... 助手: 根据之前的搜索信息，我只有北京今天的天气情况。如果你需要上海的天气，我可以现在为你搜索一下。

验证点：

1. 路由功能：简单问候是否触发了direct_answer_node？包含“搜索”关键词的问题是否触发了research_node？
2. 工具调用：在research_node执行时，观察控制台（如果verbose=True）是否打印了Action:和Observation:，确认工具被正确调用。
3. 记忆保持：在第二轮问“上海呢？”时，智能体是否引用了历史对话（“根据之前的搜索信息…”）？这证明messages状态被正确累积和传递。
4. 工作流终止：每个问题处理后，状态是否顺利到达END，并准备好接收下一个输入？

4.3 状态可视化（高级调试）

LangGraph 提供了可视化功能，可以帮助你理解工作流的执行路径。在代码中添加：

from langchain_core.messages import HumanMessage # 获取图的结构图（需要安装 graphviz） try: image_data = research_agent_graph.get_graph().draw_mermaid_png() with open("workflow_graph.png", "wb") as f: f.write(image_data) print("工作流图已保存为 workflow_graph.png") except Exception as e: print(f"生成可视化图失败: {e}，请安装 graphviz。") # 追踪单次调用的步骤 config = {"configurable": {"thread_id": "test_user_1"}} events = research_agent_graph.stream( { "messages": [HumanMessage(content="搜索LangGraph资料")], "user_input": "搜索LangGraph资料", "intermediate_steps": [], "next": "", }, config=config, stream_mode="values" ) for event in events: print(f"步骤类型: {type(event).__name__}, 状态键: {list(event.keys())}") # 可以进一步打印关键内容

5. 常见问题排查与优化实践

在开发和运行基于 LangGraph 的智能体时，你会遇到一些典型问题。下面是一个排查清单。

5.1 启动与基础运行问题

5.2 工具调用与代理逻辑问题

5.3 记忆与状态管理问题

5.4 性能与生产环境考量

1. 模型选择与优化：本地运行的轻量模型（如 3B、7B）响应快但能力有限。对于复杂任务，考虑：
   • 使用 API 调用更强大的云端模型（如 OpenAI GPT-4, Anthropic Claude）。
   • 采用模型路由：简单问题用本地小模型，复杂问题用云端大模型。这可以在 LangGraph 中通过一个路由节点实现。
2. 持久化长期记忆：当前示例的状态仅在内存中，程序重启后记忆消失。生产环境需要：
   • 集成 LangChain 的ChatMessageHistory组件，将其与 Redis、PostgreSQL 或文件系统连接。
   • 在图的开始节点读取历史，在结束节点保存历史。
3. 错误处理与鲁棒性：
   • 在所有节点函数中添加try…except，并返回明确的错误信息到状态中。
   • 设置超时和重试机制，特别是对于网络工具调用。
   • 实现Human-in-the-loop：当代理不确定或遇到关键决策时，可以暂停并询问用户。这可以通过在图中添加一个human_review节点来实现，该节点将状态中的特定信息展示给用户并等待输入。
4. 图的复杂性管理：当图变得非常复杂时，考虑使用子图（Subgraph）功能将相关节点模块化，使主图结构更清晰。

6. 扩展方向：从原型到健壮应用

本示例是一个起点。要将其发展为生产可用的智能体，可以考虑以下方向：

1. 集成更多工具：除了搜索，可以接入数据库查询、代码执行、文件读写、内部 API 调用等，打造全能助手。
2. 实现多智能体协作：使用 LangGraph 的StateGraph轻松创建多个具有不同专长的智能体（如“研究员”、“写手”、“校对员”），并通过状态流转让它们协同完成一个复杂任务。
3. 引入监督机制（Supervisor）：为工作流设置一个“监督者”节点，它不处理具体任务，只负责监控其他节点的执行状态、处理异常、决定重试或升级。这能极大提升系统的稳定性。
4. 添加流式输出：使用graph.stream()接口，实现 token 级别的流式响应，提升用户体验。
5. 构建 Web 服务：使用 FastAPI 或 Flask 将你的智能体图包装成 REST API 或 WebSocket 服务。

通过 LangGraph，你将 AI 应用的逻辑从线性的“链式思维”解放出来，进入了更灵活、更强大的“图式思维”。它要求开发者更清晰地定义状态和流程，而这正是构建复杂、可靠 AI 系统所必需的工程设计。从这个小型的本地研究助手开始，逐步扩展其能力和架构，你将能驾驭越来越复杂的智能体应用场景。