Kotaemon如何避免上下文截断？智能截取策略优化-开发者社区

Kotaemon如何避免上下文截断？智能截取策略优化

在构建现代智能对话系统时，一个看似不起眼却频频“背锅”的问题浮出水面：为什么AI明明看过文档，回答却像没看过一样？

答案往往藏在“上下文长度限制”这道无形的墙背后。大语言模型（LLM）虽然能写诗、编程、推理，但它的“注意力窗口”是有限的——无论是8K、32K还是最新的128K token，当面对冗长的对话历史或整篇PDF知识文档时，系统不得不做一件事：截断。

传统的做法简单粗暴：要么砍头，保留最近几轮对话；要么削尾，只留开头部分。结果就是，关键信息恰好被切掉，模型“失忆”，用户得到驴唇不对马嘴的回答。

Kotaemon 作为一款面向生产级部署的检索增强生成（RAG）与智能代理框架，没有选择向这一瓶颈低头。它引入了一套语义感知的智能上下文截取机制，让每一次截断都“有理有据”，最大程度保留真正重要的内容。

我们不妨设想这样一个场景：一位客户连续咨询了产品价格、保修政策、发票开具方式，最后问：“之前说的退货材料清单，能再发一遍吗？”
如果系统只是机械地保留最后三轮对话，很可能把三天前提到的“需提供订单截图和开箱视频”这类关键信息无情丢弃。而 Kotaemon 的做法完全不同。

它会先将整个对话流拆解成一个个逻辑单元——比如每一轮对话作为一个“块”（chunk）。然后，不是按时间顺序硬性取舍，而是给每个块打分。怎么打？看几个维度：

相关性：这个片段和当前问题在语义上有多接近？用嵌入向量计算余弦相似度，哪怕没出现相同词汇，只要意思相近也能被识别。
时效性：越靠近当前提问的内容，权重越高。“上次”“刚才”这类指代词触发的时间敏感信号会被放大。
角色重要性：用户主动提出的问题、确认类语句（如“我同意”“请帮我操作”）优先级高于系统通用回复。
关键词密度：是否包含“退货”“材料”“清单”等核心术语？命中越多，加分越多。

这些分数加权融合后，所有片段按得分排序。系统从高到低挑选，直到总token数逼近预设上限（通常预留20%空间给prompt模板和生成输出）。选中的片段再按原始时间顺序拼接，送入LLM。

这样做的好处显而易见：既避免了语义断裂，又确保最关键的信息始终在场。

from sentence_transformers import SentenceTransformer import numpy as np from typing import List, Dict class SmartContextTrimmer: def __init__(self, max_tokens: int = 8192, model_name: str = "all-MiniLM-L6-v2"): self.max_tokens = max_tokens self.encoder = SentenceTransformer(model_name) self.token_estimator = lambda text: len(text.split()) * 1.3 # 粗略估算token数 def score_chunks(self, chunks: List[str], query: str) -> List[Dict]: embeddings = self.encoder.encode(chunks + [query]) query_emb = embeddings[-1] chunk_embs = embeddings[:-1] # 计算语义相关性（余弦相似度） similarities = np.dot(chunk_embs, query_emb) / ( np.linalg.norm(chunk_embs, axis=1) * np.linalg.norm(query_emb) ) scored_chunks = [] for i, chunk in enumerate(chunks): tokens = self.token_estimator(chunk) relevance = float(similarities[i]) recency = 1.0 if i == len(chunks) - 1 else 0.8 # 最后一条对话更受重视 keyword_bonus = 1.2 if any(kw in chunk.lower() for kw in query.split()) else 1.0 final_score = relevance * recency * keyword_bonus scored_chunks.append({ "text": chunk, "score": final_score, "tokens": tokens, "index": i }) return sorted(scored_chunks, key=lambda x: x["score"], reverse=True) def trim(self, chunks: List[str], query: str) -> str: scored = self.score_chunks(chunks, query) selected = [] total_tokens = 0 for item in scored: if total_tokens + item["tokens"] <= self.max_tokens * 0.9: # 预留10%给prompt和生成 selected.append((item["index"], item["text"])) total_tokens += item["tokens"] # 按原始顺序恢复 selected.sort(key=lambda x: x[0]) return "\n".join([text for _, text in selected]) # 使用示例 if __name__ == "__main__": trimmer = SmartContextTrimmer(max_tokens=8192) dialog_history = [ "用户：你们公司的退货政策是什么？", "客服：我们支持7天无理由退货。", "用户：如果商品已经使用过了呢？", "客服：已使用的商品不支持无理由退货，但若存在质量问题可申请售后。", "用户：明白了，谢谢。", "用户：另外，发票怎么开？" ] current_query = "发票开具流程是怎样的？" optimized_context = trimmer.trim(dialog_history, current_query) print("优化后上下文：") print(optimized_context)

