Agent 短期会话记忆实现方案 · devcfg

1. 引言

大模型上下文窗口的 token 上限（通常是 4K–128K token），决定了单次对话中能承载的信息量有限。Agent 在执行多轮工具调用或复杂推理时，早期对话内容会随着上下文增长而被截断，导致“失忆”问题。本文专门讨论短期会话记忆的实现方案——即如何在当前会话内高效管理对话历史，使 Agent 能够维持上下文连贯性。

读者阅读后能理解：短期记忆面临的核心挑战（token 溢出、信息丢失、检索效率），掌握从简单截断到向量检索的多种工程实现路径，以及如何在 LangGraph 等框架下落地 MemorySaver、摘要压缩、向量数据库等技术方案。本文不涉及长期跨会话记忆（如用户画像、知识库持久化）的完整设计，但会在后续章节讨论二者的衔接点。

2. 核心概念：短期记忆的定义与边界

2.1 短期记忆在 Agent 记忆系统中的定位

Agent 的记忆系统通常分为三个层次：

• 工作记忆（Working Memory）：当前执行任务中的临时状态，例如“刚调用了天气 API，正在解析返回结果”。这部分数据仅存在于单次推理的上下文窗口中，随模型推理结束自动丢弃，通常由提示词中的系统消息或对话记录承载。
• 短期记忆（Short-term Memory）：同一会话内跨多个步骤的上下文。

例如用户先问“北京天气怎么样”，接着问“那上海呢”，Agent 需要记住“刚才在讨论城市天气”这一主题。短期记忆需要工程手段管理，否则 token 溢出后上下文被截断，Agent 就会丢失对话主题。

• 长期记忆（Long-term Memory）：跨会话的用户偏好、业务规则、历史交互记录。

通常借助外部存储（向量数据库、关系数据库）实现持久化。

本文聚焦于短期记忆（会话级）。它与工作记忆的区别在于：工作记忆是模型当前正在处理的短窗口（通常仅数轮），而短期记忆需要主动管理，在 token 接近上限时决定保留哪些信息、丢弃哪些信息。

2.2 短期记忆的核心挑战

短期记忆的实现面临三个相互制约的问题：

• 上下文窗口截断（token 溢出）：这是最直接的问题。当对话轮次增多，累积的消息 token 数超过模型上下文窗口时，最早的对话被截断。轻则丢失用户意图，重则导致 Agent 执行后续步骤时缺乏必要的前置信息。

• 信息丢失：截断或压缩策略必然带来信息损失。如固定窗口截断会完全丢弃旧对话；摘要压缩会丢失细节，尤其是数值、地址、具体指令等精确信息。

• 检索效率：当需要在大量历史对话中快速定位关键信息时（如用户在三轮前提到的“客户 ID”），线性扫描全部历史记录的时延与 token 成本都不可接受。需要高效的检索机制。

这三个挑战之间存在权衡：要保留更多信息，就需要更复杂的压缩或检索方案，但会引入额外的计算开销和延迟。工程实践中需要根据业务场景（高实时性 vs 高准确性）选择合适方案。

3. 基础方案：上下文窗口截断解决方案

3.1 固定窗口滑动

最简单的策略：固定保留最近 N 轮对话，丢弃所有更早的记录。实现方式如下：

• 维护一个循环队列或双端队列，容量设置为 N（例如 20 轮）。
• 每增加一轮新对话，若队列已满则弹出最早的一轮。
• 将队列中的对话消息拼接为 LLM 的 messages 列表。

优点：实现极简单，无额外计算成本，延迟几乎为零。适用于短会话场景（如单次查询、单轮工具调用）。

缺点：一旦早期对话被丢弃，信息彻底丢失。若用户在对话中突然提到“刚才的那个参数是多少”，Agent 无法回答。同时，N 的选取需要人工经验：对于工具调用密集的场景，20 轮对话可能已包含大量工具返回数据导致 token 超标；对于纯聊天场景，20 轮对话可能只占少量 token。

