5.3 上下文管理

5.3 上下文管理 — RAG知识库实战上下文窗口优化

本节导读：深入理解RAG系统中的上下文管理技术，掌握上下文窗口算术、Lost-in-the-Middle问题、上下文压缩策略等核心方法，学会如何高效利用有限的上下文空间最大化RAG回答质量。读者读完本节将能设计出生产级的上下文管理方案。

学习目标

• 理解上下文窗口的Token算术和长度预算分配方法
• 掌握Lost-in-the-Middle现象的成因和应对策略
• 学会使用摘要压缩、关键句提取、滑动窗口等上下文优化技术
• 能够实现动态上下文路由和分块策略
• 了解多轮对话中的上下文累积与衰减管理

核心概念

上下文窗口是RAG系统的物理瓶颈

无论你的检索引擎多强大，最终能喂给LLM的上下文受限于模型的上下文窗口大小。GPT-4 Turbo支持128K token，Claude 3支持200K，但这些数字需要被合理拆分：

可用上下文窗口 = 模型窗口上限 - 系统提示 - 历史对话 - 输出预留
​

举个实际例子：假设模型窗口128K token，系统提示占500 token，历史对话占2000 token，输出预留2000 token，那么留给检索上下文的空间是 128000 - 500 - 2000 - 2000 = 123,500 token。看似很多，但一篇中文技术文档平均约3000-5000 token，123,500 token大约能容纳25-40篇文档片段。如果检索返回了100个候选文档，你必须做出取舍。

这就是上下文管理的核心问题：在有限的窗口中，选择和排列哪些信息，才能最大化回答质量。

Lost-in-the-Middle：被忽视的关键发现

2023年Liu等人发表的论文揭示了一个重要现象：LLM对上下文中间位置的信息利用效率显著低于开头和结尾。这意味着如果你把最相关的文档放在上下文中间，模型可能"看不到"它。

这个发现对RAG系统的上下文排列策略有直接指导意义：最相关的文档应该放在上下文的开头和结尾，而不是简单地按检索分数从头排到尾。

环境准备 / 前置知识

import json
import re
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from collections import deque
import heapq
​

本节假设读者已理解本教程3.4节和4.2节中关于向量检索和检索策略的内容。上下文管理发生在检索之后、LLM调用之前。

分步实战

步骤1：Token算术与预算分配

1.1 上下文预算计算器

@dataclass
class ContextBudget:
    """上下文窗口预算分配"""
    model_context_window: int = 128000  # 模型总窗口（token数）
    system_prompt_tokens: int = 500
    output_reserve_tokens: int = 2000
    conversation_history_limit: int = 10  # 保留最近N轮对话
    
    def get_available_for_retrieval(self, history_tokens: int = 0) -> int:
        """计算可用于检索上下文的token数"""
        available = (
            self.model_context_window
            - self.system_prompt_tokens
            - self.output_reserve_tokens
            - history_tokens
        )
        return max(0, available)
    
    def estimate_chunk_capacity(self, avg_chunk_tokens: int = 500,
                                  history_tokens: int = 0) -> int:
        """估算可容纳的文档块数"""
        available = self.get_available_for_retrieval(history_tokens)
        return available // avg_chunk_tokens


# 使用示例
budget = ContextBudget(model_context_window=128000)
print(f"可用检索空间: {budget.get_available_for_retrieval(2000)} tokens")
print(f"预估可容纳: {budget.estimate_chunk_capacity(500, 2000)} 个文档块")
# 输出: 可用检索空间: 123500 tokens, 预估可容纳: 247 个文档块
​

1.2 中文Token粗估工具

大多数Token计算器（如tiktoken）对中文的tokenize结果不太精确，但粗估公式在工程实践中够用：

def estimate_chinese_tokens(text: str) -> int:
    """
    粗估中文文本的token数。
    经验公式：中文约1.5-2个字符/token，英文约0.25个词/token，
    代码和标点各算1个token。
    """
    chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', text))
    english_words = len(re.findall(r'[a-zA-Z]+', text))
    code_blocks = len(re.findall(r'```', text))  # 每对```约50 token
    
    tokens = (
        chinese_chars * 1.7      # 中文平均1.7 token/字
        + english_words * 1.3    # 英文平均1.3 token/词
        + code_blocks * 50       # 代码块开销
        + text.count('\n') * 1  # 换行
    )
    return int(tokens)
