第02章 对话上下文管理：短期记忆的实现
第02章 对话上下文管理：短期记忆的实现
“Token窗口就是Agent的注意力范围。超出这个范围的，Agent就当不存在。管理好这个窗口，是记忆工程的第一步。” —— LLM工程实践
所有的记忆问题，从最简单的开始：一次对话内，如何不让重要信息被"挤出"上下文窗口？
2.1 Token窗口的限制
# 理解Token窗口限制
MODEL_CONTEXT_WINDOWS = {
"gpt-4o": 128_000, # tokens
"claude-3-5-sonnet": 200_000,
"claude-3-haiku": 200_000,
"gpt-3.5-turbo": 16_385,
"llama-3-8b": 8_192,
}
# 直觉上感受一下大小：
# 1 token ≈ 0.75 个英文单词，约 1.5 个中文字
# 128K tokens ≈ 192,000 中文字 ≈ 一本200页的书
# 听起来很大，但实际上：
CONTEXT_CONSUMPTION_EXAMPLES = {
"system_prompt": "500-2000 tokens（Agent的指令、人格设定）",
"tool_schemas": "每个工具约 200-500 tokens（10个工具 = 5000 tokens）",
"user_message": "一条普通消息 50-200 tokens",
"agent_response": "一条回复 200-800 tokens",
"document_chunk": "粘贴一段代码可能 500-5000 tokens",
}
# 假设对话循环：每轮消耗 1000 tokens（问题+回答）
# 128K 窗口 = 约 128 轮对话
# 但如果中间粘贴了大量代码、文档，可能 20-30 轮就满了
# 窗口满了会发生什么？
WHAT_HAPPENS_WHEN_FULL = """
大多数LLM API：
- 直接报错（超过最大token数）
- 或者静默截断（丢弃最早的内容）
无论哪种，结果都是：Agent"失忆"了早期的重要信息
"""
2.2 滑动窗口：保留最近N条消息
# 最简单的上下文管理策略：只保留最近N条消息
from typing import Optional
from dataclasses import dataclass, field
import tiktoken # OpenAI的tokenizer库
@dataclass
class Message:
role: str # "user" / "assistant" / "system"
content: str
tokens: int = 0
class SlidingWindowContext:
"""
滑动窗口上下文管理器
- 保留最近N条消息
- 始终保留system prompt
- 支持token数量限制
"""
def __init__(
self,
max_tokens: int = 8000,
max_messages: int = 20,
model: str = "gpt-4o",
):
self.max_tokens = max_tokens
self.max_messages = max_messages
self.system_prompt: Optional[str] = None
self.messages: list[Message] = []
# 初始化tokenizer
try:
self.encoding = tiktoken.encoding_for_model(model)
except:
self.encoding = tiktoken.get_encoding("cl100k_base")
def count_tokens(self, text: str) -> int:
return len(self.encoding.encode(text))
def set_system_prompt(self, prompt: str):
self.system_prompt = prompt
def add_message(self, role: str, content: str):
tokens = self.count_tokens(content)
self.messages.append(Message(role=role, content=content, tokens=tokens))
# 超出限制时，裁剪旧消息
self._trim()
def _trim(self):
"""裁剪消息，确保在Token和数量限制内"""
system_tokens = self.count_tokens(self.system_prompt or "")
while len(self.messages) > self.max_messages:
self.messages.pop(0) # 删除最早的消息
# Token数量限制
current_tokens = system_tokens + sum(m.tokens for m in self.messages)
while current_tokens > self.max_tokens and len(self.messages) > 2:
removed = self.messages.pop(0) # 删除最早的消息
current_tokens -= removed.tokens
def get_context(self) -> list[dict]:
"""获取LLM API格式的上下文"""
messages = []
if self.system_prompt:
messages.append({"role": "system", "content": self.system_prompt})
for msg in self.messages:
messages.append({"role": msg.role, "content": msg.content})
return messages
def get_stats(self) -> dict:
"""当前上下文统计"""
total_tokens = sum(m.tokens for m in self.messages)
if self.system_prompt:
total_tokens += self.count_tokens(self.system_prompt)
return {
"messages_count": len(self.messages),
"total_tokens": total_tokens,
"remaining_capacity": self.max_tokens - total_tokens,
"usage_percent": f"{total_tokens/self.max_tokens*100:.1f}%",
}
# 使用示例
context = SlidingWindowContext(max_tokens=8000, max_messages=20)
context.set_system_prompt("你是一个专业的Python开发助手。")
# 模拟对话
context.add_message("user", "你好，我叫Charlie，在用FastAPI开发项目")
context.add_message("assistant", "你好Charlie！FastAPI是个很棒的选择...")
context.add_message("user", "帮我写一个用户认证的路由")
# ... 继续添加消息
print(context.get_stats())
# {'messages_count': 3, 'total_tokens': 350, 'remaining_capacity': 7650, ...}
2.3 重要消息固定：确保关键信息不被丢失
# 问题：滑动窗口会丢失重要的早期信息
# 解决：标记"重要消息"，不让它们被裁剪
from enum import Enum
class MessagePriority(Enum):
NORMAL = "normal" # 可以被裁剪
PINNED = "pinned" # 永不裁剪
SUMMARY = "summary" # 摘要消息（特殊处理）
@dataclass
class Message:
role: str
content: str
tokens: int = 0
priority: MessagePriority = MessagePriority.NORMAL
metadata: dict = field(default_factory=dict)
class PinningContext(SlidingWindowContext):
"""
支持消息固定的上下文管理器
"""
def add_message(self, role: str, content: str, pin: bool = False):
tokens = self.count_tokens(content)
priority = MessagePriority.PINNED if pin else MessagePriority.NORMAL
self.messages.append(Message(
role=role, content=content, tokens=tokens, priority=priority
))
self._trim()
def _trim(self):
"""裁剪时保护PINNED消息"""
system_tokens = self.count_tokens(self.system_prompt or "")
# 只裁剪NORMAL优先级的消息
normal_messages = [m for m in self.messages if m.priority == MessagePriority.NORMAL]
pinned_messages = [m for m in self.messages if m.priority == MessagePriority.PINNED]
# 先检查PINNED消息占用的token
pinned_tokens = sum(m.tokens for m in pinned_messages)
available_for_normal = self.max_tokens - system_tokens - pinned_tokens
# 从最早的NORMAL消息开始裁剪
while normal_messages:
current_normal_tokens = sum(m.tokens for m in normal_messages)
if current_normal_tokens <= available_for_normal:
break
normal_messages.pop(0)
# 重建消息列表，保持时间顺序
# 简化实现：把PINNED放前面
self.messages = pinned_messages + normal_messages
# 使用示例
ctx = PinningContext(max_tokens=4000)
ctx.set_system_prompt("你是Python助手")
# 用户介绍自己——固定这条消息
ctx.add_message("user", "我叫Charlie，在用Python开发一个广告管理系统Market Vault，技术栈是FastAPI+PostgreSQL+Redis", pin=True)
ctx.add_message("assistant", "你好Charlie！...")
# 后续对话正常添加（可能被裁剪）
for i in range(50): # 模拟大量对话
ctx.add_message("user", f"问题{i}: 关于某个技术细节...")
ctx.add_message("assistant", f"回答{i}: 详细的技术回答...")
# 无论多少轮对话，Charlie的自我介绍都会保留
context_for_llm = ctx.get_context()
assert any("Charlie" in m["content"] for m in context_for_llm)
print("Charlie的介绍被保留了！")
2.4 实时Token监控与警告
# 实用功能：在窗口快满时发出警告，让Agent主动摘要
class MonitoredContext(PinningContext):
"""带实时监控的上下文管理器"""
WARNING_THRESHOLD = 0.8 # 使用量超过80%时警告
def add_message(self, role: str, content: str, pin: bool = False):
super().add_message(role, content, pin)
self._check_capacity()
def _check_capacity(self):
stats = self.get_stats()
usage = float(stats["usage_percent"].replace("%", "")) / 100
if usage > self.WARNING_THRESHOLD:
print(f"⚠️ 上下文窗口使用率: {stats['usage_percent']} ({stats['total_tokens']}/{self.max_tokens} tokens)")
print(f" 建议：触发记忆摘要，压缩历史对话")
def needs_summarization(self) -> bool:
"""判断是否需要摘要压缩"""
stats = self.get_stats()
usage = float(stats["usage_percent"].replace("%", "")) / 100
return usage > self.WARNING_THRESHOLD
# 集成到Agent循环
async def agent_loop_with_memory(user_input: str, context: MonitoredContext, llm_client):
# 如果上下文快满了，先摘要（第06章会详细实现）
if context.needs_summarization():
summary = await summarize_context(context) # 第06章的函数
context.compress_with_summary(summary)
context.add_message("user", user_input)
response = await llm_client.chat(
messages=context.get_context()
)
context.add_message("assistant", response.content)
return response.content
本章小结
五个核心认知：
-
上下文窗口是Agent的"注意力焦点"，有限且宝贵：就算128K tokens听起来很大，但包含工具Schema、系统提示、对话历史后，有效空间会快速减少；必须主动管理
-
滑动窗口是最简单有效的短期记忆管理策略：保留最近N条消息，丢弃最早的；实现简单，效果立竿见影；适合大多数场景
-
不是所有消息都应该被平等对待：用户的背景介绍、关键决策、重要约束——这些应该被"固定"（pin），永远不裁剪；临时的闲聊可以优先裁剪
-
主动监控使用率，在满之前采取行动：不要等到窗口满了再报错；当使用率超过80%时，主动触发摘要压缩（第06章）
-
短期记忆只解决"当次对话内"的问题：对话结束后，这些信息依然会丢失；要实现跨对话记忆，需要外部存储（第03章）
核心行动：
# 今天就能做的改进：
# 如果你有现有的Agent，检查它的上下文管理：
# 1. 它是否会在长对话中"忘记"早期信息？
# 2. 有没有用户的关键背景信息（名字、项目、偏好）被丢失？
# 3. 用本章的 MonitoredContext 替换你的上下文管理代码
# 4. 标记出最重要的3条消息，加上 pin=True
本章提示词模板
模板一：分析对话中哪些信息应该被固定
以下是一段Agent对话记录：
[粘贴对话内容]
请帮我识别：
1. 哪些信息是"必须永远保留"的（用户身份、关键约束、重要决策）？
2. 哪些信息是"可以在摘要后丢弃"的（具体的技术细节讨论）？
3. 哪些信息是"应该提取到长期记忆"的（用户偏好、反复出现的模式）？
4. 如果这段对话被大幅截断，哪些损失是不可接受的？
5. 建议在system prompt中预置什么用户背景信息？
模板二：上下文压缩策略设计
我的Agent对话场景是：[描述场景]
典型对话长度：[X轮，大约Y tokens]
最重要的信息类型：[用户背景/技术讨论/决策记录等]
请帮我设计上下文管理策略：
1. 应该保留最近几条消息？（滑动窗口大小）
2. 哪类消息应该pin（永远保留）？
3. 什么时候触发摘要压缩（什么阈值）？
4. 摘要应该保留哪些核心信息？
5. 是否需要额外的"关键事实"板块（在system prompt中持续维护）？