大模型应用中的上下文窗口治理：从对话压缩、记忆分层到长上下文成本与效果平衡

在很多 LLM 应用里，效果波动并不是模型本身出了问题，常见原因反而是上下文塞得太满：历史对话越来越长，检索片段越堆越多，系统提示词还不能随便删。结果很直接，token 成本上去了，响应时间变慢，回答质量也开始飘。

我在实际项目里踩过这个坑。说实话，一开始我以为把上下文窗口开大就够了，后来实测发现并不是这么回事。上下文变长之后，账单会先告诉你问题有多明显。

这篇文章写一个可复现方案：怎么把上下文治理拆成对话压缩、记忆分层、长上下文预算控制、离线评测几块，并给出一套能直接落地的工程实现。

一、为什么上下文窗口会失控

先看一个典型的在线问答助手，请求上下文通常由几部分拼出来：

  • system prompt：角色、规则、输出格式
  • recent chat：最近几轮对话
  • long-term memory：用户画像、偏好、历史结论
  • retrieved docs：RAG 检索到的片段
  • tool results：工具调用返回内容

问题就出在这里。每一块都感觉不能删，最后拼起来经常超过预算。

我一般先不谈优化，先做一次真实统计。下面是我常用的请求级观测字段：

from dataclasses import dataclass, asdict
from typing import List, Dict

@dataclass
class ContextUsage:
    session_id: str
    request_id: str
    system_tokens: int
    recent_chat_tokens: int
    memory_tokens: int
    retrieved_doc_tokens: int
    tool_result_tokens: int
    total_prompt_tokens: int
    completion_tokens: int
    latency_ms: int
    success: bool


def log_context_usage(usage: ContextUsage):
    print(asdict(usage))

埋点后很容易看出问题分布。短句先说结论：超窗只是表象，预算失控才是根因。

我见过一条线上请求，system prompt 800 token，最近 20 轮对话 4200 token，RAG 检索塞了 8 段文档又是 3600 token，最后模型还没生成内容，请求已经快到上限。这样一来，模型不是丢信息，就是被迫截断。

二、治理目标别只盯着 token 数

很多人做上下文治理，第一反应是压缩得越狠越好。这个思路容易把系统做偏。

我建议同时盯住四个指标：

指标 含义 常见问题
prompt_tokens 每次请求输入 token 成本高、超窗
latency_p95 95 分位响应时延 上下文过长导致生成变慢
task_success_rate 任务成功率 压缩后关键信息丢失
context_hit_rate 回答中是否用到了正确上下文 检索或记忆注入失效

这里有个工程细节。压缩后 token 下降，不代表质量就稳。

比如客服助手里，用户连续 15 轮讨论退款、发票和物流，如果你把前面的“退款原因”和“订单号”压没了，后面再省 token 也没意义。上线后经常会看到一种假象：成本降了，人工接管率却涨了。

所以治理目标应该是：

  1. 给不同上下文来源设预算上限
  2. 让高价值信息留在窗口里
  3. 在成本、时延、效果之间做动态平衡

三、上下文治理的整体架构

我通常把上下文拆成四层，而不是一股脑拼接：

[系统规则层]
  固定注入，严格控长

[会话短期层]
  保留最近若干轮原始对话

[会话摘要层]
  对较早历史做压缩摘要

[长期记忆层]
  存结构化用户事实、偏好、关键事件

再叠加一个外部知识层，也就是 RAG 检索结果。构建 prompt 时按预算做装配，而不是先装满再裁剪。

一个简化版流程如下：

class ContextAssembler:
    def __init__(self, tokenizer, budget_config):
        self.tokenizer = tokenizer
        self.budget_config = budget_config

    def count_tokens(self, text: str) -> int:
        return len(self.tokenizer.encode(text))

    def assemble(self, system_prompt, recent_msgs, summary, memories, docs, tool_results):
        budget = self.budget_config["max_prompt_tokens"]
        parts = []
        used = 0

        def try_add(name, content, max_tokens):
            nonlocal used
            if not content:
                return None
            text = content if isinstance(content, str) else "\n".join(content)
            tokens = self.count_tokens(text)
            tokens = min(tokens, max_tokens)
            if used + tokens > budget:
                return None
            used += tokens
            parts.append((name, text))
            return True

        try_add("system", system_prompt, self.budget_config["system_max"])
        try_add("recent_msgs", recent_msgs, self.budget_config["recent_max"])
        try_add("summary", summary, self.budget_config["summary_max"])
        try_add("memories", memories, self.budget_config["memory_max"])
        try_add("docs", docs, self.budget_config["docs_max"])
        try_add("tool_results", tool_results, self.budget_config["tool_max"])

        return parts, used

这段代码很简化，但核心点是对的：预算前置，装配优先级明确。

四、对话压缩：不是简单摘要，而是保留可执行状态

