# 3.2 短期记忆管理

大语言模型的上下文窗口是有限的资源。如何在有限的窗口中放入最有价值的信息，是构建高效智能体的关键挑战。

## 3.2.1 概述

短期记忆通常对应于大模型的 **上下文窗口**。虽然一些模型已支持更长的上下文窗口，但高效的窗口管理算法仍然是构建生产级智能体的关键，主要基于以下三个因素：

1. **成本控制 (Cost)**：长窗口的 Token 费用是线性的，但在高并发场景下，每次请求都带上完整的历史记录会导致总体成本快速上升。
2. **延迟 (Latency)**：首字生成延迟（TTFT）与输入 Token 数成正比。过长的上下文会显著降低交互的实时性。
3. **注意力稀释 (Attention/Recall)**：随着上下文长度增加，模型对中间信息的检索能力（Recall）会下降（即 “Lost in the Middle” 现象），导致指令遵循能力变弱。

## 3.2.2 上下文限制（实践视角）

不同模型/平台的上下文上限差异很大，而且会随着版本更新而变化。工程上更建议把“上下文上限”当成一个可配置参数，并在运行时做预算与降级策略：

* **把上限当配置**：在模型配置中维护 `max_input_tokens` / `max_output_tokens`。
* **预留输出空间**：避免把输入塞满导致模型没有空间输出。
* **做降级策略**：超预算时优先摘要、再裁剪历史、最后裁剪检索内容。

## 3.2.3 上下文窗口的构成

上下文窗口由以下几个部分构成：

| 组件       | 说明         | 典型 Token 范围 |
| -------- | ---------- | ----------- |
| 系统提示词    | 身份、规则、工具定义 | 500-2k      |
| 检索的知识/记忆 | 从外部检索的相关信息 | 1k-5k       |
| 对话历史     | 之前的对话轮次    | 2k-10k      |
| 当前用户输入   | 用户本轮的请求    | 100-1k      |
| 预留给输出的空间 | 模型生成回复所需空间 | 1k-4k       |

```mermaid
pie showData
    title 上下文窗口预算分配（示意）
    "系统提示词 10% 500-2k" : 10
    "检索的知识/记忆 25% 1k-5k" : 25
    "对话历史 35% 2k-10k" : 35
    "当前用户输入 5% 100-1k" : 5
    "预留输出空间 25% 1k-4k" : 25
```

图 3-2：上下文窗口构成与预算分配（示意）

## 3.2.4 基础管理策略

上下文管理的基础是解决两个核心问题：**保留什么（选择策略）** 和 **如何精简（压缩策略）**。

**选择策略**决定保留哪些内容，常见方法包括滑动窗口、Token 预算管理、重要性采样。

### 滑动窗口

滑动窗口是最简单的选择策略：保留最近的 N 条消息，丢弃更早的历史。其核心假设是“最近的对话最相关”，适用于上下文延续性强、早期信息价值低的场景。

```python
class SlidingWindowMemory:
    def __init__(self, max_messages: int = 20):
        self.messages = []
        self.max_messages = max_messages

    def add(self, message: dict):
        self.messages.append(message)
        if len(self.messages) > self.max_messages:
            self.messages.pop(0)  # 移除最早的消息

    def get_context(self) -> List[dict]:
        return self.messages
```

**成本/风险/适用场景**：

* **成本**：极低（$O(1)$ 队列操作）。
* **风险**：可能丢失早期的关键上下文（如初始系统指令），导致任务目标漂移。
* **适用场景**：上下文延续性强、早期信息价值低的多轮闲聊。

### Token 预算管理

Token 预算管理是滑动窗口的一种变种：将度量单位从“消息数”改为“Token 数”。由于不同消息的 Token 数差异巨大，按 Token 设定上限能更精确地控制成本和上下文使用率。

