文章目录

引言

Memory 是一个存储先前交互信息的系统。对于AI Agent 而言，Memory 至关重要，因为它能让 AI Agent 记住先前与大模型交互的情况，从反馈中学习并适应用户偏好。随着 AI Agent 在处理更复杂的任务时需要与用户进行大量交互，这种能力对于提高效率和提升用户满意度都至关重要。

一. 什么是LangChain短期记忆

短期记忆（Short-term memory） 是指在单个对话会话期间临时存储和管理的上下文信息，主要用于维持对话的连续性和相关性。通过将对话信息保存到内存中，能够暂时保存用户与大模型之间的交互历史，确保大模型能够记住当前对话中的关键细节，从而做出连贯的回应。这种记忆是临时的、容量有限的，通常只保留最近几轮对话，会话结束后自动清除。

二. LangChain短期记忆几种实现形式

2.1 基于内存形式

from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from pydantic import SecretStr

llm = ChatOpenAI(
    model="kimi-k2-0711-preview",  # 模型名称
    base_url="https://api.moonshot.cn/v1",  # 设置api端点
    api_key=SecretStr("your api key")  # api密钥
)

# 创建智能体（无工具、仅绑定内存检查点）
agent = create_agent(
    model=llm,
    checkpointer=InMemorySaver(),  # 基于内存方式透传检查点实现多轮记忆
)

# 固定会话ID，保证多轮状态一致
config = {"configurable": {"thread_id": "1"}}

print("多轮对话已启动，输入'exit'退出\n")
while True:
    # 动态获取控制台输入
    user_content = input("你: ")
    if user_content.strip().lower() == "exit":
        print("AI: 对话结束～")
        break

    # 调用智能体
    response = agent.invoke(
        {"messages": [{"role": "user", "content": user_content}]},
        {"configurable": {"thread_id": "1"}},
    )

    # 提取并打印AI回复
    ai_content = [msg for msg in response["messages"] if msg.type == "ai"][-1].content
    print(f"AI: {ai_content}\n")

在这里插入图片描述

2.2 基于redis形式

使用redis存储历史对话内容，需要redis服务器安装RediSearch模块和Rejson模块

# uv add langgraph-checkpoint-redis 添加langgraph-checkpoint-redis依赖
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.redis import RedisSaver
from pydantic import SecretStr

llm = ChatOpenAI(
    model="kimi-k2-0711-preview",  # 模型名称
    base_url="https://api.moonshot.cn/v1",  # 设置api端点
    api_key=SecretStr("your api key")  # api密钥
)

REDIS_ADDRESS = "redis://:password@192.168.197.129:6379"
with     RedisSaver.from_conn_string(REDIS_ADDRESS) as checkpointer:
    checkpointer.setup()  # 在 PostgresSql数据库中自动创建表
    agent = create_agent(
        model=llm,
        checkpointer=checkpointer,
    )

# 固定会话ID，保证多轮状态一致
print("多轮对话已启动，输入'exit'退出\n")
while True:
    # 动态获取控制台输入
    user_content = input("你: ")
    if user_content.strip().lower() == "exit":
        print("AI: 对话结束～")
        break

    # 调用智能体
    response = agent.invoke(
        {"messages": [{"role": "user", "content": user_content}]},
        {"configurable": {"thread_id": "1"}},
    )

    # 提取并打印AI回复
    response["messages"][-1].pretty_print()

在这里插入图片描述

2.3 基于postgresql形式

# uv add langgraph-checkpoint-postgres 添加langgraph-checkpoint-postgres依赖
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.postgres import PostgresSaver
from pydantic import SecretStr

llm = ChatOpenAI(
    model="kimi-k2-0711-preview",  # 模型名称
    base_url="https://api.moonshot.cn/v1",  # 设置api端点
    api_key=SecretStr("your api key")  # api密钥
)

DB_URI = "postgresql://username:password@localhost:5442/postgres?sslmode=disable"
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
    checkpointer.setup()  # 在数据库中自动创建表
    agent = create_agent(
        model=llm,
        checkpointer=checkpointer,
    )

# 固定会话ID，保证多轮状态一致
print("多轮对话已启动，输入'exit'退出\n")
while True:
    # 动态获取控制台输入
    user_content = input("你: ")
    if user_content.strip().lower() == "exit":
        print("AI: 对话结束～")
        break

    # 调用智能体
    response = agent.invoke(
        {"messages": [{"role": "user", "content": user_content}]},
        {"configurable": {"thread_id": "1"}},
    )

    # 提取并打印AI回复
    response["messages"][-1].pretty_print()

三. 大模型上下文窗口限制

启用短期记忆后，长时间对话可能会超出 LLM 的上下文窗口。当对话内容超过 LLM 的最大 token 容量时，模型会丢失早期信息，导致对话连贯性断裂。以下策略通过动态管理历史记录解决该问题。

3.1 Trim messages

大多数 LLM 都有最大支持上下文窗口（以 tokens 表示），决定何时截断消息的一种方法是统计消息历史记录中的 tokens 数量，并在接近该限制时进行截断。可以使用 LangChain 开发框架的裁剪消息实用工具，并指定要从列表中保留的 tokens 数量，以及用于处理边界的策略（例如，保留最后 max_tokens 数量）。要在智能体中裁剪消息历史记录，可以使用@before_model中间件装饰器。

from langchain.messages import RemoveMessage
from langchain_openai import ChatOpenAI
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig
from typing import Any

from pydantic import SecretStr


