上下文工程最常见的线上事故，不是"模型不会答"，而是：答到一半被截断、工具返回太长把关键信息挤掉、对话越久越贵且越不稳定。根因往往很朴素：你没有把"上下文窗口"当成一种要运营的预算。

摘要：把总 token 预算拆成四个"资金池"：指令与边界、历史与记忆、检索块（RAG）、工具返回，并预留输出空间。触发阈值到达时，用"快而确定"的截断/分页优先，再用"慢但更保真"的摘要补位；工具返回只回注本轮决策所需字段，其余落库可追溯但不进窗。
关键词：Token budget；context compression；summarization；sliding window；pagination；tool output truncation

0 系列回顾

1 先回答一个关键问题：你到底在为谁花 token？

想象一下你在写一份"办事材料包"：

  • system 像"办事规则与红线"
  • 历史对话像"往来邮件"
  • 检索块像"附件证据"
  • 工具输出像"窗口打印的小票"

窗口大小固定，你必须决定"带哪些纸进大厅"。不做预算管理，最后一定会变成"把所有纸都塞进包里"，既重又乱。

2 一个可落地的预算切分法（四池 + 输出预留）

设总窗口为 (W) tokens。推荐先预留输出 (O)（否则模型容易溢出），再把输入预算 (I = W - O) 切四池：

  • 指令池（10–20%）：system 规则、输出格式、禁止项
  • 历史池（20–40%）：最近对话 + 关键事实摘要
  • 检索池（20–40%）：RAG chunks（带元数据）
  • 工具池（10–30%）：工具结果回注（只保留必要字段）

💡 理解要点：比例不是教条，关键是有硬上限，并且每池都有"溢出策略"。

🔍 实际例子：(W=32k)，预留 (O=4k)。输入 (I=28k) 可粗分：指令 4k、历史 8k、检索 10k、工具 6k。

3 三种压缩手段：截断、滑窗、摘要（以及何时用）

3.1 截断（Truncation）：最快、最可控

适用场景：

  • 历史很长但大多不重要（如闲聊、问候、已解决的技术支持）
  • 工具输出爆长（如返回了 5000 行日志或 1000 条订单记录）
  • 需要确定性和零额外成本的紧急压缩

具体做法与实现:

def truncate_history(messages, keep_last_n=10, keep_first_n=1):
    """
    截断对话历史，保留最近N轮和首条消息
    """
    if len(messages) <= keep_last_n + keep_first_n:
        return messages
    
    # 保留 system 和首轮（关键约束）
    first_part = messages[:keep_first_n]
    # 保留最近N轮（当前上下文）
    last_part = messages[-keep_last_n:]
    
    return first_part + last_part

def truncate_tool_result(result_json, max_chars=2000, essential_fields=None):
    """
    工具结果截断：只保留关键字段，其余丢弃
    """
    if essential_fields is None:
        essential_fields = ["status", "order_id", "error_code"]
    
    truncated = {}
    for field in essential_fields:
        if field in result_json:
            truncated[field] = result_json[field]
    
    # 添加截断标记
    truncated["_truncated"] = True
    truncated["_original_size"] = len(str(result_json))
    
    return truncated

截断策略的选择矩阵:

内容类型截断策略保留策略风险
System prompt绝不截断完整保留截断会丢失安全边界
当前任务绝不截断完整保留截断会导致模型不知要做什么
对话历史保留最近N轮首条 + 末N条丢失远期上下文
工具结果字段投影 + 长度截断关键字段优先丢失详细数据
RAG chunks按相关度截断Top-K最高相关丢失低相关证据

优点：

  • 无额外模型调用，零延迟
  • 行为完全可预测，便于调试
  • 实现简单，可在线性时间内完成

缺点与补救：

  • 可能丢失远处关键事实 → 使用"锚点块"补救（见第12篇）
  • 可能切断因果链 → 在截断前主动提取关键决策点保存

🔍 实际例子：客服对话已进行50轮，前30轮是寒暄和身份验证，中间15轮是问题诊断，最后5轮是当前解决方案讨论。截断策略：保留首条（身份确认）+ 中间诊断的关键节点（已提取为锚点）+ 最后5轮。

3.2 滑动窗口（Sliding window）：保结构，丢细节

核心思想：像视频播放器的"最近缓存"——新内容全保留，旧内容只留概要，而不是完全删除。

适用场景：

  • 长对话需要"覆盖面"，但细节不必全保留
  • 文档分析需要"知道大概发生过什么"，但不需要每句话
  • 渐进式信息衰减，而非断崖式截断

具体实现（双层滑动窗口）:

