共计 2320 个字符,预计需要花费 6 分钟才能阅读完成。
问题背景
在对话系统开发中,窗口上下文(Context Window)就像人类的短期记忆,它决定了 AI 能记住多少对话历史。常见的痛点包括:

- 上下文丢失:用户说 ” 接着上文的方案 ” 时,系统却忘记之前的讨论
- Token 超限:当对话超过模型的最大 token 限制(如 Claude 的 100K)时被截断
- 对话偏移:多轮交互后响应逐渐偏离原始话题
技术方案对比
固定窗口 vs 动态窗口
- 固定窗口策略
- 像固定长度的滑动窗口,始终保持最近的 N 条消息
- 优点:实现简单,内存占用稳定
-
缺点:可能过早丢弃重要早期信息
-
动态窗口策略
- 根据消息重要性动态调整保留内容
- 典型实现:基于 TF-IDF 或嵌入相似度的摘要压缩
- 优点:保留关键信息更久
- 缺点:计算开销较大
Python 实现示例
from typing import List, Dict
import tiktoken # 用于 token 计数
class ContextManager:
def __init__(self, max_tokens: int = 8000):
self.encoder = tiktoken.get_encoding("cl100k_base")
self.max_tokens = max_tokens
self.messages: List[Dict] = []
def add_message(self, role: str, content: str) -> bool:
"""添加新消息并自动修剪上下文"""
new_msg = {'role': role, 'content': content}
self.messages.append(new_msg)
while self._count_tokens() > self.max_tokens:
if len(self.messages) > 1:
self.messages.pop(0) # 移除最旧消息
else:
# 单条消息就超限的特殊处理
self.messages[0]['content'] = self._truncate_content(self.messages[0]['content'],
self.max_tokens
)
break
return True
def _count_tokens(self) -> int:
"""计算当前上下文的 token 总数"""
return sum(len(self.encoder.encode(msg['content']))
for msg in self.messages
)
def _truncate_content(self, text: str, max_tokens: int) -> str:
"""智能截断超长文本"""
tokens = self.encoder.encode(text)
if len(tokens) <= max_tokens:
return text
# 保留开头和结尾重要部分
head = self.encoder.decode(tokens[:max_tokens//2])
tail = self.encoder.decode(tokens[-(max_tokens//2):])
return f"{head}...[中间内容已截断]...{tail}"
生产环境实践
性能优化技巧
- LRU 缓存:对频繁使用的上下文片段建立缓存
- 使用
functools.lru_cache装饰器缓存摘要计算结果 -
示例:
@lru_cache(maxsize=100) def generate_summary(text: str) -> str: # 生成摘要的实现 -
分层存储:
- 热数据:保留在内存中
- 温数据:存入 Redis 等高速缓存
- 冷数据:持久化到数据库
安全防护
-
敏感信息过滤:
def sanitize_context(context): sensitive_keywords = ['密码', 'token', '身份证'] for keyword in sensitive_keywords: if keyword in context: context = context.replace(keyword, '***') return context -
会话隔离:为每个用户 / 会话分配独立的 context_id
框架集成示例
Flask 集成方案
from flask import Flask, request
app = Flask(__name__)
contexts = {} # 全局上下文存储
@app.route('/chat', methods=['POST'])
def chat():
user_id = request.json.get('user_id')
message = request.json.get('message')
# 获取或创建上下文
if user_id not in contexts:
contexts[user_id] = ContextManager()
# 添加用户消息
contexts[user_id].add_message('user', message)
# 调用 Claude API(伪代码)response = claude_api(contexts[user_id].messages)
# 添加 AI 回复到上下文
contexts[user_id].add_message('assistant', response)
return {'response': response}
延伸思考
- 如何评估不同上下文策略的对话质量影响?可以设计哪些量化指标?
- 在超长文档问答场景中,如何平衡上下文完整性和 token 消耗?
- 当需要跨会话持久化知识时,应该采用什么架构设计?
推荐工具链
- LangChain:提供现成的 ConversationBufferMemory 等组件
- Redis:适合分布式上下文存储
- HuggingFace Transformers:可用于实现高级的上下文摘要功能
正文完
发表至: 未分类
近三天内