​

步骤2：上下文排序——对抗Lost-in-the-Middle

2.1 双端排列策略

根据Lost-in-the-Middle现象，我们将高分文档放在上下文的首尾，低分文档放在中间：

class BiDirectionalContextOrderer:
    """
    双端排列器：将高分文档放在首尾，低分文档放在中间，
    对抗Lost-in-the-Middle效应。
    """
    
    def __init__(self, max_chunks: int = 20):
        self.max_chunks = max_chunks
    
    def order(self, chunks_with_scores: List[Tuple[str, float]]) -> List[str]:
        """
        Args:
            chunks_with_scores: [(文档内容, 检索分数)] 按分数降序
        Returns:
            排列后的文档列表
        """
        # 截取前N个
        selected = chunks_with_scores[:self.max_chunks]
        n = len(selected)
        
        if n <= 2:
            return [c for c, s in selected]
        
        # 双端插入：高分放首尾，低分放中间
        ordered = [None] * n
        left, right = 0, n - 1
        
        for i, (chunk, score) in enumerate(selected):
            if i % 2 == 0:
                ordered[left] = chunk
                left += 1
            else:
                ordered[right] = chunk
                right -= 1
        
        return [c for c in ordered if c is not None]
    
    def format_context(self, ordered_chunks: List[str]) -> str:
        """格式化上下文，为每个文档块编号"""
        parts = []
        for i, chunk in enumerate(ordered_chunks, 1):
            parts.append(f"[文档片段{i}]\n{chunk}")
        return "\n\n".join(parts)
​

为什么不直接按分数排序？ 按分数排序意味着分数第5-15的文档落在中间区域，而恰恰是这些中等相关性的文档最容易被模型忽略。双端排列确保模型在注意力最集中的开头和结尾看到最重要的信息。

步骤3：上下文压缩策略

当检索到的文档总量超过上下文预算时，压缩是必要的。压缩的目标是：在减少Token数量的同时，尽可能保留与问题相关的信息。

3.1 关键句提取压缩

class KeySentenceCompressor:
    """基于句子重要性的上下文压缩器"""
    
    def __init__(self, query: str, compression_ratio: float = 0.5):
        self.query = query
        self.compression_ratio = compression_ratio
        self.query_keywords = set(re.findall(r'[\w]+', query.lower()))
    
    def compress(self, text: str) -> str:
        """压缩文本到指定比例"""
        sentences = self._split_sentences(text)
        
        # 为每个句子计算重要性分数
        scored_sentences = []
        for i, sent in enumerate(sentences):
            score = self._sentence_importance(sent, i, len(sentences))
            scored_sentences.append((sent, score))
        
        # 按重要性排序，保留前N个
        target_count = max(3, int(len(sentences) * self.compression_ratio))
        scored_sentences.sort(key=lambda x: x[1], reverse=True)
        selected = scored_sentences[:target_count]
        
        # 按原文顺序恢复
        selected.sort(key=lambda x: sentences.index(x[0]))
        
        return ' '.join(s for s, _ in selected)
    
    def _split_sentences(self, text: str) -> List[str]:
        """分割句子"""
        sentences = re.split(r'[。！？；\n]', text)
        return [s.strip() for s in sentences if s.strip()]
    
    def _sentence_importance(self, sentence: str, position: int,
                              total: int) -> float:
        """计算句子重要性分数"""
        sent_words = set(re.findall(r'[\w]+', sentence.lower()))
        
        # 关键词匹配度
        keyword_overlap = len(sent_words & self.query_keywords)
        keyword_score = keyword_overlap / max(len(self.query_keywords), 1)
        
        # 位置偏好（开头和结尾的句子更可能包含核心信息）
        position_score = 1.0 - abs(position - total/2) / (total/2)
        
        # 信息密度（包含数字和专有名词的句子信息密度更高）
        has_numbers = bool(re.search(r'\d+', sentence))
        density_score = 0.3 if has_numbers else 0.1
        
        # 句子长度适中（太短信息少，太长可能冗余）
        length = len(sentence)
        length_score = min(1.0, length / 50) * (1.0 if 10 < length < 200 else 0.5)
        
        return (keyword_score * 0.4 + position_score * 0.2 +
                density_score * 0.2 + length_score * 0.2)
