5.3 上下文管理

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

本节导读：深入理解RAG系统中的上下文管理技术，从基础的上下文窗口管理到复杂的上下文优化策略，掌握如何高效利用有限的上下文空间提升RAG效果。

学习目标

• 理解上下文在RAG系统中的核心作用
• 掌握上下文窗口管理的技术和方法
• 学会优化上下文结构和内容选择
• 能够处理长文档和复杂上下文的挑战
• 了解上下文质量评估和改进方法

核心概念

上下文管理是RAG系统的关键技术环节，直接影响检索到的知识能否被有效利用和LLM的响应质量。

上下文在RAG中的重要性

上下文管理核心维度

环境准备 / 前置知识

基础依赖

import json
import re
from typing import Dict, List, Optional, Tuple, Union
from dataclasses import dataclass
from enum import Enum
import numpy as np
from collections import deque
import heapq
​

上下文管理基础配置

# 上下文管理配置
CONTEXT_CONFIG = {
    'max_context_length': 4000,  # 最大上下文长度
    'chunk_size': 500,           # 分块大小
    'overlap_size': 50,         # 分块重叠大小
    'max_chunks': 10,           # 最大分块数量
    'compression_ratio': 0.7,   # 压缩比例
    'relevance_threshold': 0.6,  # 相关性阈值
    'enable_cache': True,      # 启用缓存
    'cache_size': 1000,         # 缓存大小
}

# 优先级配置
PRIORITY_CONFIG = {
    'document_relevance': 0.4,    # 文档相关性权重
    'position_weight': 0.3,       # 位置权重
    'recency_weight': 0.2,       # 新鲜度权重
    'importance_weight': 0.1,     # 重要性权重
}

# 上下文格式模板
CONTEXT_TEMPLATES = {
    'simple': """
上下文信息：
{context}

问题：{question}
""",
    
    'structured': """
【文档上下文】
{context}

【用户问题】
{question}

【回答要求】
1. 基于提供的上下文回答
2. 答案要准确、有条理
3. 重点突出关键信息
4. 如果信息不足，明确说明

【回答】
""",
    
    'detailed': """
=== 上下文信息 ===

【主要文档】
{main_context}

【辅助信息】
{auxiliary_context}

【相关历史】
{history_context}

=== 当前问题 ===
{question}

=== 回答指导 ===
- 基于上述上下文回答问题
- 优先参考主要文档信息
- 适当引用辅助信息补充
- 考虑历史对话的连续性
"""
}
​

分步实战

步骤1：基础上下文管理

1.1 上下文长度控制

class ContextLengthManager:
    """
    上下文长度管理器
    """
    
    def __init__(self, config: Dict = None):
        self.config = config or CONTEXT_CONFIG
        self.cache = {}
        
    def fit_context(self, context: str, max_length: int = None) -> str:
        """
        适配上下文长度
        """
        max_length = max_length or self.config['max_context_length']
        
        # 如果长度合适，直接返回
        if len(context) <= max_length:
            return context
            
        # 否则进行截断或压缩
        return self._compress_context(context, max_length)
    
    def _compress_context(self, context: str, target_length: int) -> str:
        """
        压缩上下文到指定长度
        """
        # 方法1：智能截断（保留前面）
        truncated = context[:target_length]
        
        # 方法2：基于重要性的截断（如果可用重要性信息）
        # 这里可以实现更复杂的压缩策略
        
        return truncated
    
    def chunk_context(self, context: str, chunk_size: int = None, 
                     overlap: int = None) -> List[str]:
        """
        将长上下文分块
        """
        chunk_size = chunk_size or self.config['chunk_size']
        overlap = overlap or self.config['overlap_size']
        
        chunks = []
        start = 0
        
        while start < len(context):
            end = start + chunk_size
            chunk = context[start:end]
            chunks.append(chunk)
            
            # 移动到下一个块，考虑重叠
            start = end - overlap
            
            # 防止无限循环
            if start >= len(context):
                break
                
        return chunks[:self.config['max_chunks']]  # 限制最大块数
​

1.2 上下文重排序

