一、概述

模型本身是无状态，是没有记忆的。

但是为了保证上下文交流的连续性，可以给创建Agent添加记忆的功能。

1、Agent的记忆分类

• 短期记忆：当前任务或会话的上下文。

• 长期记忆：跨任务或会话的经验与知识。

2、短期记忆的工具类

AgentState （智能体状态）：Agent 在执行任务时，需要记住用户的问题、调用工具返回的结果、当前执行到哪一步、以及下一步计划等信息，这些所有信息的集合就是 AgentState。它就像一个共享的内存空间，让 Agent 在多个步骤之间保持连贯，而不是每次都从零开始。

Checkpointer（快照）：是 LangGraph 框架中负责保存和管理检查点的工具，它把图每次执行后的状态快照存下来，让 AI 应用能记住对话、断点续跑和回溯调试。它就像给 AI 工作流做的“存档”功能。每次图执行完一个步骤，Checkpointer 就会自动保存当前状态；下次用同一个 thread_id 调用时，它会把上次的状态加载回来，接着往下走，而不是重新开始。

二、代码示例

1、Checkpointer在内存中实现短期记忆

# 环境初始化
import os
import base64

from langchain.messages import HumanMessage,SystemMessage,AIMessage
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from dotenv import load_dotenv
from ipywidgets import FileUpload
from IPython.display import display, HTML
from pathlib import Path
from openai import OpenAI
import time


load_dotenv()

# 构建模型
model=init_chat_model(
    model='qwen3.8-omni-flash',
    model_provider='openai',
    base_url='https://maas.qianwenaiapi.com/compatible-mode/v1',
    api_key=os.getenv("DASHSCOPE_API_KEY","").strip(),
)

#定义工具
@tool
def getWeather(location:str) -> str:
    '''
        Get the weather conditions for a given location
        Args:
            location (str):city name of location
    '''
    import requests

     # 基本参数配置
    apiUrl ='https://apis.juhe.cn/simpleWeather/query'  # 接口请求URL
    apiKey =os.getenv("JUHE_API_KEY","").strip()  # 在个人中心->我的数据,接口名称上方查看

    # 接口请求入参配置
    requestParams = {
        'key': apiKey,
        'city': location,
        'future_num': '',
    }

    # 发起接口网络请求
    response = requests.get(apiUrl, params=requestParams)

    # 解析响应结果
    if response.status_code == 200:
        responseResult = response.json()
        # 网络请求成功。可依据业务逻辑和接口文档说明自行处理。
        # print(responseResult)
        return responseResult
    else:
        # 网络异常等因素，解析结果异常。可依据业务逻辑自行处理。
        # print('请求异常')
        return '请求异常'


# 创建智能体
agent=create_agent(
    model=model,
    tools=[getWeather],
    checkpointer= InMemorySaver(),

)

# 定义会话标识
session_id={
    "configurable":{"thread_id":"session_01"}
}

# 第1次调用
response=agent.invoke(
    {"messages":[HumanMessage(content="明天东台的天气如何？")]},
    session_id
)
print(response)

# 第二次调用
response=agent.invoke(
    {"messages":[HumanMessage(content="不好意思，我刚刚问的是哪个地方的天气？")]},
    session_id
)
print(response)

2、基于数据库的短期记忆持久化保存

从langchain官网上看，目前支持的数据库常见的有 sqlite、postgres、mongodb、cosmosdb等。

代码示例如下：

# 环境初始化
import os
import base64
import sqlite3

from anyio.lowlevel import checkpoint
from langchain.messages import HumanMessage,SystemMessage,AIMessage
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langgraph.checkpoint.memory import InMemorySaver
from dotenv import load_dotenv
from ipywidgets import FileUpload
from IPython.display import display, HTML
from pathlib import Path
from openai import OpenAI
import time
from langgraph.checkpoint.sqlite import SqliteSaver


load_dotenv()


# sqlite 初始化
conn=sqlite3.connect('resources/ckp.db',check_same_thread=False)
checkpointer=SqliteSaver(conn)

# 初始ckeckpoint 的 tables表到sqlite 非常重要
checkpointer.setup()


# 构建模型
model=init_chat_model(
    model='qwen3.8-omni-flash',
    model_provider='openai',
    base_url='https://maas.qianwenaiapi.com/compatible-mode/v1',
    api_key=os.getenv("DASHSCOPE_API_KEY","").strip(),
)

#定义工具
@tool
def getWeather(location:str) -> str:
    '''
        Get the weather conditions for a given location
        Args:
            location (str):city name of location
    '''
    import requests

     # 基本参数配置
    apiUrl ='https://apis.juhe.cn/simpleWeather/query'  # 接口请求URL
    apiKey =os.getenv("JUHE_API_KEY","").strip()  # 在个人中心->我的数据,接口名称上方查看

    # 接口请求入参配置
    requestParams = {
        'key': apiKey,
        'city': location,
        'future_num': '',
    }

    # 发起接口网络请求
    response = requests.get(apiUrl, params=requestParams)

    # 解析响应结果
    if response.status_code == 200:
        responseResult = response.json()
        # 网络请求成功。可依据业务逻辑和接口文档说明自行处理。
        # print(responseResult)
        return responseResult
    else:
        # 网络异常等因素，解析结果异常。可依据业务逻辑自行处理。
        # print('请求异常')
        return '请求异常'

# 创建智能体
agent=create_agent(
    model=model,
    tools=[getWeather],
    checkpointer= checkpointer,

)

# 定义会话标识
session_id={
    "configurable":{"thread_id":"session_02"}
}

# 第1次调用
response=agent.invoke(
    {"messages":[HumanMessage(content="明天南京的天气如何？")]},
    session_id
)
print(response)

# 第二次调用
response=agent.invoke(
    {"messages":[HumanMessage(content="不好意思，我刚刚问的是哪个地方的天气？")]},
    session_id
)
print(response)

# 第三次调用
response=agent.invoke(
    {"messages":[HumanMessage(content="明天这个地方这种天气，可以去哪边游玩？")]},
    session_id
)
print(response)