当前位置：首页 > news >正文

LLM上下文压缩：在有限窗口中保留关键信息的工程策略

news 2026/6/18 12:42:45

LLM上下文压缩：在有限窗口中保留关键信息的工程策略

一、上下文窗口的"空间焦虑"：为什么压缩不是锦上添花

大模型的上下文窗口从4K扩展到128K再到1M，但"窗口焦虑"并没有消失。原因很简单：窗口越大，推理成本越高，延迟越长。一个128K token的请求，其推理延迟和费用远超8K token的请求。在生产环境中，不可能每个请求都塞满上下文窗口。

更现实的问题是：Agent系统的上下文消耗速度远超预期。系统提示词占2-4K，工具定义占1-3K，每轮对话输入输出各占500-2000 token，工具调用结果可能占1-5K。一个10轮对话的Agent，上下文轻松突破30K token。如果同时维护多个Agent的协作上下文，窗口消耗会更快。

上下文压缩的核心目标不是"塞更多内容"，而是"用更少token保留等价信息"。这和传统的文本压缩不同——不是无损还原每个字，而是保留对后续推理有用的信息。这是一个信息论和工程实践交叉的问题。

二、上下文压缩的核心策略与机制

2.1 三种压缩策略对比

flowchart TD A[上下文压缩策略] --> B[截断策略<br/>Truncation] A --> C[摘要策略<br/>Summarization] A --> D[结构化提取<br/>Structured Extraction] B --> B1[实现简单] B --> B2[信息损失不可控] B --> B3[适用: 闲聊场景] C --> C1[信息保留较好] C --> C2[需要额外LLM调用] C --> C3[适用: 长对话] D --> D1[信息精确保留] D --> D2[需要预定义Schema] D --> D3[适用: 任务型对话]

截断策略最简单但最粗暴——直接丢弃早期消息。摘要策略用LLM对早期对话生成摘要，保留语义但丢失细节。结构化提取策略从对话中抽取关键实体和关系，以JSON等结构化格式存储，信息精确但需要预定义Schema。

2.2 混合压缩策略设计

生产环境中最有效的方案是混合策略：对最近N轮对话保留原文，对更早的对话生成摘要，同时维护一个结构化的关键信息表。

flowchart LR subgraph "压缩前上下文" M1[第1-5轮对话] M2[第6-10轮对话] M3[第11-15轮对话] M4[第16-20轮对话] end subgraph "压缩后上下文" S1[摘要: 第1-5轮] S2[摘要: 第6-10轮] K[关键信息表] M3R[原文: 第11-15轮] M4R[原文: 第16-20轮] end M1 --> S1 M2 --> S2 M1 --> K M2 --> K M3 --> M3R M4 --> M4R

三、生产级上下文压缩器实现

3.1 关键信息提取器

package contextcomp import ( "context" "encoding/json" "fmt" "strings" ) // KeyInfo 关键信息条目 type KeyInfo struct { Category string `json:"category"` // 信息类别：user_profile/preference/decision/fact Key string `json:"key"` // 信息键 Value string `json:"value"` // 信息值 Source string `json:"source"` // 来源轮次 Priority int `json:"priority"` // 优先级（1-5，5最高） } // KeyInfoExtractor 关键信息提取器 type KeyInfoExtractor struct { client LLMClient schema []InfoCategory } // InfoCategory 信息类别定义 type InfoCategory struct { Name string `json:"name"` Description string `json:"description"` Keys []string `json:"keys"` } // NewKeyInfoExtractor 创建提取器 func NewKeyInfoExtractor(client LLMClient) *KeyInfoExtractor { return &KeyInfoExtractor{ client: client, schema: defaultSchema(), } } func defaultSchema() []InfoCategory { return []InfoCategory{ { Name: "user_profile", Description: "用户基本信息", Keys: []string{"name", "role", "company", "location"}, }, { Name: "preference", Description: "用户偏好", Keys: []string{"style", "language", "format", "detail_level"}, }, { Name: "decision", Description: "已做出的决策", Keys: []string{"chosen_option", "rejected_option", "reason"}, }, { Name: "fact", Description: "关键事实和数据", Keys: []string{"number", "date", "constraint", "requirement"}, }, } } // Extract 从消息中提取关键信息 func (e *KeyInfoExtractor) Extract(ctx context.Context, messages []Message) ([]KeyInfo, error) { // 构建对话文本 var sb strings.Builder for _, msg := range messages { sb.WriteString(fmt.Sprintf("[%s]: %s\n", msg.Role, msg.Content)) } // 构建Schema描述 schemaJSON, _ := json.Marshal(e.schema) prompt := fmt.Sprintf(`从以下对话中提取关键信息，按照给定的Schema结构返回。 信息类别Schema： %s 对话内容： %s 请以JSON数组格式返回提取结果，每个条目包含category/key/value/source/priority字段。 只提取明确提到的信息，不要推断。priority: 1=可选信息, 3=重要信息, 5=关键决策信息`, schemaJSON, sb.String()) resp, err := e.client.Chat(ctx, []Message{{Role: "user", Content: prompt}}) if err != nil { return nil, fmt.Errorf("key info extraction failed: %w", err) } var infos []KeyInfo if err := json.Unmarshal([]byte(resp), &infos); err != nil { return nil, fmt.Errorf("parse key info failed: %w", err) } return infos, nil }