class ContextReRanker:
    """
    上下文重排序器
    """
    
    def __init__(self, config: Dict = None):
        self.config = config or PRIORITY_CONFIG
        
    def rank_contexts(self, contexts: List[str], query: str, 
                     scores: List[float] = None) -> List[Tuple[str, float]]:
        """
        对上下文进行重排序
        """
        if scores is None:
            scores = self._calculate_scores(contexts, query)
            
        # 将上下文和分数配对
        context_score_pairs = list(zip(contexts, scores))
        
        # 按分数降序排序
        sorted_pairs = sorted(context_score_pairs, key=lambda x: x[1], reverse=True)
        
        return sorted_pairs
    
    def _calculate_scores(self, contexts: List[str], query: str) -> List[float]:
        """
        计算上下文分数
        """
        scores = []
        query_words = set(query.lower().split())
        
        for i, context in enumerate(contexts):
            # 计算相关性分数
            context_words = set(context.lower().split())
            relevance_score = len(query_words & context_words) / len(query_words)
            
            # 位置分数（越靠前分数越高）
            position_score = 1.0 / (i + 1)
            
            # 长度分数（适中的长度得分更高）
            length_score = min(1.0, len(context) / 500)
            
            # 综合分数
            total_score = (
                relevance_score * self.config['document_relevance'] +
                position_score * self.config['position_weight'] +
                length_score * self.config['importance_weight']
            )
            
            scores.append(total_score)
            
        return scores
​

步骤2：高级上下文管理

2.1 智能上下文构建

class IntelligentContextBuilder:
    """
    智能上下文构建器
    """
    
    def __init__(self, config: Dict = None):
        self.config = config or CONTEXT_CONFIG
        self.length_manager = ContextLengthManager(config)
        self.reranker = ContextReRanker(config)
        
    def build_optimal_context(self, contexts: List[str], query: str, 
                             history: str = "") -> str:
        """
        构建最优上下文
        """
        # 1. 重排序上下文
        ranked_contexts = self.reranker.rank_contexts(contexts, query)
        
        # 2. 分块处理长上下文
        processed_chunks = []
        for context, _ in ranked_contexts:
            chunks = self.length_manager.chunk_context(context)
            processed_chunks.extend(chunks)
        
        # 3. 添加历史上下文
        if history:
            processed_chunks.append(history)
        
        # 4. 选择最重要的块
        selected_chunks = self._select_important_chunks(
            processed_chunks, query, self.config['max_context_length']
        )
        
        # 5. 构建最终上下文
        final_context = self._build_final_context(selected_chunks)
        
        return final_context
    
    def _select_important_chunks(self, chunks: List[str], query: str, 
                                max_length: int) -> List[str]:
        """
        选择最重要的上下文块
        """
        # 计算每个块的重要性
        chunk_scores = []
        query_words = set(query.lower().split())
        
        for i, chunk in enumerate(chunks):
            chunk_words = set(chunk.lower().split())
            relevance = len(query_words & chunk_words) / len(query_words)
            
            # 综合考虑相关性和位置
            score = relevance * (1.0 / (i + 1))
            chunk_scores.append((chunk, score))
        
        # 按分数排序并选择最重要的块
        chunk_scores.sort(key=lambda x: x[1], reverse=True)
        
        # 选择直到达到长度限制
        selected = []
        current_length = 0
        
        for chunk, score in chunk_scores:
            if current_length + len(chunk) <= max_length:
                selected.append(chunk)
                current_length += len(chunk)
            else:
                break
                
        return selected
    
    def _build_final_context(self, chunks: List[str]) -> str:
        """
        构建最终上下文
        """
        if not chunks:
            return ""
            
        # 按重要性排序
        chunks.sort(key=len, reverse=True)  # 长的在前
        
        # 构建结构化上下文
        context_parts = []
        
        for i, chunk in enumerate(chunks[:5]):  # 最多5个主要部分
            context_parts.append(f"""
【文档片段 {i+1}】
{chunk}
""")
        
        return "\n".join(context_parts)
​

2.2 动态上下文管理

