对话越聊越蠢？AI Agent 长对话记忆管理的工程化方案

一、Token 账单与上下文遗忘：长对话 Agent 的双重困境

大模型 Agent 落地到真实业务后，第一个撞上的墙不是模型能力不够，而是"对话管理"出了问题。具体表现有两类：一类是"越聊越贵"——每轮对话把完整历史塞进 Prompt，Token 消耗随轮次线性增长，10 轮对话的 Token 成本是第 1 轮的 5 倍以上；另一类是"越聊越蠢"——上下文窗口被早期无关内容占满，模型对近期关键信息的注意力被稀释，回复质量明显下降。

以一个客服 Agent 为例：用户前 3 轮咨询退货政策，第 4 轮开始讨论换货流程，到第 8 轮时模型已经忘了用户在第 2 轮提到的订单号。这不是模型能力问题，而是记忆管理策略缺失导致的工程问题。

长对话记忆管理的本质是：在有限的上下文窗口内，用最低的 Token 成本，保留对当前任务最有价值的信息。这不是一个算法问题，而是一个工程架构问题。

二、记忆分层架构：从工作记忆到长期存储

人类大脑的记忆系统分为工作记忆、短期记忆和长期记忆三层。Agent 的记忆架构可以类比设计，但需要适配 LLM 的技术约束。

graph TB
    subgraph 记忆分层架构
        WM[工作记忆 Working Memory<br/>当前对话上下文窗口]
        SM[短期记忆 Short-term Memory<br/>近期对话摘要 + 关键实体]
        LM[长期记忆 Long-term Memory<br/>向量数据库 + 结构化知识]
    end

    User[用户输入] --> WM
    WM -->|窗口溢出时压缩| SM
    SM -->|关键信息持久化| LM
    LM -->|检索召回| WM

    WM --> LLM[大模型推理]
    LLM --> Response[生成回复]
    Response --> WM

    style WM fill:#e1f5fe
    style SM fill:#fff3e0
    style LM fill:#e8f5e9

工作记忆（Working Memory）：直接填充在 LLM Prompt 中的上下文，受模型窗口大小限制。这是模型"此刻能看到"的全部信息。管理策略的核心是：保留什么、丢弃什么、如何压缩。

短期记忆（Short-term Memory）：最近 N 轮对话的摘要和关键实体提取。当工作记忆窗口即将溢出时，将早期对话压缩为摘要存入短期记忆，释放 Token 空间。摘要不是简单的截断，而是保留语义关键信息的压缩。

长期记忆（Long-term Memory）：持久化存储在向量数据库中的历史知识。当用户提到"上次讨论的那个方案"时，通过语义检索从长期记忆中召回相关片段，注入工作记忆。长期记忆的召回精度决定了 Agent 的"记忆力"。

三、生产级记忆管理器实现

3.1 核心数据结构

from dataclasses import dataclass, field
from typing import Optional
import time
import hashlib
import json

@dataclass
class Message:
    """单条对话消息"""
    role: str          # user / assistant / system
    content: str
    timestamp: float = field(default_factory=time.time)
    token_count: int = 0
    msg_id: str = ""

    def __post_init__(self):
        if not self.msg_id:
            # 用内容哈希 + 时间戳生成唯一 ID，避免重复
            raw = f"{self.role}:{self.content}:{self.timestamp}"
            self.msg_id = hashlib.md5(raw.encode()).hexdigest()[:12]

@dataclass
class SummaryBlock:
    """对话摘要块：多轮对话压缩后的语义保留"""
    summary: str           # 压缩后的摘要文本
    original_msg_ids: list # 被压缩的原始消息 ID 列表
    key_entities: list     # 提取的关键实体（订单号、人名等）
    token_count: int = 0
    created_at: float = field(default_factory=time.time)

@dataclass
class Entity:
    """关键实体：需要在上下文中持续保留的信息"""
    name: str              # 实体名称
    value: str             # 实体值
    source_msg_id: str     # 来源消息 ID
    updated_at: float = field(default_factory=time.time)