4.1 什么情况下该压缩

我一般不会每轮都做摘要，因为摘要本身也要花钱，还会引入信息漂移。更稳的做法是基于阈值触发：

  • 最近原始对话超过 2500 token
  • 会话轮数超过 12 轮
  • 新一轮请求前预计总 prompt 超过预算的 80%

阈值要实测。别拍脑袋。

4.2 摘要应该存什么

很多摘要 prompt 只会让模型“总结一下上文”，最后产物像会议纪要，看起来完整，实际没法给后续轮次直接用。

更实用的是生成状态摘要，把后续推理真正需要的信息抽出来，比如：

  • 当前任务目标
  • 已确认事实
  • 未解决问题
  • 用户明确偏好或限制条件
  • 后续待执行动作

我常用的摘要模板如下：

请将以下多轮对话压缩为可供后续助手继续工作的状态摘要。
要求：
1. 保留用户目标、约束、已确认事实、关键中间结论
2. 保留后续回答必须依赖的时间、数字、实体名
3. 删除寒暄、重复表述、无关追问
4. 输出 JSON，字段包括：
   - user_goal
   - constraints
   - confirmed_facts
   - unresolved_questions
   - pending_actions
5. 不要编造对话中未出现的信息

4.3 摘要校验不能省

摘要一旦写错，后面几轮都会被带偏。所以我会加一个轻量校验器，检查数字、订单号、日期、实体名是否在原文出现过。

import re


def extract_key_items(text: str):
    numbers = re.findall(r"\b\d+(?:\.\d+)?\b", text)
    dates = re.findall(r"\b\d{4}-\d{2}-\d{2}\b", text)
    order_ids = re.findall(r"\b[A-Z0-9]{8,20}\b", text)
    return set(numbers + dates + order_ids)


def validate_summary(source_text: str, summary_text: str):
    source_items = extract_key_items(source_text)
    summary_items = extract_key_items(summary_text)
    hallucinated = summary_items - source_items
    return {
        "is_valid": len(hallucinated) == 0,
        "hallucinated_items": list(hallucinated)
    }

这个校验不算很强，但在线上已经能挡住一批低级错误。尤其是订单场景，少一个数字就会出事故。

4.4 一个压缩前后对比

下面给一个简化的实测风格数据：

方案 历史对话 token 总输入 token p95 时延 任务成功率
不压缩，保留 18 轮原文 4680 7310 4210ms 82.4%
只做通用摘要 1260 3980 2890ms 79.8%
做状态摘要 + 最近 6 轮原文 1810 4320 3010ms 85.1%

结果很有意思。简单摘要最省 token，但成功率掉了；状态摘要加最近原文，整体更稳。

五、记忆分层：哪些该进长期记忆，哪些只留在会话里

很多应用上线一段时间后，会把“记忆”做成另一个黑洞。什么都存，什么都召回，最后又把上下文撑爆。

我的做法是把记忆分成两类：

5.1 会话级记忆

只在当前 session 有效，比如：

  • 当前问题背景
  • 当前任务进度
  • 临时约束
  • 这轮工具调用的中间结果

这类数据可以跟会话绑定，过期后清理。别留太久。

5.2 用户级长期记忆

只存跨会话依然有效的信息，而且要尽量结构化。典型字段：

{
  "user_id": "u_1024",
  "preferences": {
    "language": "zh-CN",
    "tone": "concise"
  },
  "persistent_facts": [
    {"key": "industry", "value": "manufacturing", "confidence": 0.92},
    {"key": "role", "value": "procurement_manager", "confidence": 0.87}
  ],
  "last_updated_at": "2026-04-15T10:20:00Z"
}

这里要卡得严一点。用户随口说一句“我今天赶时间”，不该直接变成长期记忆；“默认给我输出表格”和“我只看中文回复”这种，才值得保留。

5.3 记忆写入策略

我会给记忆写入加一个判定层，不满足条件就不落库：

def should_write_memory(candidate):
    if candidate["type"] not in {"preference", "persistent_fact"}:
        return False
    if candidate.get("confidence", 0) < 0.8:
        return False
    if len(candidate.get("value", "")) > 200:
        return False
    return True

短句很关键：不是所有历史都配叫记忆。

5.4 记忆召回策略

召回时不要全量注入。我一般会按“相关性 + 新鲜度 + 置信度”排序，取前 N 条。

def rank_memories(memories, query_embedding, now_ts):
    ranked = []
    for m in memories:
        relevance = m.get("relevance", 0.0)
        confidence = m.get("confidence", 0.0)
        freshness = m.get("freshness", 0.0)
        score = 0.6 * relevance + 0.3 * confidence + 0.1 * freshness
        ranked.append((score, m))
    ranked.sort(key=lambda x: x[0], reverse=True)
    return [m for _, m in ranked[:5]]

