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

3.2.4 基础管理策略

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

选择策略

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

滑动窗口

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

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

优点：简单、低成本 缺点：可能丢失重要的早期上下文

Token 预算管理

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

import tiktoken

class TokenBudgetMemory:
    def __init__(self, max_tokens: int = 4000, model: str = "<llm-model>"):
        self.messages = []
        self.max_tokens = max_tokens
        self.encoder = tiktoken.encoding_for_model(model)
    
    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):
        while self.count_tokens(self.messages) > self.max_tokens:
            # 保留 system 消息，移除最早的用户/助手消息

            for i, msg in enumerate(self.messages):
                if msg['role'] != 'system':
                    self.messages.pop(i)
                    break

重要性采样

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

class ImportanceSamplingMemory:
    def __init__(self, max_tokens: int = 4000):
        self.messages = []
        self.importance_scores = []
        self.max_tokens = max_tokens
    
    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 _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 将历史对话压缩为简洁摘要。核心思路是：保留最近几轮完整对话（用于上下文延续），更早的历史则压缩为摘要（保留语义）。适用于长对话场景。

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]
        
        # 生成摘要

        summary_prompt = f"""请将以下对话历史压缩为简洁的摘要，保留关键信息：

{self._format_messages(to_summarize)}

摘要："""
        
        new_summary = self.llm.generate(summary_prompt)
        
        # 合并新旧摘要

        if self.summary:
            self.summary = f"{self.summary}\n{new_summary}"
        else:
            self.summary = new_summary
        
        self.messages = recent
    
    def get_context(self) -> List[dict]:
        context = []
        if self.summary:
            context.append({
                "role": "system",
                "content": f"[对话历史摘要]\n{self.summary}"
            })
        context.extend(self.messages)
        return context

词元压缩

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

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']

相关性过滤

[!NOTE] 这里的相关性过滤是指在 上下文窗口内 进行的轻量级筛选。对于海量数据的检索，请参考 3.4 RAG 系统设计与优化 和 3.6 上下文工程。

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

def extract_relevant_info(history: List[dict], current_query: str) -> str:
    """从历史中提取与当前查询相关的信息"""
    
    prompt = f"""以下是对话历史：

{format_history(history)}

当前用户问题：{current_query}

请提取与当前问题相关的关键信息（如有的话）："""

    return llm.generate(prompt)

知识抽取

知识抽取将非结构化对话转化为结构化知识（如实体、事实、待办事项）。这样可以更紧凑地存储信息，也便于后续查询和引用。适用于需要长期维护用户信息或任务状态的场景。

def extract_entities_and_facts(messages: List[dict]) -> dict:
    prompt = """从以下对话中提取结构化信息：

{conversation}

请按以下格式输出：
- 用户信息：[姓名、偏好、背景等]
- 讨论的主题：[主题列表]
- 达成的结论：[结论列表]
- 待办事项：[任务列表]
"""
    
    result = llm.generate(prompt)
    return parse_structured_info(result)

3.2.5 高级管理策略

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

3.2.5.1 会话生命周期管理

[!TIP] 最佳实践：保持对话简短、专注，每个对话只做一件事。

"醉酒"现象

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

重启阈值

不要等到 Token 耗尽才重启。

• 最佳实践：当上下文开始明显膨胀、且历史信息边际价值下降时，就应该考虑重启会话或切分任务。

• 成本考量：长对话的总生成成本会快速增长（因为每次请求都需要携带更长的历史作为输入）。

任务拆解

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

1. 调研对话：只负责读代码，输出计划。

2. 实现对话：带着计划（作为系统提示词或第一条消息）进行编码。

3. 测试对话：带着代码变更进行测试和修复。

3.2.5.2 动态预算分配

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

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"])

3.2.5.3 优先级上下文队列

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

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):
        """priority 越大越重要"""
        heappush(self.items, (-priority, tokens, content))
    
    def build_context(self) -> str:
        """构建不超过预算的最优上下文"""
        selected = []
        current_tokens = 0
        
        # 按优先级取出

        temp = []
        while self.items:
            priority, tokens, content = heappop(self.items)
            temp.append((priority, tokens, content))
            
            if current_tokens + tokens <= self.max_tokens:
                selected.append(content)
                current_tokens += tokens
        
        # 恢复堆

        for item in temp:
            heappush(self.items, item)
        
        return "\n\n".join(selected)

3.2.6 预算分配经验法则

组件	建议占比	说明
系统提示词	5-10%	精简但完整
检索知识	20-40%	根据任务动态调整
对话历史	20-30%	摘要 + 最近几轮
当前输入	5-15%	通常用户输入不长
输出预留	20-30%	确保模型有足够空间输出

3.2.7 监控和优化

持续监控上下文使用情况是优化的基础。通过记录每次请求的上下文大小、输出长度和任务类型，可以发现使用模式、识别浪费并调整预算分配策略。

class ContextMonitor:
    def __init__(self):
        self.stats = []
    
    def log(self, context_tokens: int, output_tokens: int, task_type: str):
        self.stats.append({
            "context": context_tokens,
            "output": output_tokens,
            "task": task_type,
            "timestamp": datetime.now()
        })
    
    def analyze(self):
        """分析上下文使用模式"""
        avg_context = sum(s["context"] for s in self.stats) / len(self.stats)
        avg_output = sum(s["output"] for s in self.stats) / len(self.stats)
        
        print(f"平均上下文使用：{avg_context:.0f} tokens")
        print(f"平均输出长度：{avg_output:.0f} tokens")
        print(f"建议输出预留：{avg_output * 1.5:.0f} tokens")

---

下一节: 3.3 长期记忆与向量数据库