```python
import tiktoken

class TokenBudgetMemory:
    def __init__(self, max_tokens: int = 4000, model: str = "gpt-5.5"):
        self.messages = []
        self.max_tokens = max_tokens
        try:
            self.encoder = tiktoken.encoding_for_model(model)
        except KeyError:
            self.encoder = tiktoken.get_encoding("o200k_base")

    def count_tokens(self, messages: List[dict]) -> int:
        total = 0
        for msg in messages:
            total += len(self.encoder.encode(msg['content']))
            total += 4  # 消息格式开销
        return total

    def add(self, message: dict):
        self.messages.append(message)
        self._trim()

    def _trim(self):
        current_tokens = self.count_tokens(self.messages)
        while current_tokens > self.max_tokens:
            # 找到第一个非 system 消息的索引
            try:
                idx = next(i for i, m in enumerate(self.messages) if m['role'] != 'system')
                # 简单估算减少的 Token 数，避免全量重新计算
                removed_msg = self.messages.pop(idx)
                current_tokens -= (len(self.encoder.encode(removed_msg['content'])) + 4)
            except StopIteration:
                break
```

**成本/风险/适用场景**：

* **成本**：中等（由于每次添加都要进行编码计算，时间复杂度为 $O(N)$）。
* **风险**：若不加选择地按 Token 截断，可能随机破坏 JSON 结构或代码块的完整性。
* **适用场景**：对 API 消耗敏感、输入长度极为不可控的商业化应用。

### 重要性采样

不是所有消息都同等重要。重要性采样为每条消息打分，当需要精简时优先丢弃低分消息。这样可以保留关键决策和重要指令，丢弃闲聊和确认性回复。

```python
class ImportanceSamplingMemory(TokenBudgetMemory):
    def __init__(self, max_tokens: int = 4000, model: str = "gpt-5.5"):
        super().__init__(max_tokens, model)
        self.importance_scores = []

    def add(self, message: dict, importance: float = 0.5):
        """
        importance: 0.0-1.0，越高越重要
        - 1.0: 关键决策、用户明确要求记住的内容
        - 0.7: 重要的技术细节
        - 0.5: 普通对话
        - 0.3: 闲聊、确认性回复
        """
        self.messages.append(message)
        self.importance_scores.append(importance)
        self._trim()

    def _over_budget(self):
        return self.count_tokens(self.messages) > self.max_tokens

    def _trim(self):
        while self._over_budget():
            # 找到重要性最低的非系统消息

            min_idx = -1
            min_score = float('inf')
            for i, (msg, score) in enumerate(zip(self.messages, self.importance_scores)):
                if msg['role'] != 'system' and score < min_score:
                    min_score = score
                    min_idx = i

            if min_idx >= 0:
                self.messages.pop(min_idx)
                self.importance_scores.pop(min_idx)
```

**成本/风险/适用场景**：

* **成本**：较高（需要额外维护打分逻辑，甚至需要 LLM 辅助打分）。
* **风险**：打分规则如果未经过充分回归测试，容易造成关键依赖消息的误删。
* **适用场景**：需要长期遵守系统约束、且对话中常夹杂冗余信息的长线规划智能体。

**压缩策略**决定如何精简内容，常见方法包括对话摘要、词元压缩、相关性过滤、知识抽取。这些技术的完整实现可参考 [3.6 上下文工程](/agentic_ai_guide/di-yi-bu-fen-dan-ti-zhi-neng-jia-gou/03_memory/3.6_context_engineering.md)，此处着重介绍核心思路。

### 对话摘要

对话摘要使用 LLM 将历史对话压缩为简洁摘要。核心思路是：保留最近几轮完整对话（用于上下文延续），更早的历史则压缩为摘要（保留语义）。适用于长对话场景。

```python
class SummaryMemory:
    def __init__(self, llm, summary_threshold: int = 10):
        self.llm = llm
        self.messages = []
        self.summary = None
        self.summary_threshold = summary_threshold

    def add(self, message: dict):
        self.messages.append(message)
        if len(self.messages) > self.summary_threshold:
            self._compress()

    def _compress(self):
        recent = self.messages[-3:]           # 保留最近几条
        to_summarize = self.messages[:-3]     # 更早的压缩为摘要
        new_summary = self.llm.generate(
            f"请将以下对话压缩为简洁摘要：\n{self._format_messages(to_summarize)}"
        )
        self.summary = f"{self.summary}\n{new_summary}" if self.summary else new_summary
        self.messages = recent

    def get_context(self) -> List[dict]:
        ctx = [{"role": "system", "content": f"[历史摘要]\n{self.summary}"}] if self.summary else []
        return ctx + self.messages
```

**成本/风险/适用场景**：

* **成本**：高（额外增加一次 LLM 摘要调用的延迟与 Token 费用）。
* **风险**：摘要过程本质是有损压缩，频繁摘要不仅引入语义漂移（Semantic Drift），还容易遗漏细节线索。
* **适用场景**：生命周期极长的个人助理或客服对话（需配合关键实体抽取使用）。

