AI Agent开发实战⑱｜上下文压缩与选择：让LLM看到最有价值的信息-平芜编程栈

AI Agent开发实战⑱｜上下文压缩与选择：让LLM看到最有价值的信息

检索到了50篇文档，但LLM的上下文窗口只能塞5篇。选哪5篇？平均选会漏掉关键信息，全塞进去会爆Token。上下文压缩和选择策略就是解决这个矛盾：用最少的Token承载最多的信息。

一、上下文窗口的困境

典型场景： 检索返回：50篇文档，共30000字符 LLM上下文：4000 tokens（约6000字符） 可用窗口：4000 - 1000（Query+输出预留）= 3000 tokens 问题： - 50篇文档塞不进去 - 随机选会漏掉关键信息 - 每篇都压缩会丢失细节

上下文管理要解决三个问题：

选择：从50篇里选哪几篇？
压缩：如何在不丢失信息的前提下压缩？
排序：关键信息放在上下文的什么位置？

二、上下文选择策略

2.1 基于分数的选择

最简单：按检索分数选Top-K。

defselect_by_score(docs:list[dict],k:int=5)->list[dict]:"""按分数选择Top-K"""sorted_docs=sorted(docs,key=lambdax:x["score"],reverse=True)returnsorted_docs[:k]

问题：分数高的文档可能内容重复，浪费窗口。

2.2 基于多样性的选择（MMR）

Maximal Marginal Relevance：选既相关又多样的文档。

importnumpyasnpdefmmr_selection(docs:list[dict],query_vec:np.ndarray,doc_vecs:np.ndarray,k:int=5,lambda_param:float=0.7)->list[dict]:""" MMR选择 lambda_param: 相关性权重（1.0=只看相关性，0.0=只看多样性） 推荐：0.7（相关性和多样性的平衡） """selected=[]selected_indices=[]remaining=list(range(len(docs)))for_inrange(k):ifnotremaining:breakbest_score=-np.inf best_idx=Noneforidxinremaining:# 相关性：与Query的相似度relevance=np.dot(doc_vecs[idx],query_vec)/(np.linalg.norm(doc_vecs[idx])*np.linalg.norm(query_vec))# 多样性：与已选文档的最大相似度（越小越多样）ifselected_indices:max_similarity=max(np.dot(doc_vecs[idx],doc_vecs[s])/(np.linalg.norm(doc_vecs[idx])*np.linalg.norm(doc_vecs[s]))forsinselected_indices)else:max_similarity=0# MMR分数mmr_score=lambda_param*relevance-(1-lambda_param)*max_similarityifmmr_score>best_score:best_score=mmr_score best_idx=idxifbest_idxisnotNone:selected.append(docs[best_idx])selected_indices.append(best_idx)remaining.remove(best_idx)returnselected

2.3 基于覆盖度的选择

选择能覆盖最多查询关键词的文档组合。

defcoverage_selection(docs:list[dict],query:str,k:int=5)->list[dict]:"""基于关键词覆盖度选择"""# 提取查询关键词query_keywords=set(jieba.cut(query))selected=[]covered_keywords=set()whilelen(selected)<kanddocs:# 找能覆盖最多新关键词的文档best_doc=Nonebest_new_coverage=0fordocindocs:doc_keywords=set(jieba.cut(doc["content"]))new_coverage=len(doc_keywords&query_keywords-covered_keywords)ifnew_coverage>best_new_coverage:best_new_coverage=new_coverage best_doc=docifbest_doc:selected.append(best_doc)docs.remove(best_doc)# 更新已覆盖关键词covered_keywords|=set(jieba.cut(best_doc["content"]))&query_keywordselse:breakreturnselected

三、上下文压缩策略

3.1 摘要压缩

用LLM生成文档摘要。

classContextCompressor:"""上下文压缩器"""def__init__(self,llm):self.llm=llmdefcompress_doc(self,doc:str,query:str,max_length:int=200)->str:"""压缩单个文档"""iflen(doc)<=max_length:returndoc prompt=f""" 用户查询：{query}文档内容：{doc}请提取与用户查询最相关的内容，压缩到{max_length}字以内。 要求： 1. 保留关键信息 2. 保留具体数字、名称 3. 不要添加原文没有的信息 压缩结果： """response=self.llm.invoke(prompt)returnresponse.content.strip()[:max_length]defcompress_batch(self,docs:list[str],query:str,max_total_length:int=2000)->list[str]:"""批量压缩"""# 每个文档的预算长度budget_per_doc=max_total_length//len(docs)compressed=[]fordocindocs:compressed.append(self.compress_doc(doc,query,budget_per_doc))returncompressed

3.2 提取式压缩

提取关键句子，不重新生成。

