LLM 能力集成：多轮对话的上下文压缩与长文本处理策略

LLM 能力集成多轮对话的上下文压缩与长文本处理策略一、Token 账单与注意力衰减长对话场景的双重困境在将 LLM 能力集成到业务系统时多轮对话的上下文管理是一个绕不开的工程难题。一方面随着对话轮次增加输入 Token 数线性增长API 调用成本随之飙升——一个 20 轮的客服对话Token 消耗可能达到初始请求的 5 倍以上。另一方面研究表明当上下文长度超过模型有效注意力范围时模型对中间位置信息的提取准确率显著下降即所谓的Lost in the Middle现象。这两个问题形成了一个矛盾为了保持对话连贯性需要尽可能多的上下文但过多的上下文既增加成本又降低生成质量。简单的滑动窗口方案虽然控制了 Token 数量却会丢失早期对话中的关键信息——用户在第一轮提到的偏好到第十轮时可能已经被窗口淘汰。如何在压缩上下文的同时保留关键信息是 LLM 集成工程中的核心挑战。二、上下文压缩的三大策略摘要、检索与结构化提取上下文压缩的核心思路是将原始对话信息转化为更紧凑的表示形式同时保留对后续对话有用的语义信息。三种主流策略各有适用场景。flowchart LR subgraph 原始对话 M1[第1轮: 用户偏好] M2[第2-5轮: 闲聊] M3[第6-8轮: 问题排查] M4[第9轮: 解决方案] M5[第10轮: 新问题] end subgraph 策略一: 滚动摘要 M1 -- S1[摘要1: 用户偏好闲聊要点] M2 -- S1 S1 -- S2[摘要2: 偏好排查进展] M3 -- S2 S2 -- S3[摘要3: 偏好已解决问题] M4 -- S3 S3 -- CTX1[压缩上下文: 摘要3 第10轮] M5 -- CTX1 end subgraph 策略二: 检索增强 M1 -- V1[向量化存储] M2 -- V2[向量化存储] M3 -- V3[向量化存储] M4 -- V4[向量化存储] M5 -- Q[查询向量] Q -- |Top-K| CTX2[压缩上下文: Top-K片段 第10轮] V1 -- Q V3 -- Q V4 -- Q end subgraph 策略三: 结构化提取 M1 -- E1[偏好: 暗色主题] M3 -- E2[问题: 登录失败] M4 -- E3[方案: 重置密码] E1 -- CTX3[压缩上下文: 结构化摘要 第10轮] E2 -- CTX3 E3 -- CTX3 end滚动摘要每隔 N 轮对话调用 LLM 对早期对话生成摘要用摘要替代原始消息。优点是信息损失可控缺点是每次摘要调用本身消耗 Token且摘要错误会累积传播。检索增强将所有历史对话向量化存储每轮对话时根据当前问题检索最相关的 K 个片段。优点是无需丢弃任何信息缺点是检索噪声可能引入无关上下文。结构化提取在对话过程中实时提取关键实体、用户偏好、已解决问题等结构化信息维护一个动态更新的状态表。优点是信息密度最高缺点是需要针对业务场景定制提取规则。三、混合压缩引擎的实现# context_compressor.py — 多轮对话上下文压缩引擎 import time import json from dataclasses import dataclass, field from typing import Optional import numpy as np dataclass class ConversationTurn: 单轮对话的数据模型 turn_id: int role: str # user / assistant content: str timestamp: float field(default_factorytime.time) token_count: int 0 entities: list[dict] field(default_factorylist) # 提取的结构化实体 summary: Optional[str] None embedding: Optional[np.ndarray] None class RollingSummarizer: 滚动摘要每 N 轮生成一次摘要替代原始对话 def __init__(self, summarize_fn, window_size: int 6, max_summary_tokens: int 200): self._summarize_fn summarize_fn self.window_size window_size self.max_summary_tokens max_summary_tokens self._pending_turns: list[ConversationTurn] [] self._summaries: list[str] [] def add_turn(self, turn: ConversationTurn) - None: 添加一轮对话达到窗口大小时触发摘要 self._pending_turns.append(turn) if len(self._pending_turns) self.window_size: self._generate_summary() def _generate_summary(self) - None: 对累积的对话轮次生成摘要 conversation_text \n.join( f{t.role}: {t.content} for t in self._pending_turns ) prompt ( f请将以下对话压缩为不超过{self.max_summary_tokens}Token的摘要 f保留关键决策、用户偏好和未解决问题\n\n{conversation_text} ) summary self._summarize_fn(prompt) self._summaries.append(summary) # 为已摘要的轮次标记避免重复处理 for turn in self._pending_turns: turn.summary summary self._pending_turns.clear() def get_context(self, current_turn: ConversationTurn, max_tokens: int 4096) - list[dict]: 构建压缩后的上下文历史摘要 未摘要轮次 当前轮 context [] # 历史摘要部分 for summary in self._summaries: context.append({ role: system, content: f[对话摘要] {summary} }) # 未摘要的待处理轮次 for turn in self._pending_turns: context.append({ role: turn.role, content: turn.content }) # 当前轮 context.append({ role: current_turn.role, content: current_turn.content }) return context def flush(self) - None: 强制将剩余待处理轮次生成摘要 if self._pending_turns: self._generate_summary() class StructuredExtractor: 结构化提取从对话中实时提取实体和状态 def __init__(self, extract_fn, schema: dict): self._extract_fn extract_fn self.schema schema # 定义需要提取的字段和类型 self._state: dict {} def extract(self, turn: ConversationTurn) - list[dict]: 从一轮对话中提取结构化信息 prompt ( f从以下对话中提取结构化信息按 JSON Schema 输出\n fSchema: {json.dumps(self.schema, ensure_asciiFalse)}\n f对话: {turn.role}: {turn.content}\n f仅输出 JSON不要解释。 ) try: result self._extract_fn(prompt) entities json.loads(result) if isinstance(entities, dict): entities [entities] turn.entities entities # 更新全局状态 for entity in entities: entity_type entity.get(type, unknown) entity_key f{entity_type}:{entity.get(name, entity.get(id, ))} self._state[entity_key] { value: entity, updated_at: time.time(), source_turn: turn.turn_id, } return entities except (json.JSONDecodeError, Exception): return [] def get_state_summary(self) - str: 生成当前结构化状态的文本摘要 if not self._state: return lines [[当前状态]] for key, info in self._state.items(): entity info[value] lines.append(f- {key}: {json.dumps(entity, ensure_asciiFalse)}) return \n.join(lines) class HybridCompressor: 混合压缩引擎结合摘要、检索和结构化提取 def __init__(self, summarize_fn, extract_fn, embed_fn, entity_schema: dict, max_context_tokens: int 4096): self.summarizer RollingSummarizer(summarize_fn, window_size6) self.extractor StructuredExtractor(extract_fn, entity_schema) self._embed_fn embed_fn self._turns: list[ConversationTurn] [] self._embeddings: list[np.ndarray] [] self.max_context_tokens max_context_tokens def process_turn(self, role: str, content: str) - list[dict]: 处理一轮新对话返回压缩后的上下文 turn ConversationTurn( turn_idlen(self._turns), rolerole, contentcontent, token_countlen(content), # 简化估算 ) # 1. 结构化提取 self.extractor.extract(turn) # 2. 向量化存储 if self._embed_fn: turn.embedding self._embed_fn(content) self._embeddings.append(turn.embedding) # 3. 滚动摘要 self.summarizer.add_turn(turn) self._turns.append(turn) # 4. 构建混合上下文 return self._build_context(turn) def _build_context(self, current_turn: ConversationTurn) - list[dict]: 构建混合压缩上下文 context [] token_budget self.max_context_tokens # 优先级1: 结构化状态摘要信息密度最高 state_summary self.extractor.get_state_summary() if state_summary: context.append({role: system, content: state_summary}) token_budget - len(state_summary) # 优先级2: 滚动摘要保留对话主线 for summary in self.summarizer._summaries: if token_budget 0: break context.append({role: system, content: f[摘要] {summary}}) token_budget - len(summary) # 优先级3: 语义检索补充填充关键细节 if self._embed_fn and current_turn.embedding is not None and token_budget 0: relevant self._semantic_search(current_turn.embedding, top_k3) for turn in relevant: if token_budget 0: break context.append({ role: system, content: f[相关历史] {turn.role}: {turn.content} }) token_budget - turn.token_count # 当前轮对话 context.append({ role: current_turn.role, content: current_turn.content }) return context def _semantic_search(self, query_embedding: np.ndarray, top_k: int 3) - list[ConversationTurn]: 基于向量相似度检索相关历史对话 if len(self._embeddings) 2: return [] matrix np.array(self._embeddings[:-1]) # 排除当前轮 query_norm query_embedding / (np.linalg.norm(query_embedding) 1e-8) matrix_norm matrix / (np.linalg.norm(matrix, axis1, keepdimsTrue) 1e-8) similarities matrix_norm query_norm top_indices np.argsort(similarities)[-top_k:][::-1] return [self._turns[i] for i in top_indices if similarities[i] 0.6]混合压缩引擎按优先级组合三种策略结构化状态优先信息密度最高滚动摘要次之保留对话主线语义检索补充填充关键细节。这种分层策略在保证关键信息不丢失的前提下将 20 轮对话的 Token 消耗压缩到原始的 30%-40%。四、压缩策略的精度损失与延迟代价上下文压缩本质上是一种有损压缩必然引入信息损失。滚动摘要的累积误差是最突出的问题——每次摘要都是对前一次摘要的再压缩经过多轮后早期对话中的细节可能被完全抹去。缓解方案是在摘要 Prompt 中显式要求保留未解决问题和用户偏好两类信息并在结构化提取层做兜底。延迟代价同样需要关注。每次结构化提取和摘要生成都需要额外的 LLM 调用在串行模式下会增加 500ms-2s 的响应延迟。生产环境中应将提取和摘要操作异步化——用户请求先基于当前上下文生成响应提取和摘要在后台异步完成供下一轮对话使用。适用边界方面混合压缩策略适用于客服、咨询等长对话场景。对于简单的指令式交互如代码补全、翻译滑动窗口即可满足需求引入压缩引擎反而增加了不必要的复杂度。五、总结多轮对话的上下文压缩是 LLM 工程化落地的关键环节。滚动摘要、检索增强和结构化提取三种策略各有优劣混合使用可以在 Token 压缩率和信息保留率之间取得平衡。落地时建议优先实现结构化提取投入产出比最高再逐步引入滚动摘要和语义检索。异步化是控制延迟的核心手段压缩操作不应阻塞主响应路径。