class DynamicContextManager:
    """
    动态上下文管理器
    """
    
    def __init__(self, config: Dict = None):
        self.config = config or CONTEXT_CONFIG
        self.context_history = deque(maxlen=10)  # 保留最近10次上下文
        self.user_preferences = {}
        
    def manage_conversation_context(self, query: str, retrieved_contexts: List[str],
                                  session_id: str = "default") -> str:
        """
        管理对话上下文
        """
        # 获取用户历史偏好
        preferences = self.user_preferences.get(session_id, {})
        
        # 构建当前上下文
        current_context = self._build_conversation_context(
            query, retrieved_contexts, preferences
        )
        
        # 添加到历史
        self.context_history.append({
            'query': query,
            'context': current_context,
            'timestamp': time.time()
        })
        
        # 更新用户偏好
        self._update_user_preferences(session_id, query, current_context)
        
        return current_context
    
    def _build_conversation_context(self, query: str, 
                                  contexts: List[str], 
                                  preferences: Dict) -> str:
        """
        构建对话上下文
        """
        # 基于偏好选择上下文风格
        style = preferences.get('style', 'structured')
        
        # 构建上下文
        if style == 'detailed':
            context = self._build_detailed_context(query, contexts)
        elif style == 'simple':
            context = self._build_simple_context(query, contexts)
        else:
            context = self._build_structured_context(query, contexts)
        
        # 应用长度限制
        if len(context) > self.config['max_context_length']:
            context = self.length_manager.fit_context(context)
            
        return context
    
    def _build_detailed_context(self, query: str, contexts: List[str]) -> str:
        """
        构建详细上下文
        """
        # 选择最相关的上下文
        ranked = self.reranker.rank_contexts(contexts, query)
        main_context = ranked[0][0] if ranked else ""
        auxiliary_contexts = [c for c, s in ranked[1:3]]  # 取前3个
        
        # 添加历史上下文
        recent_history = self._get_recent_history()
        
        # 按模板格式化
        template = CONTEXT_TEMPLATES['detailed']
        return template.format(
            main_context=main_context,
            auxiliary_context="\n\n".join(auxiliary_contexts),
            history_context=recent_history,
            question=query
        )
    
    def _build_structured_context(self, query: str, contexts: List[str]) -> str:
        """
        构建结构化上下文
        """
        # 选择最相关的上下文
        ranked = self.reranker.rank_contexts(contexts, query)
        main_context = "\n\n".join([c for c, s in ranked[:2]])  # 取前2个
        
        # 按模板格式化
        template = CONTEXT_TEMPLATES['structured']
        return template.format(
            context=main_context,
            question=query
        )
    
    def _build_simple_context(self, query: str, contexts: List[str]) -> str:
        """
        构建简单上下文
        """
        # 取最相关的上下文
        ranked = self.reranker.rank_contexts(contexts, query)
        main_context = ranked[0][0] if ranked else ""
        
        # 按模板格式化
        template = CONTEXT_TEMPLATES['simple']
        return template.format(
            context=main_context,
            question=query
        )
    
    def _update_user_preferences(self, session_id: str, query: str, context: str):
        """
        更新用户偏好
        """
        if session_id not in self.user_preferences:
            self.user_preferences[session_id] = {
                'style': 'structured',
                'context_length_preference': 'medium'
            }
        
        # 分析查询类型
        if '代码' in query or '实现' in query:
            self.user_preferences[session_id]['style'] = 'detailed'
        elif '简单' in query or '快速' in query:
            self.user_preferences[session_id]['style'] = 'simple'
            
        # 记录上下文长度偏好
        if len(context) > self.config['max_context_length'] * 0.8:
            self.user_preferences[session_id]['context_length_preference'] = 'long'
        elif len(context) < self.config['max_context_length'] * 0.3:
            self.user_preferences[session_id]['context_length_preference'] = 'short'
    
    def _get_recent_history(self) -> str:
        """
        获取最近的对话历史
        """
        recent = list(self.context_history)[-3:]  # 最近3次
        history_text = ""
        
        for i, item in enumerate(recent):
            history_text += f"""
历史对话 {i+1}:
问题：{item['query']}
上下文：{item['context'][:200]}...
"""
        
        return history_text
​

完整示例

完整的上下文管理系统