3.2 记忆管理器核心逻辑

class AgentMemoryManager:
    """Agent 长对话记忆管理器"""

    def __init__(
        self,
        max_working_tokens: int = 6000,   # 工作记忆 Token 上限
        summary_threshold: float = 0.8,    # 触发压缩的阈值比例
        max_recent_messages: int = 6,      # 始终保留的最近消息数
    ):
        self.max_working_tokens = max_working_tokens
        self.summary_threshold = summary_threshold
        self.max_recent_messages = max_recent_messages

        # 三层记忆
        self.working_memory: list[Message] = []      # 工作记忆
        self.short_term_memory: list[SummaryBlock] = []  # 短期记忆
        self.entities: dict[str, Entity] = {}         # 关键实体表

    def add_message(self, role: str, content: str, token_count: int) -> None:
        """添加消息到工作记忆，并检查是否需要压缩"""
        msg = Message(role=role, content=content, token_count=token_count)
        self.working_memory.append(msg)

        # 提取关键实体（订单号、金额等结构化信息）
        self._extract_entities(msg)

        # 检查工作记忆是否接近溢出
        current_tokens = sum(m.token_count for m in self.working_memory)
        if current_tokens >= self.max_working_tokens * self.summary_threshold:
            self._compress_working_memory()

    def get_context_for_llm(self) -> list[dict]:
        """组装发送给 LLM 的完整上下文"""
        context = []

        # 1. 注入关键实体（始终保留在上下文头部）
        if self.entities:
            entity_text = "关键信息：\n" + "\n".join(
                f"- {e.name}: {e.value}" for e in self.entities.values()
            )
            context.append({"role": "system", "content": entity_text})

        # 2. 注入短期记忆摘要
        if self.short_term_memory:
            summary_parts = []
            for block in self.short_term_memory[-3:]:  # 最近 3 个摘要块
                summary_parts.append(block.summary)
            if summary_parts:
                summary_text = "历史对话摘要：\n" + "\n".join(summary_parts)
                context.append({"role": "system", "content": summary_text})

        # 3. 注入工作记忆中的近期对话
        for msg in self.working_memory:
            context.append({"role": msg.role, "content": msg.content})

        return context

    def _compress_working_memory(self) -> None:
        """压缩工作记忆：将早期对话转为摘要，保留最近消息"""
        if len(self.working_memory) <= self.max_recent_messages:
            return  # 消息数不够，不需要压缩

        # 分割：待压缩部分 + 保留部分
        split_idx = len(self.working_memory) - self.max_recent_messages
        to_compress = self.working_memory[:split_idx]
        to_keep = self.working_memory[split_idx:]

        # 调用 LLM 生成摘要（这里用简化逻辑示意）
        summary_text = self._generate_summary(to_compress)

        # 收集被压缩消息中的关键实体
        compressed_ids = [m.msg_id for m in to_compress]
        related_entities = [
            e.name for e in self.entities.values()
            if e.source_msg_id in compressed_ids
        ]

        summary_block = SummaryBlock(
            summary=summary_text,
            original_msg_ids=compressed_ids,
            key_entities=related_entities,
            token_count=len(summary_text) // 2,  # 粗估 Token 数
        )

        self.short_term_memory.append(summary_block)
        self.working_memory = to_keep

    def _generate_summary(self, messages: list[Message]) -> str:
        """调用 LLM 对历史对话生成摘要"""
        # 生产环境应调用 LLM API，这里用简化逻辑
        conversation = "\n".join(
            f"{m.role}: {m.content[:100]}" for m in messages
        )
        # 实际调用：llm_client.chat("请用200字以内总结以下对话的关键信息", conversation)
        return f"[摘要] 对话涉及 {len(messages)} 轮交互，关键实体已提取"

    def _extract_entities(self, msg: Message) -> None:
        """从消息中提取关键实体（订单号、金额等）"""
        import re
        # 订单号模式：ORD-XXXX 或纯数字订单号
        order_patterns = [
            r'ORD-\w+',
            r'订单号[：:]\s*(\d+)',
            r'订单\s*(\d{10,})',
        ]
        for pattern in order_patterns:
            matches = re.findall(pattern, msg.content)
            for match in matches:
                entity_key = f"order_{match}"
                self.entities[entity_key] = Entity(
                    name="订单号",
                    value=match,
                    source_msg_id=msg.msg_id,
                )