@before_model
def trim_messages(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
    # 获取当前所有消息
    messages = state["messages"]

    # 如果消息条数<=3不需要裁剪
    if len(messages) <= 3:
        return None  # No changes needed

    # 始终保留第一条消息（通常是系统提示）
    first_msg = messages[0]
    # 根据消息总数决定保留多少最近消息
    # 偶数条：保留最后3条，奇数条：保留最后4条
    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
        ]
    }


llm = ChatOpenAI(
    model="kimi-k2-0711-preview",  # 模型名称
    base_url="https://api.moonshot.cn/v1",  # 设置api端点
    api_key=SecretStr("your api key")  # api密钥
)

agent = create_agent(
    model=llm,
    middleware=[trim_messages],
    checkpointer=InMemorySaver(),
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}

agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()

在这里插入图片描述

3.2 Delete messages

开发者可从图状态（graph state）中删除消息来管理消息历史。这在需要移除特定消息或清空完整消息历史的场景下尤为实用。若要从图状态中删除消息，可使用 RemoveMessage工具。为确保RemoveMessage 生效，需使用一个配置了 add_messages 归约器的状态键。默认的 AgentState已内置此配置。

from langchain.messages import RemoveMessage
from langchain.agents import create_agent, AgentState
from langchain.agents.middleware import after_model
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.runtime import Runtime
from langchain_core.runnables import RunnableConfig
from pydantic import SecretStr


@after_model
def delete_old_messages(state: AgentState, runtime: Runtime) -> dict | None:
    """Remove old messages to keep conversation manageable."""
    messages = state["messages"]
    if len(messages) > 2:
        # remove the earliest two messages
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:2]]}
    return None

llm = ChatOpenAI(
    model="kimi-k2-0711-preview",  # 模型名称
    base_url="https://api.moonshot.cn/v1",  # 设置api端点
    api_key=SecretStr("your api key")  # api密钥
)

agent = create_agent(
    model=llm,
    tools=[],
    system_prompt="Please be concise and to the point.",
    middleware=[delete_old_messages],
    checkpointer=InMemorySaver(),
)

config: RunnableConfig = {"configurable": {"thread_id": "1"}}

for event in agent.stream(
    {"messages": [{"role": "user", "content": "hi! I'm bob"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

for event in agent.stream(
    {"messages": [{"role": "user", "content": "what's my name?"}]},
    config,
    stream_mode="values",
):
    print([(message.type, message.content) for message in event["messages"]])

在这里插入图片描述

3.3 Summarize messages

对消息进行截断或删除操作的潜在问题在于通过对消息队列执行裁剪或者删除操作，可能导致关键信息丢失。因此，一些应用场景更适合采用一种更先进的实现方式，即通过对话模型对历史消息进行智能摘要。

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langchain_core.runnables import RunnableConfig
from pydantic import SecretStr

checkpointer = InMemorySaver()


llm = ChatOpenAI(
    model="kimi-k2-0711-preview",  # 模型名称
    base_url="https://api.moonshot.cn/v1",  # 设置api端点
    api_key=SecretStr("your api key")  # api密钥
)

agent = create_agent(
    model=llm,
    tools=[],
    middleware=[
    		 # 使用SummarizationMiddleware方法 langchain版本 >= v1.1.0
        SummarizationMiddleware(
            model=llm,
            trigger=("tokens", 100),
            keep=("messages", 2)
        )
    ],
    checkpointer=checkpointer,
)

config: RunnableConfig = {"configurable": {"thread_id": "2"}}
agent.invoke({"messages": "hi, my name is bob"}, config)
agent.invoke({"messages": "write a short poem about cats"}, config)
agent.invoke({"messages": "now do the same but for dogs"}, config)
final_response = agent.invoke({"messages": "what's my name?"}, config)

final_response["messages"][-1].pretty_print()

在这里插入图片描述

四. 总结

本文通过基础示例介绍了 LangChain 短期记忆的使用方法。简而言之，短期记忆是 LangChain 中让大模型记住当前聊天内容的核心工具。它通过将对话历史临时保存下来，并在每次交互时自动将这些记忆信息注入给大模型的提示中，从而使原本“无状态”的大模型能够感知上下文，做出连贯且符合对话逻辑的回应。它提供了多种灵活策略，让开发者可以根据场景平衡性能与效果，是构建流畅对话应用的关键基础组件。

Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

Kubernetes CoreDNS 解析超时排查（ndots 与 NodeLocal DNSCache）

CoreDNS 超时先用 ClusterIP 直连排除数据面，再读 Pod的ndots 与 search。默认 ndots:5 让短名查询放大数倍，AAAA 再乘一倍。指标上区分 NXDOMAIN 瀑布和真正的 SERVFAIL/p99。NodeLocal DNSCache 降 RTT、不消灭放大；根上仍是 FQDN 或谨慎的 dnsConfig。改 DNS 全局配置必须可回滚，禁止先停 Core

avatar 2048 AI社区
cover

AIGC课程为什么从99元到数万元都有：先把“买工具”与“买结果”分开

avatar 2048 AI社区

AI视频总结工具有哪些？2026年5款热门工具横向评测

本文从客观角度介绍5款常见的AI视频总结工具：Ai好记、NoteAi、BibiGPT、通义听悟、NotebookLM。每款工具按同一套维度说明：支持的内容来源、转写与总结能力、操作方式、页面呈现，以及适用场景。文末附功能对比表和常见问题，供读者按需选择。

avatar 2048 AI社区