3.2 基于 token 阈值的截断

比固定轮次更精细的做法：跟踪当前 messages 列表的总 token 数，当超过阈值时，从最早的消息开始逐条丢弃，直到总 token 数低于阈值。

• 阈值通常设为模型最大上下文窗口的 70%–80%，预留一定 token 给后续的回复生成。
• 丢弃策略可以是“丢弃最早的单条消息”或“丢弃最早的若干条”。

优点：能更高效地利用上下文窗口（固定轮次可能窗口未满就丢弃，或窗口溢出却未及时丢弃）。适合对话长度不均的场景。

缺点：与固定窗口一样，丢弃后信息彻底丢失。对于长对话场景，两种方案都不够理想。

这两种方案适用于原型验证、对历史信息要求不高的短期交互（如客服场景，仅需知道当前轮次的问题）。若业务需要频繁引用历史信息，应使用更高级的方案。

4. 中间方案：对话历史摘要压缩技术

4.1 原理

摘要压缩的核心思路：当对话历史即将超过 token 阈值时，调用 LLM 将当前 messages 列表中最早的若干条消息（或全部消息）压缩为一段简要摘要，用这条摘要替换原始消息。摘要占据的 token 数远小于原始消息，从而在上下文中“腾出空间”。

实现流程：

1. 持续收集对话 messages。
2. 当 messages 总 token 数接近阈值（例如 80%）时，触发压缩操作。
3. 将需要压缩的历史消息（通常是最早的部分）发送给 LLM，prompt 示例：“请将以下对话内容压缩为一段摘要，保留关键信息（如用户指定参数、工具执行结果、结论等），字数不超过 XXX token。

”
4. LLM 返回摘要文本。将原始历史消息替换为摘要消息（通常作为 system 消息或 assistant 消息，表明这是历史摘要）。
5. 继续后续对话，新的消息追加到列表中。

4.2 工程实践要点

压缩触发时机：建议基于总 token 数动态触发，而非固定轮次。例如设置 token 阈值为 max_tokens * 0.8，当实际 token 超过该阈值时执行压缩。若每次压缩后 token 仍接近上限，可执行多轮压缩（压缩一部分后，再压缩另一部分）。

压缩存储位置：摘要可以存储在内存中的 messages 列表里（如上所述），也可以持久化到外部缓存（Redis、SQLite）以备后续查询。外部存储的好处是，如果用户要求“回到话题 X，我们从那里开始”，可以通过检索外部缓存恢复部分历史。

压缩更新策略：有两种做法：

• 增量压缩：每次只压缩最早的一段（例如最早 5 轮对话），保留其余历史。优点是不需要重新摘要全部历史，节省 token。缺点是摘要之间可能存在信息重叠或遗漏。
• 全量压缩：在首次触发时压缩全部历史，后续每次更新只压缩增量部分。LLM 可以基于上一次摘要，给出新的增量摘要，类似于“增量式更新”。LangGraph 的实现中，压缩节点可以在每次状态更新后判断是否触发全量或增量压缩。

4.3 典型实现：LangGraph 中的摘要节点

在 LangGraph 中，可以通过 MemorySaver 配合自定义节点来实现摘要压缩。MemorySaver 负责持久化对话状态，而压缩节点负责在每次状态更新后判断 token 是否超标，若是则进行压缩。

from langgraph.checkpoint import MemorySaver
from langgraph.graph import StateGraph, END
from typing import Dict, List, Optional

class CompressionNode:
    def __init__(self, max_tokens: int = 2000):
        self.max_tokens = max_tokens
        self.memory = MemorySaver()

    def should_compress(self, messages: List[Dict]) -> bool:
        total = sum(len(m["content"]) for m in messages)  # 简化计算
        return total > self.max_tokens

