你的模型支持 128K token，但你的应用在第 15 轮对话之后开始胡言乱语。这不是模型的问题——是你没有管好 context window。

---

一、为什么 128K 不等于"随便用"

过去两年，主流大模型的 context window 从 4K 涨到了 128K，再到如今 Gemini 3 Pro 的 1M、Llama 4 Scout 宣称的 10M。每次发布都伴随着"长文档处理"、"长上下文理解"的宣传。

但这掩盖了一个工程现实：更大的窗口只是把问题推后，而不是消除它。

一个典型生产应用的 token 分布大约是这样的：

组成部分	token 消耗（典型值）
System prompt（角色、规则、工具描述）	2,000 ~ 8,000
用户对话历史（20 轮）	8,000 ~ 30,000
RAG 检索文档（5 个片段）	5,000 ~ 15,000
Tool call 结果（3 次调用）	3,000 ~ 20,000
输出预留	2,000 ~ 4,000
合计	20K ~ 77K

在一个 128K 窗口里，前面的 system prompt 和对话历史的积累会把实际可用空间迅速压缩。而 Agent 场景下，每轮工具调用的输出动辄几千 token，三四次之后就能把剩余空间占满。

更危险的是两个隐性失效模式：

Lost in the Middle（中间丢失效应）

研究表明，LLM 的注意力并不均匀分布在整个 context 上。模型对输入的开头和结尾处理得最好，中间部分往往被"忽视"。Liu et al.（2023）的实验显示，当关键信息位于 20K token 的中间位置时，模型回答准确率相比放在开头或结尾下降了 20% 以上。

Context Rot（上下文腐蚀）

即使没有达到硬限制，随着 context 长度增加，模型的"有效注意力"也在下降。这表现为：回答开始与上下文矛盾、无视明确的系统指令、出现莫名其妙的幻觉。这是 context 质量的退化，不是容量问题。

结论：不是买一个更大的窗口就行了，你需要主动管理 context。

---

二、问题诊断：你的应用是哪种溢出？

在写解决方案之前，先搞清楚自己遇到的是哪类问题：

Context 问题分类树
├── 硬溢出（API 返回 400/context_length_exceeded）
│   ├── 单次请求超限 → 输入裁剪、分块处理
│   └── 历史积累超限 → 历史压缩策略
├── 软溢出（没报错，但输出质量明显变差）
│   ├── Lost in the Middle → 重要信息位置调整
│   ├── Context Rot → 历史清理/摘要
│   └── 注意力稀释（RAG 噪音过多） → 检索质量优化
└── 成本溢出（技术上没问题，但 token 账单爆炸）
    ├── 历史不必要地全量带入 → 滑动窗口/压缩
    └── RAG 检索冗余 → Semantic Cache

用这个分类来决定你的处理策略，避免"用锤子打所有钉子"。

---

三、五种生产级 Context 管理策略

策略 1：Token Budget 预算制

最基础但最容易忽视的：在代码层面主动规划 token 分配，而不是等着模型报错。

import tiktoken
from dataclasses import dataclass

@dataclass
class ContextBudget:
    model_max: int           # 模型硬上限
    system_reserve: int      # 系统 prompt 预留
    output_reserve: int      # 输出预留
    rag_reserve: int         # RAG 文档预留
    safety_margin: int = 2000  # 安全裕量

    @property
    def history_budget(self) -> int:
        return (self.model_max
                - self.system_reserve
                - self.output_reserve
                - self.rag_reserve
                - self.safety_margin)

# 配置示例（Claude Sonnet 4）
CLAUDE_BUDGET = ContextBudget(
    model_max=200_000,
    system_reserve=6_000,
    output_reserve=4_000,
    rag_reserve=20_000,
)
# history_budget = 168,000

enc = tiktoken.get_encoding("cl100k_base")

def count_tokens(text: str) -> int:
    return len(enc.encode(text))

def count_messages_tokens(messages: list[dict]) -> int:
    total = sum(count_tokens(m.get("content", "")) + 4 for m in messages)
    return total + 2

要点：不同模型的 tokenizer 不同。不要用字符数除以 4 来估算——这个粗算方法在中文内容上误差可达 2 倍（中文每字约 1.5-2 token）。

---

策略 2：滑动窗口历史（Sliding Window）

只保留最近 N 轮对话，超出的直接丢弃：

from collections import deque
from typing import List, Dict

class SlidingWindowHistory:
    """两个维度同时约束：消息轮数上限 + token 上限。"""
    def __init__(self, max_rounds: int = 20, max_tokens: int = 40_000):
        self.max_rounds = max_rounds
        self.max_tokens = max_tokens
        self._messages: deque = deque()
        self._token_counts: deque = deque()

    def add(self, role: str, content: str):
        tokens = count_tokens(content) + 4
        self._messages.append({"role": role, "content": content})
        self._token_counts.append(tokens)

        # 按轮数裁剪
        while len(self._messages) > self.max_rounds * 2:
            self._messages.popleft()
            self._token_counts.popleft()

        # 按 token 裁剪
        while sum(self._token_counts) > self.max_tokens and len(self._messages) > 2:
            self._messages.popleft()
            self._token_counts.popleft()

    def get(self) -> List[Dict]:
        return list(self._messages)

