大模型上下文压缩：长对话场景下的 Token 省流策略

大模型上下文压缩长对话场景下的 Token 省流策略一、长对话 Token 消耗为什么是推理成本的黑洞大模型推理的计费单位是 Token对话场景中每轮请求携带完整历史上下文。10 轮对话后上下文膨胀到数千 Token20 轪后翻倍。如果服务端不做压缩每轮推理的输入 Token 数线性增长推理成本同步增长。更严重的是上下文超过模型窗口上限后要么截断导致信息丢失要么直接拒绝请求。基础设施不需要漂亮话Token 消耗是可计量的成本不是模型很聪明所以可以接受的问题。一个日均 100 万次推理请求的服务平均上下文从 200 Token 增长到 2000 Token推理成本翻 10 倍。上下文压缩的目标在保留必要信息的前提下减少每轮请求的 Token 数控制推理成本增长曲线。压缩策略分三类截断丢弃旧信息、摘要浓缩旧信息、滑动窗口只保留最近几轮。每类策略各有适用场景组合使用效果更好。二、上下文压缩策略与 Token 管理模型三种压缩策略的原理和适用场景flowchart TB subgraph Input[原始对话上下文] direction LR R1[第1轮: 用户提问] R2[第2轮: 模型回答] R3[第3-10轮: 对话历史] R10[第11轮: 当前请求] end subgraph Strategy[压缩策略] direction TB S1[截断策略br/丢弃超出窗口的旧轮次] S2[摘要策略br/压缩历史为摘要段落] S3[滑动窗口br/只保留最近 N 轮] end subgraph Output[压缩后上下文] direction LR O1[摘要段落br/(~200 Token)] O2[最近 3 轮完整对话br/(~600 Token)] O3[当前请求br/(~100 Token)] end Input -- Strategy -- Output style S1 fill:#fce4ec style S2 fill:#e8f5e9 style S3 fill:#fff3e0 style Output fill:#e3f2fd策略Token 减少信息保留适用场景截断高直接丢弃低旧信息完全丢失信息时效性强、旧轮不重要滑动窗口中保留最近 N 轮中近期信息完整通用场景默认策略摘要高压缩为短文本中高语义保留细节丢失长对话、需要全局上下文组合策略摘要压缩旧历史 滑动窗口保留近期完整对话。这是推理服务中最常用的方案Token 减少约 60%同时保留关键语义信息。三、上下文压缩引擎实现# context_compressor.py — 大模型上下文压缩引擎 import json import logging import re from typing import List, Dict, Optional from dataclasses import dataclass, field logging.basicConfig(levellogging.INFO) logger logging.getLogger(context-compressor) dataclass class ConversationRound: 单轮对话数据结构 round_id: int role: str # user or assistant content: str token_count: int # 该轮的 Token 估算 dataclass class CompressedContext: 压缩后的上下文结构 summary: str # 历史摘要 recent_rounds: List[ConversationRound] # 保留的近期完整对话 current_request: str # 当前用户请求 total_tokens: int # 压缩后总 Token 数 class ContextCompressor: 上下文压缩引擎组合摘要滑动窗口策略 def __init__( self, max_context_tokens: int 4000, # 上下文 Token 上限 recent_window_size: int 3, # 滑动窗口保留轮数 summary_max_tokens: int 300, # 摘要最大 Token 数 tokenizer_estimate_ratio: float 1.3, # 中文字符到 Token 的估算比例 ): self.max_context_tokens max_context_tokens self.recent_window_size recent_window_size self.summary_max_tokens summary_max_tokens self.tokenizer_estimate_ratio tokenizer_estimate_ratio def estimate_tokens(self, text: str) - int: 估算文本的 Token 数简易估算生产环境替换为真实 tokenizer # 中文文本字符数 × 1.3 约等于 Token 数 # 英文文本单词数 × 1.3 约等于 Token 数 # 混合文本按字符计算保守估算 char_count len(text) return int(char_count * self.tokenizer_estimate_ratio) def compress( self, rounds: List[ConversationRound], current_request: str, existing_summary: Optional[str] None, ) - CompressedContext: 压缩对话上下文 1. 保留最近 N 轮完整对话 2. 旧历史压缩为摘要 3. 确保总 Token 数在上限内 if not rounds: return CompressedContext( summaryexisting_summary or , recent_rounds[], current_requestcurrent_request, total_tokensself.estimate_tokens(current_request), ) # 分割近期轮次和旧历史 split_idx max(0, len(rounds) - self.recent_window_size) old_rounds rounds[:split_idx] recent_rounds rounds[split_idx:] # 计算当前 Token 占用 current_tokens self.estimate_tokens(current_request) recent_tokens sum(r.token_count for r in recent_rounds) summary_tokens self.estimate_tokens(existing_summary or ) total_without_old current_tokens recent_tokens summary_tokens # 如果已在上限内不需要额外压缩 if total_without_old self.max_context_tokens: logger.info(f上下文未超限: {total_without_old} {self.max_context_tokens}) return CompressedContext( summaryexisting_summary or , recent_roundsrecent_rounds, current_requestcurrent_request, total_tokenstotal_without_old, ) # 超限需要压缩旧历史为摘要 logger.info(f上下文超限: {total_without_old} {self.max_context_tokens}) # 生成摘要如果有旧历史需要压缩 if old_rounds or existing_summary: new_summary self._generate_summary(old_rounds, existing_summary) summary_tokens self.estimate_tokens(new_summary) else: new_summary summary_tokens 0 # 再次检查 Token 上限 total_tokens current_tokens recent_tokens summary_tokens # 仍然超限缩小滑动窗口 while total_tokens self.max_context_tokens and len(recent_rounds) 1: removed recent_rounds.pop(0) total_tokens - removed.token_count logger.info(f移除近期轮次: round{removed.round_id}, f当前 tokens{total_tokens}) # 最后兜底截断当前请求极端情况 if total_tokens self.max_context_tokens: logger.warning(f上下文仍然超限: {total_tokens}截断摘要) # 截断摘要到上限 max_summary_tokens self.max_context_tokens - current_tokens - recent_tokens new_summary self._truncate_text(new_summary, max_summary_tokens) summary_tokens self.estimate_tokens(new_summary) total_tokens current_tokens recent_tokens summary_tokens return CompressedContext( summarynew_summary, recent_roundsrecent_rounds, current_requestcurrent_request, total_tokenstotal_tokens, ) def _generate_summary( self, new_rounds: List[ConversationRound], existing_summary: Optional[str], ) - str: 生成对话摘要。 生产环境中调用摘要模型如小参数模型生成 此处提供规则式摘要作为兜底方案。 # 规则式摘要提取关键信息点 key_points [] if existing_summary: key_points.append(existing_summary) for round in new_rounds: if round.role user: # 提取用户提问的核心意图取首句 first_sentence self._extract_first_sentence(round.content) key_points.append(f用户提问: {first_sentence}) else: # 提取回答的关键结论取首句 first_sentence self._extract_first_sentence(round.content) key_points.append(f回答要点: {first_sentence}) # 拼接摘要控制在目标 Token 内 summary_text 历史对话摘要\n \n.join(key_points) return self._truncate_text(summary_text, self.summary_max_tokens) def _extract_first_sentence(self, text: str) - str: 提取文本首句中文以句号/问号/感叹号分割 # 匹配首个完整句子 match re.match(r[^。\n][。]?, text) if match: sentence match.group(0) # 限制单句长度 if len(sentence) 80: return sentence[:80] ... return sentence # 无标点的长文本截断到 50 字 if len(text) 50: return text[:50] ... return text def _truncate_text(self, text: str, max_tokens: int) - str: 按 Token 上限截断文本 estimated self.estimate_tokens(text) if estimated max_tokens: return text # 按 Token 估算反推最大字符数 max_chars int(max_tokens / self.tokenizer_estimate_ratio) return text[:max_chars] ... def format_for_inference(self, compressed: CompressedContext) - List[Dict]: 将压缩上下文格式化为模型输入 messages messages [] # 摘要作为系统提示的一部分 if compressed.summary: messages.append({ role: system, content: f以下是之前的对话摘要请在此基础上继续对话\n{compressed.summary}, }) # 近期完整对话 for round in compressed.recent_rounds: messages.append({ role: round.role, content: round.content, }) # 当前请求 messages.append({ role: user, content: compressed.current_request, }) return messages推理服务侧集成压缩引擎# inference_with_compression.py — 推理服务集成上下文压缩 from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Optional from context_compressor import ContextCompressor, ConversationRound app FastAPI(titleInference with Compression) # 初始化压缩引擎 compressor ContextCompressor( max_context_tokens4000, recent_window_size3, summary_max_tokens300, ) class InferenceRequest(BaseModel): 推理请求包含完整对话历史 prompt: str # 当前用户请求 history: List[dict] [] # 对话历史 session_id: Optional[str] None # 会话 ID existing_summary: Optional[str] None # 上次压缩的摘要 class InferenceResponse(BaseModel): 推理响应包含压缩后的摘要客户端缓存 result: str # 推理结果 compressed_summary: str # 新摘要客户端下次请求时传入 tokens_used: int # 实际使用的 Token 数 compression_ratio: float # 压缩比例 app.post(/v1/inference, response_modelInferenceResponse) async def inference(req: InferenceRequest): 带压缩的推理接口 # 构造 ConversationRound 列表 rounds [] for i, h in enumerate(req.history): content h.get(content, ) rounds.append(ConversationRound( round_idi 1, roleh.get(role, user), contentcontent, token_countcompressor.estimate_tokens(content), )) # 压缩上下文 compressed compressor.compress( roundsrounds, current_requestreq.prompt, existing_summaryreq.existing_summary, ) # 格式化为模型输入 messages compressor.format_for_inference(compressed) # 调用推理模型此处为示例实际替换为模型调用 # result call_model(messages) result f基于 {compressed.total_tokens} Token 的上下文生成的回复 # 计算压缩比例 original_tokens sum(r.token_count for r in rounds) compressor.estimate_tokens(req.prompt) compression_ratio 1.0 - (compressed.total_tokens / original_tokens) if original_tokens 0 else 0.0 return InferenceResponse( resultresult, compressed_summarycompressed.summary, tokens_usedcompressed.total_tokens, compression_ratiocompression_ratio, )四、压缩策略的边界与权衡场景一摘要质量依赖摘要模型。规则式摘要提取首句是最简方案质量有限。生产环境中用小参数模型如 1B 参数的摘要模型生成语义摘要成本比主推理模型低两个数量级。摘要模型的延迟约 50ms可接受。摘要缓存到客户端下次请求时传入避免每轮重复生成。场景二关键信息的丢失。截断和摘要都会丢失信息某些场景中旧轮次包含关键指令如请用中文回答、请只输出代码。解法提取系统级指令单独保存不参与压缩。压缩引擎只处理对话内容指令作为 system prompt 独立传入。场景三Token 估算的精度。简易估算字符数 × 瀹与真实 tokenizer 的偏差约 10-20%。生产环境必须使用真实 tokenizer 计算 Token 数否则上下文可能超限或浪费空间。tiktoken 库支持 OpenAI 模型的精确计算其他模型用对应的 tokenizer 实现。场景四多轮压缩的累积误差。每轮压缩都基于上一轮的摘要多轮压缩后摘要可能偏离原始语义。解法每隔 N 轮如 5 轮重新从完整历史生成摘要而非增量拼接。摘要刷新频率根据对话质量要求设定。五、总结大模型长对话的 Token 消耗随轮数线性增长上下文压缩控制推理成本。组合策略摘要压缩旧历史 滑动窗口保留近期对话是通用方案Token 减少约 60%。压缩引擎的实现要点滑动窗口保留最近 3 轮完整对话旧历史压缩为 200-300 Token 的摘要总上下文不超过模型窗口上限。摘要生成用小参数模型降低成本摘要缓存在客户端避免重复生成。系统级指令不参与压缩单独保留。Token 计算使用真实 tokenizer不能用简易估算替代。多轮压缩的累积误差通过定期刷新摘要缓解。上下文压缩的效果用 Token 数和推理成本衡量数据可追踪。