def compress(self, messages: List[Dict]) -> List[Dict]:
        # 调用 LLM 压缩最早的部分
        oldest_chunk = messages[:5]  # 示例：压缩最早 5 条
        summary_prompt = f"请将以下对话摘要为一段: {oldest_chunk}"
        # 实际需调用 LLM 生成摘要
        summary = "<LLM generated summary>"
        # 替换最早几条为一条摘要
        return [{"role": "system", "content": f"历史摘要: {summary}"}] + messages[5:]

from langgraph.graph import StateGraph

class AgentState:
    messages: List[Dict]
    token_count: int  # 可选：预计算 token 数

graph = StateGraph(AgentState)
graph.add_node("chat", chat_node)           # 正常对话节点
graph.add_node("compress", compress_node)   # 压缩节点
graph.add_conditional_edge("chat", "compress", should_compress)
graph.add_edge("compress", "chat")

注意：摘要压缩会丢失原始对话的精确细节。工程中需要评估业务场景：对于需要精确复述“用户在三轮前说‘使用优惠码 ABCD’”，摘要可能丢失该信息。如果精度要求高，应考虑下一节的向量检索方案。

5. 实战代码：基于 LangGraph 的 MemorySaver 会话级记忆

5.1 MemorySaver 是什么

MemorySaver 是 LangGraph 内置的检查点存储器，用于保存 Agent 在每一步执行后的状态快照（包括 messages、当前运行的中间变量等）。它默认使用 SQLite 作为后端存储，也可以配置为内存存储。作用：

• 当 Agent 因错误或中断需要恢复时，可从最近的检查点重放。
• 在多轮对话中，它能自动维护 messages 列表，无需手动维护历史。

5.2 核心代码实现

以下示例展示如何创建 Agent 并注入 MemorySaver，实现会话级短期记忆的自动管理。

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver
from typing import TypedDict, List, Dict, Any

class AgentState(TypedDict):
    messages: List[Dict[str, Any]]
    next: str

def chatbot(state: AgentState) -> AgentState:
    # 模拟 LLM 回复：从 state["messages"] 中生成回复
    last_user_message = state["messages"][-1]["content"]
    reply = f"你说了: {last_user_message}"
    state["messages"].append({"role": "assistant", "content": reply})
    return state

# 初始化一个简单的图：仅聊天节点
graph = StateGraph(AgentState)
graph.add_node("chat", chatbot)
graph.set_entry_point("chat")
graph.add_edge("chat", END)

# 注入 MemorySaver
memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# 模拟多轮对话
config = {"configurable": {"thread_id": "session-123"}}
for user_msg in ["你好", "今天天气如何？", "请帮我查一下上海到北京的航班"]:
    app.invoke({"messages": [{"role": "user", "content": user_msg}]}, config)
    # 打印当前状态的 messages，可以看到历史持续累积
    current_state = app.get_state(config)
    print(current_state.values["messages"])

5.3 MemorySaver 的工作原理

• 每次 invoke() 调用后，MemorySaver 将当前的 AgentState 持久化到 SQLite 文件中（默认路径为当前目录下的 memory.db）。

• thread_id 用于隔离不同会话的记忆；同一 thread_id 的对话属于同一个会话，共享记忆；不同 thread_id 之间的记忆隔离。

• 在代码中无需手动管理消息历史；只需保证 AgentState 中包含 messages 列表，MemorySaver 会自动记录每次状态更新后的完整 messages 列表。

5.4 注意事项与调优

• 存储膨胀：MemorySaver 会保存每一步的完整状态快照，若对话很长（数百轮），数据库会迅速膨胀。生产环境中建议定期清理旧检查点，或限制最大检查点数量。LangGraph 提供了 MaxHistory 参数或手动删除检查点的方法。
• 与摘要压缩结合：可以在 chatbot 节点内部实现摘要逻辑，当 messages 长度超过阈值时，将最早对话压缩为摘要。

MemorySaver 会保存压缩后的 messages，下一次调用时状态就是压缩后的版本。

