上下文管理最佳实践
本文档整理上下文窗口管理的最佳实践，帮助在有限上下文中最大化信息价值。
核心挑战
txt
┌─────────────────────────────────────────────────────┐
│ 上下文管理挑战 │
├─────────────────────────────────────────────────────┤
│ │
│ 输入受限 ──────────────────── 输出受限 │
│ │ │ │
│ ↓ ↓ │
│ ┌─────────────────────────────────────────┐ │
│ │ 上下文窗口限制 │ │
│ │ (4K / 8K / 32K / 128K / 1M) │ │
│ └─────────────────────────────────────────┘ │
│ │ │ │
│ ↓ ↓ │
│ 信息选择 ──────────────────── 信息压缩 │
│ │
│ 挑战： │
│ - 如何选择最相关的信息？ │
│ - 如何压缩信息而不丢失关键内容？ │
│ - 如何管理长期记忆？ │
│ - 如何处理超长文档？ │
│ │
└─────────────────────────────────────────────────────┘模型上下文窗口参考
| 模型 | 上下文窗口 | 输入限制 | 输出限制 |
|---|---|---|---|
| GPT-4 Turbo | 128K | ~128K tokens | 4K tokens |
| GPT-4 | 8K / 32K | ~8K/32K tokens | 4K tokens |
| GPT-3.5 Turbo | 16K | ~16K tokens | 4K tokens |
| Claude 3 Opus | 200K | ~200K tokens | 4K tokens |
| Claude 3 Sonnet | 200K | ~200K tokens | 4K tokens |
| Gemini 1.5 Pro | 1M | ~1M tokens | 8K tokens |
策略 1：上下文优先级
优先级框架
yaml
上下文优先级分层:
P0 - 必须保留:
- 系统指令
- 当前用户请求
- 关键约束和规则
- 格式要求
P1 - 高优先级:
- 最近 N 轮对话
- 关键实体和定义
- 当前任务状态
- 重要决策记录
P2 - 中优先级:
- 相关历史上下文
- 检索到的文档片段
- 中间推理过程
P3 - 低优先级:
- 详细的历史记录
- 完整的原始文档
- 冗余信息
P4 - 可丢弃:
- 已处理的信息
- 不相关的历史
- 错误和重试记录实现示例
python
class ContextPrioritizer:
def __init__(self, max_tokens: int, tokenizer):
self.max_tokens = max_tokens
self.tokenizer = tokenizer
def prioritize(self, context: dict) -> str:
"""按优先级构建上下文"""
result = []
current_tokens = 0
# P0: 系统指令（必须保留）
system = context.get("system", "")
system_tokens = self._count_tokens(system)
if current_tokens + system_tokens <= self.max_tokens:
result.append(system)
current_tokens += system_tokens
# P0: 当前请求（必须保留）
request = context.get("request", "")
request_tokens = self._count_tokens(request)
if current_tokens + request_tokens <= self.max_tokens:
result.append(request)
current_tokens += request_tokens
# P1: 关键信息
key_info = context.get("key_info", [])
for info in key_info:
info_tokens = self._count_tokens(info)
if current_tokens + info_tokens <= self.max_tokens * 0.8: # 留 20% 余量
result.append(info)
current_tokens += info_tokens
# P2: 相关上下文
relevant_context = context.get("relevant_context", [])
for ctx in relevant_context:
ctx_tokens = self._count_tokens(ctx)
if current_tokens + ctx_tokens <= self.max_tokens * 0.9:
result.append(ctx)
current_tokens += ctx_tokens
return "\n\n".join(result)
def _count_tokens(self, text: str) -> int:
"""计算 token 数量"""
return len(self.tokenizer.encode(text))
```txt
## 策略 2：上下文压缩
### 压缩技术
```python
class ContextCompressor:
def __init__(self, model):
self.model = model
def compress(self, text: str, target_ratio: float = 0.5) -> str:
"""压缩文本到目标比例"""
if target_ratio >= 1.0:
return text
prompt = f"""
Compress the following text to approximately {target_ratio * 100}% of its original length.
Preserve all key information, entities, and relationships.
Remove redundancy and verbosity.
Original text:
{text}
Compressed text:
"""
return self.model.generate(prompt)
def summarize_sections(self, text: str, section_size: int = 1000) -> str:
"""分段摘要"""
sections = self._split_text(text, section_size)
summaries = []
for section in sections:
summary = self._summarize_section(section)
summaries.append(summary)
return "\n\n".join(summaries)
def extract_key_points(self, text: str, max_points: int = 5) -> list[str]:
"""提取关键点"""
prompt = f"""
Extract the top {max_points} key points from the following text.
Each point should be a single, concise sentence.
Text:
{text}
Key points (one per line):
"""
result = self.model.generate(prompt)
return result.strip().split("\n")
def _split_text(self, text: str, size: int) -> list[str]:
"""分割文本"""
words = text.split()
sections = []
current = []
current_size = 0
for word in words:
current.append(word)
current_size += len(word) + 1
if current_size >= size:
sections.append(" ".join(current))
current = []
current_size = 0
if current:
sections.append(" ".join(current))
return sections
def _summarize_section(self, section: str) -> str:
"""摘要单个段落"""
prompt = f"Summarize the following in 1-2 sentences:\n\n{section}"
return self.model.generate(prompt)压缩策略选择
| 策略 | 压缩比 | 信息保留 | 适用场景 |
|---|---|---|---|
| 关键点提取 | 10-20% | 高 | 事实性文本 |
| 分段摘要 | 20-40% | 中高 | 长文档 |
| 语义压缩 | 40-60% | 中 | 对话历史 |
| 截断 | 可变 | 低 | 最后手段 |
策略 3：记忆管理
记忆层次架构
txt
┌─────────────────────────────────────────────────────┐
│ 记忆层次架构 │
├─────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 工作记忆 (Working Memory) │ │
│ │ 当前上下文窗口 │ │
│ │ 容量: 4K - 1M tokens │ │
│ └─────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 短期记忆 (Short-term Memory) │ │
│ │ 最近 N 轮对话 │ │
│ │ 容量: 最近 10-20 轮 │ │
│ └─────────────────────────────────────────────┘ │
│ ↓ │
│ ┌─────────────────────────────────────────────┐ │
│ │ 长期记忆 (Long-term Memory) │ │
│ │ 向量数据库 / 知识库 │ │
│ │ 容量: 无限 │ │
│ └─────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────┘实现示例
python
from dataclasses import dataclass
from datetime import datetime
from typing import Optional
import json
@dataclass
class Memory:
content: str
timestamp: datetime
importance: float # 0-1
type: str # "fact", "event", "preference", "decision"
metadata: dict
class MemoryManager:
def __init__(self, vector_store, summarizer, max_working_tokens=8000):
self.vector_store = vector_store
self.summarizer = summarizer
self.max_working_tokens = max_working_tokens
self.working_memory: list[Memory] = []
self.short_term_memory: list[Memory] = []
def add_memory(self, content: str, memory_type: str, importance: float = 0.5):
"""添加新记忆"""
memory = Memory(
content=content,
timestamp=datetime.now(),
importance=importance,
type=memory_type,
metadata={}
)
# 添加到工作记忆
self.working_memory.append(memory)
# 检查是否需要压缩
if self._get_working_tokens() > self.max_working_tokens:
self._compress_working_memory()
# 存储到长期记忆
self.vector_store.add(memory)
def recall(self, query: str, top_k: int = 5) -> list[Memory]:
"""检索相关记忆"""
# 从长期记忆检索
long_term_results = self.vector_store.search(query, top_k)
# 从短期记忆匹配
short_term_results = [
m for m in self.short_term_memory
if query.lower() in m.content.lower()
]
# 合并和去重
all_results = long_term_results + short_term_results
unique_results = self._deduplicate(all_results)
# 按重要性和时间排序
sorted_results = sorted(
unique_results,
key=lambda m: (m.importance, m.timestamp),
reverse=True
)
return sorted_results[:top_k]
def consolidate(self):
"""将工作记忆整合到短期记忆"""
for memory in self.working_memory:
if memory.importance > 0.7:
self.short_term_memory.append(memory)
# 限制短期记忆大小
if len(self.short_term_memory) > 20:
# 移除最不重要的
self.short_term_memory.sort(key=lambda m: m.importance, reverse=True)
self.short_term_memory = self.short_term_memory[:20]
# 清空工作记忆
self.working_memory = []
def _compress_working_memory(self):
"""压缩工作记忆"""
if not self.working_memory:
return
# 按重要性排序
self.working_memory.sort(key=lambda m: m.importance, reverse=True)
# 保留高重要性记忆
high_importance = [m for m in self.working_memory if m.importance > 0.8]
# 压缩低重要性记忆
low_importance = [m for m in self.working_memory if m.importance <= 0.8]
if low_importance:
combined = "\n".join([m.content for m in low_importance])
summary = self.summarizer.compress(combined, 0.3)
summary_memory = Memory(
content=summary,
timestamp=datetime.now(),
importance=0.7,
type="summary",
metadata={"original_count": len(low_importance)}
)
high_importance.append(summary_memory)
self.working_memory = high_importance
def _get_working_tokens(self) -> int:
"""计算工作记忆的 token 数"""
total = 0
for memory in self.working_memory:
total += len(memory.content.split()) * 1.3 # 粗略估计
return int(total)
def _deduplicate(self, memories: list[Memory]) -> list[Memory]:
"""去重"""
seen = set()
result = []
for memory in memories:
key = memory.content[:100] # 使用前 100 字符作为 key
if key not in seen:
seen.add(key)
result.append(memory)
return result策略 4：滑动窗口
实现示例
python
class SlidingWindow:
def __init__(self, max_messages: int = 20, overlap: int = 2):
self.max_messages = max_messages
self.overlap = overlap
self.messages = []
def add_message(self, role: str, content: str):
"""添加消息"""
self.messages.append({"role": role, "content": content})
# 检查是否需要滑动
if len(self.messages) > self.max_messages:
self._slide()
def get_context(self) -> list[dict]:
"""获取当前上下文"""
return self.messages.copy()
def _slide(self):
"""滑动窗口"""
# 保留最近的 max_messages - overlap 条消息
# 以及 overlap 条旧消息的摘要
keep_count = self.max_messages - self.overlap
# 提取要压缩的旧消息
old_messages = self.messages[:self.overlap]
old_summary = self._summarize_old(old_messages)
# 构建新的消息列表
self.messages = [
{"role": "system", "content": f"[历史摘要] {old_summary}"},
*self.messages[keep_count:]
]
def _summarize_old(self, messages: list[dict]) -> str:
"""摘要旧消息"""
# 简化实现，实际应该调用 LLM
topics = []
for msg in messages:
# 提取关键实体
topics.append(msg["content"][:50])
return " | ".join(topics)滑动窗口变体
python
class SmartSlidingWindow:
"""智能滑动窗口：根据重要性滑动"""
def __init__(self, max_tokens: int = 8000):
self.max_tokens = max_tokens
self.messages = []
self.importance_scores = []
def add_message(self, role: str, content: str, importance: float = 0.5):
"""添加消息"""
self.messages.append({
"role": role,
"content": content,
"tokens": self._estimate_tokens(content)
})
self.importance_scores.append(importance)
# 检查是否需要压缩
if self._get_total_tokens() > self.max_tokens:
self._smart_compress()
def _smart_compress(self):
"""智能压缩"""
while self._get_total_tokens() > self.max_tokens * 0.8:
# 找到最不重要的消息
min_idx = self.importance_scores.index(min(self.importance_scores))
# 压缩或删除
if self.importance_scores[min_idx] < 0.3:
# 直接删除
del self.messages[min_idx]
del self.importance_scores[min_idx]
else:
# 压缩
self.messages[min_idx]["content"] = self._compress_content(
self.messages[min_idx]["content"]
)
self.messages[min_idx]["tokens"] = self._estimate_tokens(
self.messages[min_idx]["content"]
)
self.importance_scores[min_idx] *= 0.8 # 降低重要性
def _estimate_tokens(self, text: str) -> int:
"""估算 token 数"""
return int(len(text.split()) * 1.3)
def _get_total_tokens(self) -> int:
"""计算总 token 数"""
return sum(msg["tokens"] for msg in self.messages)
def _compress_content(self, content: str) -> str:
"""压缩内容"""
# 简化实现
return content[:len(content) // 2] + "..."策略 5：检索增强
检索策略
python
class RetrievalAugmentedContext:
def __init__(self, retriever, reranker, max_context_tokens=6000):
self.retriever = retriever
self.reranker = reranker
self.max_context_tokens = max_context_tokens
def build_context(self, query: str, system_prompt: str) -> str:
"""构建检索增强的上下文"""
# 1. 初始检索
initial_results = self.retriever.search(query, top_k=20)
# 2. 重排序
reranked_results = self.reranker.rerank(query, initial_results, top_k=5)
# 3. 构建上下文
context_parts = [system_prompt]
current_tokens = self._estimate_tokens(system_prompt)
for doc in reranked_results:
doc_tokens = self._estimate_tokens(doc["content"])
if current_tokens + doc_tokens <= self.max_context_tokens:
context_parts.append(f"\n[Reference]\n{doc['content']}")
current_tokens += doc_tokens
else:
break
context_parts.append(f"\n[Query]\n{query}")
return "\n".join(context_parts)
def _estimate_tokens(self, text: str) -> int:
"""估算 token 数"""
return int(len(text.split()) * 1.3)检索优先级
yaml
检索优先级:
高优先级:
- 精确匹配的实体
- 最近的交互记录
- 用户明确引用的内容
- 关键定义和规则
中优先级:
- 语义相似的文档
- 相关主题的内容
- 历史决策记录
低优先级:
- 模糊匹配的内容
- 过时的信息
- 冗余的内容
检索配置:
初始检索量: 20-50
重排序后保留: 3-5
最小相关分数: 0.7
多样性权重: 0.2策略 6：结构化上下文
上下文模板
python
class StructuredContextBuilder:
def __init__(self, template: str):
self.template = template
def build(self, **kwargs) -> str:
"""构建结构化上下文"""
return self.template.format(**kwargs)
# 系统指令模板
SYSTEM_TEMPLATE = """
# 角色定义
{role_definition}
# 核心能力
{capabilities}
# 约束条件
{constraints}
# 输出格式
{output_format}
"""
# 对话上下文模板
CONVERSATION_TEMPLATE = """
# 当前状态
任务: {current_task}
进度: {progress}
# 相关上下文
{relevant_context}
# 最近交互
{recent_messages}
# 当前请求
{current_request}
"""
# RAG 上下文模板
RAG_TEMPLATE = """
# 任务说明
{task_description}
# 检索结果
{retrieved_documents}
# 用户问题
{user_question}
# 回答要求
1. 基于检索结果回答
2. 标注信息来源
3. 不确定时明确说明
"""使用示例
python
# 创建结构化上下文
builder = StructuredContextBuilder(RAG_TEMPLATE)
context = builder.build(
task_description="根据提供的文档回答用户问题",
retrieved_documents="""
[文档1] 公司成立于 2020 年，主要业务是 AI 解决方案...
[文档2] 公司的 AI 产品包括智能客服、智能推荐...
[文档3] 公司在 2023 年获得 A 轮融资...
""",
user_question="公司的主要产品有哪些？",
current_request="请详细回答用户的问题"
)最佳实践总结
上下文管理清单
markdown
## 上下文管理检查清单
### 必须项
- [ ] 明确上下文窗口限制
- [ ] 实现优先级排序
- [ ] 设置压缩策略
- [ ] 定义保留规则
### 推荐项
- [ ] 实现记忆管理
- [ ] 使用滑动窗口
- [ ] 检索增强上下文
- [ ] 结构化上下文模板
### 可选项
- [ ] 动态上下文调整
- [ ] 多级缓存
- [ ] 上下文预热
- [ ] 上下文分析
### 避免事项
- [ ] 上下文溢出
- [ ] 无限增长的历史
- [ ] 无优先级的检索
- [ ] 无压缩策略Token 预算分配
yaml
Token 预算分配建议:
总预算: 100%
系统指令: 10-15%
- 角色定义
- 核心规则
- 输出格式
检索上下文: 40-50%
- 检索到的文档
- 相关历史
- 关键实体
对话历史: 20-30%
- 最近 N 轮对话
- 历史摘要
用户请求: 10-15%
- 当前问题
- 补充说明
输出预留: 10-20%
- 生成空间
- 安全边际不同场景配置
| 场景 | 上下文预算 | 检索量 | 历史保留 | 压缩策略 |
|---|---|---|---|---|
| 问答 | 系统 10% + 检索 60% + 请求 15% + 输出 15% | 5-10 条 | 最近 3 轮 | 检索压缩 |
| 对话 | 系统 10% + 历史 50% + 请求 20% + 输出 20% | 3-5 条 | 最近 10 轮 | 滑动窗口 |
| 写作 | 系统 10% + 检索 30% + 请求 30% + 输出 30% | 3-5 条 | 最近 5 轮 | 关键点提取 |
| 分析 | 系统 10% + 检索 50% + 请求 20% + 输出 20% | 10-20 条 | 最近 3 轮 | 分段摘要 |
参考来源
- Lost in the Middle: https://arxiv.org/abs/2307.03172
- Long Context Windows: https://arxiv.org/abs/2306.11001
- StreamingLLM: https://arxiv.org/abs/2309.17453
- Memorizing Transformers: https://arxiv.org/abs/2203.08913