我自己踩过一个坑：长期记忆一多，旧偏好会压过当前意图。后来把会话短期层优先级提到长期记忆前面，这个问题缓了很多。

六、长上下文不是越长越好：做预算控制比盲目扩窗更有效

很多团队在模型支持 64k、128k 甚至更长窗口后，会自然地认为问题解决了。实测下来，这个想法很容易失真。

原因很简单：

  • 长上下文会带来更高的推理成本
  • 检索噪声也会跟着放大
  • 模型对远距离信息的利用率并不稳定

所以我会做一层上下文预算管理器，根据请求类型动态分配 token，而不是固定写死。

6.1 一个简单预算表

场景 system recent chat summary memory docs tool
通用问答 800 1200 600 400 1800 600
文档问答 700 800 500 300 3200 700
任务执行 Agent 900 1400 700 500 1200 1800

有预算表之后，再根据问题类型做路由。比如用户问“帮我解释这份制度差旅报销部分”，就把 docs 预算拉高；如果用户是在做多轮任务执行，就给 recent chat 和 tool result 更多空间。

6.2 动态裁剪实现

def allocate_budget(intent: str):
    plans = {
        "qa": {"max_prompt_tokens": 5400, "system_max": 800, "recent_max": 1200, "summary_max": 600, "memory_max": 400, "docs_max": 1800, "tool_max": 600},
        "doc_qa": {"max_prompt_tokens": 6200, "system_max": 700, "recent_max": 800, "summary_max": 500, "memory_max": 300, "docs_max": 3200, "tool_max": 700},
        "agent": {"max_prompt_tokens": 6500, "system_max": 900, "recent_max": 1400, "summary_max": 700, "memory_max": 500, "docs_max": 1200, "tool_max": 1800},
    }
    return plans.get(intent, plans["qa"])

6.3 裁剪顺序也要设计

我一般不会先砍 system prompt，也不会先砍最近对话。一个更稳的顺序是：

  1. 先压缩低分检索片段
  2. 再裁剪冗长工具返回
  3. 再把较早对话合并进摘要
  4. 最后才收缩长期记忆注入数量

这背后的原因很现实。系统规则通常是行为边界，最近几轮对话是当前意图，真要删也应该最后动。

七、RAG 场景里，上下文治理和召回治理要一起做

在文档问答里，很多人把“回答不准”归因到模型，其实常见问题是检索片段太碎、太多、太重复。然后这些片段又全被塞进上下文。

这时别急着加大窗口。先查片段质量。

7.1 检索结果去重与合并

def deduplicate_docs(docs):
    seen = set()
    result = []
    for d in docs:
        key = (d.get("doc_id"), d.get("chunk_text", "")[:120])
        if key in seen:
            continue
        seen.add(key)
        result.append(d)
    return result

7.2 相邻 chunk 合并

如果检索命中了同一文档的相邻段落，我更倾向先做合并，再交给模型。这样能减少边界断裂。

def merge_adjacent_chunks(chunks):
    chunks = sorted(chunks, key=lambda x: (x["doc_id"], x["chunk_idx"]))
    merged = []
    for c in chunks:
        if not merged:
            merged.append(c.copy())
            continue
        last = merged[-1]
        if c["doc_id"] == last["doc_id"] and c["chunk_idx"] == last["chunk_idx"] + 1:
            last["chunk_text"] += "\n" + c["chunk_text"]
            last["chunk_idx"] = c["chunk_idx"]
        else:
            merged.append(c.copy())
    return merged

7.3 文档片段再压缩

对特别长的法规、手册类文档，我会多做一步“查询相关摘要”，把每段压成与问题相关的证据句，而不是整段照搬。

这个步骤有收益，但有个小缺点：如果摘要模型质量一般，容易漏掉边角条件。所以我只在原始文档很长、且预算确实不够时启用，不会默认全开。

八、线上治理少不了降级策略

再好的预算设计，也会遇到极端输入。比如用户一次贴 3 万字合同，或者工具返回一个超长表格。

这时要有明确降级动作，不然服务就会抖。

8.1 常见降级方案

  • 检测用户输入过长，先做分段理解
  • 工具返回只保留关键字段和统计结果
  • RAG 结果只保留 top-k 高分证据
  • 超预算时提示用户聚焦问题范围

8.2 一个简单的超预算处理器

def over_budget_fallback(context_parts, token_counter, max_tokens):
    compact_parts = []
    used = 0
    priority = ["system", "recent_msgs", "summary", "docs", "memories", "tool_results"]

    ordered = sorted(context_parts, key=lambda x: priority.index(x[0]))
    for name, text in ordered:
        tokens = token_counter(text)
        if used + tokens <= max_tokens:
            compact_parts.append((name, text))
            used += tokens
            continue

        # 对长文本截断保底
        remaining = max_tokens - used
        if remaining > 100:
            compact_text = text[: max(remaining * 2, 200)]
            compact_parts.append((name, compact_text))
            used = max_tokens
            break

    return compact_parts