• 并发控制：SQLite 默认不支持高并发写入。在高并发的生产环境（多个线程/进程同时操作同一文件），应考虑换用 PostgresSaver（LangGraph 提供的 PostgreSQL 后端）或使用 Redis 等支持并发的存储。

6. 进阶方案：向量数据库短期会话检索

6.1 适用场景

摘要压缩会丢失细节，固定窗口截断会直接丢弃。如果业务场景需要对历史对话进行精确回溯（例如用户问“上个月第三个对话中提到的客户需求是什么”），上述方案均不适用。此时需要引入检索机制：将多轮对话拆分为若干语义块，嵌入为向量存储到向量数据库，用户提出问题时根据语义相似度检索相关片段。

6.2 实现思路

1. 分块与嵌入：将当前会话的对话历史按轮或按语义段落切分为块（chunk），每块包含若干轮对话。每块使用 embedding 模型（如 text-embedding-3-small）生成向量表示，存储到内存中的向量库（如 chromadb 的 EphemeralClient）或 Redis 向量库中。

2. 检索：当用户提出新问题时，将问题也嵌入为向量，与存储的对话块向量计算余弦相似度，召回 top-K 最相关的块。

3. 注入上下文：将召回的块文本连同当前对话 messages 一同注入 LLM，作为上下文。通常将召回内容拼接为一个系统消息或用户消息。

6.3 对比摘要方案

维度	摘要压缩	向量检索
信息保留	丢失细节（尤其精确数值、长文本）	保留原始文本，但仅召回相关片段
延迟	低（仅一次 LLM 压缩调用）	中等（嵌入 + 检索 + LLM 解析）
关键信息找回	取决于摘要质量，易遗漏	语义匹配准确，但可能召回不相关片段
实现复杂度	低	中等（需管理向量库、嵌入调用）
适用场景	对话轮次多但不需要精确回看	需要精确访问历史中的特定信息

选型建议：对于大多数企业应用，混合使用更合理——对近期对话使用 MemorySaver + 摘要压缩（保证低延迟），同时将每轮对话向量化存储在短期向量库中，当用户提问涉及历史详情时启用检索。即“短窗口记忆 + 可检索历史”的组合。

7. 企业级 Agent 记忆架构：多层次短期记忆设计

7.1 轻量级 vs 企业级方案对比

维度	轻量级方案	企业级方案
存储介质	内存、Redis 缓存、SQLite	分布式缓存（Redis Cluster）+ 关系数据库 + 向量数据库
记忆层次	仅当前会话	工作记忆（内存）→ 短期记忆（Redis/SQLite）→ 长期记忆（RAG 向量库）
权限与隔离	简单 session ID 隔离	用户级、角色级、租户级隔离；支持审计日志
版本管理	无	支持状态回滚、检查点恢复
清理策略	TTL 自动过期	自动归档 + 策略化清理（按时间、按重要性）

7.2 企业级多层次设计示例

• 工作记忆层：当前推理轮次的上下文，存在于 LLM 输入 prompt 中，由 LangGraph 的 StateGraph 管理，不持久化。
• 短期记忆层：最近 N 个会话（例如最近 10 次交互），使用 Redis 存储，key 格式为 session:{user_id}:short_term，value 为序列化的压缩摘要 + 向量索引指针。

设置 TTL（如 7 天），过期后自动删除。

• 长期记忆层：跨会话的重要信息（用户偏好、关键业务结论），通过 RAG 流程存储在向量数据库（如 Pinecone、Milvus）中，由单独的长期记忆服务管理。

版本管理与隔离：每个 user_id + session_id 对应独立的短期记忆 key。在多用户系统中，需要确保短期记忆不互相污染。LangGraph 的 thread_id 机制天然支持会话隔离，企业级可在上层封装一层用户会话映射（例如将 user_id 映射到 thread_id）。

8. 实战对比：轻量级 vs 企业级短期记忆集成

8.1 轻量级示例：基于 Redis 的简单存储