classExtractiveCompressor:"""提取式压缩器"""def__init__(self,sentence_embedder):self.embedder=sentence_embedderdefcompress(self,doc:str,query:str,max_sentences:int=3)->str:"""提取关键句子"""# 分句sentences=[s.strip()forsindoc.split('。')ifs.strip()]iflen(sentences)<=max_sentences:returndoc# 计算每个句子与查询的相似度query_vec=self.embedder.embed(query)sentence_vecs=[self.embedder.embed(s)forsinsentences]scores=[np.dot(sv,query_vec)/(np.linalg.norm(sv)*np.linalg.norm(query_vec))forsvinsentence_vecs]# 选Top-K句子top_indices=np.argsort(scores)[::-1][:max_sentences]top_indices=sorted(top_indices)# 保持原文顺序selected=[sentences[i]foriintop_indices]return'。'.join(selected)+'。'

3.3 LLM自适应压缩

让LLM自己决定压缩策略。

classAdaptiveCompressor:"""自适应压缩器"""def__init__(self,llm):self.llm=llmdefcompress(self,docs:list[str],query:str,max_tokens:int=2000)->str:"""自适应压缩多文档"""# 合并文档all_text="\n\n---\n\n".join([f"文档{i+1}：{doc}"fori,docinenumerate(docs)])prompt=f""" 用户查询：{query}以下是与查询相关的多个文档片段：{all_text}请从中提取与查询最相关的信息，整合成一段连贯的文字。 要求： 1. 总长度不超过{max_tokens}字 2. 保留所有关键信息（数字、名称、结论） 3. 去除重复内容 4. 按逻辑组织，不要简单拼接 整合结果： """response=self.llm.invoke(prompt)returnresponse.content

四、上下文排序策略

研究表明，LLM对上下文不同位置的信息关注度不同。

4.1 Lost in the Middle问题

研究发现： - 开头的信息：关注度高 - 中间的信息：关注度低（Lost in the Middle） - 结尾的信息：关注度中等 建议：关键信息放在开头或结尾，不要埋在中间

4.2 实现策略

defreorder_context(docs:list[dict],strategy:str="relevance_first")->list[dict]:"""重新排序上下文"""ifstrategy=="relevance_first":# 相关性高的放前面returnsorted(docs,key=lambdax:x["score"],reverse=True)elifstrategy=="relevance_both_ends":# 相关性高的放两端sorted_docs=sorted(docs,key=lambdax:x["score"],reverse=True)n=len(sorted_docs)result=[]foriinrange(n):ifi%2==0:result.append(sorted_docs[i//2])else:result.append(sorted_docs[n-1-i//2])returnresultelifstrategy=="important_first":# 包含关键信息的放前面returnsorted(docs,key=lambdax:x.get("importance",0),reverse=True)returndocs

五、实测对比

5.1 测试设置

测试数据：-文档：每轮检索返回50篇-查询：100个测试查询-LLM：GPT-4-Turbo（4096tokens上下文）-评估：Answer Accuracy

5.2 选择策略对比

选择策略	Accuracy	平均Token数	说明
Top-K（k=5）	68.2%	1850	基线
MMR（λ=0.7）	72.1%	1920	+3.9%
覆盖度选择	70.5%	1780	+2.3%

5.3 压缩策略对比

压缩策略	Accuracy	Token消耗	信息保留
不压缩	68.2%	1850	100%
摘要压缩	65.3%	820	82%
提取式压缩	67.1%	950	89%
自适应压缩	69.8%	1100	91%

关键发现：

MMR选择效果最好，但Token略高
自适应压缩在压缩和信息保留之间平衡最好

六、完整方案集成

classContextManager:"""上下文管理器：选择+压缩+排序"""def__init__(self,llm,embedder,max_tokens:int=3000):self.llm=llm self.embedder=embedder self.max_tokens=max_tokensdefprocess(self,docs:list[dict],query:str)->str:"""处理上下文"""# 第一步：选择（MMR）selected=self._select(docs,query,k=10)# 第二步：压缩compressed=self._compress(selected,query)# 第三步：排序reordered=self._reorder(compressed)# 第四步：格式化context=self._format(reordered)returncontextdef_select(self,docs:list[dict],query:str,k:int)->list[dict]:"""MMR选择"""query_vec=self.embedder.embed(query)doc_vecs=[self.embedder.embed(d["content"])fordindocs]returnmmr_selection(docs,query_vec,np.array(doc_vecs),k=k)def_compress(self,docs:list[dict],query:str)->list[dict]:"""压缩"""# 计算每篇文档的预算budget=self.max_tokens//len(docs)compressor=ExtractiveCompressor(self.embedder)compressed=[]fordocindocs:iflen(doc["content"])>budget:doc["content"]=compressor.compress(doc["content"],query,max_sentences=3)compressed.append(doc)returncompresseddef_reorder(self,docs:list[dict])->list[dict]:"""排序：相关性高的放两端"""returnreorder_context(docs,strategy="relevance_both_ends")def_format(self,docs:list[dict])->str:"""格式化"""return"\n\n".join([f"[{i+1}]{doc['content']}"fori,docinenumerate(docs)])