原文：https://www.mstx.cn/harness/427.html · 创建：2026-09-06 · 更新：2026-09-06 · 阅读：1104

---

一个无法回避的物理限制

每一个LLM都有一个上下文窗口——它能"看到"的最大token数量。这个窗口是Agent短期记忆的物理边界。无论你的记忆管理策略多么精妙，你都无法让LLM"记住"超过窗口大小的信息。

这个限制对Agent的影响远比聊天应用严重。在聊天应用中，上下文窗口只包含用户和AI的对话。但在Agent中，上下文窗口需要容纳：

• 系统提示词：定义Agent的角色、行为规则、边界
• 工具Schema：每个工具的名称、描述、参数定义
• 对话历史：用户输入、LLM响应、中间推理
• 工具调用和结果：每次工具调用的参数和返回值

一个典型的Agent在5轮循环后，上下文可能已经消耗了数千token。10轮循环后，可能已经触及窗口上限。当上下文窗口满了，你面临一个艰难的选择：丢弃什么？

这就是短期记忆管理的核心问题：在有限的窗口内，让最重要的信息留在LLM的视野中。

上下文窗口的分配策略

在讨论如何"截断"之前，先理解上下文的构成。一次典型的Agent循环中，上下文窗口的分配大致如下：

组成部分	说明	Token估算
系统提示词（System Prompt）	Agent的角色定义、行为规则	~500-2000
工具Schema（Tool Schemas）	每个工具的JSON定义（名称、描述、参数）	~200-1000
对话历史（Conversation History）	用户输入、LLM的响应（文本或工具调用）、工具执行结果	变化最大
预留空间（Reserved Space）	LLM的输出token、安全边际（防止截断）	~500-1000

关键洞察：系统提示词和工具Schema是固定开销——每次LLM调用都需要完整发送。对话历史是可变开销——随着循环进行不断增长。当总token数接近窗口上限时，只能从对话历史中做减法。

朴素的截断方法及其问题

最直观的截断方法是：保留最近的消息，丢弃最旧的消息。

def truncate_messages_simple(messages: list, max_messages: int = 20) -> list:
    """简单截断：只保留最近max_messages条消息"""
    if len(messages) <= max_messages:
        return messages
    return messages[-max_messages:]

这个方法简单，但有一个严重的问题：它可能丢弃最关键的信息。

考虑以下场景：

[系统提示词]         ← 不应被丢弃
[用户："帮我查订单"]  ← 任务的起点，包含关键信息
[LLM：调用查询工具]   ← 工具调用
[工具结果：订单详情]  ← 关键数据
[LLM：调用发货工具]   ← 工具调用
[工具结果：发货成功]  ← 关键结果
[LLM："已完成发货"]   ← 回答
[用户："查一下物流"]  ← 新的用户输入
...（大量中间消息）...
[LLM：调用物流工具]   ← 当前正在进行的操作

如果只保留最近20条消息，可能把最初的用户输入和最早的订单数据丢弃了。LLM仍然能看到"当前在查物流"，但它不知道用户在查哪个订单的物流——因为订单号在很早之前的消息中。

简单截断的假设是错误的：它假设"最近的信息最重要"，但在Agent的多轮循环中，任务的关键信息往往分布在对话的各个位置。

基于消息重要性的管理策略

更好的方法是：不是按时间截断，而是按重要性选择保留什么。

不同角色的消息有不同的重要性：

消息类型	重要性	理由
系统提示词	最高	定义Agent的行为，永远不能丢弃
用户输入	高	包含任务目标和关键信息
工具调用请求	中	LLM的决策，通常可以由工具结果推断
工具执行结果	中-高	关键数据，后续推理依赖于此
LLM的中间回答	低-中	通常可以从后续消息中推断
错误信息	高	告诉LLM什么不能做

基于这个认知，设计一个分层管理的短期记忆系统：

from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from enum import Enum

class MessageImportance(Enum):
    """消息的重要性级别"""
    CRITICAL = 5    # 永远保留：系统提示词、核心任务目标
    HIGH = 4        # 优先保留：用户输入、工具结果
    MEDIUM = 3      # 常规保留：工具调用、中间回答
    LOW = 2         # 可丢弃：重复信息、已被替代的信息
    MINIMAL = 1     # 最先丢弃：冗余确认、简单确认