import redis
import json

class RedisShortTermMemory:
    def __init__(self, redis_url: str = "redis://localhost:6379/0"):
        self.client = redis.from_url(redis_url)
        self.ttl = 3600  # 1 小时

    def save(self, session_id: str, messages: list):
        key = f"session_memory:{session_id}"
        self.client.setex(key, self.ttl, json.dumps(messages))

    def load(self, session_id: str) -> list:
        key = f"session_memory:{session_id}"
        data = self.client.get(key)
        return json.loads(data) if data else []

def append(self, session_id: str, role: str, content: str):
        messages = self.load(session_id)
        messages.append({"role": role, "content": content})
        self.save(session_id, messages)

# 使用示例
mem = RedisShortTermMemory()
session_id = "user-123-session-1"
mem.append(session_id, "user", "你好")
mem.append(session_id, "assistant", "你好，有什么可以帮你？")
print(mem.load(session_id))

适用场景：单机或小型团队的原型、内部工具。Redis 提供数毫秒级的读写延迟。

8.2 企业级示例：LangGraph + Chroma 向量检索

from langgraph.graph import StateGraph, END
from langgraph.checkpoint import MemorySaver
import chromadb
from chromadb.utils import embedding_functions

class VectorMemoryAgent:
    def __init__(self, max_tokens: int = 3000):
        self.memory_saver = MemorySaver()
        self.chroma_client = chromadb.EphemeralClient()
        self.collection = self.chroma_client.create_collection(
            name="short_term",
            embedding_function=embedding_functions.OpenAIEmbeddingFunction()
        )
        self.session_id = None

def get_agent(self):
        # 构建 LangGraph Agent
        graph = StateGraph(AgentState)
        graph.add_node("chat", self.chat_with_memory)
        graph.set_entry_point("chat")
        graph.add_edge("chat", END)
        return graph.compile(checkpointer=self.memory_saver)

def chat_with_memory(self, state: AgentState) -> AgentState:
        # 1. 如果当前问题需要回顾历史，从 chroma 检索相关片段
        last_question = state["messages"][-1]["content"]
        similar_chunks = self.collection.query(
            query_texts=[last_question],
            n_results=3
        )
        if similar_chunks:
            # 将检索结果追加为 system 消息
            context = "\n".join(similar_chunks["documents"][0])
            state["messages"].insert(-1, {"role": "system", "content": f"相关历史: {context}"})
        # 2. 调用 LLM 生成回复（此处略）
        # 3. 将当前轮对话嵌入并存入 chroma
        self.collection.add(
            ids=[f"{self.session_id}-{len(state['messages'])}"],
            documents=[last_question + " " + state["messages"][-1]["content"]],
        )
        return state

8.3 选型建议

• 从实现复杂度：轻量级方案可以两天内对接完成；企业级方案需要构建向量库、设计权限隔离，通常需要 1–2 周。
• 延迟：轻量级方案（Redis get/set）< 5ms；向量检索方案（嵌入 + 检索）通常 50–200ms，但检索本身是异步的，可以在用户思考时预检索。
• 成本：向量检索会增加 embedding 模型 API 调用和向量库存储费用。

若会话量不大（日活 < 1000），用内存或 SQLite 即可。

• 可维护性：轻量级方案难以扩展；企业级方案需要专人的部署与运维。

9. 踩坑记录与性能优化

9.1 常见陷阱

• 摘要丢失关键信息：LLM 压缩摘要时可能忽略具体的数字、地址、工具返回的原始数据。解决：在压缩 prompt 中明确要求“保留所有数字和专有名词”，或者对包含数值的对话块不压缩，直接放到向量库中。
• 向量检索维度选择不当：使用 1536 维的 OpenAI embedding 模型性能尚可，但如果使用大维度的开源模型（如 4096 维），检索延迟会明显增高。

建议根据数据规模选择维度：小数据集（< 10 万条）用 768 维即可。

