多轮对话的上下文管理——不是把历史都塞进去那么简单
2026/4/30大约 8 分钟
多轮对话的上下文管理——不是把历史都塞进去那么简单
适读人群：在做多轮对话应用的开发者 | 阅读时长：约 13 分钟 | 核心价值：生产环境里的上下文管理真实方案，解决 Token 超限和信息取舍问题
有个用户给我们发来一个 Bug 报告，截图里是一段聊天记录：他和我们的 AI 助手谈了一个多小时的项目方案，聊到后面，AI 开始「失忆」——明明前面讨论过的需求，后面又问一遍。
我去看了代码，问题一眼就找到了：上下文管理逻辑写的是「当历史消息超过 8000 个 token 时，删掉最早的几条」。
看起来没问题，但有个缺陷：它删掉的是「最早的几条」，而那几条往往是最重要的——用户在对话开始时说的背景信息、项目需求、约束条件。后面那些「好的」「明白了」「继续」之类的确认消息反而被保留下来，但它们没有任何实质内容。
这就是上下文管理里「看起来合理、实际上错的」最典型案例。
上下文管理要解决的核心矛盾
多轮对话的上下文管理本质上是一个在有限空间里做信息取舍的问题。
约束条件：
- LLM 的 context window 有限（哪怕是 128k token 的模型，大量历史记录也会撑满）
- 用的 token 越多，成本越高，延迟也越高
- 丢失重要历史信息，AI 会「失忆」，体验很差
这个矛盾没有完美解法，只有根据场景做的工程取舍。
信息的重要性不是按时间顺序排列的
这是最关键的认知：对话里的信息重要性和时间顺序不相关。
通常，重要性分布是这样的：
- 高重要性：用户在对话初期说的目标、背景、约束（「我要做一个面向老年人的 App」「预算有限制」「必须兼容 iOS 12」）
- 高重要性：关键决策节点（「我们决定用方案 B」「放弃这个功能」）
- 中等重要性：最近几轮的对话（当前话题的上下文）
- 低重要性：中间的过渡性发言（「好的」「继续」「嗯嗯」「明白了」）
按时间顺序删除历史，正好把高重要性的内容删掉，留了低重要性的内容。
我现在用的分层管理方案
把对话历史分成三层：
┌─────────────────────────────┐
│ 系统摘要层（永久保留） │ ← 用户基础信息、对话目标摘要
│ 约 200-500 tokens │
├─────────────────────────────┤
│ 重要片段层（按重要性保留） │ ← 关键决策、重要信息点
│ 约 500-1000 tokens │
├─────────────────────────────┤
│ 近期对话层（滑动窗口） │ ← 最近 N 轮完整对话
│ 约 2000-4000 tokens │
└─────────────────────────────┘层一：系统摘要层
在对话开始时（或每隔 N 轮）由 LLM 生成一个摘要，固定注入到 System Prompt 里：
SUMMARY_PROMPT = """
基于以下对话历史，生成一个简洁的摘要，包含：
1. 用户的主要目标和背景
2. 已确认的关键决策
3. 重要的约束条件
摘要不超过200字，使用第三人称。
对话历史：
{conversation_history}
"""
async def generate_conversation_summary(history: list, llm_client) -> str:
history_text = "\n".join([
f"{msg['role']}: {msg['content']}"
for msg in history
])
response = await llm_client.chat.completions.create(
model="gpt-4o-mini", # 用便宜的小模型做摘要
messages=[
{"role": "user", "content": SUMMARY_PROMPT.format(
conversation_history=history_text
)}
],
max_tokens=300,
temperature=0
)
return response.choices[0].message.content层二：重要片段层
对每条消息打分，保留高分消息：
from dataclasses import dataclass
from typing import List
import re
@dataclass
class MessageWithScore:
role: str
content: str
importance_score: float
turn_index: int
def score_message_importance(message: dict, turn_index: int, total_turns: int) -> float:
"""
给消息打重要性分数 (0-1)
打分维度：
1. 消息长度（长消息通常更有信息量）
2. 是否包含关键词（决策词、约束词）
3. 位置权重（早期消息有背景信息价值）
4. 是否是用户消息（通常比AI回复更重要）
"""
content = message['content']
role = message['role']
score = 0.0
# 长度分数（对数缩放，避免太长的消息权重过大）
import math
length_score = min(math.log(len(content) + 1) / 10, 0.3)
score += length_score
# 关键词分数
decision_keywords = [
'决定', '确认', '选择', '采用', '放弃', '取消', '方案',
'decided', 'confirmed', 'choose', 'adopt', 'cancel',
'需求', '要求', '必须', '不能', '限制', '约束',
'requirement', 'must', 'cannot', 'constraint'
]
keyword_count = sum(1 for kw in decision_keywords if kw.lower() in content.lower())
keyword_score = min(keyword_count * 0.05, 0.3)
score += keyword_score
# 位置权重：开头的消息（前20%）给额外加分
if turn_index / max(total_turns, 1) < 0.2:
score += 0.2
# 角色权重：用户消息略高于AI消息
if role == 'user':
score += 0.1
# 过于短的消息（确认性回复）降分
if len(content.strip()) < 20:
score *= 0.3
return min(score, 1.0)
def select_important_messages(
history: List[dict],
token_budget: int,
min_recent_turns: int = 6
) -> List[dict]:
"""
从历史中选取重要消息，控制在token预算内
保证最近min_recent_turns轮不被删除
"""
if not history:
return []
total_turns = len(history)
# 打分
scored_messages = []
for i, msg in enumerate(history):
score = score_message_importance(msg, i, total_turns)
scored_messages.append(MessageWithScore(
role=msg['role'],
content=msg['content'],
importance_score=score,
turn_index=i
))
# 最近N轮必须保留
recent_start = max(0, total_turns - min_recent_turns * 2) # *2因为每轮有user和assistant
must_keep = scored_messages[recent_start:]
candidates = scored_messages[:recent_start]
# 按重要性排序候选消息
candidates.sort(key=lambda x: x.importance_score, reverse=True)
# 估算token数（粗略：1 token ≈ 4字符）
def estimate_tokens(text: str) -> int:
return len(text) // 3 # 中文用3更准确
selected = list(must_keep)
used_tokens = sum(estimate_tokens(m.content) for m in selected)
# 从高分到低分添加候选消息，直到budget用完
for candidate in candidates:
candidate_tokens = estimate_tokens(candidate.content)
if used_tokens + candidate_tokens <= token_budget:
selected.append(candidate)
used_tokens += candidate_tokens
# 按原始顺序排列（打乱顺序会让对话失去连贯性）
selected.sort(key=lambda x: x.turn_index)
return [{'role': m.role, 'content': m.content} for m in selected]完整的上下文管理器
import json
from datetime import datetime
from typing import Optional
class ConversationContextManager:
def __init__(
self,
llm_client,
redis_client,
max_context_tokens: int = 8000,
summary_interval: int = 20, # 每20轮更新一次摘要
):
self.llm = llm_client
self.redis = redis_client
self.max_context_tokens = max_context_tokens
self.summary_interval = summary_interval
async def load_context(self, session_id: str) -> dict:
"""加载完整会话上下文"""
data = await self.redis.get(f"ctx:{session_id}")
if data:
return json.loads(data)
return {
'session_id': session_id,
'summary': None,
'full_history': [], # 完整历史（用于生成摘要）
'turn_count': 0,
'user_profile': {}, # 用户基础信息
'key_facts': [], # 提取的关键事实
'created_at': datetime.now().isoformat()
}
async def save_context(self, session_id: str, context: dict):
"""保存上下文，TTL 24小时"""
await self.redis.setex(
f"ctx:{session_id}",
86400, # 24小时
json.dumps(context, ensure_ascii=False)
)
async def add_message(self, session_id: str, role: str, content: str):
"""添加新消息到上下文"""
context = await self.load_context(session_id)
context['full_history'].append({
'role': role,
'content': content,
'timestamp': datetime.now().isoformat()
})
context['turn_count'] += 1
# 每隔summary_interval轮更新摘要
if context['turn_count'] % self.summary_interval == 0:
context['summary'] = await generate_conversation_summary(
context['full_history'][-40:], # 用最近40条生成摘要
self.llm
)
await self.save_context(session_id, context)
async def build_messages_for_llm(
self,
session_id: str,
current_message: str
) -> list:
"""
构建发给LLM的消息列表
在token预算内选取最有价值的历史
"""
context = await self.load_context(session_id)
# 计算各部分的token预算
summary_tokens = 500
recent_history_tokens = 3000
key_facts_tokens = 300
current_message_tokens = len(current_message) // 3 + 100
remaining_budget = (self.max_context_tokens
- summary_tokens
- key_facts_tokens
- current_message_tokens)
# 构建系统消息（包含摘要和关键事实）
system_parts = []
if context['summary']:
system_parts.append(f"## 对话背景摘要\n{context['summary']}")
if context['key_facts']:
facts_text = '\n'.join(f"- {f}" for f in context['key_facts'][-10:])
system_parts.append(f"## 已确认的关键信息\n{facts_text}")
# 从完整历史里智能选取消息
selected_history = select_important_messages(
context['full_history'][:-0] if context['full_history'] else [],
token_budget=remaining_budget
)
# 构建最终消息列表
messages = selected_history + [
{'role': 'user', 'content': current_message}
]
return messages, '\n\n'.join(system_parts)
async def extract_and_store_key_facts(self, session_id: str, assistant_response: str):
"""从最新回复中提取关键事实并持久化"""
context = await self.load_context(session_id)
# 用小模型快速判断是否有关键事实需要记录
# 这一步可以做得很轻量，不需要每条消息都提取
recent_messages = context['full_history'][-4:]
if len(recent_messages) < 2:
return
last_user_msg = next(
(m for m in reversed(recent_messages) if m['role'] == 'user'),
None
)
if last_user_msg and any(
kw in last_user_msg['content']
for kw in ['决定', '确认', '选择', '要求', '必须']
):
# 有关键决策词，提取事实
extract_prompt = f"""
从以下用户消息中提取关键决策或约束（如果有的话），用一句话表述，没有则返回空字符串：
用户消息：{last_user_msg['content']}
"""
response = await self.llm.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": extract_prompt}],
max_tokens=100,
temperature=0
)
fact = response.choices[0].message.content.strip()
if fact and len(fact) > 5:
context['key_facts'].append(fact)
# 最多保留50个关键事实
context['key_facts'] = context['key_facts'][-50:]
await self.save_context(session_id, context)用户信息持久化：跨会话记忆
除了单次会话的上下文管理，还有一个需求是「跨会话记忆」——用户昨天说了他是做电商的，今天新开一个会话，AI 应该还记得。
这个用独立的用户档案来实现：
class UserProfileManager:
def __init__(self, db_client):
self.db = db_client
async def get_user_profile(self, user_id: str) -> dict:
profile = await self.db.query_one(
"SELECT * FROM user_profiles WHERE user_id = %s",
(user_id,)
)
if profile:
return {
'background': profile.get('background', ''),
'preferences': json.loads(profile.get('preferences', '{}')),
'expertise_level': profile.get('expertise_level', 'unknown'),
'previous_topics': json.loads(profile.get('previous_topics', '[]'))
}
return {}
def build_profile_context(self, profile: dict) -> str:
"""把用户档案转成注入System Prompt的文本"""
if not profile:
return ''
parts = []
if profile.get('background'):
parts.append(f"用户背景：{profile['background']}")
if profile.get('expertise_level') != 'unknown':
parts.append(f"技术水平：{profile['expertise_level']}")
if profile.get('preferences'):
prefs = profile['preferences']
if prefs.get('language_style'):
parts.append(f"偏好沟通风格：{prefs['language_style']}")
if parts:
return "## 用户信息\n" + "\n".join(parts)
return ''几个不做但很多人在做的错误
错误一：无脑截断历史
「超过 8k token 就删最早的几条」——前文说了，这会删掉最重要的背景信息。
错误二：把所有历史都压缩成摘要
有人把全部历史都摘要化，只保留摘要。问题是摘要会丢失细节，用户前面说的具体需求（「按钮颜色要用 #FF6B35」这种细节）在摘要里很可能消失。
正确做法是摘要和关键片段并存，不是单纯替换。
错误三：对话框架和存储不分离
把完整历史直接存在 Session 里，Session 在 Redis 里，TTL 只有几小时。用户第二天回来继续对话，上下文全没了。
长期对话历史要存数据库，Redis 只做热数据缓存。
错误四：不考虑 token 计费
用了摘要、关键事实、完整历史三层，如果不控制 token 总量，反而比直接用全历史更贵。每次构建 LLM 请求时要算清楚实际 token 数。
这套方案适合什么场景
适合：
- 用户会话平均超过 10 轮的应用
- 有明确任务目标的对话（项目规划、需求分析、代码协作）
- 需要跨会话记忆的助手类产品
不适合：
- 简单的单次问答（每次都是独立的，没有历史关联需求）
- 纯工具调用场景（不是对话，是任务执行）