### 词元压缩

词元压缩使用专门的压缩模型（如 LLMLingua）移除冗余 Token，同时保留核心语义。其原理是：利用小模型评估每个 Token 的信息量，移除低信息量 Token。适用于压缩大段文档或检索结果。

```python
from llmlingua import PromptCompressor

compressor = PromptCompressor(model_name="microsoft/llmlingua-2-bert-base-multilingual-cased-meetingbank")

def compress_context(context: str, target_ratio: float = 0.5) -> str:
    """将上下文压缩到指定比例"""
    result = compressor.compress_prompt(
        context,
        rate=target_ratio,
        force_tokens=['重要', '关键'],  # 强制保留的词
    )
    return result['compressed_prompt']
```

### 相关性过滤

> **说明**： 这里的相关性过滤是指在 **上下文窗口内** 进行的轻量级筛选。对于海量数据的检索，请参考 [3.4 RAG 系统设计与优化](/agentic_ai_guide/di-yi-bu-fen-dan-ti-zhi-neng-jia-gou/03_memory/3.4_rag_advanced.md) 和 [3.6 上下文工程](/agentic_ai_guide/di-yi-bu-fen-dan-ti-zhi-neng-jia-gou/03_memory/3.6_context_engineering.md)。

相关性过滤根据当前查询，从历史中只提取相关信息。其核心思路是：用 LLM 判断哪些历史内容与当前问题相关，只保留相关部分。适用于历史中包含多个不相关话题的场景。

```python
def extract_relevant_info(history: List[dict], current_query: str) -> str:
    """用 LLM 从历史中只提取与当前查询相关的信息"""
    prompt = f"对话历史：\n{format_history(history)}\n\n当前问题：{current_query}\n\n请提取相关关键信息："
    return llm.generate(prompt)
```

### 知识抽取

知识抽取将非结构化对话转化为结构化知识（实体、事实、待办事项），以更紧凑的方式存储，便于后续查询。适用于需要长期维护用户信息或任务状态的场景。典型做法是让 LLM 从对话中提取“用户信息、讨论主题、达成结论、待办事项”等字段，输出结构化 JSON。

## 3.2.5 高级管理策略

随着智能体复杂度的提升，仅仅依靠基础算法（如滑动窗口）已经难以满足需求。需要引入更智能的动态管理策略。

### 会话生命周期管理

> **最佳实践**：尽量控制单次会话的目标范围；当任务开始分叉或上下文明显膨胀时，优先拆成多个独立会话处理。

### “醉酒”现象

随着上下文积累，智能体会表现出类似醉酒的症状：遗忘指令、逻辑混乱、开始产生幻觉。这是因为信噪比随着无关信息的积累而降低。

### 重启阈值

不要等到 Token 耗尽才重启。

* **最佳实践**：当上下文开始明显膨胀、且历史信息边际价值下降时，就应该考虑重启会话或切分任务。
* **成本考量**：长对话的总生成成本会快速增长（因为每次请求都需要携带更长的历史作为输入）。

### 任务拆解

将大任务拆解为一组相互关联的短对话：

1. **调研对话**：只负责读代码，输出计划。
2. **实现对话**：带着计划（作为系统提示词或第一条消息）进行编码。
3. **测试对话**：带着代码变更进行测试和修复。

### 动态预算分配

不同任务对上下文组件的需求不同。代码生成需要更多代码示例，对话类任务需要更多历史记录，分析任务需要更多背景资料。动态分配的核心思路是：根据任务类型预定义不同的预算分配方案，然后根据当前任务选择合适的方案。

```python
class DynamicContextManager:
    def __init__(self, total_budget: int = 8000):
        self.total_budget = total_budget

    def allocate(self, task_type: str) -> dict:
        """根据任务类型分配上下文预算"""

        allocations = {
            "code_generation": {
                "system": 500,
                "knowledge": 3000,  # 需要更多代码示例
                "history": 1000,
                "output": 3500
            },
            "conversation": {
                "system": 500,
                "knowledge": 1000,
                "history": 4000,   # 需要更多对话历史
                "output": 2500
            },
            "analysis": {
                "system": 500,
                "knowledge": 4000, # 需要更多背景资料
                "history": 1000,
                "output": 2500
            }
        }

        return allocations.get(task_type, allocations["conversation"])
```

