【LangChain_V1.0从入门到实践】第7章 记忆(Memory)

记忆是一种能记录以往交互信息的系统。对于AI Agent而言，记忆至关重要，因为它能让智能体记住之前的交互过程、从反馈中学习，并适应用户的偏好。当智能体需要处理更复杂的任务时，这种记忆能力对于提升效率和满足用户需求方面都不可或缺。短期记忆能让你的应用程序记住单一线程或单次对话内的过往交互信息。对话历史时短期记忆最常见的形式。

超长对话对当前大语言模型构成挑战：完整的对话历史可能无法装进大模型的上下文窗口，进而导致上下文丢失或者出错。即便你的模型支持完整的上下文长度，大多数大型语言模型在处理长上下文时表现依然欠佳。它们会被过时 或偏离主题的内容 “分散注意力”，同时还会面临响应速度变慢、成本升高的问题。

7.1. 短期记忆基本使用方法

短期记忆让应用程序能记住单个对话线程中的历史交互。在创建智能体时，通过设置
checkpointer参数来启用记忆功能。

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool

load_dotenv()

llm = init_chat_model(
    "deepseek:deepseek-chat",
    temperature=0.1,
    max_retries=2,
    timeout=None
)

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气晴朗"

#创建检查点
checkpoint = InMemorySaver()

#创建agent
agent = create_agent(
    model=llm,
    tools=[get_weather],
    checkpointer=checkpoint
)

# 使用相同thread_id维持对话上下文
config = {"configurable": {"thread_id": "example_thread_1"}}
response1 = agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}, config)
print(response1)
response2 = agent.invoke({"messages": [{"role": "user", "content": "我刚才说了我叫什么？"}]}, config)
print(response2)

执行结果：

说明：

从执行结果看，第二次执行会话，将第一次请求的Messages也带上了。Checkpoint采用InMemorySaver，将消息存储在内存中，简单场景测试可以，生产环境需要存储到外部存储设备中。

7.2. 记忆持久化

对于需要长期运行和可靠记忆的应用，推荐使用数据库进行持久化。以下以 PostgreSQL 为例：

安装依赖

pip install langgraph-checkpoint-postgres
pip uninstall psycopg2 psycopg-binary psycopg
pip install psycopg-binary
pip install psycopg

示例脚本

from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent
from langgraph.checkpoint.postgres import PostgresSaver
from langchain.tools import tool

load_dotenv()

llm = init_chat_model(
    "deepseek:deepseek-chat",
    temperature=0.1,
    max_retries=2,
    timeout=None
)

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气晴朗"

# 数据库连接字符串
DB_URI = "postgresql://postgres:12345@localhost:5432/postgres"

# 创建agent
"""
在Python中，with语句是一个非常实用的语法结构，主要用于简化资源管理，确保像文件、网络连接或数据库会话这样的资源在使用后能被正确、及时地释放，即使在操作过程中发生了异常也是如此。
"""
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup() # 初始化数据库
    agent = create_agent(
        model=llm,
        tools=[get_weather],
        checkpointer=checkpointer
    )
    # 使用相同thread_id维持对话上下文
    config = {"configurable": {"thread_id": "example_thread_1"}}
    response1 = agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}, config)
    print(response1)
    response2 = agent.invoke({"messages": [{"role": "user", "content": "我刚才说了我叫什么？"}]}, config)
    print(response2)

执行结果

postgreSQL数据库中已创建4表，并且表中已写入消息数据。

7.3. 定制化记忆

LangChain 1.0 通过
AgentState管理短期记忆，这是一个可扩展的状态字典。默认包含
messages键，用于存储历史对话消息。开发者可以通过添加额外的字段来扩展AgentState。扩展
AgentState的核心意义在于，它让智能体从一个仅能记忆对话内容的聊天机器人，升级为一个能够在复杂任务中保持上下文、记录中间过程并存储自定义信息的真正意义上的“智能执行体”。下表清晰地对比了默认状态与扩展状态的主要差异。如果采用了持久化存储组件，扩展后的字段信息也会一并存储。

特性维度	默认 AgentState (仅有 messages)	扩展后的 AgentState (自定义字段)
记忆范围	仅限对话历史	对话历史 + 任务状态 + 用户偏好 + 中间结果等
适用任务	简单的一问一答、短对话	多步骤复杂任务、需持久化特定信息的场景
状态管理	自动追加消息	可精细控制不同字段的更新逻辑（如追加、覆盖）
信息利用率	模型只能看到对话历史	模型和工具可读取/写入更丰富的结构化信息

示例脚本

"""
持久化扩展记忆信息
"""
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent,AgentState
from langgraph.checkpoint.postgres import PostgresSaver
from langchain.tools import tool

load_dotenv()

llm = init_chat_model(
    "deepseek:deepseek-chat",
    temperature=0.1,
    max_retries=2,
    timeout=None
)

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气晴朗"