3.2 分层压缩引擎

package contextcomp import ( "context" "sort" "time" ) // CompressionConfig 压缩配置 type CompressionConfig struct { MaxTokens int // 上下文窗口最大token数 ReservedTokens int // 为输出预留的token数 RecentTurns int // 保留最近N轮原文 SummaryChunkSize int // 每次摘要的对话轮数 MinPriority int // 保留的最低优先级 } // CompressionResult 压缩结果 type CompressionResult struct { SystemPrompt string // 系统提示词 KeyInfoTable []KeyInfo // 关键信息表 Summaries []string // 早期对话摘要 RecentMessages []Message // 最近N轮原文 TotalTokens int // 压缩后总token数 } // Compressor 上下文压缩引擎 type Compressor struct { config CompressionConfig extractor *KeyInfoExtractor summarizer *Summarizer counter TokenCounter } // NewCompressor 创建压缩引擎 func NewCompressor(config CompressionConfig, client LLMClient, counter TokenCounter) *Compressor { return &Compressor{ config: config, extractor: NewKeyInfoExtractor(client), summarizer: NewSummarizer(client), counter: counter, } } // Compress 执行上下文压缩 func (c *Compressor) Compress(ctx context.Context, systemPrompt string, history []Message) (*CompressionResult, error) { result := &CompressionResult{ SystemPrompt: systemPrompt, } availableTokens := c.config.MaxTokens - c.config.ReservedTokens usedTokens := c.counter.Count(systemPrompt) // 第一步：提取关键信息表 keyInfos, err := c.extractor.Extract(ctx, history) if err != nil { return nil, err } // 过滤低优先级信息 var filteredInfos []KeyInfo for _, info := range keyInfos { if info.Priority >= c.config.MinPriority { filteredInfos = append(filteredInfos, info) } } result.KeyInfoTable = filteredInfos // 计算关键信息表token infoText := c.formatKeyInfoTable(filteredInfos) usedTokens += c.counter.Count(infoText) // 第二步：分割历史消息 totalTurns := len(history) / 2 // 每轮=1条user+1条assistant recentStartIdx := len(history) - c.config.RecentTurns*2 if recentStartIdx < 0 { recentStartIdx = 0 } // 保留最近N轮原文 result.RecentMessages = history[recentStartIdx:] for _, msg := range result.RecentMessages { usedTokens += c.counter.Count(msg.Content) } // 第三步：对早期消息生成摘要 if recentStartIdx > 0 { oldMessages := history[:recentStartIdx] // 分块摘要 chunkSize := c.config.SummaryChunkSize * 2 for i := 0; i < len(oldMessages); i += chunkSize { end := i + chunkSize if end > len(oldMessages) { end = len(oldMessages) } chunk := oldMessages[i:end] summary, err := c.summarizer.Summarize(ctx, chunk) if err != nil { // 摘要失败，跳过该块 continue } summaryTokens := c.counter.Count(summary) if usedTokens+summaryTokens > availableTokens { break // 空间不足，停止添加摘要 } result.Summaries = append(result.Summaries, summary) usedTokens += summaryTokens } } result.TotalTokens = usedTokens return result, nil } // formatKeyInfoTable 将关键信息表格式化为文本 func (c *Compressor) formatKeyInfoTable(infos []KeyInfo) string { if len(infos) == 0 { return "" } // 按类别分组 grouped := make(map[string][]KeyInfo) for _, info := range infos { grouped[info.Category] = append(grouped[info.Category], info) } var result string result += "【关键信息表】\n" for category, items := range grouped { result += category + ":\n" for _, item := range items { result += fmt.Sprintf(" - %s: %s\n", item.Key, item.Value) } } return result } // ToMessages 将压缩结果转换为LLM可消费的消息列表 func (r *CompressionResult) ToMessages() []Message { var messages []Message // 系统提示词 + 关键信息表 systemContent := r.SystemPrompt if infoText := r.formatKeyInfoTable(r.KeyInfoTable); infoText != "" { systemContent += "\n\n" + infoText } messages = append(messages, Message{Role: "system", Content: systemContent}) // 早期摘要 for _, summary := range r.Summaries { messages = append(messages, Message{ Role: "system", Content: fmt.Sprintf("【早期对话摘要】%s", summary), }) } // 最近原文 messages = append(messages, r.RecentMessages...) return messages }

3.3 摘要生成器

package contextcomp import ( "context" "fmt" "strings" ) // Summarizer 对话摘要生成器 type Summarizer struct { client LLMClient } func NewSummarizer(client LLMClient) *Summarizer { return &Summarizer{client: client} } // Summarize 对一组消息生成摘要 func (s *Summarizer) Summarize(ctx context.Context, messages []Message) (string, error) { var sb strings.Builder for _, msg := range messages { sb.WriteString(fmt.Sprintf("[%s]: %s\n", msg.Role, msg.Content)) } prompt := fmt.Sprintf(`请对以下对话片段生成简洁摘要，要求： 1. 保留用户的核心需求和意图 2. 保留已确认的具体信息（数字、日期、名称等） 3. 保留重要的决策和结论 4. 丢弃闲聊和重复内容 5. 摘要长度不超过原文的30%% 对话内容： %s`, sb.String()) return s.client.Chat(ctx, []Message{{Role: "user", Content: prompt}}) }