LangGraph实战：用MemorySaver+ChatGPT API快速搭建一个能记住上下文的天气查询机器人

LangGraph实战用MemorySaverChatGPT API快速搭建能记住上下文的天气机器人最近在开发对话式AI应用时我发现很多初学者都面临一个共同痛点如何让机器人记住之前的对话上周帮朋友调试一个天气查询项目时我们仅用2小时就基于LangGraph的MemorySaver功能实现了上下文记忆。下面分享这个实战案例的完整构建过程。1. 环境准备与工具定义1.1 初始化项目环境首先创建项目目录并安装核心依赖mkdir weather-bot cd weather-bot pip install langgraph langchain-openai python-dotenv在项目根目录创建.env文件存储API密钥OPENAI_API_KEY你的API密钥1.2 构建天气查询工具我们使用LangChain的tool装饰器创建天气查询函数。实际项目中可以接入真实天气API这里先用模拟数据演示from langchain_core.tools import tool tool def get_weather(location: str, date: str 今天) - str: 获取指定地点和日期的天气信息 参数 location: 查询地点如上海 date: 查询日期默认今天 返回 格式化天气信息字符串 return f{date}{location}天气晴25℃东南风3级提示工具函数的docstring会直接影响LLM对工具的理解建议明确参数类型和返回格式2. 构建对话状态机2.1 定义状态结构LangGraph使用TypedDict定义对话状态。我们的天气机器人只需要维护消息历史from typing import Annotated, TypedDict from langgraph.graph.message import add_messages class State(TypedDict): messages: Annotated[list, add_messages] # 自动处理消息追加2.2 创建核心节点定义两个关键节点模型节点调用ChatGPT处理用户输入工具节点执行天气查询等具体操作from langchain_openai import ChatOpenAI from langgraph.prebuilt import ToolNode # 初始化模型并绑定工具 llm ChatOpenAI(modelgpt-3.5-turbo) bound_llm llm.bind_tools([get_weather]) # 模型节点 def call_model(state: State): response bound_llm.invoke(state[messages]) return {messages: response} # 工具节点 tool_node ToolNode([get_weather])3. 实现记忆功能3.1 配置MemorySaverLangGraph的MemorySaver让状态持久化变得简单from langgraph.graph import StateGraph from langgraph.checkpoint.memory import MemorySaver # 创建状态图 graph StateGraph(State) graph.add_node(agent, call_model) graph.add_node(tools, tool_node) # 设置条件转移 def should_continue(state: State): last_msg state[messages][-1] return tools if hasattr(last_msg, tool_calls) else __end__ graph.add_edge(tools, agent) graph.add_conditional_edges(agent, should_continue) # 启用记忆功能 memory MemorySaver() app graph.compile(checkpointermemory)3.2 线程隔离实践通过thread_id区分不同会话实现多用户隔离config {configurable: {thread_id: user123}} # 用户唯一标识 # 第一轮对话 input_msg HumanMessage(content今天北京天气怎样) for event in app.stream({messages: [input_msg]}, config): print(event[messages][-1].content) # 第二轮对话机器人记得之前的内容 input_msg HumanMessage(content我刚才问了什么) for event in app.stream({messages: [input_msg]}, config): print(event[messages][-1].content) # 输出您刚才询问了今天北京的天气4. 性能优化与扩展4.1 内存管理技巧当处理大量会话时可以添加自动清理机制from datetime import datetime, timedelta class TimedMemorySaver(MemorySaver): def __init__(self, ttltimedelta(hours1)): self.ttl ttl super().__init__() def get(self, config): thread_id config[configurable][thread_id] if thread_id in self.store: timestamp, _ self.store[thread_id] if datetime.now() - timestamp self.ttl: del self.store[thread_id] return super().get(config)4.2 支持更多工具类型扩展机器人能力只需添加新工具函数tool def get_air_quality(city: str) - str: 获取城市空气质量指数 return f{city}AQI78良 # 更新工具列表 tools [get_weather, get_air_quality] bound_llm llm.bind_tools(tools)5. 常见问题排查在实际部署中可能会遇到问题1工具调用不触发检查工具函数的参数类型提示验证模型是否成功绑定工具print(bound_llm.tools)问题2记忆丢失确保每次请求使用相同的thread_id检查MemorySaver实例是否被意外重建问题3响应延迟高尝试减小max_tokens参数使用streamTrue实现渐进式响应这个项目最让我惊喜的是LangGraph的状态管理设计——用200行代码就实现了传统需要复杂数据库支持的对话记忆功能。最近在客户项目中我们基于这个原型扩展出了支持10工具的客服系统MemorySaver在原型验证阶段节省了大量开发时间。