### 优先级上下文队列

当有多个信息源竞争有限的上下文空间时，可以使用优先级队列管理。其核心思路是：为每个内容块分配优先级，在预算内优先纳入高优先级内容。这种方法适用于需要从多个来源（检索结果、历史、工具定义等）动态组装上下文的场景。

```python
from heapq import heappush, heappop

class PriorityContextManager:
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
        self.items = []  # 最小堆：(-priority, tokens, content)

    def add(self, content: str, priority: int, tokens: int):
        heappush(self.items, (-priority, tokens, content))

    def build_context(self) -> str:
        """按优先级贪心选取，直到预算耗尽"""
        selected, used = [], 0
        for neg_pri, tokens, content in sorted(self.items):
            if used + tokens <= self.max_tokens:
                selected.append(content)
                used += tokens
        return "\n\n".join(selected)
```

## 3.2.6 案例研究：渐进式记忆架构

下面是一套面向编码 Agent 的**渐进式记忆架构示例**，用于说明如何把低成本缓存、会话笔记、压缩摘要和子智能体隔离组合起来。公开文档通常只确认某些能力（如项目记忆、子智能体、会话恢复或工具结果管理），不应把这里的层级、目录名、清理周期或“做梦”机制理解为 Claude Code 的官方内部实现。

| 层级 | 机制                    | 触发条件              | 成本            | 用途         |
| -- | --------------------- | ----------------- | ------------- | ---------- |
| L1 | 工具结果磁盘缓存 + 2KB 预览     | 工具调用完成            | 极低            | 避免重复调用     |
| L2 | 微压缩（60分钟自动清理）         | 时间阈值              | 低             | 清理过期临时数据   |
| L3 | Session Memory（结构化笔记） | 手动 `/memory save` | 低（零 API 成本）   | 跨会话恢复工作状态  |
| L4 | 完整压缩（9 小节摘要）          | 上下文逼近上限           | 中（一次性 LLM 调用） | 保留完整会话语义   |
| L5 | 自动记忆提取（跨会话知识库）        | 检测稳定知识点           | 中             | 累积可复用知识    |
| L6 | “做梦”机制（后台整合）          | 累积多个会话            | 高（独立 agent）   | 跨会话知识整合与优化 |
| L7 | 子智能体通信（Branch Agent）  | 任务分发              | 中（共享缓存）       | 并行执行与隔离上下文 |

关键设计哲学：每层拦截可能导致下层触发的问题，从而避免成本螺旋。例如，若 L1 缓存有效命中率足够高，L2-L4 的压缩操作就能大幅延后甚至避免。

## 3.2.7 监控和优化

持续监控上下文使用情况是优化的基础。建议在（后续 [第 9 章 AgentOps](/agentic_ai_guide/di-san-bu-fen-gong-cheng-shi-jian-yu-luo-di/09_agentops.md)）的可观测性体系中，把以下维度设为核心监控指标：

* **上下文长度（Context Length）**：防范输入突然膨胀引发的 OOM 或超预算熔断。
* **摘要触发次数（Summary Triggers）**：频次过高说明窗口太小或压缩策略过于激进，极大增加延迟与语义漂移风险。
* **检索片段命中率（Retrieval Hit Rate）**：评估注入上下文的外部知识是否被模型真正引用或采纳。
* **目标漂移报警（Goal Drift Alert）**：若发生连续多次的推理重试或偏离初始用户指令，通常意味着上游的上下文裁剪不慎丢弃了核心意图。

具体的监控实现方案，请参考 [第 9 章 AgentOps](/agentic_ai_guide/di-san-bu-fen-gong-cheng-shi-jian-yu-luo-di/09_agentops/9.3_observability.md) 的可观测性体系。

***

**下一节**: [3.3 长期记忆与向量数据库](/agentic_ai_guide/di-yi-bu-fen-dan-ti-zhi-neng-jia-gou/03_memory/3.3_vector_databases.md)


---

# Agent Instructions: Querying This Documentation

If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter:

```
GET https://yeasy.gitbook.io/agentic_ai_guide/di-yi-bu-fen-dan-ti-zhi-neng-jia-gou/03_memory/3.2_short_term_memory.md?ask=<question>
```

The question should be specific, self-contained, and written in natural language.
The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.