生产环境里我会比这更细，至少要按 token 而不是字符裁剪。但思路就是这样，先保主干，再保细节。

九、怎么评估上下文治理到底有没有效果

如果没有评测，最后很可能只剩“感觉快了一点”这种结论。这个不够用。

我一般做两层评估：离线回放 + 线上 A/B。

9.1 离线回放数据集

从真实日志抽样，覆盖这几类样本：

  • 短对话，几乎不需要压缩
  • 长对话，历史依赖强
  • 文档问答，检索片段长
  • Agent 任务，工具结果冗长

每条样本保留：用户输入、历史对话、检索文档、工具返回、标准答案或人工判分结果。

9.2 评测指标

指标 说明
token 降幅 新方案相对基线节省多少输入 token
latency 降幅 p50/p95 时延变化
answer_consistency 压缩前后回答语义是否一致
key_fact_retention 关键事实是否仍被保留
task_success_rate 真实任务是否完成

9.3 一个简单评测脚本

def evaluate_case(case, pipeline):
    result = pipeline.run(case)
    return {
        "case_id": case["case_id"],
        "prompt_tokens": result["prompt_tokens"],
        "latency_ms": result["latency_ms"],
        "success": result["success"],
        "key_fact_retention": result["key_fact_retention"],
    }


def compare_pipelines(cases, baseline, candidate):
    rows = []
    for case in cases:
        b = evaluate_case(case, baseline)
        c = evaluate_case(case, candidate)
        rows.append({
            "case_id": case["case_id"],
            "baseline_tokens": b["prompt_tokens"],
            "candidate_tokens": c["prompt_tokens"],
            "baseline_success": b["success"],
            "candidate_success": c["success"],
            "baseline_retention": b["key_fact_retention"],
            "candidate_retention": c["key_fact_retention"],
        })
    return rows

9.4 一组可参考的实测结果

下面这组数据来自一个多轮业务问答助手的回放实验，样本量 500：

方案 平均 prompt token p95 时延 关键事实保留率 任务成功率
基线：最近 20 轮 + top8 文档全塞 6840 4380ms 91.2% 81.6%
方案A：只做摘要 3920 2960ms 84.7% 77.9%
方案B：状态摘要 + 记忆分层 + docs 预算控制 4410 3090ms 92.5% 85.4%

我自己当时看到这组结果，没想到最省 token 的方案反而不是上线选择。原因很简单，线上系统不是做压缩比赛，最终还是看整体收益。

十、一个可直接参考的落地清单

如果你准备在现有应用里接入上下文治理，我建议按这个顺序推进：

10.1 先补观测

先拿到每次请求的：

  • 各上下文来源 token 占比
  • 超预算比例
  • 截断发生位置
  • 请求时延
  • 任务成功率或人工兜底率

没有这些数据，后面的优化很容易偏。

10.2 再做预算表

按场景区分预算，不同任务不同配额。固定一个总上限，然后给 system、recent chat、summary、memory、docs、tool 分配额度。

10.3 接着做状态摘要

从“总结全文”改成“保留可执行状态”，并加最基础的事实校验。

10.4 最后做记忆分层

把长期有效信息单独沉淀，别把整段历史都塞进 memory。召回时按相关性、置信度、新鲜度排序，只取少量高分项。

十一、结尾

上下文窗口治理，本质上不是单纯压 token，也不是一味追求更长窗口。工程上更有用的做法，是把上下文看成一种受限资源：谁该进窗口，进多少，什么时候压缩，超预算时怎么退让，都要提前设计。

真在线上跑起来后，你会发现一个很实际的规律：保留正确的信息，比保留更多的信息更值钱。

如果你现在的系统已经出现这些信号——成本上涨、长对话质量变差、RAG 文档越塞越多、Agent 工具结果越来越长——那可以先从预算埋点和状态摘要开始做，改动不算大，但通常能很快看到对比结果。

后面如果你愿意，我可以继续写一篇配套文章，把“上下文治理的离线评测集怎么构建、关键事实保留率怎么标”拆开讲，给一套更细的可复现方案。

# windows # 服务器 # 数据库 # golang
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

神经网络基础知识-分享教材

用一条线把"线性方程 → 非线性激活 → 神经元 → 正向传播 → 反向传播 → CNN → RNN → LSTM → NLP → 大模型"串起来

avatar AtomGit开源社区
cover

【多模态大模型系列·第 06 篇·终篇】未来与挑战：世界模型·具身智能·AGI——多模态的终局在哪里

avatar AtomGit开源社区
cover

越会“想“越爱乱来？聊聊大模型的“推理陷阱

avatar AtomGit开源社区