在实际 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 的基础状态State一个共享的、可变的字典在工作流的所有节点间传递和更新。它定义了整个工作流的“记忆”和上下文。通常你会定义一个 Pydantic 模型或 TypedDict 来明确状态的 schema包含如messages对话历史、intermediate_steps工具执行结果、next下一步指示等字段。节点Node工作流中的一个步骤。它是一个函数接收当前状态作为输入执行某些操作如调用 LLM、运行工具并返回一个更新后的状态字典或包含更新字段的字典。关键点节点只负责修改状态中它关心的部分。边Edge决定工作流控制流的条件逻辑。边连接节点它根据当前状态的值决定下一个应该执行的节点是谁。边可以是固定的总是跳转到某个节点也可以是条件式的根据状态中的某个字段值如next字段决定路由。图Graph由节点和边构成的网络。在 LangGraph 中你需要先创建一个StateGraph实例然后添加节点和边最后编译compile成一个可执行的计算图。这种基于状态图的模型使得实现循环、分支、并行和人工干预Human-in-the-loop等模式变得直观。1.3 为什么需要“长期记忆”在热搜词中“langgraph 长期记忆”被频繁提及。在传统对话中LLM 通常只看到有限的上下文窗口。“记忆”机制的核心是突破这个窗口限制让智能体能够记住更早的交互、学到的知识或做出的决策。在 LangGraph 的上下文中记忆主要通过两种方式实现工作流状态State作为短期或会话记忆在单次工作流执行过程中持续存在并更新。外部记忆存储通过 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中记录依赖langgraph0.0.40 langchain0.1.0 langchain-community0.0.10 ollama0.1.0 pydantic2.0.0 # 其他工具可能需要的库如 duckduckgo-search duckduckgo-search4.1.03. 构建具备记忆的研究助手智能体我们将分步骤构建智能体的核心组件定义状态、创建工具、实现节点、编排图。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(modelllama3.2:3b, temperature0) # 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(agentagent, toolsresearch_agent_tools, verboseFalse, handle_parsing_errorsTrue)现在实现具体的节点函数。节点函数接收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(contentresult[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(contentresponse.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()图结构解析工作流从route节点开始。route节点根据用户输入将state[‘next’]设置为”research”或”direct_answer”。add_conditional_edges方法会读取这个next值并将工作流导向对应的节点。research节点和direct_answer节点执行完毕后都通过add_edge连接到END表示工作流结束。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(contentuser_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}, indent2, defaultstr)) except KeyboardInterrupt: print(\n程序被中断。) break except Exception as e: print(f\n处理过程中出现错误: {e}) # 可以选择重置状态或保留部分状态 # current_state[messages].append(AIMessage(contentf抱歉处理时出错: {e})) if __name__ __main__: main()现在运行python main.py你就可以与你的本地研究助手对话了。它会根据问题复杂度决定是否调用搜索工具并且能记住整个对话历史。4. 关键配置、参数详解与运行验证4.1 模型与工具配置详解Ollama 模型选择ChatOllama(model”llama3.2:3b”, temperature0)model必须与ollama list中显示的模型名称一致。对于研究任务建议使用 7B 或更大参数量的模型以获得更好效果但需要更多显存/内存。temperature控制生成随机性。0 表示确定性最强适合需要准确答案的任务更高的值如 0.7-1.0会使输出更有创造性。在工具调用场景较低的温度有助于稳定性和准确性。AgentExecutor 参数verboseTrue在开发时开启可以打印详细的思考过程和工具调用日志便于调试。handle_parsing_errorsTrue当 LLM 输出的内容无法被解析为有效的工具调用或答案时这个选项可以防止整个程序崩溃而是将错误信息返回给 LLM 让其重试。生产环境强烈建议开启。max_iterations限制代理的最大思考-行动循环次数防止陷入无限循环。默认值通常为 15对于复杂任务可以调高。工具定义tool装饰器会自动为函数生成描述LLM 根据描述决定是否使用。描述 (docstring) 必须清晰准确说明工具的用途、输入和输出。4.2 运行验证与调试启动程序后进行多轮对话测试用户: 你好。 智能体正在思考... 助手: 你好很高兴为你服务。我是一个AI助手可以帮你解答问题或进行对话。有什么我可以帮助你的吗 用户: 搜索一下今天北京天气。 智能体正在思考... # 此处你会看到 verbose 模式下打印的思考、行动、观察过程 助手: 根据最新的搜索结果今天北京天气晴朗最高气温约25°C最低气温约15°C风力较小适合户外活动。 用户: 上海呢 智能体正在思考... 助手: 根据之前的搜索信息我只有北京今天的天气情况。如果你需要上海的天气我可以现在为你搜索一下。验证点路由功能简单问候是否触发了direct_answer_node包含“搜索”关键词的问题是否触发了research_node工具调用在research_node执行时观察控制台如果verboseTrue是否打印了Action:和Observation:确认工具被正确调用。记忆保持在第二轮问“上海呢”时智能体是否引用了历史对话“根据之前的搜索信息…”这证明messages状态被正确累积和传递。工作流终止每个问题处理后状态是否顺利到达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: , }, configconfig, stream_modevalues ) for event in events: print(f步骤类型: {type(event).__name__}, 状态键: {list(event.keys())}) # 可以进一步打印关键内容5. 常见问题排查与优化实践在开发和运行基于 LangGraph 的智能体时你会遇到一些典型问题。下面是一个排查清单。5.1 启动与基础运行问题问题现象可能原因检查与解决ModuleNotFoundError: No module named ‘langgraph’依赖未正确安装。1. 确认虚拟环境已激活。2. 运行pip install langgraph langchain langchain-community。3. 检查requirements.txt文件是否存在版本冲突。OllamaConnectionError或模型加载失败Ollama 服务未启动或模型未拉取。1. 在终端运行ollama serve启动服务。2. 运行ollama list确认所需模型存在。3. 使用ollama pull model-name拉取模型。智能体对任何输入都无反应或立即结束图的结构定义错误可能缺少入口点或边。1. 检查set_entry_point是否设置正确。2. 检查add_edge和add_conditional_edges的节点名称是否与add_node时一致。3. 使用print(graph.get_graph().draw_mermaid())输出文本图检查结构。错误State is missing required keys状态State的初始字典缺少TypedDict中定义的必需键。确保invoke传入的初始状态字典包含所有在AgentState中定义的键如messages,user_input,intermediate_steps,next即使值为空列表或空字符串。5.2 工具调用与代理逻辑问题问题现象可能原因检查与解决智能体不调用工具总是直接回答。1. 提示词Prompt未明确指示使用工具。2. 工具描述不够清晰LLM 不理解何时使用。3. 模型能力不足。1. 检查react_prompt_template确保有类似“你可以使用以下工具{tools}”和“Action:”的明确指示。2. 优化工具的docstring说明何时以及如何使用该工具。3. 尝试更换更强的基础模型。工具调用格式解析错误 (OutputParserException)。LLM 生成的文本不符合Action:和Action Input:的格式。1. 开启handle_parsing_errorsTrue。2. 在提示词中提供更清晰、更严格的格式示例。3. 使用verboseTrue查看 LLM 的原始输出检查其是否偏离格式。工具调用陷入无限循环。代理反复调用同一个工具无法得出最终答案。1. 设置AgentExecutor的max_iterations如 10。2. 在提示词中增加限制如“最多只能进行3次搜索”。3. 检查工具返回的结果是否有效无效结果可能导致 LLM 困惑并重试。5.3 记忆与状态管理问题问题现象可能原因检查与解决智能体似乎“忘记”了之前的对话。1. 状态中的messages列表没有被正确累积。2. 每次invoke都使用了全新的初始状态覆盖了历史。1. 确认状态定义中messages字段使用了Annotated[List[BaseMessage], operator.add]。2. 在主循环中确保将上一次的final_state作为下一次invoke的输入如示例代码所示。3. 检查节点函数是否错误地重置了messages列表。上下文长度超限错误。对话历史 (messages) 太长超过了模型的上下文窗口。1. 实现记忆摘要定期用 LLM 将长对话历史总结成一段摘要替换旧消息。2. 使用滑动窗口只保留最近 N 轮对话。3. 使用支持更长上下文的模型。intermediate_steps累积导致状态臃肿。在多轮复杂对话后intermediate_steps列表可能变得非常大。1. 对于非必要场景可以在工作流结束时或每轮对话后清空intermediate_steps。2. 如果后续节点不需要完整的步骤历史可以修改状态 Schema不将其定义为operator.add。5.4 性能与生产环境考量模型选择与优化本地运行的轻量模型如 3B、7B响应快但能力有限。对于复杂任务考虑使用 API 调用更强大的云端模型如 OpenAI GPT-4, Anthropic Claude。采用模型路由简单问题用本地小模型复杂问题用云端大模型。这可以在 LangGraph 中通过一个路由节点实现。持久化长期记忆当前示例的状态仅在内存中程序重启后记忆消失。生产环境需要集成 LangChain 的ChatMessageHistory组件将其与 Redis、PostgreSQL 或文件系统连接。在图的开始节点读取历史在结束节点保存历史。错误处理与鲁棒性在所有节点函数中添加try…except并返回明确的错误信息到状态中。设置超时和重试机制特别是对于网络工具调用。实现Human-in-the-loop当代理不确定或遇到关键决策时可以暂停并询问用户。这可以通过在图中添加一个human_review节点来实现该节点将状态中的特定信息展示给用户并等待输入。图的复杂性管理当图变得非常复杂时考虑使用子图Subgraph功能将相关节点模块化使主图结构更清晰。6. 扩展方向从原型到健壮应用本示例是一个起点。要将其发展为生产可用的智能体可以考虑以下方向集成更多工具除了搜索可以接入数据库查询、代码执行、文件读写、内部 API 调用等打造全能助手。实现多智能体协作使用 LangGraph 的StateGraph轻松创建多个具有不同专长的智能体如“研究员”、“写手”、“校对员”并通过状态流转让它们协同完成一个复杂任务。引入监督机制Supervisor为工作流设置一个“监督者”节点它不处理具体任务只负责监控其他节点的执行状态、处理异常、决定重试或升级。这能极大提升系统的稳定性。添加流式输出使用graph.stream()接口实现 token 级别的流式响应提升用户体验。构建 Web 服务使用 FastAPI 或 Flask 将你的智能体图包装成 REST API 或 WebSocket 服务。通过 LangGraph你将 AI 应用的逻辑从线性的“链式思维”解放出来进入了更灵活、更强大的“图式思维”。它要求开发者更清晰地定义状态和流程而这正是构建复杂、可靠 AI 系统所必需的工程设计。从这个小型的本地研究助手开始逐步扩展其能力和架构你将能驾驭越来越复杂的智能体应用场景。