​

3.2 分层上下文策略

不是所有检索到的文档都需要完整塞入上下文。分层策略的核心思想是：核心文档全文保留，辅助文档只保留摘要或关键段落。

class LayeredContextBuilder:
    """分层上下文构建器"""
    
    def __init__(self, budget: ContextBudget):
        self.budget = budget
        self.orderer = BiDirectionalContextOrderer()
    
    def build(self, query: str, retrieval_results: List[Dict],
              history_tokens: int = 0) -> str:
        """
        Args:
            query: 用户查询
            retrieval_results: [{
                'content': '...', 'score': 0.95,
                'metadata': {'source': '...', 'title': '...'}
            }]
            history_tokens: 对话历史占用的token数
        """
        available_tokens = self.budget.get_available_for_retrieval(history_tokens)
        
        # 第一层：高分文档（score > 0.8）全文保留
        # 第二层：中分文档（0.5 < score < 0.8）提取关键段落
        # 第三层：低分文档（score < 0.5）仅保留标题和摘要
        
        layer1 = [(r['content'], r['score'])
                  for r in retrieval_results if r['score'] > 0.8]
        layer2 = [(self._extract_key_paragraphs(r['content'], query), r['score'])
                  for r in retrieval_results if 0.5 < r['score'] <= 0.8]
        layer3 = [(f"[来源: {r['metadata'].get('title', '未知')}]", r['score'])
                  for r in retrieval_results if r['score'] <= 0.5]
        
        # 合并并排列
        all_chunks = layer1 + layer2 + layer3
        ordered = self.orderer.order(all_chunks)
        formatted = self.orderer.format_context(ordered)
        
        # 如果仍然超预算，进一步截断
        estimated_tokens = estimate_chinese_tokens(formatted)
        if estimated_tokens > available_tokens:
            formatted = self._truncate_to_budget(formatted, available_tokens)
        
        return formatted
    
    def _extract_key_paragraphs(self, text: str, query: str) -> str:
        """提取与查询最相关的段落"""
        compressor = KeySentenceCompressor(query, compression_ratio=0.4)
        return compressor.compress(text)
    
    def _truncate_to_budget(self, text: str, max_tokens: int) -> str:
        """按预算截断文本"""
        # 粗估：中文约1.7 token/字，取安全值2.0
        max_chars = max_tokens // 2
        if len(text) <= max_chars:
            return text
        
        # 按段落截断，避免截断句子中间
        paragraphs = text.split('\n\n')
        result = []
        current_len = 0
        
        for para in paragraphs:
            if current_len + len(para) > max_chars:
                break
            result.append(para)
            current_len += len(para)
        
        return '\n\n'.join(result) + '\n\n[上下文已截断，部分文档未完整展示]'
​

步骤4：多轮对话的上下文管理

多轮对话中，上下文会随着对话轮次增长而膨胀。如果不加管理，几轮对话后检索上下文就会被历史对话挤占。

class ConversationContextManager:
    """多轮对话上下文管理器"""
    
    def __init__(self, budget: ContextBudget, max_history_rounds: int = 5):
        self.budget = budget
        self.max_history_rounds = max_history_rounds
        self.history: deque = deque(maxlen=max_history_rounds)
        self.topic_summary: str = ""
    
    def add_exchange(self, user_msg: str, assistant_msg: str):
        """记录一轮对话"""
        self.history.append({
            'user': user_msg,
            'assistant': assistant_msg
        })
    
    def get_history_text(self) -> Tuple[str, int]:
        """获取对话历史文本和预估token数"""
        if not self.history:
            return "", 0
        
        lines = []
        for i, exchange in enumerate(self.history):
            lines.append(f"用户: {exchange['user']}")
            lines.append(f"助手: {exchange['assistant']}")
            lines.append("")
        
        text = "\n".join(lines)
        tokens = estimate_chinese_tokens(text)
        return text, tokens
    
    def build_multi_turn_context(self, query: str,
                                  retrieval_results: List[Dict]) -> str:
        """构建多轮对话的完整上下文"""
        history_text, history_tokens = self.get_history_text()
        
        # 构建分层上下文
        builder = LayeredContextBuilder(self.budget)
        retrieval_context = builder.build(query, retrieval_results, history_tokens)
        
        # 组装完整prompt
        full_context = f"""对话历史：
{history_text}

参考文档：
{retrieval_context}"""
        
        return full_context