3.3 长期记忆：向量检索召回

import numpy as np

class LongTermMemory:
    """基于向量数据库的长期记忆存储"""

    def __init__(self, embedding_dim: int = 1536):
        self.embedding_dim = embedding_dim
        # 生产环境使用 Milvus / Qdrant / Pinecone
        self.vectors: list[np.ndarray] = []
        self.documents: list[dict] = []

    def store(self, doc: dict, embedding: np.ndarray) -> None:
        """存储文档及其向量表示"""
        self.vectors.append(embedding)
        self.documents.append(doc)

    def recall(self, query_embedding: np.ndarray, top_k: int = 3) -> list[dict]:
        """根据查询向量召回最相关的文档"""
        if not self.vectors:
            return []

        # 余弦相似度计算
        query_norm = query_embedding / np.linalg.norm(query_embedding)
        similarities = []
        for vec in self.vectors:
            vec_norm = vec / np.linalg.norm(vec)
            sim = np.dot(query_norm, vec_norm)
            similarities.append(sim)

        # 取 Top-K
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.documents[i] for i in top_indices if similarities[i] > 0.7]
        # 0.7 是召回阈值，低于此值的不注入上下文，避免噪声

四、记忆管理的代价：延迟、一致性与召回噪声

记忆分层架构解决了 Token 成本和上下文遗忘问题，但引入了新的工程权衡。

摘要压缩的信息损失。LLM 生成的摘要必然丢失细节。当用户追问"我之前说的那个具体金额是多少"时，摘要中可能已经没有这个数字。解决方案是：关键实体（金额、订单号、日期）不进入摘要流程，而是单独提取到实体表中，始终保留在工作记忆头部。但这增加了实体提取的准确率依赖——提取遗漏的信息就真的丢了。

长期记忆的召回噪声。向量检索基于语义相似度，但"相似"不等于"相关"。用户说"我要退货"，可能召回三个月前关于退货政策的对话，而实际上用户关心的是当前的退货流程。解决方案是给长期记忆加时间衰减因子，近期存储的文档权重更高。

多轮压缩的累积误差。每次压缩都是一次有损操作，多次压缩后摘要可能偏离原始语义。生产环境中建议限制压缩次数，当短期记忆摘要块超过一定数量时，将最早的摘要块合并为更高层级的摘要。

延迟开销。每次对话需要执行：实体提取 + 压缩判断 + 向量检索 + 上下文组装。这些步骤的延迟叠加后，首字响应时间（TTFT）可能增加 200-500ms。对于实时性要求高的场景，需要将实体提取和向量检索做成异步流水线。

适用边界：短对话（5 轮以内）不需要记忆管理，直接全量上下文即可；中等对话（5-30 轮）需要工作记忆压缩 + 实体提取；长对话（30 轮以上）需要完整三层架构。

五、总结

Agent 长对话记忆管理的核心思路是分层：工作记忆负责当前推理，短期记忆负责近期摘要，长期记忆负责历史召回。三层之间的流转规则——何时压缩、保留什么、如何召回——决定了 Agent 的"记忆质量"。

落地路线建议：第一步，实现工作记忆的 Token 计数和溢出检测，这是最基础的成本控制；第二步，实现关键实体提取，确保结构化信息不因压缩而丢失；第三步，引入 LLM 摘要压缩，将早期对话转为语义摘要；第四步，接入向量数据库实现长期记忆召回，加时间衰减因子降低噪声；第五步，将实体提取和向量检索异步化，控制 TTFT 延迟增量在 300ms 以内。

记忆管理不是可选优化，而是 Agent 从 Demo 走向生产的必经之路。没有记忆管理的 Agent，就像一个没有笔记本的客服——短期还能应付，时间一长必然出错。