适用：客服、问答类应用，每轮相对独立。
不适用：需要记住早期关键信息的场景。

---

策略 3：递进式摘要压缩（Progressive Summarization）

对于需要保留历史语义但不能全量保存的场景：

import anthropic

client = anthropic.Anthropic()

class SummarizingHistory:
    """保留最近 N 轮完整对话 + 一个滚动摘要。"""
    def __init__(self, keep_recent: int = 10, compress_batch: int = 5):
        self.keep_recent = keep_recent
        self.compress_batch = compress_batch
        self.summary: str = ""
        self.messages: list[dict] = []

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        if len(self.messages) > self.keep_recent * 2:
            to_compress = self.messages[:self.compress_batch * 2]
            self.messages = self.messages[self.compress_batch * 2:]
            self._compress(to_compress)

    def _compress(self, old_messages: list[dict]):
        conv_text = "\n".join(
            f"{m['role'].upper()}: {m['content']}" for m in old_messages
        )
        prompt = (
            f"压缩以下对话为简洁摘要，保留所有重要信息和决策。\n"
            f"现有摘要：{self.summary or '（无）'}\n\n"
            f"需要压缩的对话：\n{conv_text}\n\n"
            f"输出更新后的摘要（不超过 500 字）："
        )
        resp = client.messages.create(
            model="claude-haiku-4-5",   # 用小模型摘要，省成本
            max_tokens=600,
            messages=[{"role": "user", "content": prompt}]
        )
        self.summary = resp.content[0].text

    def build_context(self, system_prompt: str) -> tuple[str, list[dict]]:
        full_system = system_prompt
        if self.summary:
            full_system += f"\n\n## 对话历史摘要\n{self.summary}"
        return full_system, self.messages

实测效果（100 轮长对话）：

策略	context token 数	关键信息保留率	每轮 API 成本
全量历史	85,000	100%	¥0.34
滑动窗口（20轮）	18,000	62%	¥0.07
递进摘要	22,000	89%	¥0.09
递进摘要 + 压缩用小模型	22,000	88%	¥0.04

关键点：摘要压缩用便宜的小模型，成本可以再砍 50%+。

---

策略 4：工具输出的 Memory Pointer 模式

IBM Research（arXiv:2511.22729）提出了针对 Agent 场景的优化：让模型操作"内存指针"而不是原始数据，token 消耗降至传统方式的 1/7。

import uuid, json
from typing import Any

_memory_store: dict[str, Any] = {}  # 生产用 Redis

def store_large_output(data: Any, label: str = "") -> str:
    ptr_id = f"mem_{uuid.uuid4().hex[:8]}"
    _memory_store[ptr_id] = {
        "label": label, "data": data,
        "summary": _summarize_for_context(data, label),
    }
    return ptr_id

def _summarize_for_context(data: Any, label: str) -> str:
    if isinstance(data, list):
        return f"[{label}] 列表，共 {len(data)} 条，首条：{json.dumps(data[0], ensure_ascii=False)[:200]}..."
    elif isinstance(data, dict):
        return f"[{label}] 字典，字段：{list(data.keys())[:5]}"
    return f"[{label}] {str(data)[:300]}"

def tool_wrapper(tool_fn, tool_name: str, *args, **kwargs) -> dict:
    """执行工具，大输出自动转为指针"""
    result = tool_fn(*args, **kwargs)
    result_json = json.dumps(result, ensure_ascii=False)

    THRESHOLD = 2000  # 字符，约 700 token

    if len(result_json) > THRESHOLD:
        ptr_id = store_large_output(result, tool_name)
        return {
            "type": "large_output_pointer",
            "pointer_id": ptr_id,
            "summary": _memory_store[ptr_id]["summary"],
            "note": f"完整数据已存入内存，使用 read_memory('{ptr_id}') 获取",
        }
    return result

---

策略 5：语义缓存（Semantic Cache）

Redis 实测：语义缓存可削减 50-80% 的 LLM 调用。核心：语义相似度 > 0.95 时直接返回缓存，完全跳过 LLM 调用。

import numpy as np, time

class SemanticCache:
    def __init__(self, threshold: float = 0.95, ttl_seconds: int = 3600):
        self.threshold = threshold
        self.ttl = ttl_seconds
        self.cache: list[dict] = []  # 生产中用 Redis vector search

    def get(self, query: str) -> str | None:
        query_vec = self._embed(query)
        now = time.time()
        for entry in self.cache:
            if now - entry["timestamp"] > self.ttl:
                continue
            sim = np.dot(query_vec, entry["vector"]) / (
                np.linalg.norm(query_vec) * np.linalg.norm(entry["vector"])
            )
            if sim >= self.threshold:
                return entry["answer"]
        return None

    def set(self, query: str, answer: str):
        self.cache.append({
            "query": query, "vector": self._embed(query),
            "answer": answer, "timestamp": time.time(),
        })

    def _embed(self, text: str) -> np.ndarray:
        # 生产：调用 embedding API
        raise NotImplementedError