​

关键设计决策：限制历史轮次而非历史长度。 按长度限制会导致旧对话被截断到一半，造成信息碎片化。按轮次限制则保留完整的对话轮次，语义更连贯。建议保留3-5轮，超过5轮的早期对话可以通过摘要压缩为一句话。

步骤5：对话主题漂移检测

在长对话中，用户的话题可能逐渐偏移。如果上下文管理仍然保留早期不相关的话题历史，不仅浪费Token，还可能误导LLM的回答方向。主题漂移检测通过比较当前查询与历史查询的词汇重叠度来判断话题是否发生了显著变化：

class TopicDriftDetector:
    """对话主题漂移检测器"""
    
    def __init__(self, drift_threshold: float = 0.3):
        self.drift_threshold = drift_threshold
    
    def detect_drift(self, current_query: str,
                      history_queries: List[str]) -> Dict:
        """检测话题是否发生了漂移"""
        if not history_queries:
            return {'drifted': False, 'score': 0.0}
        
        current_words = set(re.findall(r'[\w]+', current_query.lower()))
        
        # 计算与最近N轮对话的词汇重叠度
        overlap_scores = []
        for hist_query in history_queries[-3:]:
            hist_words = set(re.findall(r'[\w]+', hist_query.lower()))
            if current_words and hist_words:
                overlap = len(current_words & hist_words) / len(current_words | hist_words)
                overlap_scores.append(overlap)
        
        avg_overlap = sum(overlap_scores) / len(overlap_scores) if overlap_scores else 0
        
        return {
            'drifted': avg_overlap < self.drift_threshold,
            'score': avg_overlap,
            'suggestion': ('建议清空早期对话历史'
                           if avg_overlap < self.drift_threshold
                           else '话题连贯，继续保留历史')
        }
​

当检测到话题漂移时，建议清空对话历史或压缩为一句摘要，将Token空间让给新的检索上下文。这是很多生产RAG系统忽视的细节，但对长会话场景影响很大。举个例子：用户先问了5轮关于Redis的问题，然后突然切换到问Kafka，如果不检测漂移，Redis的对话历史会挤占Kafka相关文档的上下文空间，导致回答质量下降。

完整示例

端到端上下文管理Pipeline

class ContextManagementPipeline:
    """
    完整的上下文管理Pipeline：
    检索结果 → 分层选择 → 双端排列 → 压缩适配 → 输出
    """
    
    def __init__(self, model_context_window: int = 128000):
        self.budget = ContextBudget(model_context_window)
        self.orderer = BiDirectionalContextOrderer(max_chunks=20)
        self.compressor = KeySentenceCompressor
        self.builder = LayeredContextBuilder(self.budget)
        self.conversation_mgr = ConversationContextManager(self.budget)
    
    def process(self, query: str, retrieval_results: List[Dict],
               is_multi_turn: bool = False) -> Dict:
        """
        处理完整的上下文管理流程
        
        Returns:
            {
                'context': str,       # 最终上下文文本
                'metadata': {
                    'total_chunks': int,     # 检索到的总数
                    'selected_chunks': int,   # 选中进入上下文的数量
                    'estimated_tokens': int, # 预估token数
                    'compression_applied': bool, # 是否进行了压缩
                    'strategy': str           # 使用的策略
                }
            }
        """
        total_chunks = len(retrieval_results)
        
        if is_multi_turn:
            context = self.conversation_mgr.build_multi_turn_context(
                query, retrieval_results
            )
        else:
            context = self.builder.build(query, retrieval_results)
        
        estimated_tokens = estimate_chinese_tokens(context)
        
        return {
            'context': context,
            'metadata': {
                'total_chunks': total_chunks,
                'selected_chunks': min(total_chunks, self.orderer.max_chunks),
                'estimated_tokens': estimated_tokens,
                'budget_available': self.budget.get_available_for_retrieval(),
                'strategy': 'multi_turn_layered' if is_multi_turn else 'single_turn_layered'
            }
        }