class SlidingWindowCompressor:
    def __init__(self, 
                 recent_window_size=5,      # 最近5轮全保留
                 summary_window_size=10,    # 再往前10轮做摘要
                 token_budget_recent=2000,   # 近期窗口预算
                 token_budget_summary=1500):  # 摘要窗口预算
        self.recent_size = recent_window_size
        self.summary_size = summary_window_size
        self.budget_recent = token_budget_recent
        self.budget_summary = token_budget_summary
    
    def compress(self, messages, current_turn_idx):
        """
        滑动窗口压缩：
        1. 最近窗口：完整保留
        2. 中间窗口：提取结构化摘要
        3. 远古窗口：完全丢弃或只做极简记忆
        """
        # 1. 最近窗口（全保留）
        recent_start = max(0, current_turn_idx - self.recent_size)
        recent_messages = messages[recent_start:current_turn_idx]
        
        # 2. 中间窗口（需要摘要）
        summary_start = max(0, recent_start - self.summary_size)
        summary_end = recent_start
        
        if summary_end > summary_start:
            to_summarize = messages[summary_start:summary_end]
            summary = self._extract_structured_summary(to_summarize)
        else:
            summary = None
        
        # 3. 组装
        result = []
        if summary:
            result.append({
                "role": "system",
                "content": f"[对话摘要] {summary}",
                "is_summary": True
            })
        result.extend(recent_messages)
        
        return result
    
    def _extract_structured_summary(self, messages):
        """
        提取结构化摘要（关键事实，不是散文）
        """
        key_facts = []
        
        for msg in messages:
            # 提取关键实体和决策
            if msg.get("extracted_facts"):
                key_facts.extend(msg["extracted_facts"])
        
        # 去重并按时间排序
        seen = set()
        unique_facts = []
        for fact in key_facts:
            fact_key = (fact.get("entity"), fact.get("value"))
            if fact_key not in seen:
                seen.add(fact_key)
                unique_facts.append(fact)
        
        # 格式化为结构化摘要
        summary_parts = []
        for fact in unique_facts[-5:]:  # 只保留最近5个关键事实
            summary_parts.append(f"- {fact['entity']}={fact['value']} ({fact.get('time', 'unknown')})")
        
        return "\n".join(summary_parts) if summary_parts else "(无关键决策)"

滑动窗口 vs 截断的对比:

维度截断滑动窗口
保留方式断崖式删除渐进式衰减
远处信息完全丢失保留摘要
实现复杂度简单中等（需要摘要提取）
Token成本低中（摘要本身也占Token）
适用场景远端确实不重要远端"可能还有用"
延迟零低（摘要可异步做）

最佳实践：

  • 滑动窗口的"摘要"必须是结构化的（键值对），而不是自由文本
  • 关键实体（order_id、user_id）必须在摘要中显式出现，不能隐含在叙述中
  • 摘要可异步生成（上一轮对话结束时预计算），不阻塞当前轮次

💡 理解要点：滑动窗口不是"折中"，而是一种信息分层策略——远处信息降级为"索引"，近处信息保持"原文"。

3.3 摘要（Summarization）：最慢，但信息密度最高

何时必须用摘要：

  • 远端内容包含复杂因果链（如多轮故障排查、法律条款推导）
  • 远端内容包含约束条件（如"如果X则Y，否则Z"）
  • 远端内容需要保留情感或语气（如用户已明确表达的偏好或不满）
  • 纯截断或滑窗会导致决策错误（如不知道用户已同意某条款）

摘要的工程实现:

