AI Agent 长对话管理：上下文窗口溢出的工程解法

一、对话越长越笨：Agent 上下文管理的真实困境

大模型 Agent 在短对话场景下表现尚可，但当对话轮次超过 20 轮、上下文逼近 Token 上限时，问题集中爆发：模型开始遗忘早期约定、重复执行已完成的工具调用、甚至产生与上下文矛盾的回复。这不是模型能力不足，而是上下文管理策略的缺失。

以一个客服 Agent 为例：用户在第 3 轮提供了订单号，第 15 轮又问物流状态，Agent 却要求重新提供订单号——因为中间 12 轮的对话已经把原始信息挤出了有效注意力范围。更严重的是，当上下文逼近模型窗口上限时，API 调用的 Token 费用线性增长，但回复质量反而下降，ROI 急剧恶化。

长对话管理的核心矛盾是：可用上下文窗口是有限的，但业务对话的信息量是无限增长的。必须有一套工程化的策略，在有限窗口内保留高价值信息、淘汰低价值信息，同时控制 Token 成本。

二、上下文窗口的运作机制与溢出策略

理解上下文管理，先要理解大模型如何消费上下文窗口。每次 API 调用时，模型接收的是完整的消息列表——包括系统提示、历史对话、工具调用结果。这个列表的总 Token 数不能超过模型上下文窗口（如 GPT-4o 的 128K、Claude 的 200K）。

graph LR
    subgraph 上下文窗口组成
        A[系统提示 System Prompt] --> B[对话历史 Conversation History]
        B --> C[工具调用结果 Tool Results]
        C --> D[当前用户输入 Current Input]
    end

    subgraph 溢出处理策略
        E[滑动窗口截断] --> F[摘要压缩]
        F --> G[向量检索补充]
        G --> H[分层记忆架构]
    end

    D -->|Token 超限| E

四种主流的溢出处理策略各有适用场景：

滑动窗口截断是最简单的策略——保留最近 N 轮对话，丢弃更早的。优点是实现简单，缺点是丢失关键信息。适合闲聊型 Agent，不适合需要长期记忆的任务型 Agent。

摘要压缩是用 LLM 对早期对话生成摘要，用摘要替代原文。保留语义的同时大幅减少 Token 占用。但摘要本身有信息损失，且额外的 LLM 调用增加了延迟和成本。

向量检索补充是将对话历史存入向量数据库，每次对话时检索最相关的片段注入上下文。适合知识密集型场景，但检索质量依赖 Embedding 模型，且增加了系统复杂度。

分层记忆架构是综合方案——短期记忆保留最近对话，中期记忆存储摘要，长期记忆使用向量检索。三层协同，在 Token 预算内最大化信息密度。

三、生产级分层记忆系统实现

以下实现基于分层记忆架构，包含 Token 预算管理和自动摘要压缩。

from dataclasses import dataclass, field
from typing import Optional
import tiktoken
import time

@dataclass
class Message:
    role: str          # user / assistant / tool
    content: str
    token_count: int = 0
    timestamp: float = field(default_factory=time.time)
    metadata: dict = field(default_factory=dict)

