# LangGraph实战：用Python构建一个真正“有记忆”的AI聊天机器人 最近在折腾AI应用开发时，我发现了一个挺有意思的现象：很多基于大模型的聊天机器人，聊着聊着就“失忆”了。你上一秒刚告诉它你的名字，下一秒它就能一脸无辜地问你是谁。这种体验，对于想打造专业级对话应用的开发者来说，无疑是致命的。直到我开始深入使用**LangGraph**，这个基于LangChain构建的、专门用于处理有状态工作流的库，才真正找到了解决多轮对话记忆难题的钥匙。 今天，我想和你分享的，不是又一个“Hello World”式的教程，而是一个从零开始，用Python和LangGraph构建一个具备完整对话记忆能力的聊天机器人的实战过程。我们会从最核心的“状态”概念入手，一步步搭建起一个能记住对话历史、能根据上下文灵活响应的智能体。无论你是想为自己的产品添加一个智能客服，还是单纯对AI应用开发感兴趣，相信这篇手把手的指南都能给你带来一些实实在在的启发。 ## 1. 理解LangGraph的核心：状态、节点与工作流 在开始敲代码之前，我们得先搞明白LangGraph到底在解决什么问题。传统的LangChain链（Chain）本质上是**有向无环图（DAG）**，数据从输入到输出，单向流动，没有“回头路”。这对于一次性的问答任务很合适，但对于聊天这种典型的**多轮、有状态**的交互，就显得力不从心了。 LangGraph的核心突破在于引入了**“状态（State）”** 和**“循环（Cycles）”**。你可以把整个对话过程想象成一个不断演进的状态机。每一次用户输入，都会触发这个状态机运行一步，更新其内部状态（比如追加新的对话消息），然后根据新的状态决定下一步做什么（是直接回复，还是调用某个工具查询信息）。 ### 1.1 状态（State）：对话的记忆体 在LangGraph中，状态是一个自定义的字典（或TypedDict），它定义了工作流需要记住的所有信息。对于聊天机器人，最核心的状态就是对话消息列表。 ```python from typing import Annotated, List from typing_extensions import TypedDict from langchain_core.messages import BaseMessage from operator import add class ChatState(TypedDict): """定义聊天机器人的状态结构""" # 关键部分：messages字段使用Annotated和add操作符 # 这意味着每次节点返回新的消息时，会自动追加到现有列表末尾，而不是覆盖 messages: Annotated[List[BaseMessage], add] # 你可以在这里添加其他需要跟踪的状态，比如用户ID、会话主题等 user_id: str ``` 这里有个关键技巧：`Annotated[List[BaseMessage], add]`。`add`是一个归约器（reducer），它告诉LangGraph，当多个节点都更新`messages`字段时，应该使用`list.append`的方式将新消息**追加**到列表后面，而不是用新列表替换旧列表。这正是实现对话历史累积的魔法所在。 ### 1.2 节点（Node）与边（Edge）：构建对话逻辑 节点是工作流中的基本执行单元，每个节点都是一个Python函数，它接收当前状态，执行一些操作（比如调用大模型），然后返回一个包含部分状态更新的字典。边则定义了节点之间的流转路径。 一个最简单的聊天机器人工作流可能只包含一个节点：调用大模型。但更实用的机器人会包含条件分支，例如判断是否需要调用外部工具。 ```python from langgraph.graph import StateGraph, START, END # 1. 初始化一个状态图，指定状态类型为我们定义的ChatState builder = StateGraph(ChatState) # 2. 定义节点函数 def call_llm(state: ChatState): """节点：调用大语言模型生成回复""" # 获取完整的对话历史 history = state['messages'] # 这里模拟调用LLM，实际中会替换为真实的模型调用 # 假设模型返回一条新的AIMessage from langchain_core.messages import AIMessage fake_response = AIMessage(content="这是根据您的历史消息生成的回复。") # 返回状态更新：将模型回复追加到消息列表 return {"messages": [fake_response]} # 3. 将节点添加到图中 builder.add_node("llm_node", call_llm) # 4. 添加边：从开始 -> 节点 -> 结束 builder.add_edge(START, "llm_node") builder.add_edge("llm_node", END) # 5. 编译图，得到一个可执行的应用 app = builder.compile() ``` 这个最简单的图已经具备了状态管理能力。当我们传入一个包含初始消息的状态时，`call_llm`节点会接收到包含该消息的状态，并返回新的回复消息。由于`messages`字段使用了`add`归约器，回复消息会被自动追加，形成历史。 > **提示**：`START`和`END`是LangGraph内置的特殊节点，分别代表工作流的开始和结束。你不需要定义它们，直接使用即可。 ## 2. 为机器人注入“记忆”：检查点（Checkpointer）实战 上面我们构建的图在单次调用中能维护状态，但一旦调用结束，状态就消失了。下次再调用`app.invoke()`，它又是一个“全新失忆”的机器人。为了让记忆能**跨多次交互持久化**，我们需要引入**检查点（Checkpointer）**。 LangGraph提供了多种检查点后端，比如内存型的`MemorySaver`（适合短期、开发测试）和数据库型的`AsyncSqliteSaver`（适合生产环境长期存储）。这里我们用`MemorySaver`来演示。 ### 2.1 使用MemorySaver实现会话记忆 ```python from langgraph.checkpoint.memory import MemorySaver from langchain_community.chat_models import ChatTongyi # 示例使用通义千问，可替换为任何LangChain支持的模型 from langchain_core.messages import HumanMessage, AIMessage # 假设你已经有了一个模型实例 # model = ChatTongyi(model="qwen-max", api_key="your_key") # 为了示例，我们创建一个模拟模型函数 def mock_model_invoke(messages): """模拟大模型调用，根据最后一条用户消息生成回复""" last_msg = messages[-1].content if "名字" in last_msg: return AIMessage(content="我记得你刚才提到了名字，但我需要你明确告诉我你叫什么。") elif "我叫" in last_msg: # 简单解析出名字 name = last_msg.replace("我叫", "").strip() return AIMessage(content=f"好的，{name}，我记住你了！") else: return AIMessage(content="这是一个普通的回复。") # 修改之前的call_llm节点函数，接入模拟模型 def call_llm_with_memory(state: ChatState): messages = state['messages'] response = mock_model_invoke(messages) return {"messages": [response]} # 重新构建图 builder = StateGraph(ChatState) builder.add_node("llm_node", call_llm_with_memory) builder.add_edge(START, "llm_node") builder.add_edge("llm_node", END) # 关键步骤：创建MemorySaver实例，并在编译时传入 memory = MemorySaver() app_with_memory = builder.compile(checkpointer=memory) ``` 现在，我们的应用具备了记忆能力。但如何区分不同用户的对话呢？这就需要用到`config`参数中的`thread_id`。 ### 2.2 通过thread_id管理独立对话线程 `thread_id`是一个唯一标识符，用于区分不同的对话会话。同一个`thread_id`下的所有调用共享同一份检查点状态。 ```python # 配置一个对话线程ID config_zhang = {"configurable": {"thread_id": "user_zhangsan_001"}} # 第一轮对话 initial_state = { "messages": [HumanMessage(content="你好！")], "user_id": "zhangsan" } result1 = app_with_memory.invoke(initial_state, config=config_zhang) print(result1["messages"][-1].content) # 输出：这是一个普通的回复。 # 第二轮对话：直接传入新消息，状态会自动从检查点恢复 new_state = { "messages": [HumanMessage(content="我叫张三。")], # 注意：这里只传了新消息 # user_id 也会从检查点中恢复，所以这里可以不传，或者传了也会被合并 } result2 = app_with_memory.invoke(new_state, config=config_zhang) print(result2["messages"][-1].content) # 输出：好的，张三，我记住你了！ # 查看此时完整的状态消息列表 print(f"历史消息数: {len(result2['messages'])}") # 应该是3条：你好 + 回复1 + 我叫张三 + 回复2 for msg in result2['messages']: print(f"{type(msg).__name__}: {msg.content}") ``` > **注意**：在启用检查点后调用`invoke`时，我们传入的`state`实际上是**对已保存状态的更新**，而不是完全替换。LangGraph会将传入的更新与检查点中保存的状态合并（根据归约器规则），然后执行图。这就是记忆得以延续的原理。 ### 2.3 不同线程的隔离性 你可以轻松创建多个独立的对话线程，它们之间的记忆互不干扰。 ```python config_li = {"configurable": {"thread_id": "user_lisi_002"}} # 李四开始一个新对话 result_li = app_with_memory.invoke( {"messages": [HumanMessage(content="我的名字是什么？")], "user_id": "lisi"}, config=config_li ) print(result_li["messages"][-1].content) # 输出：我记得你刚才提到了名字...（但李四的对话历史是空的，所以模型实际上没有“刚才”的记忆） ``` 这个例子清晰地展示了`thread_id`的隔离作用。张三的对话历史（包含“我叫张三”）只保存在`thread_id: user_zhangsan_001`的检查点下，李四的线程完全访问不到。 ## 3. 构建复杂对话逻辑：条件分支与工具调用 一个只会简单回复的机器人显然不够看。现实中，机器人需要能判断用户意图，决定是直接回答，还是去查询天气、搜索资料、执行计算等。这就需要用到**条件边（Conditional Edges）**和**工具调用（Tool Calling）**。 ### 3.1 设计一个能调用工具的机器人 假设我们要构建一个能查询天气的机器人。工作流如下： 1. **Agent节点**：接收用户消息，由大模型判断是否需要调用天气查询工具。如果需要，模型会返回一个“工具调用请求”。 2. **条件路由**：检查上一步的结果，如果包含工具调用，则路由到`tools`节点；否则，直接结束。 3. **Tools节点**：执行具体的工具（如查询天气API），并返回结果。 4. 工具执行结果返回给`agent`节点，由模型将工具结果整合成自然语言回复给用户。 首先，我们定义一个简单的天气查询工具（这里用模拟数据）： ```python from langchain_core.tools import tool import json @tool def get_weather(city: str): """根据城市名称查询当前天气情况。""" # 模拟一个天气数据源 weather_data = { "北京": "晴，15°C，西北风2级", "上海": "多云，18°C，东南风1级", "广州": "阵雨，25°C，南风3级", "成都": "阴，12°C，微风", } return weather_data.get(city, f"未找到{city}的天气信息。") ``` ### 3.2 实现带条件分支的工作流 我们需要使用`add_conditional_edges`来创建分支。路由函数根据当前状态决定下一个节点。 ```python from typing import Literal from langgraph.prebuilt import ToolNode from langchain_core.messages import AIMessage, ToolMessage # 假设我们有一个绑定了工具的模型 # 这里为了演示，创建一个模拟的“绑定工具的模型”行为 def mock_agent_node(state: ChatState): """模拟Agent节点：分析用户输入，决定是否调用工具。""" last_message = state['messages'][-1] user_input = last_message.content if hasattr(last_message, 'content') else "" # 简单的意图判断 if any(word in user_input for word in ["天气", "weather", "气候"]): # 模拟模型决定调用工具，并生成工具调用请求 # 这里需要解析出城市，实际中可由LLM完成 city = "北京" # 简化处理 tool_call_id = "call_weather_001" # 返回一个包含工具调用的AIMessage tool_call_msg = AIMessage( content="", # 模型思考内容可为空或包含中间推理 tool_calls=[{ "name": "get_weather", "args": {"city": city}, "id": tool_call_id, }] ) return {"messages": [tool_call_msg]} else: # 不需要工具，直接生成回复 reply = AIMessage(content=f“我收到你的消息：'{user_input}'。这是一个普通回复。”) return {"messages": [reply]} def route_after_agent(state: ChatState) -> Literal['tools', END]: """路由函数：判断Agent节点后该去哪里""" last_message = state['messages'][-1] # 如果最后一条消息是AIMessage且包含工具调用，则去tools节点 if isinstance(last_message, AIMessage) and hasattr(last_message, 'tool_calls') and last_message.tool_calls: return 'tools' # 否则，工作流结束 return END # 构建图 builder = StateGraph(ChatState) # 添加节点 builder.add_node("agent", mock_agent_node) # 使用LangGraph预建的ToolNode，它会自动执行消息中的工具调用 tool_node = ToolNode([get_weather]) # 传入工具列表 builder.add_node("tools", tool_node) # 设置入口点 builder.set_entry_point("agent") # 添加条件边：从agent节点出来，根据route_after_agent函数的结果决定去向 builder.add_conditional_edges( "agent", route_after_agent, { "tools": "tools", # 如果函数返回'tools'，则前往'tools'节点 END: END # 如果函数返回END，则直接结束 } ) # 添加普通边：tools节点执行完后，必须回到agent节点，让模型总结工具结果 builder.add_edge("tools", "agent") # 注意：这里agent -> tools -> agent 形成了一个循环。 # 但第二次进入agent节点时，路由函数会看到最新的状态（包含ToolMessage）， # 此时last_message是ToolMessage，不包含tool_calls，所以会路由到END，结束循环。 ``` 这个结构实现了经典的“思考-行动-观察”循环。Agent思考后决定行动（调用工具），Tools节点执行行动并返回观察结果，Agent再根据观察结果进行下一步思考或最终回复。 ### 3.3 编译并测试复杂工作流 ```python # 编译图，并启用记忆 memory = MemorySaver() weather_bot = builder.compile(checkpointer=memory) config = {"configurable": {"thread_id": "weather_chat_1"}} # 测试1：询问天气 state1 = {"messages": [HumanMessage(content="今天北京天气怎么样？")]} result1 = weather_bot.invoke(state1, config=config) print("=== 测试1：询问天气 ===") # 打印最后一条消息，应该是模型整合了天气信息后的回复 # 由于我们的mock_agent_node在收到ToolMessage后直接回复的逻辑没写全，这里可能看不到最终回复。 # 一个更完整的实现需要在agent节点中处理ToolMessage。 # 让我们完善一下mock_agent_node，使其能处理来自tools节点的ToolMessage def improved_agent_node(state: ChatState): last_message = state['messages'][-1] # 情况1：上一条是用户消息，需要判断意图 if isinstance(last_message, HumanMessage): user_input = last_message.content if any(word in user_input for word in ["天气", "weather"]): city = "北京" # 简化 tool_call_msg = AIMessage( content="", tool_calls=[{"name": "get_weather", "args": {"city": city}, "id": "call_1"}] ) return {"messages": [tool_call_msg]} else: return {"messages": [AIMessage(content="我可以帮你查天气哦。")]} # 情况2：上一条是ToolMessage（工具执行结果），需要生成最终回复 elif isinstance(last_message, ToolMessage): # 从ToolMessage中提取结果 tool_result = last_message.content final_reply = AIMessage(content=f“查询完成。天气情况是：{tool_result}”) return {"messages": [final_reply]} # 情况3：其他情况（如上一条是AIMessage），理论上不会进入，直接返回空更新 return {"messages": []} # 更新图并重新测试 builder.update_node("agent", improved_agent_node) weather_bot_v2 = builder.compile(checkpointer=memory) print("\n=== 使用改进版Agent测试 ===") result_v2 = weather_bot_v2.invoke( {"messages": [HumanMessage(content="查询一下北京天气")]}, config={"configurable": {"thread_id": "test_v2"}} ) for msg in result_v2['messages']: print(f"{type(msg).__name__}: {getattr(msg, 'content', 'N/A')}") ``` 输出可能会显示类似这样的消息流： ``` HumanMessage: 查询一下北京天气 AIMessage: N/A (但内部有tool_calls) ToolMessage: 晴，15°C，西北风2级 AIMessage: 查询完成。天气情况是：晴，15°C，西北风2级 ``` 这表明工作流成功地走完了`agent -> tools -> agent`的循环，并给出了最终答案。 ## 4. 项目化与部署考量 当我们有了一个功能完整的机器人工作流后，接下来要考虑的就是如何将它集成到真实的项目中，比如一个Web API服务。这里有几个关键点需要注意。 ### 4.1 状态序列化与检查点后端选择 在开发环境，`MemorySaver`很方便，但它将状态保存在进程内存中，服务重启后记忆会全部丢失，且无法在多个服务进程间共享。对于生产环境，你需要一个持久化的检查点后端。 **AsyncSqliteSaver** 是一个不错的选择，它将状态存储在SQLite数据库中。你也可以实现自定义的后端，比如连接到PostgreSQL、Redis或任何你喜欢的存储。 ```python # 示例：使用AsyncSqliteSaver (异步) import asyncio from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver async def run_with_sqlite(): async with AsyncSqliteSaver.from_conn_string("chat_memory.db") as sqlite_memory: # 使用同一个builder，但编译时换用sqlite检查点 app = builder.compile(checkpointer=sqlite_memory) config = {"configurable": {"thread_id": "sqlite_user_1"}} # 注意：需要使用异步调用 result = await app.ainvoke( {"messages": [HumanMessage(content="测试持久化记忆")]}, config=config ) print(result['messages'][-1].content) # 即使程序重启，只要thread_id相同，就能恢复对话状态 # asyncio.run(run_with_sqlite()) ``` ### 4.2 集成到FastAPI Web服务 将LangGraph应用封装成API是常见的做法。下面是一个简单的FastAPI示例框架： ```python from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional import uuid app = FastAPI(title="LangGraph Chatbot API") # 定义请求/响应模型 class ChatRequest(BaseModel): message: str thread_id: Optional[str] = None # 如果不提供，则创建新会话 user_id: Optional[str] = "anonymous" class ChatResponse(BaseModel): reply: str thread_id: str full_history: List[str] # 可选，返回简化版历史 # 假设我们已经有一个编译好的、带检查点的LangGraph应用实例：chatbot_app # chatbot_app = builder.compile(checkpointer=production_memory) @app.post("/chat", response_model=ChatResponse) async def chat_endpoint(request: ChatRequest): if not request.thread_id: # 生成新的会话ID request.thread_id = str(uuid.uuid4()) config = {"configurable": {"thread_id": request.thread_id}} try: # 调用LangGraph工作流 result = chatbot_app.invoke( { "messages": [HumanMessage(content=request.message)], "user_id": request.user_id }, config=config ) # 提取最后一条AI回复 last_msg = result['messages'][-1] if hasattr(last_msg, 'content'): reply_text = last_msg.content else: reply_text = "[模型未返回文本内容]" # 简化历史，只提取文本 history_texts = [] for msg in result['messages']: if hasattr(msg, 'content') and msg.content: prefix = "用户: " if isinstance(msg, HumanMessage) else "助手: " history_texts.append(prefix + msg.content) return ChatResponse( reply=reply_text, thread_id=request.thread_id, full_history=history_texts ) except Exception as e: raise HTTPException(status_code=500, detail=f"处理请求时出错: {str(e)}") # 启动命令：uvicorn main:app --reload ``` 这个API提供了最基本的聊天功能，并自动管理会话线程。前端应用只需要在每次请求时带上相同的`thread_id`，就能维持连续的对话。 ### 4.3 性能优化与监控 当你的机器人上线后，下面这些实践可以帮助你更好地管理和优化它： * **使用LangSmith进行跟踪**：LangGraph与LangSmith无缝集成。在编译应用时传入LangSmith的配置，可以记录每一次图执行、每个节点的输入输出、耗时等，对于调试复杂的工作流和优化性能至关重要。 ```python from langsmith import Client import os os.environ["LANGCHAIN_TRACING_V2"] = "true" os.environ["LANGCHAIN_PROJECT"] = "My-Chatbot-Production" # 编译时无需额外操作，只要环境变量设置正确，链路会自动上报到LangSmith ``` * **设置超时与错误处理**：在节点函数中，对模型调用和工具调用添加超时逻辑，避免单个请求阻塞整个工作流。LangGraph的状态检查点机制本身也支持错误恢复，你可以在某个节点失败后，从上一个检查点重试。 * **状态设计优化**：不是所有数据都需要放在状态里被永久记忆。对于非常大的消息历史，考虑只保留最近N轮对话的原始消息，或者将更早的历史总结成一段摘要存储在状态中，以节省存储空间和降低模型上下文长度压力。这可以通过在状态中增加一个`conversation_summary`字段，并设计一个专门的“总结节点”来实现。 构建一个真正实用、健壮的AI聊天机器人，LangGraph提供的状态管理和工作流编排能力只是一个强大的起点。如何设计更智能的对话逻辑、如何与业务系统深度集成、如何保障服务的稳定性和可扩展性，这些都是更值得深入探索的课题。从我自己的经验来看，从这个小而美的记忆机器人开始，逐步迭代和扩展，是探索AI应用开发这条道路的一个非常踏实的方式。