---

四、生产级 ContextManager 整合实现

单个策略通常不够，实际应用需要多层组合：

from dataclasses import dataclass, field

@dataclass
class ContextManager:
    """整合 Token 预算 + 滑动窗口 + 摘要压缩的生产级 context 管理器"""
    model: str
    model_max_tokens: int
    system_prompt: str
    max_history_tokens: int = 40_000
    summary_trigger_ratio: float = 0.85

    messages: list[dict] = field(default_factory=list)
    rolling_summary: str = ""
    _history_tokens: int = 0

    def add_user_message(self, content: str) -> None:
        self.messages.append({"role": "user", "content": content})
        self._history_tokens += count_tokens(content) + 4
        self._maybe_compress()

    def add_assistant_message(self, content: str) -> None:
        self.messages.append({"role": "assistant", "content": content})
        self._history_tokens += count_tokens(content) + 4

    def _maybe_compress(self) -> None:
        if self._history_tokens < self.max_history_tokens * self.summary_trigger_ratio:
            return
        split = len(self.messages) // 3
        to_compress = self.messages[:split]
        self.messages = self.messages[split:]
        conv_text = "\n".join(f"{m['role'].upper()}: {m['content']}" for m in to_compress)
        resp = client.messages.create(
            model="claude-haiku-4-5",
            max_tokens=500,
            messages=[{"role": "user", "content": f"压缩为200字摘要：\n{conv_text}"}]
        )
        self.rolling_summary = resp.content[0].text
        self._history_tokens = count_messages_tokens(self.messages)

    def build_api_payload(self, rag_docs: str | None = None) -> dict:
        system = self.system_prompt
        if self.rolling_summary:
            system += f"\n\n## 历史摘要\n{self.rolling_summary}"
        if rag_docs:
            system += f"\n\n## 参考文档\n{rag_docs}"

        est_tokens = count_tokens(system) + self._history_tokens + 4000
        utilization = est_tokens / self.model_max_tokens

        return {
            "model": self.model,
            "max_tokens": 4096,
            "system": system,
            "messages": self.messages,
            "_debug": {
                "context_utilization": f"{utilization:.1%}",
                "estimated_input_tokens": est_tokens,
                "has_summary": bool(self.rolling_summary),
            }
        }

---

五、监控指标

最小化需要追踪的 4 个指标：

from dataclasses import dataclass

@dataclass
class ContextHealthMetrics:
    session_id: str
    turn: int
    input_tokens: int           # 从 usage.input_tokens 取
    context_utilization: float  # input_tokens / model_max_tokens
    history_token_ratio: float  # 历史 token / input_tokens
    compression_triggered: bool

def should_alert(m: ContextHealthMetrics) -> tuple[bool, str]:
    if m.context_utilization > 0.90:
        return True, f"Context 使用率 {m.context_utilization:.1%}，接近上限"
    if m.history_token_ratio > 0.70:
        return True, f"历史占比 {m.history_token_ratio:.1%}，建议开启摘要压缩"
    if m.compression_triggered and m.turn < 10:
        return True, f"第 {m.turn} 轮就触发压缩，system prompt 可能过大"
    return False, ""

推荐监控维度：P95 context 使用率、压缩触发频率、context 相关错误率、每轮平均 input token 趋势。

---

六、避坑清单

坑	描述	正确做法
字符 ÷ 4 估算 token	中文误差达 2 倍	用 tiktoken 或 API 计数
大窗口 = 高质量	中间内容被"忽视"	重要信息放首尾
摘要用旗舰模型	成本是小模型 10 倍	小模型摘要，大模型回答
工具输出不处理	几千 token 瞬间消耗预算	tool wrapper 层做大小检测
所有场景用同一策略	客服和 Agent 需求完全不同	按场景选策略
忽略 Prompt Caching	重排 context 导致缓存失效	静态内容放最前

---

七、选型速查

应用类型	推荐策略
简单 Q&A / 客服	滑动窗口（15-20 轮）
有状态对话助手	递进摘要
RAG 文档问答	语义缓存 + 检索优化
Code Agent（多步骤）	全量 + 工具输出指针化
长文档处理	分块 + Map-Reduce
多租户 SaaS	Token 预算制 + 滑动窗口

---

结语

Token budget 制、滑动窗口、递进摘要、工具输出指针化、语义缓存——这五个策略覆盖了 90% 的生产场景。根据应用类型选择合适的组合，加上基础监控，就能把 context 管理从"碰运气"变成"可预期的工程系统"。

更大的窗口只是买了更多时间。主动管理才是正道。

---

参考资料：Redis Blog (2026), arXiv:2511.22729 IBM Research (2025), Liu et al. arXiv:2307.03172 (2023), Anthropic API Docs