当前位置：首页 > news >正文

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

news 2026/5/12 18:34:12

LangGraph实战：用MemorySaver+ChatGPT 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(model="gpt-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 配置MemorySaver

LangGraph的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(checkpointer=memory)

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, ttl=timedelta(hours=1)): 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}AQI：78（良）" # 更新工具列表 tools = [get_weather, get_air_quality] bound_llm = llm.bind_tools(tools)