这段代码展示了一个轻量级实现的核心逻辑。实际在 Kotaemon 中，这套机制更加灵活：评分引擎支持插件化扩展，开发者可以加入情感分析、实体识别、合规过滤等维度；同时支持缓存预编码结果，减少重复计算开销，保证端到端响应延迟可控。

但这还没完。真正的挑战往往出现在长期多轮对话中。

试想一个技术支持会话持续了几十轮，涉及多个问题切换、反复确认、跳转话题。如果每次都保留全部原始记录，很快就会超出上下文极限。Kotaemon 的应对策略是“渐进式压缩”——不是简单删除，而是智能归档。

它的对话管理器内置了一个摘要模块。当历史轮次超过阈值（例如6轮），最早的几轮会被自动提炼成一句简洁陈述：

[归档] 用户咨询了产品A的价格与保修期，被告知价格为¥2999，保修两年。

这种摘要不是随便生成的。它聚焦于用户意图和系统回应的关键点，舍弃寒暄和重复表达。更重要的是，系统还会标记“关键事件”：比如用户提供了手机号、确认购买意向、发起投诉等行为对应的原始语句会被强制保留在上下文中，确保后续流程不会丢失上下文锚点。

class DialogueSummarizer: def __init__(self): self.summary_prompt = """ 请将以下对话内容总结为一句简洁陈述，突出用户关注点和系统回应要点： {dialogue} 总结： """ def generate_summary(self, dialogue: List[str]) -> str: full_text = "\n".join(dialogue) # 此处可接入LLM API 或本地小模型进行摘要生成 # 示例简化处理： return f"[归档] 用户咨询商品信息，系统回复了价格与保修政策。" class DialogueManager: def __init__(self, max_raw_turns: int = 6): self.history = [] self.archived_summary = "" self.summarizer = DialogueSummarizer() self.max_raw_turns = max_raw_turns def add_turn(self, speaker: str, text: str): self.history.append(f"{speaker}：{text}") # 超出长度时触发归档 if len(self.history) > self.max_raw_turns: old_portion = self.history[:len(self.history) - self.max_raw_turns + 1] self.archived_summary += self.summarizer.generate_summary(old_portion) self.history = self.history[-self.max_raw_turns:] def get_context(self) -> str: context_parts = [] if self.archived_summary: context_parts.append(self.archived_summary) context_parts.extend(self.history) return "\n".join(context_parts) # 使用示例 dm = DialogueManager(max_raw_turns=4) dm.add_turn("用户", "我想了解一下你们的会员制度") dm.add_turn("客服", "我们提供金卡和银卡两种会员...") dm.add_turn("用户", "金卡有什么权益？") dm.add_turn("客服", "金卡享受免运费、专属折扣等...") dm.add_turn("用户", "那积分怎么兑换？") print(dm.get_context())

这样的设计实现了“记忆的层次化”：近期细节完整保留，远期背景以摘要形式沉淀。就像人类大脑一样，在不丢失主线的前提下释放认知资源。

在 Kotaemon 的整体架构中，这套智能截取机制位于检索之后、生成之前，扮演着“信息守门人”的角色：

[用户输入] ↓ [意图识别 & 状态追踪] ↓ [知识检索 | 工具调用] → 返回候选文档/API结果 ↓ [智能上下文构建器] ├─ 分块处理 ├─ 多维度评分 ├─ 动态选择 └─ 上下文重组 ↓ [LLM Prompt 组装] → 注入优化后上下文 ↓ [大语言模型生成] ↓ [格式化输出]

它不仅处理对话历史，还能统一整合来自外部知识库的检索结果、工具调用返回的数据结构化信息，形成一个高度浓缩、语义连贯的上下文视图。

工程实践中，我们也总结了一些关键经验：