@dataclass
class ManagedMessage:
    """带重要性标记的消息"""
    role: str
    content: str
    importance: MessageImportance
    tool_call_id: Optional[str] = None
    token_estimate: int = 0
    timestamp: float = 0.0

    def to_api_format(self) -> Dict[str, Any]:
        """转换为LLM API所需的消息格式"""
        msg = {"role": self.role, "content": self.content}
        if self.tool_call_id:
            msg["tool_call_id"] = self.tool_call_id
        return msg

class WorkingMemory:
    """短期记忆：管理当前会话的上下文窗口"""

    def __init__(self, max_tokens: int = 8000):
        self.max_tokens = max_tokens
        self.messages: List[ManagedMessage] = []
        self.system_prompt: Optional[ManagedMessage] = None

    def set_system_prompt(self, content: str):
        """设置系统提示词（永远保留）"""
        self.system_prompt = ManagedMessage(
            role="system",
            content=content,
            importance=MessageImportance.CRITICAL,
            token_estimate=self._estimate_tokens(content),
        )

    def add_user_message(self, content: str):
        """添加用户消息"""
        self.messages.append(ManagedMessage(
            role="user",
            content=content,
            importance=MessageImportance.HIGH,
            token_estimate=self._estimate_tokens(content),
        ))

    def add_assistant_message(self, content: str,
                              importance: MessageImportance = MessageImportance.MEDIUM):
        """添加LLM的消息"""
        self.messages.append(ManagedMessage(
            role="assistant",
            content=content,
            importance=importance,
            token_estimate=self._estimate_tokens(content),
        ))

    def add_tool_result(self, tool_call_id: str, content: str,
                        importance: MessageImportance = MessageImportance.HIGH):
        """添加工具执行结果"""
        self.messages.append(ManagedMessage(
            role="tool",
            content=content,
            tool_call_id=tool_call_id,
            importance=importance,
            token_estimate=self._estimate_tokens(content),
        ))

    def get_context(self) -> List[Dict[str, Any]]:
        """获取当前上下文（在窗口限制内）"""
        context = []
        current_tokens = 0

        # 系统提示词始终在首位
        if self.system_prompt:
            context.append(self.system_prompt.to_api_format())
            current_tokens += self.system_prompt.token_estimate

        # 预留输出空间
        reserved_output_tokens = 1024
        available_tokens = self.max_tokens - current_tokens - reserved_output_tokens

        # 按重要性排序选择消息
        selected = self._select_messages_within_budget(available_tokens)
        context.extend(selected)

        return context

    def _select_messages_within_budget(self, budget: int) -> List[Dict[str, Any]]:
        """在token预算内选择最重要的消息"""
        # 按重要性排序
        sorted_messages = sorted(
            self.messages,
            key=lambda m: (m.importance.value, m.timestamp),
            reverse=True,  # 重要性高的在前
        )

        selected = []
        used_tokens = 0

        for msg in sorted_messages:
            if used_tokens + msg.token_estimate > budget:
                continue
            selected.append(msg)
            used_tokens += msg.token_estimate

        # 恢复原始的时间顺序
        selected.sort(key=lambda m: m.timestamp)

        return [m.to_api_format() for m in selected]

    def _estimate_tokens(self, text: str) -> int:
        """粗略估算token数"""
        # 中文：约1.5-2字符/token
        # 英文：约4字符/token
        # 混合估算：取平均值约2.5字符/token
        return max(1, len(text) // 2)

    def get_stats(self) -> Dict[str, Any]:
        """获取当前记忆的统计信息"""
        total_tokens = sum(m.token_estimate for m in self.messages)
        if self.system_prompt:
            total_tokens += self.system_prompt.token_estimate

        return {
            "message_count": len(self.messages),
            "total_estimated_tokens": total_tokens,
            "window_capacity": self.max_tokens,
            "utilization": f"{total_tokens / self.max_tokens * 100:.1f}%",
            "importance_distribution": {
                level.name: sum(1 for m in self.messages if m.importance == level)
                for level in MessageImportance
            }
        }

这个实现的核心思想是：不是按时间截断，而是按重要性选择。

但等等——这个实现有一个问题：它把对话的顺序打乱了。 LLM的对话历史是有顺序的——用户先说了什么，LLM后回应了什么。如果只按重要性排序而忽略时间顺序，LLM看到的将是一堆乱序的消息，无法理解对话的因果链条。

更好的做法：保持时间顺序，但在窗口不足时从最不重要的开始丢弃：

def _select_messages_within_budget(self, budget: int) -> List[Dict[str, Any]]:
    """在token预算内选择消息（保持时间顺序）"""
    # 计算当前总token数
    total_tokens = sum(m.token_estimate for m in self.messages)

    if total_tokens <= budget:
        # 全部保留
        return [m.to_api_format() for m in self.messages]

    # 需要丢弃一些消息
    # 从重要性最低的开始标记丢弃
    messages_to_keep = [True] * len(self.messages)
    current_tokens = total_tokens

    # 按重要性从低到高依次丢弃
    for importance_level in [MessageImportance.MINIMAL,
                              MessageImportance.LOW,
                              MessageImportance.MEDIUM]:
        for i, msg in enumerate(self.messages):
            if not messages_to_keep[i]:
                continue
            if msg.importance == importance_level:
                messages_to_keep[i] = False
                current_tokens -= msg.token_estimate
                if current_tokens <= budget:
                    # 已满足预算
                    break
        if current_tokens <= budget:
            break

    # 如果仍然超出预算（只剩HIGH和CRITICAL），需要更激进的策略
    if current_tokens > budget:
        # 从最旧的HIGH消息开始丢弃
        for i, msg in enumerate(self.messages):
            if not messages_to_keep[i]:
                continue
            if msg.importance == MessageImportance.HIGH:
                messages_to_keep[i] = False
                current_tokens -= msg.token_estimate
                if current_tokens <= budget:
                    break

    # 按原始顺序返回保留的消息
    return [self.messages[i].to_api_format()
            for i in range(len(self.messages)) if messages_to_keep[i]]

这个实现保持了时间顺序，同时优先丢弃重要性低的消息。但丢弃时有一个重要细节：工具调用和工具结果是成对的。丢弃了工具调用而保留工具结果，LLM会看到一条"悬空"的工具结果，不知道它对应什么调用。

处理成对消息：

def _find_related_message_indices(self, messages: List[ManagedMessage],
                                   target_index: int) -> List[int]:
    """找到与目标消息相关的所有消息的索引"""
    target = messages[target_index]
    related = [target_index]

    # 如果目标是工具结果，找到对应的工具调用
    if target.role == "tool" and target.tool_call_id:
        for i, msg in enumerate(messages):
            if (msg.role == "assistant" and
                target.tool_call_id in msg.content):
                related.append(i)

    # 如果目标是工具调用，找到对应的工具结果
    if target.role == "assistant" and "tool_calls" in target.content:
        # 简化处理：假设紧跟在后面的tool消息是相关的
        for i in range(target_index + 1, min(target_index + 5, len(messages))):
            if messages[i].role == "tool":
                related.append(i)

    return related

消息摘要：更优雅的"压缩"

丢弃不是唯一的策略。更好的方式是压缩：将多条消息合并为一条摘要。

比如，一个工具返回了5000字的原始数据，LLM之后的推理只用到了其中的几个关键数字。与其在后续循环中重复发送这5000字，不如将它替换为一条摘要：

class WorkingMemory:
    def compress_messages(self, start_index: int, end_index: int,
                          summary: str):
        """将索引范围内的消息压缩为一条摘要"""
        if start_index < 0 or end_index >= len(self.messages):
            raise ValueError("无效的消息索引范围")

        # 创建摘要消息
        summary_msg = ManagedMessage(
            role="system",
            content=f"[前文摘要] {summary}",
            importance=MessageImportance.HIGH,
            token_estimate=self._estimate_tokens(summary),
            timestamp=self.messages[end_index].timestamp,
        )

        # 替换原消息
        self.messages = (
            self.messages[:start_index] +
            [summary_msg] +
            self.messages[end_index + 1:]
        )

    def auto_compress(self, summary_func=None):
        """自动压缩旧消息"""
        if self.get_stats()["total_estimated_tokens"] < self.max_tokens * 0.7:
            return  # 使用率低于70%，不需要压缩

        # 找到最早的N条消息
        n_to_compress = max(1, len(self.messages) // 3)
        old_messages = self.messages[:n_to_compress]

        if not old_messages:
            return

        # 生成摘要
        if summary_func:
            summary = summary_func(old_messages)
        else:
            # 简单的摘要：拼接前几段
            summary = self._simple_summary(old_messages)

        # 执行压缩
        self.compress_messages(0, n_to_compress - 1, summary)

    def _simple_summary(self, messages: List[ManagedMessage]) -> str:
        """简单的摘要：提取关键信息"""
        user_inputs = [m.content for m in messages if m.role == "user"]
        tool_results = [m.content[:100] for m in messages if m.role == "tool"]

        summary_parts = []
        if user_inputs:
            summary_parts.append(f"用户请求：{user_inputs[-1]}")
        if tool_results:
            summary_parts.append(f"工具返回（截断）：{'; '.join(tool_results[:3])}")

        return " | ".join(summary_parts) if summary_parts else "前文对话内容"

摘要的质量是压缩策略的核心。简单的拼接摘要（如上例）虽然减少了token消耗，但可能丢失关键信息。更好的方式是用LLM来生成摘要——但这又消耗额外的token。这是一个权衡：

• 简单摘要：节省token但丢失信息多
• LLM摘要：信息保留好但消耗额外token

实际应用中的折衷：只在上下文使用率超过阈值时触发LLM摘要，并且只摘要最早的消息（这些消息通常与当前推理的关联最弱）。

一个实用的短期记忆管理策略

综合以上讨论，一个实用的短期记忆管理系统应该做以下事情：

第一：分类管理消息。 每条消息都被赋予重要性级别。系统提示词是CRITICAL，用户输入和工具结果是HIGH，中间推理是MEDIUM。

第二：监控窗口使用率。 在每次添加消息时检查窗口使用率。设置两个阈值：

• 警告阈值（如70%）：触发自动压缩
• 紧急阈值（如90%）：触发主动丢弃

第三：优先压缩而非丢弃。 压缩保留了信息的"骨架"，丢弃则完全丢失信息。只有在压缩后仍然超出窗口时，才考虑丢弃LOW和MINIMAL重要性的消息。

第四：记录被丢弃的内容。 当不得不丢弃消息时，在日志中记录丢弃了什么。这对于调试Agent行为至关重要——如果Agent"忘了"某个关键信息，你需要知道是窗口管理把它丢弃了，还是LLM自身的问题。

监控窗口使用情况

窗口使用情况应该被持续监控和记录：

class MemoryMonitor:
    """监控短期记忆的使用情况"""

    def __init__(self, working_memory: WorkingMemory):
        self.working_memory = working_memory
        self.peak_utilization = 0.0
        self.compressions = 0
        self.discards = 0
        self.utilization_history = []

    def check(self):
        """检查当前使用率并记录"""
        stats = self.working_memory.get_stats()
        utilization = stats["total_estimated_tokens"] / stats["window_capacity"]

        self.peak_utilization = max(self.peak_utilization, utilization)
        self.utilization_history.append(utilization)

        return utilization

    def get_report(self) -> Dict[str, Any]:
        """生成使用报告"""
        return {
            "peak_utilization": f"{self.peak_utilization * 100:.1f}%",
            "compressions_performed": self.compressions,
            "messages_discarded": self.discards,
            "utilization_trend": self.utilization_history[-10:] if self.utilization_history else [],
        }

本节小结

短期记忆管理的核心是在有限的上下文窗口内做最优的信息选择。本节讨论了：

• 窗口的构成：系统提示词、工具Schema、对话历史、预留空间
• 简单截断的问题：丢弃了可能关键的历史信息
• 基于重要性的管理：给消息分级，优先丢弃低重要性消息
• 压缩作为替代：将多条消息合并为摘要，保留信息"骨架"
• 成对消息的处理：工具调用和结果必须一起保留或丢弃
• 监控的重要性：持续跟踪窗口使用率，在需要时触发管理策略

核心原则：短期记忆管理不是"丢弃旧消息"那么简单，而是在信息完整性和窗口限制之间做持续的权衡。好的管理策略让Agent在窗口不足时仍然保持对任务关键信息的"可见性"。

在下一节中，我们将讨论长期记忆——如何让Agent跨会话记住信息，使用向量数据库和检索机制实现持久化的知识存储。