​

常见问题 FAQ

Q1：上下文窗口到底该留多少给检索结果？

A：经验法则：将可用空间的70-80%留给检索上下文，20-30%留给对话历史。具体比例取决于应用场景：客服问答场景对话轮次多，历史可以占30%；文档分析场景基本是单轮查询，历史可以只留10%。关键是算清楚Token预算，不要靠感觉。

Q2：Lost-in-the-Middle问题在所有模型上都存在吗？

A：不是同等程度存在。研究表明，GPT-4和Claude 3系列受影响较小（但仍存在），而较小的开源模型（7B-13B）受影响更明显。如果你的目标模型是开源小模型，上下文排列策略就更加重要。一个简单的验证方法：把关键信息分别放在开头、中间、结尾测试三组，看回答质量差异。

Q3：摘要压缩和关键句提取哪种方法更好？

A：各有优劣。摘要压缩需要调用LLM（增加成本和延迟），但生成的摘要更自然、更连贯。关键句提取是纯规则方法，速度快、零成本，但输出可能不太通顺。我的建议：对于核心文档（高分文档）用摘要压缩保证质量，对于辅助文档（中低分）用关键句提取控制成本。

最佳实践与避坑

最佳实践

1. 先算预算再写代码：明确可用Token数，避免上下文溢出后才发现
2. 双端排列是免费的性能提升：不增加任何成本，只需改变排列顺序，就能利用LLM的开头-结尾注意力偏好
3. 分层而非一刀切：高分全文、中分提取、低分丢弃，比统一截断效果好
4. 监控Token使用率：定期检查上下文利用率，过低说明检索量不足，过高说明有溢出风险
5. 对话历史按轮次限制：保留完整对话轮次，避免截断造成信息碎片
6. 预留安全余量：实际Token消耗通常比预估多10-20%，建议保留15%的缓冲空间

常见陷阱

• 忽视系统提示的开销：复杂的系统提示可能占用500-1000 token，必须纳入预算计算
• 所有文档同等对待：把高分和低分文档混在一起全部塞入，稀释了关键信息
• 按原始顺序排列：检索引擎返回的顺序未必是LLM的最佳阅读顺序
• 过度压缩核心文档：为了塞更多文档而把最重要的文档压缩到面目全非

本节小结

本节深入探讨了RAG系统中的上下文管理技术，从Token算术到Lost-in-the-Middle现象，从分层压缩到话题漂移检测，构建了一套完整的上下文管理方法论。核心收获有三点：第一，上下文窗口是物理瓶颈，必须用Token算术做预算分配，不要靠感觉估算；第二，Lost-in-the-Middle现象要求我们对上下文排列做专门优化，双端排列是不增加任何成本就能获得的性能提升；第三，分层压缩策略（高分文档全文保留、中分文档提取关键段落、低分文档仅保留标题）比一刀切的统一截断更智能。此外，多轮对话场景中的话题漂移检测是容易被忽视但影响很大的细节。这些技术组合使用，能显著提升RAG系统在有限上下文窗口下的回答质量。下一节我们将讨论响应生成优化，关注LLM如何把精心管理的上下文转化为高质量的用户回答。

延伸阅读

• 官方文档：LangChain上下文管理模块文档 v0.3 版本（文字描述，不带链接）
• 相关论文：Lost in the Middle: How Language Models Use Long Contexts（文字描述，不带链接）
• 相关章节：本教程4.2节检索策略设计、5.2节提示工程实践（文字描述，不带链接）

关键词：RAG知识库实战, 上下文管理, Token算术, Lost-in-the-Middle, 上下文压缩, 分层策略
难度：进阶
预计阅读：25分钟