class ContextSummarizer:
    def __init__(self, llm_client, summary_trigger_threshold=0.85):
        self.llm = llm_client
        self.trigger_threshold = summary_trigger_threshold  # 85%预算时触发摘要
    
    def should_summarize(self, current_tokens, max_tokens):
        """判断是否应触发摘要（基于预算阈值）"""
        return current_tokens / max_tokens > self.trigger_threshold
    
    def summarize_turns(self, messages_to_summarize, 
                        preserve_entities=None,
                        preserve_constraints=None):
        """
        对多轮对话生成结构化摘要
        """
        if preserve_entities is None:
            preserve_entities = ["order_id", "user_id", "ticket_id", 
                                "product_id", "amount", "deadline"]
        
        if preserve_constraints is None:
            preserve_constraints = ["if", "unless", "must not", 
                                   "requires", "prohibited"]
        
        # 构建摘要提示
        summary_prompt = self._build_summary_prompt(
            messages_to_summarize,
            preserve_entities,
            preserve_constraints
        )
        
        # 调用LLM生成摘要（可异步）
        summary = self.llm.generate(summary_prompt)
        
        # 后处理：确保关键实体存在
        validated_summary = self._validate_summary(summary, preserve_entities)
        
        return validated_summary
    
    def _build_summary_prompt(self, messages, entities, constraints):
        """构建结构化摘要提示"""
        messages_text = "\n".join([
            f"{m['role']}: {m['content'][:200]}..."  # 截断显示
            for m in messages
        ])
        
        return f"""请将以下对话历史总结为结构化摘要，用于后续轮次决策：

对话历史：
{messages_text}

要求：
1. 保留以下关键实体（必须显式列出）：{', '.join(entities)}
2. 保留约束条件（如{if/unless/must not等逻辑）
3. 保留已确认的事实和用户明确表达的偏好
4. 标注未决问题（如有）
5. 格式：键值对列表，不要写成长段落

输出格式示例：
- user_id: U12345
- confirmed_order: ORD-789
- user_preference: "不接受电话沟通，只接受邮件"
- constraint: "if status=shipped then cannot cancel"
- unresolved: "退款金额待确认"
- sentiment: "user frustrated about delay"
"""
    
    def _validate_summary(self, summary, required_entities):
        """验证摘要是否包含必需实体"""
        missing = []
        for entity in required_entities:
            if entity not in summary:
                missing.append(entity)
        
        if missing:
            # 补充缺失的实体（从原始消息重新提取）
            summary += f"\n\n[补充] missing entities noted: {', '.join(missing)}"
        
        return summary

摘要 vs 其他压缩手段的决策树:

是否需要压缩？
├── 否 → 保持原样
└── 是 → 远端内容是否重要？
    ├── 否 → 截断（最快）
    ├── 是，但只需"存在性" → 滑动窗口（中等）
    └── 是，且需要因果/约束/情感 → 摘要（最慢但最保真）

摘要的成本与收益分析:

成本项数值/描述
额外延迟100-500ms（取决于模型和长度）
额外Token输入：需发送待摘要内容；输出：摘要本身
信息损失约10-30%（取决于内容复杂度）
实现复杂度高（需要LLM调用、错误处理、验证）
收益项描述
信息密度比原文高3-5倍（Token效率）
决策质量保留关键约束，减少幻觉
可追溯性结构化摘要便于审计

混合策略（推荐）:

大多数生产系统应采用三层压缩：

def compress_context_full(messages, budget, llm_client):
    """
    三层压缩：截断 + 滑动窗口 + 摘要
    """
    # Layer 1: 截断（零成本，去噪声）
    messages = truncate_history(messages, keep_last_n=20)
    
    # Layer 2: 滑动窗口（低成本，保近期）
    compressor = SlidingWindowCompressor(
        recent_window_size=5,
        summary_window_size=10
    )
    messages = compressor.compress(messages, current_turn_idx=len(messages))
    
    # Layer 3: 摘要（按需触发，保因果）
    current_tokens = estimate_tokens(messages)
    if current_tokens > budget.total * 0.8:
        # 只摘要最远处的窗口
        summarizer = ContextSummarizer(llm_client)
        oldest_messages = messages[1:5]  # 跳过system
        summary = summarizer.summarize_turns(oldest_messages)
        
        # 替换为摘要
        messages = [messages[0]] + [{"role": "system", "content": summary}] + messages[5:]
    
    return messages

💡 理解要点：

  • 不要一上来就用摘要——先截断、再滑窗、最后才用LLM做摘要
  • 摘要的输出格式必须结构化（键值对），方便下游使用
  • 摘要可以异步预计算（在对话间隙做），减少当前轮延迟
  • 保留原始消息的引用（如summary_of_turns_5_to_15），便于追溯

🔍 实际例子：技术支持对话，第1-3轮是问题描述，第4-8轮是逐步排查（每步都有if-then逻辑），第9-12轮是当前解决方案。压缩策略：

  1. 截断：去掉问候语和已确认的身份验证细节
  2. 滑窗：第4-8轮提取为"排查步骤摘要"（保留每步的条件和结果）
  3. 保持：第9-12轮原文（当前解决方案需要完整上下文）

4 工具长返回怎么处理：截断、分页与"只回注字段"

4.1 问题：工具返回的JSON可能"爆长"

想象你调用"查询订单详情"工具，它返回了包含2000+ tokens的巨型JSON：订单基本信息、50项商品详情、完整物流历史、支付流水、客服备注、内部标记……如果全塞进上下文，会立即挤占其他重要信息的空间，让模型注意力分散，同时增加成本和延迟。