# 1. 定义自定义状态，例如，为旅行规划智能体添加目的地和预算字段
class TravelPlanningState(AgentState):
    destination: str
    budget: float
    confirmed_itinerary: list  # 存储已确认的行程项

# 数据库连接字符串
DB_URI = "postgresql://postgres:12345@localhost:5432/postgres"

# 2. 创建agent
"""
在Python中，with语句是一个非常实用的语法结构，主要用于简化资源管理，确保像文件、网络连接或数据库会话这样的资源在使用后能被正确、及时地释放，即使在操作过程中发生了异常也是如此。
"""
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup() # 初始化数据库
    agent = create_agent(
        model=llm,
        tools=[get_weather],
        state_schema=TravelPlanningState, # 指定自定义状态
        checkpointer=checkpointer
    )
    # 使用相同thread_id维持对话上下文
    config = {"configurable": {"thread_id": "example_thread_123"}}
    response1 = agent.invoke({
        "messages": [{"role": "user", "content": "我想规划一次去日本的旅行。旅行时间是5天4晚。"}],
        "destination": "日本",  # 设置初始目的地
        "budget": 15000.0,  # 设置初始预算
        "confirmed_itinerary": []  # 初始化行程列表
    }, config)
    print(response1)
    response2 = agent.invoke({"messages": [{"role": "user", "content": "刚才提供给我规划是什么？"}]}, config)
    print(response2)

7.4. 长对话记忆管理

当对话长度超过模型上下文窗口时，需要采用特殊策略管理记忆。策略有：

记忆修剪策略；记忆摘要策略；记忆删除策略；

7.4.1. 记忆修剪策略

修剪策略通过保留最近的关键消息来控制上下文长度：

"""
修剪消息
"""
from dotenv import load_dotenv
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent, AgentState
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.runtime import Runtime

load_dotenv()

llm = init_chat_model(
    "deepseek:deepseek-chat",
    temperature=0.1,
    max_retries=2,
    timeout=None
)

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气晴朗"

@before_model
def trim_messages(state: AgentState, runtime: Runtime):
    """保留最近几条消息以适应上下文窗口"""
    messages = state["messages"]
    if len(messages) <= 3:
        return  None
    first_msg = messages[0]
    recent_messages = messages[-3:] if len(messages) % 2 == 0 else messages[-4:]
    new_messages = [first_msg] + recent_messages
    return {
        "messages": [
            RemoveMessage(id=REMOVE_ALL_MESSAGES),
            *new_messages
        ]
    }

#创建检查点
checkpoint = InMemorySaver()

#创建agent
agent = create_agent(
    model=llm,
    tools=[get_weather],
    middleware=[trim_messages],
    checkpointer=checkpoint
)

# 使用相同thread_id维持对话上下文
config = {"configurable": {"thread_id": "example_thread_2"}}
response1 = agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}, config)
print(response1)
response2 = agent.invoke({"messages": [{"role": "user", "content": "给我讲个笑话？"}]}, config)
print(response2)
response3 = agent.invoke({"messages": [{"role": "user", "content": "帮我写个关于程序员的唐诗"}]}, config)
print(response3)
response4 = agent.invoke({"messages": [{"role": "user", "content": "帮我写个关于雪的唐诗"}]}, config)
print(response4)
response5 = agent.invoke({"messages": [{"role": "user", "content": "帮我分析一下最近港股大盘情况"}]}, config)
print(response5)

7.4.2. 记忆删除策略

当记忆特别多时，你希望删除部分或者全部清空，则可以选择此策略。示例代码如下：

"""
删除记忆
"""
from dotenv import load_dotenv
from langchain.agents.middleware import before_model
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent, AgentState
from langchain.messages import RemoveMessage
from langgraph.checkpoint.memory import InMemorySaver
from langchain.tools import tool
from langgraph.runtime import Runtime

load_dotenv()

llm = init_chat_model(
    "deepseek:deepseek-chat",
    temperature=0.1,
    max_retries=2,
    timeout=None
)

@tool
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气晴朗"

@before_model
def delete_messages(state: AgentState, runtime: Runtime):
    """保留最近几条消息以适应上下文窗口"""
    messages = state["messages"]
    if len(messages) > 2:
        # remove the earliest two messages
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
    else:
        return None


#创建检查点
checkpoint = InMemorySaver()

#创建agent
agent = create_agent(
    model=llm,
    tools=[get_weather],
    middleware=[delete_messages],
    checkpointer=checkpoint
)

# 使用相同thread_id维持对话上下文
config = {"configurable": {"thread_id": "example_thread_2"}}
response1 = agent.invoke({"messages": [{"role": "user", "content": "我叫张三"}]}, config)
print(response1)
response2 = agent.invoke({"messages": [{"role": "user", "content": "给我讲个笑话？"}]}, config)
print(response2)
response3 = agent.invoke({"messages": [{"role": "user", "content": "帮我写个关于程序员的唐诗"}]}, config)
print(response3)
response4 = agent.invoke({"messages": [{"role": "user", "content": "帮我写个关于雪的唐诗"}]}, config)
print(response4)