• MemorySaver 未清理过期会话：在高频对话场景下，SQLite 文件会持续膨胀。配置 MaxHistory 参数或定期执行 memory.delete_checkpoints(session_id) 清理。

建议在应用启动时加定时任务，扫描 TTL 过期的会话并清除。

• 向量库与 MemorySaver 数据不一致：向量库中的对话片段可能在 MemorySaver 中被压缩或截断。建议将压缩后的摘要文本更新到向量库中，或者同步删除被截断的检索条目。

9.2 性能优化

• 异步写入缓存：对于向量检索场景，将嵌入和写入异步化，避免阻塞对话流程。可使用 celery 或局部 ThreadPoolExecutor 处理。

• 摘要生成使用更小模型：压缩摘要不依赖高精度，可以使用 gpt-4o-mini 或本地小模型（如 Qwen2-1.5B）降低成本。

• 设置 TTL 自动清理：无论使用 Redis 还是 SQLite，都设置会话 TTL（例如 1 天），到期自动删除。避免存储无限制增长。

9.3 跨会话短期记忆串联

在实际业务中，用户可能在多个会话中处理同一个任务（如长表单填写）。此时需要短期记忆穿透多个会话。建议方案：

1. 在 thread_id 中携带用户 ID 和任务 ID：user-{id}-task-{task_id}。
2. 当用户重新开始会话时，通过用户 ID + 任务 ID 查询上一个短期记忆的摘要。
3. 将摘要作为初始系统消息注入新会话。

这种设计本质上是短期记忆与长期记忆的融合——将跨会话的少量信息提升为“短期型长期记忆”。

10. 总结与拓展

10.1 三种短期记忆方案的适用场景总结

方案	适用场景	核心代价
上下文窗口截断	短会话、原型验证	信息完全丢失
对话历史摘要压缩	中等长度对话（10–50 轮），能接受细节丢失	丢失细节，一次 LLM 调用
向量数据库检索	长对话，需要精确回溯，知识密集型场景	额外延迟 + API 成本

实际生产推荐混合路径：MemorySaver 管理完整 messages，使用摘要压缩控制 token 长度，同时将每轮对话向量化存储，以备精确检索。这样兼顾了低延迟与高召回。

10.2 拓展方向

• 短期记忆与长期记忆的自动分级：开发重要性评分模型，根据对话内容自动判断哪些信息值得提升到长期记忆（例如用户明确设置的偏好、关键业务决策），其余信息则在 TTL 后自动清理。可以在 chat_with_memory 节点结束后，并行执行一个异步的“记忆评估”任务。
• 多模态短期记忆：当前方案主要处理文本。

对于涉及图像、表格等非文本模态的 Agent（如文档型 Agent），需考虑图像压缩与检索。

• LangGraph 的高级检查点配置：官方文档中介绍了 PostgresSaver（高并发）、SqliteSaver 的自定义表结构，以及检查点的保存策略（如仅保留最后 10 个检查点）。

生产环境下的参考配置是：使用 PostgresSaver，设置 max_checkpoints_per_thread=20，并结合定时清理 SQL 任务。

10.3 推荐进一步阅读

• LangGraph 官方文档：MemorySaver 与 PostgresSaver 的配置示例
• Redis Stack 文档：向量数据库模块 RediSearch 的集成案例
• 论文《MemGPT: Towards LLMs as Operating Systems》中关于虚拟上下文窗口的分层管理设计

短期会话记忆是 Agent 工程化落地的关键模块。从简单截断到向量检索，再到企业级层级设计，每个方案都对应特定的资源约束和业务需求。实践中建议优先采用 LangGraph 的 MemorySaver 并配合定制的摘要节点，先解决语境连贯性问题，再根据实际性能瓶颈逐步引入向量检索。

总结

通过本文的学习，相信你已经对「Agent短期会话记忆实现」有了更深入的理解。建议结合实际项目多加练习。如有疑问，欢迎交流！