解决思路：不是所有字段都参与本轮决策。根据当前任务意图，只保留关键字段（字段投影），其余要么丢弃，要么降级为"可追溯但不进窗"。

4.2 两步操作：字段投影 + 长度截断

第一步：字段投影（Field Projection）

根据用户意图，只提取决策必需的字段。

示例：用户问"我的订单到哪了？"

# 原始工具返回（约2000 tokens）
full_result = {
    "order_id": "ORD-123456",
    "user_id": "U-789",
    "status": "shipped",
    "items": [{"sku": "SKU-001", "name": "iPhone...", "price": 8999, ...}, ...],  # 50项商品
    "shipping": {
        "carrier": "SF Express",
        "tracking_number": "SF1234567890",
        "estimated_delivery": "2026-01-18",
        "history": [{"status": "picked_up", "time": "...", "location": "..."}, ...]  # 20条记录
    },
    "payment": {"method": "credit_card", "total": 12997, "breakdown": {...}},
    "customer_service_notes": "Customer called on...",  # 长文本
    "internal_flags": ["vip_customer", "priority_shipping"]
}

# 字段投影：只提取与"查物流"相关的字段（约100 tokens）
def project_for_tracking(full_result):
    return {
        "order_id": full_result["order_id"],
        "status": full_result["status"],                    # "shipped"
        "tracking_number": full_result["shipping"]["tracking_number"],
        "estimated_delivery": full_result["shipping"]["estimated_delivery"],
        "carrier": full_result["shipping"]["carrier"],
        "latest_status": full_result["shipping"]["history"][-1]["status"]  # 只保留最新状态
    }

# 投影后结果：
# {
#   "order_id": "ORD-123456",
#   "status": "shipped",
#   "tracking_number": "SF1234567890",
#   "estimated_delivery": "2026-01-18",
#   "carrier": "SF Express",
#   "latest_status": "out_for_delivery"
# }
第二步：长度截断（深度保险）

即使投影后，某个字段本身可能仍很长（如商品描述）。需要递归截断：

def deep_truncate(obj, depth=0, max_depth=2, max_str_len=200):
    """递归截断嵌套结构"""
    if depth > max_depth:
        return "[nested object truncated]"
    
    if isinstance(obj, str) and len(obj) > max_str_len:
        return obj[:max_str_len] + "...[truncated]"
    
    if isinstance(obj, dict):
        return {k: deep_truncate(v, depth+1) for k, v in obj.items()}
    
    if isinstance(obj, list):
        # 列表只保留前3项
        truncated_list = [deep_truncate(item, depth+1) for item in obj[:3]]
        if len(obj) > 3:
            truncated_list.append(f"...[{len(obj)-3} more items]")
        return truncated_list
    
    return obj

4.3 任务意图决定保留哪些字段

同一个工具返回，不同用户意图需要保留的字段不同：

用户意图保留的关键字段丢弃的字段
“我的订单到哪了？”status, tracking_number, estimated_deliverypayment, items详情, customer_service_notes
“我要退款”order_id, status, created_at(判断是否超期), payment.totalshipping.history, items.attributes
“订单里买了什么？”items(但只保留sku和name，不要attributes)shipping, payment.breakdown
“总金额多少？”payment.totalitems, shipping, internal_flags
“为什么还没发货？”status, created_at, customer_service_notes(查看是否有延迟说明)payment.breakdown, items价格详情

💡 理解要点：投影规则应与意图识别或路由节点联动——当模型判定用户要"查物流"时，自动调用project_for_tracking；当判定要"退款"时，调用project_for_refund。

4.4 完整的工具截断流程（含元数据标记）

def truncate_tool_result(full_result, task_intent, max_total_chars=2000):
    """
    工具结果截断的完整流程
    
    1. 字段投影：按任务意图选择关键字段
    2. 深度截断：对每个字段做长度限制
    3. 添加元数据：标记被截断，便于追溯
    """
    # 定义不同意图的字段选择规则
    field_selector = {
        "check_delivery": ["order_id", "status", "shipping"],
        "refund_request": ["order_id", "status", "created_at", "payment"],
        "list_items": ["order_id", "items"],
        "check_total": ["order_id", "payment.total"],
        "general_inquiry": ["order_id", "status", "created_at"]
    }
    
    # Step 1: 字段投影
    essential_keys = field_selector.get(task_intent, ["order_id", "status"])
    projected = {k: full_result.get(k) for k in essential_keys if k in full_result}
    
    # Step 2: 深度截断
    projected = deep_truncate(projected)
    
    # Step 3: 添加截断元数据（便于调试和追溯）
    projected["_truncated"] = True                    # 标记已截断
    projected["_original_size"] = len(str(full_result))  # 原始大小
    projected["_projection_rule"] = task_intent       # 使用的投影规则
    projected["_full_data_available_at"] = f"db://orders/{full_result.get('order_id')}"
    
    return projected