7.4.3. 记忆摘要策略

如上所示，修剪或删除消息的问题在于，您可能会因消息队列的清理而丢失信息。因此，一些应用程序会受益于更复杂的方法，例如使用聊天模型来汇总消息历史记录。

要在Agent中汇总消息历史记录，可以使用内置中间件
SummarizationMiddleware

示例代码：

"""
内置中间件：SummarizationMiddleware
SummarizationMiddleware - 智能摘要
"""
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent

# 加载环境变量
load_dotenv()

"""
基于类定义中间件
"""
from langchain.agents.middleware import SummarizationMiddleware
from langgraph.checkpoint.memory import InMemorySaver

llm = init_chat_model(
    "deepseek:deepseek-chat",
    temperature=0.1,
    max_retries=2,
    timeout=None
)

summary_middleware = SummarizationMiddleware(
    model=llm,  # 用于摘要的模型
    max_tokens_before_summary=2000,  # 触发摘要的token阈值
    messages_to_keep=10  # 保留的最新消息数量
)

checkpointer = InMemorySaver()

agent = create_agent(
    model=llm,
    tools=[],
    middleware=[
        summary_middleware
    ],
    checkpointer=checkpointer,  # 检查点是启用记忆功能的必要条件
)

# 使用示例：在多轮对话中，中间件会自动管理上下文长度
config = {"configurable": {"thread_id": "example_thread_1"}}

# 第一轮对话
agent.invoke({"messages": [{"role": "user", "content": "你好，请介绍下你自己。"}]}, config=config)
# 第二轮对话
agent.invoke({"messages": [{"role": "user", "content": "帮我详细介绍一下LangChain 1.0版本"}]}, config=config)
# 第三轮对话
agent.invoke({"messages": [{"role": "user", "content": "作为新手，我该如何学习AI 应用开发？AI应用开发涉及到哪些知识点？"}]}, config=config)
# 第四轮对话
agent.invoke({"messages": [{"role": "user", "content": "请给我一个关于如何使用LangChain 1.0进行AI 应用开发的详细教程。"}]}, config=config)
# 当累计Token数超过3000时，中间件自动触发摘要
agent.invoke({"messages": [{"role": "user", "content": "我们刚才聊到的第一个主题是什么？"}]}, config=config)

7.5. 记忆访问与管理

7.5.1. 通过工具访问记忆

借助ToolRuntime参数在工具中访问短期记忆（状态）：

"""
通过工具访问短期记忆
"""
from dotenv import load_dotenv
from langchain.agents import create_agent, AgentState
from langchain.chat_models import init_chat_model
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import ToolRuntime

load_dotenv()

llm = init_chat_model(
    "deepseek:deepseek-chat",
    temperature=0.1,
    max_retries=2,
    timeout=None
)

class CustomState(AgentState):
    user_id: str

@tool
def get_user_info(
    runtime: ToolRuntime
) -> str:
    """Look up user info."""
    user_id = runtime.state["user_id"]
    return "User is John Smith" if user_id == "user_123" else "Unknown user"

#创建检查点
checkpoint = InMemorySaver()

#创建agent
agent = create_agent(
    model=llm,
    tools=[get_user_info],
    state_schema=CustomState,
    checkpointer=checkpoint
)

# 使用相同thread_id维持对话上下文
config = {"configurable": {"thread_id": "example_thread_3"}}
result = agent.invoke({
    "messages": "look up user information",
    "user_id": "user_123"
},config)
print(result["messages"][-1].content) #输出：User is John Smith.

7.5.2. 通过中间件动态管理记忆

中间件提供了在模型调用前后干预记忆处理的机制：

"""
动态提示词工程 (@dynamic_prompt)
"""
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import  after_model, dynamic_prompt, hook_config,ModelRequest
from langchain.messages import HumanMessage,AIMessage
import time

# 加载环境变量
load_dotenv()

# 1. 定义自定义状态，例如，为旅行规划智能体添加目的地和预算字段
class UserState(AgentState):
    user_name: str

@dynamic_prompt
def dynamic_conversation_strategy(request: ModelRequest) -> str:
    """基于记忆状态生成动态系统提示"""
    user_name = request.state.get("user_name", "用户")
    conversation_length = len(request.state.get("messages", []))

    if conversation_length > 10:
        return f"你正在与{user_name}进行深度交流，请提供详细回答"
    else:
        return f"你是{user_name}的助手，请保持回答简洁"


llm = init_chat_model(
    "deepseek:deepseek-chat",
    temperature=0.1,
    max_retries=2,
    timeout=None
)

agent = create_agent(
    model=llm,
    tools=[],
    state_schema=UserState,
    middleware=[dynamic_conversation_strategy]
)

# 正确的配置方式
config = {"configurable": {"thread_id": "user_123"}}
initial_state = {
    "messages": [HumanMessage(content="Hi, my name is Bob")],
    "user_name": "Bob"
}

response = agent.invoke(initial_state, config=config)
print(response)