class ConversationMemory:
    """分层对话记忆管理器

    设计思路：
    - 短期记忆：保留最近 N 轮完整对话，保证即时上下文连贯
    - 中期记忆：对超出短期窗口的对话自动生成摘要
    - Token 预算：硬性约束，任何情况下不超限
    """

    def __init__(
        self,
        model_name: str = "gpt-4o",
        max_total_tokens: int = 120000,      # 留 8K 给输出
        short_term_rounds: int = 10,          # 短期保留最近 10 轮
        summary_max_tokens: int = 2000,       # 摘要最大 Token 数
        llm_client = None,                    # LLM 客户端，用于生成摘要
    ):
        self.model_name = model_name
        self.max_total_tokens = max_total_tokens
        self.short_term_rounds = short_term_rounds
        self.summary_max_tokens = summary_max_tokens
        self.llm_client = llm_client
        self.encoding = tiktoken.encoding_for_model(model_name)

        # 短期记忆：完整对话消息列表
        self.short_term: list[Message] = []
        # 中期记忆：历史摘要
        self.summaries: list[str] = []
        # 系统提示（固定占用，不参与淘汰）
        self.system_prompt: Optional[str] = None

    def count_tokens(self, text: str) -> int:
        """精确计算 Token 数，而非估算"""
        return len(self.encoding.encode(text))

    def set_system_prompt(self, prompt: str) -> None:
        self.system_prompt = prompt

    def add_message(self, role: str, content: str, metadata: dict = None) -> None:
        """添加消息并触发记忆管理"""
        msg = Message(
            role=role,
            content=content,
            token_count=self.count_tokens(content),
            metadata=metadata or {},
        )
        self.short_term.append(msg)
        self._manage_memory()

    def _manage_memory(self) -> None:
        """核心记忆管理逻辑：超限时压缩早期对话"""
        # 计算当前总 Token 占用
        total = self._calculate_total_tokens()

        if total <= self.max_total_tokens:
            return

        # 策略：将超出短期窗口的早期对话压缩为摘要
        while len(self.short_term) > self.short_term_rounds and total > self.max_total_tokens:
            # 取出最早的一轮对话（user + assistant）
            overflow_messages = []
            while len(self.short_term) > self.short_term_rounds:
                overflow_messages.append(self.short_term.pop(0))

            if overflow_messages:
                summary = self._generate_summary(overflow_messages)
                self.summaries.append(summary)
                total = self._calculate_total_tokens()

    def _generate_summary(self, messages: list[Message]) -> str:
        """使用 LLM 对早期对话生成摘要

        摘要提示词的关键：要求保留实体名、数值、决策结论，
        丢弃寒暄、重复确认等低信息密度内容
        """
        if not self.llm_client:
            # 降级策略：无 LLM 客户端时，拼接关键信息
            return self._fallback_summary(messages)

        conversation_text = "\n".join(
            f"{m.role}: {m.content}" for m in messages
        )

        prompt = (
            "请将以下对话压缩为一段摘要。要求：\n"
            "1. 保留所有实体名称、数值、日期、订单号等关键信息\n"
            "2. 保留用户的核心诉求和已做出的决策\n"
            "3. 丢弃寒暄、重复确认、格式化输出等低价值内容\n"
            "4. 摘要不超过 300 字\n\n"
            f"对话内容：\n{conversation_text}"
        )

        # 调用 LLM 生成摘要，设置超时防止挂起
        try:
            response = self.llm_client.chat.completions.create(
                model=self.model_name,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=500,
                timeout=10,
            )
            return response.choices[0].message.content
        except Exception as e:
            # LLM 调用失败时降级为基础拼接
            return self._fallback_summary(messages)

    def _fallback_summary(self, messages: list[Message]) -> str:
        """降级摘要：提取用户消息的关键片段"""
        user_msgs = [m.content[:100] for m in messages if m.role == "user"]
        return " | ".join(user_msgs) if user_msgs else "[对话已压缩]"

    def _calculate_total_tokens(self) -> int:
        """计算当前上下文的总 Token 占用"""
        total = 0
        if self.system_prompt:
            total += self.count_tokens(self.system_prompt)
        for summary in self.summaries:
            total += self.count_tokens(summary)
        for msg in self.short_term:
            total += msg.token_count
        return total

    def build_messages(self) -> list[dict]:
        """构建发送给 LLM 的完整消息列表

        组装顺序：系统提示 → 历史摘要 → 短期对话
        摘要放在系统提示之后，确保模型优先关注近期对话
        """
        messages = []

        if self.system_prompt:
            messages.append({"role": "system", "content": self.system_prompt})

        # 将所有摘要合并为一条系统消息注入
        if self.summaries:
            combined = "\n\n".join(
                f"[历史对话摘要 {i+1}]\n{s}"
                for i, s in enumerate(self.summaries)
            )
            messages.append({
                "role": "system",
                "content": f"以下是之前对话的摘要：\n{combined}",
            })

        # 短期记忆：完整对话
        for msg in self.short_term:
            messages.append({"role": msg.role, "content": msg.content})

        return messages

关键设计决策说明：

Token 计算使用 tiktoken 而非估算。 字符数与 Token 数的映射因模型而异，中文场景下 1 个汉字约 1.5-2 个 Token。估算误差会导致实际调用时超限或预算浪费。

摘要生成有降级策略。 LLM 调用可能因网络、限流等原因失败，此时退回到基础拼接而非抛出异常，保证对话不中断。

摘要注入位置在系统提示之后、短期对话之前。 这个位置确保模型在处理当前对话时，历史摘要作为背景知识存在，但不会干扰对近期上下文的理解。

四、分层记忆的代价与适用边界

分层记忆架构并非没有代价，需要在多个维度做权衡：

摘要的信息损失不可逆。 一旦对话被压缩为摘要，原始措辞、语气、隐含语义都会丢失。如果后续对话需要精确引用早期对话的原文，摘要无法满足。对策是：对关键实体（订单号、金额、日期）做结构化提取，存入独立的键值存储，不依赖摘要。

额外的 LLM 调用增加延迟和成本。 每次摘要生成是一次额外的 API 调用，增加 500ms-2s 的延迟。在对话密集场景下，可以异步生成摘要——先截断对话保证当前请求正常响应，后台异步压缩。

Token 预算分配需要调优。 系统提示、摘要、短期对话各占多少 Token，没有通用最优解。系统提示过长会挤压对话空间，摘要过多会稀释近期上下文的注意力权重。建议按 1:2:7 的比例分配，并根据实际效果调整。

适用场景： 任务型 Agent（客服、工单处理、技术支持），对话轮次通常超过 10 轮且需要跨轮次引用信息。不适用场景： 单轮问答、创意写作等短对话场景，分层记忆的复杂度不值得。

五、总结

AI Agent 的长对话管理本质是在有限 Token 预算内最大化信息密度的工程问题。分层记忆架构通过短期完整对话 + 中期摘要 + 长期检索的三层结构，在上下文窗口溢出时仍能保留关键信息。生产实现中，Token 精确计算、摘要降级策略、预算分配比例是需要重点打磨的细节。架构选型上，短对话场景用滑动窗口即可，只有对话轮次持续增长且需要跨轮次记忆的任务型 Agent，才值得引入分层记忆的额外复杂度。