为什么添加元数据很重要：

  • _truncated: True 让模型知道"还有更多信息可用，如果需要可以问我"
  • _full_data_available_at 提供溯源链接，便于开发者复查
  • _projection_rule 便于排查"为什么这次只给了这么少字段"

4.5 分页回注（Pagination-in-the-loop）

当用户确实需要"列表的所有项"（如"把过去一年所有订单列出来"），不要一次返回5000条，而是：

  1. 先问筛选条件/时间范围 —— 缩小数据集
  2. 提供分页参数 —— limit/offset 或 _links.next（与第29篇"大结果集裁剪与分页"呼应）
  3. 每轮只拿一页 —— 模型按需请求"下一页"
  4. 每页内仍做字段投影 —— 即使分页了，单条记录也只保留必要字段

🔍 实际例子：用户问"过去一年的订单"。正确姿势：

  • 第1轮：返回"共156条，最近10条摘要（只含order_id+status+date）"
  • 用户：“查看3月份的那些”
  • 第2轮：分页查询3月（缩小到23条），再展示10条
  • 用户：“第2页”
  • 第3轮：offset=10，再返回10条

这样每轮上下文只增加~200 tokens，而不是一次性+5000 tokens。

4.6 关键原则总结

  1. 投影先于截断：先决定"哪些字段要"（业务决策），再决定"每个字段留多长"（技术限制）
  2. 任务驱动选择：不同意图 = 不同投影规则，不要一刀切
  3. 添加截断标记：_truncated: True 让模型知道信息不完整
  4. 提供溯源链接：_full_data_available_at 便于复查和调试
  5. 绝不截断的字段：status, order_id, error_code 等决策必须字段要完整保留
  6. 分页优于全量：即使需要列表，也用分页控制每轮增量

💡 理解要点：工具截断不是"随便删掉一些内容"，而是根据当前任务意图，有选择地保留决策所需信息。这样做可以把2000 tokens的工具返回压缩到100-200 tokens，节省的预算可以用来放更多检索块或对话历史。

5 一个最小可运行的压缩策略（伪代码）

def assemble_context(budget, system, tools_desc, anchors, history, rag_chunks, tool_results, task):
    # 1) 预留输出空间
    input_budget = budget.total - budget.reserve_output

    # 2) 固定层先放
    parts = [system, tools_desc, anchors]

    # 3) 工具结果：先投影再截断
    projected = project_tool_results(tool_results)          # 只留必要字段
    projected = truncate(projected, budget.tool_pool)

    # 4) 检索块：去重、按权威/新鲜度排序、再截断
    rag = dedup_and_rank(rag_chunks)
    rag = truncate(rag, budget.rag_pool)

    # 5) 历史：最近N轮 + 更早摘要
    recent = take_last_turns(history, n=budget.history_turns)
    memory = summarize_if_needed(history, threshold=0.85)    # 达阈值再摘要

    parts.extend([memory, rag, projected, recent, task])
    return parts

💡 理解要点：先确定池子上限，再决定压缩方式；不要"先拼完再看超不超"。

6 小结

  • Token 预算要像成本一样运营：切池 + 硬上限 + 溢出策略。
  • 压缩顺序建议：先截断/分页（快且确定），再摘要（慢但更保真）。
  • 工具输出的默认策略应是"字段投影"，而不是"全量回注"。

参考资料

  • Lost in the Middle: How Language Models Use Long Contexts. [参考：arXiv:2307.03172]
  • OpenAI Function Calling Guide（工具调用与结果回注的结构）。[参考：https://platform.openai.com/docs/guides/function-calling]
  • OpenAI Prompt Caching（长前缀/静态指令置前，有助于降本与稳定拼装）。[参考：https://platform.openai.com/docs/guides/prompt-caching]
# 人工智能 # 语言模型 # 深度学习
Logo
乐奇 Rokid 开放社区

更多推荐

cover

极刻觅镜 | AI眼镜周报｜Rokid四款硬件发布与Android XR工具转稳定版

avatar 乐奇 Rokid 开放社区
cover

极刻觅镜 | AI眼镜日报｜北京发布七项消费举措、韩国高考排查AI眼镜

avatar 乐奇 Rokid 开放社区
cover

极刻觅镜 | AI眼镜日报｜京东智能眼镜西安站活动与千问AI眼镜OTA升级

avatar 乐奇 Rokid 开放社区