上下文窗口的切分法则
一、先把预算定下来
窗口 128K token。但实测超过 32K 之后模型注意力开始下降——后面的约束条件会被"遗忘"。所以我把 32K 当作实际天花板，128K 只当安全余量。
定预算的第一步：搞清楚每一类内容花多少 token。
# 上下文预算表 —— 这是我几轮调试后定下来的数字
CONTEXT_BUDGET = {
"system_prompt": 1500, # 角色定义 + 行为规范
"tool_schemas": 500, # 动态注入的工具 Schema（见上一篇文章）
"conversation": 8000, # 最近 N 轮对话（超过则压缩）
"rag_results": 1000, # 向量检索 top-5 格式化后
"tool_results": 1500, # 单轮工具调用返回值的上限
"reserved": 5000, # 永远留着的缓冲，不触碰
}
# 合计：17500 token，占 32K 软限制的 55%
核心原则就一条：每一类内容都有硬上限，超了就裁，没有例外。 最容易失控的是 conversation 和 tool_results，下面逐个讲怎么管。
二、Context Builder：拼接上下文的统一入口
不让各处代码自行拼接上下文，所有上下文拼装走同一个函数：
import tiktoken
enc = tiktoken.encoding_for_model("gpt-4") # DeepSeek 兼容此编码
class ContextBuilder:
def __init__(self):
self.sections = [] # [(name, content, priority), ...]
self.total_tokens = 0
def add(self, name: str, content: str, priority: int):
"""
priority: 1=绝对不能裁, 2=尽量保留, 3=溢出了优先裁
"""
tokens = len(enc.encode(content))
self.sections.append((name, content, priority, tokens))
self.total_tokens += tokens
def build(self, max_tokens: int = 32000) -> str:
"""组装最终上下文，超出 max_tokens 则从低优先级开始裁"""
if self.total_tokens <= max_tokens:
return self._assemble(self.sections)
# 超出预算了 —— 按优先级从低到高裁切
trimmed = []
remaining_budget = max_tokens
# 高优先级(1)先塞，中优先级(2)再塞，低优先级(3)最后塞
for priority in [1, 2, 3]:
for name, content, pri, tokens in self.sections:
if pri != priority:
continue
if tokens <= remaining_budget:
trimmed.append((name, content, pri, tokens))
remaining_budget -= tokens
else:
# 塞不下了就截断内容
truncated = self._truncate_text(content, remaining_budget)
trimmed.append((name, truncated, pri, remaining_budget))
remaining_budget = 0
break
if remaining_budget <= 0:
break
return self._assemble(trimmed)
def _truncate_text(self, text: str, max_tokens: int) -> str:
"""按 token 数截断文本，在句号处断开避免截断一半"""
tokens = enc.encode(text)
if len(tokens) <= max_tokens:
return text
# 取前 max_tokens 个 token，然后找到最后一个完整句子
truncated = enc.decode(tokens[:max_tokens])
last_period = truncated.rfind("。")
if last_period > len(truncated) // 2:
return truncated[:last_period + 1]
return truncated + "…"
def _assemble(self, sections) -> str:
return "\n\n---\n\n".join(content for _, content, _, _ in sections)
每一轮对话都从头 build，不存在"加着加着就超了"的情况。 预算是硬约束。
三、对话历史压缩：什么时候压、怎么压
对话历史是增长最快的区域。每轮用户+助手+tool call 消耗 500-800 token，十轮就 8000 token。必须压缩。
我的策略：保留最近 4 轮完整对话，更早的轮次用"约束摘要"替代。
class ConversationCompressor:
def __init__(self, max_full_rounds: int = 4):
self.max_full_rounds = max_full_rounds
def compress(self, messages: list[dict]) -> list[dict]:
"""
messages 是完整的对话历史，每条是 {"role": ..., "content": ...}
返回压缩后的消息列表
"""
if len(messages) <= self.max_full_rounds * 2:
return messages # 太短，不压缩
# 切分：最近 4 轮保留完整内容
recent = messages[-(self.max_full_rounds * 2):]
old = messages[:-(self.max_full_rounds * 2)]
# 从旧对话中提取约束条件
constraints = self._extract_constraints(old)
# 生成摘要
summary = self._summarize(old, constraints)
# 组装：摘要放在最前面，最近对话接在后面
compressed = [
{"role": "system", "content": f"[历史对话摘要] {summary}"}
]
compressed.extend(recent)
return compressed
def _extract_constraints(self, messages: list[dict]) -> list[str]:
"""
从旧对话中提取用户明确提出的约束条件。
这不是让 LLM 做开放性总结，而是用非常窄的 prompt 让 LLM 只输出约束列表。
"""
user_msgs = [m["content"] for m in messages if m["role"] == "user"]
combined = "\n".join(f"- {m}" for m in user_msgs)
prompt = f"""从以下用户消息中提取所有对旅行的明确约束条件。
只输出约束条件列表，每行一条。如果没有约束条件，输出"无"。
约束条件包括但不限于：预算金额、日期范围、交通工具偏好、饮食禁忌、住宿要求。
用户消息：
{combined}
约束条件列表："""
response = call_llm(prompt, max_tokens=200)
return [line.strip("- ") for line in response.split("\n") if line.strip()]
def _summarize(self, messages: list[dict], constraints: list[str]) -> str:
"""生成极简摘要，核心目标是保留约束条件"""
constraints_text = "\n".join(f" - {c}" for c in constraints)
# 只取对话中用户和助手的文本，去掉 tool call 的 JSON
dialog_text = self._extract_dialog_only(messages)
prompt = f"""将以下对话压缩为一段不超过 150 字的摘要。
重点保留：用户问了什么、Agent 给出了什么关键信息。
必须保留以下约束条件：
{constraints_text}
对话内容：
{dialog_text}
摘要："""
return call_llm(prompt, max_tokens=200)
def _extract_dialog_only(self, messages: list[dict]) -> str:
"""只提取 user/assistant 的文本内容，丢弃 tool call 的 JSON"""
lines = []
for m in messages:
if m["role"] == "user":
lines.append(f"用户: {m['content']}")
elif m["role"] == "assistant" and isinstance(m.get("content"), str):
lines.append(f"助手: {m['content'][:200]}") # 每条回复最多取 200 字
return "\n".join(lines)
压缩效果的一个实例：
压缩前 8 轮旧对话: ~4200 token
压缩后摘要 + 约束: ~250 token
压缩比: 94%
约束条件提取是这个方案最关键的一步。丢了一般对话内容可以接受，丢了"预算 5000"或者"不吃辣"就得重来。
四、工具返回值的截断
search_hotels 返回 10 个酒店，每个酒店几十个字段。如果原封不动塞进上下文，轻轻松松吃掉 2000 token。GhatGPT 实际上只看名称、价格、评分这三个字段。
截断函数：
def truncate_tool_result(tool_name: str, result: dict, max_tokens: int = 500) -> str:
"""
工具返回值截断，按工具类型采用不同的保留策略。
返回 JSON 字符串，确保在 max_tokens 以内。
"""
if tool_name == "search_hotels":
# 每家酒店只保留 4 个关键字段
essential_fields = ["name", "price_per_night", "rating", "short_description"]
stripped = []
for hotel in result.get("hotels", [])[:10]: # 最多 10 家
stripped.append({k: hotel.get(k, "") for k in essential_fields})
# 描述截断到 80 字
if stripped[-1]["short_description"]:
stripped[-1]["short_description"] = stripped[-1]["short_description"][:80]
return json.dumps({"hotels": stripped}, ensure_ascii=False)
elif tool_name == "search_flights":
# 航班只保留班次号、起降时间、价格
essential_fields = ["flight_no", "departure_time", "arrival_time", "price", "airline"]
stripped = []
for flight in result.get("flights", [])[:5]:
stripped.append({k: flight.get(k, "") for k in essential_fields})
return json.dumps({"flights": stripped}, ensure_ascii=False)
# 通用策略：截断到 max_tokens
raw = json.dumps(result, ensure_ascii=False)
tokens = enc.encode(raw)
if len(tokens) <= max_tokens:
return raw
return enc.decode(tokens[:max_tokens]) + "…[已截断]"
对应的 ContextBuilder 调用：
builder = ContextBuilder()
# system prompt: priority=1（绝不能裁）
builder.add("system", system_prompt, priority=1)
# 工具 Schema: priority=2
builder.add("tools", tools_json, priority=2)
# 对话历史: priority=2
builder.add("conversation", conversation_text, priority=2)
# RAG 结果: priority=3（先裁这个）
builder.add("rag", rag_text, priority=3)
# 最新工具结果: priority=1（当前轮要用，不能裁）
builder.add("tool_result", latest_tool_result, priority=1)
final_context = builder.build(max_tokens=32000)
优先级设计逻辑：越靠近当前轮越重要，越远的越可以裁。 这条规则贯穿整个上下文管理。
五、RAG 检索结果格式化
向量库里存的 chunk 带着 markdown 格式、多余换行、代码块标记。直接塞给 LLM 浪费 token。我写了一个格式化器：
import re
def format_rag_results(results: list[dict], top_k: int = 5, max_total_tokens: int = 1000) -> str:
"""
格式化 RAG 检索结果。
做了三件事：
1. 只保留 top-k
2. 去掉 markdown 标记
3. 每条加来源标记
"""
formatted = []
token_count = 0
for r in results[:top_k]:
# 去 markdown 格式
text = strip_markdown_noise(r["content"])
# 构建条目
entry = f"[来源: {r['source']} | 相关度: {r['score']:.3f}]\n{text}"
entry_tokens = len(enc.encode(entry))
if token_count + entry_tokens > max_total_tokens:
break
formatted.append(entry)
token_count += entry_tokens
return "\n\n---\n\n".join(formatted)
def strip_markdown_noise(text: str) -> str:
"""去掉对 LLM 没用的 markdown 噪音"""
# 去掉代码块标记
text = re.sub(r'```\w*\n?', '', text)
# 去掉图片语法
text = re.sub(r'!\[.*?\]\(.*?\)', '[图片]', text)
# 多余空白行压缩
text = re.sub(r'\n{3,}', '\n\n', text)
# 去掉纯符号分隔线
text = re.sub(r'^[-=_]{3,}\s*$', '', text, flags=re.MULTILINE)
return text.strip()
效果：检索结果从 ~1400 token 降到 ~800 token，去掉的全是 LLM 不需要的格式标记。
六、Token 监控：别等超了才发现
上下文管理最大的问题是看不见——你没法肉眼判断现在用了多少 token。我写了一个简单的监控：
class TokenMonitor:
def __init__(self, warn_threshold: int = 25000):
self.warn_threshold = warn_threshold
def check(self, builder: ContextBuilder) -> dict:
"""返回当前上下文的 token 分布"""
report = {
"total": builder.total_tokens,
"sections": {},
"warning": None,
}
for name, _, priority, tokens in builder.sections:
report["sections"][name] = {
"tokens": tokens,
"priority": priority,
"pct": round(tokens / builder.total_tokens * 100, 1),
}
if builder.total_tokens > self.warn_threshold:
report["warning"] = f"上下文已达 {builder.total_tokens} token，超过 {self.warn_threshold} 警戒线"
return report
每次调 LLM 之前在日志里打一行：
monitor = TokenMonitor(warn_threshold=25000)
ctx_report = monitor.check(builder)
logger.info(f"[TokenMonitor] total={ctx_report['total']} "
f"sys={ctx_report['sections']['system_prompt']['pct']}% "
f"conv={ctx_report['sections']['conversation']['pct']}% "
f"rag={ctx_report['sections']['rag']['pct']}%")
if ctx_report["warning"]:
logger.warning(ctx_report["warning"])
七、窗口溢出三级降级策略
当 builder.total_tokens > 32000 时，按以下顺序裁切：
def apply_degradation(builder: ContextBuilder) -> ContextBuilder:
"""
三级降级策略，每级做完重新检查，够了就停。
"""
if builder.total_tokens <= 32000:
return builder
# 降级 1: 移除最旧的 RAG 结果（priority=3 的内容先裁）
builder = _strip_old_rag(builder)
if builder.total_tokens <= 32000:
logger.info("降级1: 移除旧 RAG 结果后，窗口恢复正常")
return builder
# 降级 2: 压缩对话历史到只保留最近 2 轮完整 + 摘要
builder = _aggressive_compress(builder)
if builder.total_tokens <= 32000:
logger.info("降级2: 激进压缩对话后，窗口恢复正常")
return builder
# 降级 3: 缩短 system prompt，只保留角色定义
builder = _strip_system_prompt(builder)
logger.warning("降级3: 已缩短 system prompt，Agent 行为可能不稳定")
return builder
def _strip_old_rag(builder: ContextBuilder) -> ContextBuilder:
"""移除最早添加的 RAG 结果"""
# 找到所有 rag section，只保留最新添加的两条
rag_sections = [(i, s) for i, s in enumerate(builder.sections) if s[0] == "rag"]
if len(rag_sections) <= 2:
return builder
# 保留最后两条
keep_indices = {rag_sections[-1][0], rag_sections[-2][0]}
builder.sections = [s for i, s in enumerate(builder.sections)
if s[0] != "rag" or i in keep_indices]
builder.total_tokens = sum(s[3] for s in builder.sections)
return builder
def _aggressive_compress(builder: ContextBuilder) -> ContextBuilder:
"""将对话历史压缩到最近 2 轮"""
compressor = ConversationCompressor(max_full_rounds=2)
# 从 builder.sections 中找到 conversation section 并压缩
# ... (具体实现省略，思路同上文的 ConversationCompressor)
return builder
def _strip_system_prompt(builder: ContextBuilder) -> ContextBuilder:
"""system prompt 只保留最核心的角色定义，砍掉示例和格式说明"""
for i, (name, content, pri, tokens) in enumerate(builder.sections):
if name == "system_prompt":
# 只保留第一段（角色定义），砍掉后面的格式说明和示例
core = content.split("\n\n")[0]
builder.sections[i] = (name, core, pri, len(enc.encode(core)))
break
builder.total_tokens = sum(s[3] for s in builder.sections)
return builder
触发频率很低——跑了几个月，降级 1 触发过几次（长对话 + 大量 RAG），降级 2 触发过两次（多模态图片 + 长对话），降级 3 从来没触发过。这个分布说明预算分配基本合理。
八、完整的 build_context 流程
把上面所有组件串起来，就是每次调 LLM 之前调用的函数：
def build_context(
user_input: str,
conversation_history: list[dict],
rag_results: list[dict],
tool_results: list[dict],
matched_tools: list[str],
target_agent: str,
) -> str:
builder = ContextBuilder()
# 1. 系统提示（动态裁切版，不是每次都一样）
system_prompt = build_agent_prompt(target_agent, matched_tools)
builder.add("system_prompt", system_prompt, priority=1)
# 2. 对话历史（压缩后的）
compressor = ConversationCompressor(max_full_rounds=4)
compressed_conv = compressor.compress(conversation_history)
conv_text = "\n".join(m["content"] for m in compressed_conv if isinstance(m.get("content"), str))
builder.add("conversation", conv_text, priority=2)
# 3. RAG 结果（格式化 + top-5）
rag_text = format_rag_results(rag_results, top_k=5, max_total_tokens=1000)
builder.add("rag", rag_text, priority=3)
# 4. 工具返回值（截断后的）
for tr in tool_results[-2:]: # 只保留最近 2 个工具结果
truncated = truncate_tool_result(tr["tool"], tr["result"])
builder.add("tool_result", truncated, priority=2)
# 5. 触发降级检查
if builder.total_tokens > 32000:
builder = apply_degradation(builder)
return builder.build(max_tokens=32000)
九、投入产出
几个月的运行数据：
| 指标 | 数字 |
|---|---|
| 平均每轮上下文 token | 8K-12K |
| 复杂多轮对话峰值 | 25K-28K |
| 降级 1 触发率 | ~3% |
| 降级 2 触发率 | <1% |
| 降级 3 触发率 | 0% |
| 模型"遗忘约束"的投诉 | 改造前 ~1/15 次对话，改造后 ~1/60 次 |
核心经验就两条：
- 设硬上限——每一类内容都有 token 配额，超了就裁，不要"弹性"
- 按距离定优先级——离当前轮越近的内容越重要，越远越可以压缩