class CompleteContextSystem:
    """
    完整的上下文管理系统
    """
    
    def __init__(self, config: Dict = None):
        self.config = config or CONTEXT_CONFIG
        self.length_manager = ContextLengthManager(config)
        self.reranker = ContextReRanker(config)
        self.intelligent_builder = IntelligentContextBuilder(config)
        self.dynamic_manager = DynamicContextManager(config)
        
    def process_context(self, contexts: List[str], query: str, 
                       use_dynamic: bool = False, session_id: str = "default",
                       history: str = "") -> Dict:
        """
        处理上下文的完整流程
        """
        result = {
            'original_contexts': contexts,
            'processed_context': None,
            'context_score': 0,
            'compression_ratio': 1.0,
            'processing_method': '',
            'metadata': {}
        }
        
        if use_dynamic:
            # 使用动态上下文管理
            processed = self.dynamic_manager.manage_conversation_context(
                query, contexts, session_id
            )
            result['processing_method'] = 'dynamic'
            result['metadata']['session_id'] = session_id
        else:
            # 使用智能上下文构建
            processed = self.intelligent_builder.build_optimal_context(
                contexts, query, history
            )
            result['processing_method'] = 'intelligent'
            result['metadata']['history_used'] = bool(history)
        
        # 计算上下文分数
        result['context_score'] = self._evaluate_context_quality(
            processed, query, contexts
        )
        
        # 计算压缩比例
        original_total = sum(len(c) for c in contexts)
        result['compression_ratio'] = len(processed) / original_total if original_total > 0 else 1.0
        
        result['processed_context'] = processed
        
        return result
    
    def _evaluate_context_quality(self, processed: str, query: str, 
                               original: List[str]) -> float:
        """
        评估上下文质量
        """
        # 相关性分数
        query_words = set(query.lower().split())
        processed_words = set(processed.lower().split())
        relevance = len(query_words & processed_words) / len(query_words) if query_words else 0.5
        
        # 完整性分数（覆盖了多少原始内容）
        original_words = set()
        for context in original:
            original_words.update(context.lower().split())
        
        coverage = len(original_words & processed_words) / len(original_words) if original_words else 0.5
        
        # 长度合理性分数（既不太长也不太短）
        ideal_length = self.config['max_context_length'] * 0.7
        length_score = 1.0 - abs(len(processed) - ideal_length) / ideal_length
        length_score = max(0, min(1, length_score))
        
        # 综合分数
        total_score = (relevance * 0.4 + coverage * 0.3 + length_score * 0.3)
        
        return total_score
    
    def get_context_statistics(self, processed_context: str) -> Dict:
        """
        获取上下文统计信息
        """
        return {
            'length': len(processed_context),
            'word_count': len(processed_context.split()),
            'paragraph_count': len(processed_context.split('\n\n')),
            'character_ratio': len(processed_context) / self.config['max_context_length'],
            'has_images': '```' in processed_context,  # 检查是否包含图片
            'has_code': '```python' in processed_context,  # 检查是否包含代码
            'complexity_score': self._calculate_complexity(processed_context)
        }
    
    def _calculate_complexity(self, context: str) -> float:
        """
        计算上下文的复杂度
        """
        # 基于句子长度和词汇复杂度计算
        sentences = context.split('.')
        avg_sentence_length = len(context.split()) / len(sentences) if sentences else 0
        
        # 复杂词汇比例
        words = context.split()
        complex_words = len([w for w in words if len(w) > 8])
        complex_ratio = complex_words / len(words) if words else 0
        
        return (avg_sentence_length / 30 + complex_ratio) / 2  # 归一化到0-1
​

常见问题 FAQ

Q1：如何处理超长上下文（超过模型窗口限制）？

A：处理超长上下文的策略：

1. 智能截断：保留最相关的部分，舍弃次要内容
2. 上下文压缩：使用摘要技术压缩冗长内容
3. 分层处理：将上下文分为核心和辅助部分
4. 渐进式上下文：先提供概览，再逐步深入细节
5. 缓存策略：缓存重要的历史上下文

Q2：如何确保上下文中的关键信息不被遗漏？

A：确保关键信息保留的方法：

1. 关键词识别：识别查询中的关键词并优先保留相关内容
2. 重要性评分：为不同内容块分配重要性分数
3. 多重备份：在多个上下文块中重复关键信息
4. 位置优化：将关键信息放在上下文的前部
5. 用户偏好学习：基于用户反馈学习重要性模式

Q3：如何平衡上下文的完整性和长度限制？

A：平衡完整性和长度的策略：

1. 动态调整：根据查询复杂度动态调整上下文长度
2. 优先级排序：按重要性对内容进行排序
3. 渐进式展示：先展示核心信息，提供深入选项
4. 压缩算法：使用智能压缩减少信息损失
5. 多版本上下文：提供不同详细程度的上下文版本

最佳实践与避坑

最佳实践

1. 用户个性化：根据用户偏好调整上下文风格
2. 动态优化：持续学习和改进上下文管理策略
3. 监控反馈：监控上下文质量和用户满意度
4. 版本控制：维护上下文管理系统的版本历史
5. A/B测试：比较不同上下文策略的效果

常见陷阱

• 过度压缩：压缩过度导致重要信息丢失
• 静态管理：不考虑用户的动态需求变化
• 忽视上下文质量：只关注长度而忽略内容质量
• 缺乏测试：没有充分的上下文质量测试
• 个性化过度：过度个性化导致上下文僵化

本节小结

本节深入讲解了RAG系统中的上下文管理技术：

1. 基础概念：理解上下文在RAG系统中的核心作用和管理维度
2. 技术实现：掌握了上下文长度控制、重排序等核心技术
3. 智能构建：学习了如何构建最优上下文的策略和方法
4. 动态管理：掌握了对话上下文的管理和个性化技术
5. 质量评估：学会了如何评估和优化上下文质量

上下文管理是RAG系统成功的关键，需要结合具体应用场景进行精细化设计。下一节我们将探讨响应生成优化的技术。

延伸阅读

• 官方文档：OpenAI上下文窗口管理指南（文字描述，不带链接）
• 相关章节：本教程5.4节响应生成优化（文字描述，不带链接）

关键词：RAG知识库实战, 上下文管理, 上下文窗口, 长文档处理, 上下文优化, 上下文压缩
难度：进阶
预计阅读：50分钟