2.3 上下文窗口处理 — 大模型应用开发从零到一 关键词短语

本节导读：深入理解大模型的上下文窗口机制，掌握长文本处理和上下文管理的高级技术，构建高效的知识管理和对话系统。

学习目标

• 理解上下文窗口的基本概念和重要性
• 掌握不同模型的上下文窗口限制
• 学会长文本的分割和处理策略
• 掌握上下文连贯性的维护方法
• 了解跨文档上下文管理的最佳实践

核心概念

什么是上下文窗口

上下文窗口是语言模型能够同时处理的最大Token数量。它决定了模型在生成回应时能够"记住"多少信息：

主流模型的上下文窗口对比：

• OpenAI GPT-4: 128K tokens (约96K英文字符)
• OpenAI GPT-4 Turbo: 128K tokens (约96K英文字符)
• OpenAI GPT-3.5: 16K tokens (约12K英文字符)
• Claude-3: 200K tokens (约160K英文字符)
• Claude-3.5: 200K tokens (约160K英文字符)

上下文窗口的重要性

上下文窗口直接影响AI应用的质量和功能：

1. 知识连贯性：保持多轮对话的逻辑连贯性
2. 长文档理解：处理和分析长篇文档内容
3. 复杂任务处理：需要大量背景信息的复杂任务
4. 多模态交互：同时处理文本、图像、代码等多种信息

环境准备 / 前置知识

基础配置

# 上下文管理工具安装
pip install tiktoken langchain openai

# 环境检查
import tiktoken
import openai
import langchain

print("上下文管理工具配置完成")
​

模型上下文信息查询

def get_context_window_info(model_name: str) -> dict:
    """获取模型上下文窗口信息"""
    context_info = {
        "gpt-4": {"max_tokens": 128000, "encoding": "cl100k_base"},
        "gpt-4-turbo": {"max_tokens": 128000, "encoding": "cl100k_base"},
        "gpt-3.5-turbo": {"max_tokens": 16384, "encoding": "cl100k_base"},
        "claude-3": {"max_tokens": 200000, "encoding": "p50k_base"},
        "claude-3-5": {"max_tokens": 200000, "encoding": "p50k_base"},
    }
    
    return context_info.get(model_name, {"max_tokens": 8192, "encoding": "cl100k_base"})

# 查询当前模型信息
model_info = get_context_window_info("gpt-4")
print(f"GPT-4上下文窗口: {model_info['max_tokens']} tokens")
print(f"编码方式: {model_info['encoding']}")
​

分步实战

步骤 1：上下文窗口管理基础

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
上下文窗口管理基础工具
"""

import tiktoken
from typing import Dict, List, Any, Optional
from dataclasses import dataclass

@dataclass
class ContextInfo:
    """上下文信息"""
    max_tokens: int
    used_tokens: int
    remaining_tokens: int
    usage_percent: float
    
    def is_over_limit(self) -> bool:
        return self.used_tokens >= self.max_tokens
    
    def is_near_limit(self, threshold: float = 0.9) -> bool:
        return self.usage_percent >= threshold

class ContextManager:
    """基础上下文管理器"""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.encoding = self._get_encoding()
        self.max_context = self._get_max_context()
        
    def _get_encoding(self) -> tiktoken.Encoding:
        """获取模型编码"""
        model_mappings = {
            "gpt-4": "cl100k_base",
            "gpt-4-turbo": "cl100k_base",
            "gpt-3.5-turbo": "cl100k_base",
            "claude-3": "p50k_base",
            "claude-3.5": "p50k_base",
        }
        encoding_name = model_mappings.get(self.model_name, "cl100k_base")
        return tiktoken.get_encoding(encoding_name)
    
    def _get_max_context(self) -> int:
        """获取最大上下文长度"""
        context_limits = {
            "gpt-4": 128000,
            "gpt-4-turbo": 128000,
            "gpt-3.5-turbo": 16384,
            "claude-3": 200000,
            "claude-3.5": 200000,
        }
        return context_limits.get(self.model_name, 8192)
    
    def count_tokens(self, text: str) -> int:
        """计算Token数量"""
        return len(self.encoding.encode(text)) if text else 0
    
    def analyze_context(self, texts: List[str], reserved_space: int = 1000) -> ContextInfo:
        """
        分析上下文使用情况
        
        Args:
            texts: 文本列表
            reserved_space: 预留空间用于响应
            
        Returns:
            上下文信息
        """
        total_tokens = sum(self.count_tokens(text) for text in texts)
        used_with_reserved = total_tokens + reserved_space
        
        return ContextInfo(
            max_tokens=self.max_context,
            used_tokens=total_tokens,
            remaining_tokens=self.max_context - total_tokens,
            usage_percent=total_tokens / self.max_context
        )
    
    def check_context_usage(self, texts: List[str], reserved_space: int = 1000) -> Dict[str, Any]:
        """
        检查上下文使用情况
        
        Returns:
            使用情况报告
        """
        context_info = self.analyze_context(texts, reserved_space)
        
        status = "正常"
        recommendations = []
        
        if context_info.is_over_limit():
            status = "超出限制"
            recommendations.append("需要压缩文本或分段处理")
        elif context_info.is_near_limit():
            status = "接近限制"
            recommendations.append("考虑压缩文本或预留更多空间")
        elif context_info.usage_percent > 0.8:
            recommendations.append("使用率较高，建议优化")
        
        return {
            "status": status,
            "max_tokens": context_info.max_tokens,
            "used_tokens": context_info.used_tokens,
            "remaining_tokens": context_info.remaining_tokens,
            "usage_percent": context_info.usage_percent,
            "recommendations": recommendations
        }

# 使用示例
def basic_context_example():
    """基础上下文管理示例"""
    manager = ContextManager("gpt-4")
    
    test_texts = [
        "这是一段短文本",
        "这是一段中等长度的文本，包含了更多的信息和内容。",
        "这是一段非常长的文本，包含了很多详细的信息和解释，可能会接近或超过上下文窗口的限制。"
    ]
    
    print("=== 基础上下文管理测试 ===")
    
    for i, text in enumerate(test_texts, 1):
        print(f"\n文本 {i}:")
        print(f"内容: {text[:50]}{'...' if len(text) > 50 else ''}")
        
        analysis = manager.analyze_context([text])
        print(f"Token数量: {analysis.used_tokens}")
        print(f"使用率: {analysis.usage_percent:.2%}")
        
        check = manager.check_context_usage([text])
        print(f"状态: {check['status']}")

if __name__ == "__main__":
    basic_context_example()
​

步骤 2：长文本分割策略

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
长文本分割策略
"""

from typing import List, Dict, Any, Tuple
from dataclasses import dataclass
import tiktoken

@dataclass
class TextSegment:
    """文本片段"""
    id: int
    content: str
    token_count: int
    overlap_tokens: int = 0
    
    def get_overlap_ratio(self, total_tokens: int) -> float:
        return self.overlap_tokens / total_tokens if total_tokens > 0 else 0

class TextSplitter:
    """文本分割器"""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.encoding = tiktoken.encoding_for_model(model_name)
        self.context_window = self._get_context_window()
    
    def _get_context_window(self) -> int:
        """获取上下文窗口大小"""
        context_limits = {
            "gpt-4": 128000,
            "gpt-4-turbo": 128000,
            "gpt-3.5-turbo": 16384,
            "claude-3": 200000,
            "claude-3.5": 200000,
        }
        return context_limits.get(self.model_name, 8192)
    
    def count_tokens(self, text: str) -> int:
        """计算Token数量"""
        return len(self.encoding.encode(text)) if text else 0
    
    def split_by_tokens(self, text: str, max_tokens: int = None, overlap_ratio: float = 0.1) -> List[TextSegment]:
        """
        按Token数量分割文本
        """
        if max_tokens is None:
            max_tokens = self.context_window // 4
        
        overlap_tokens = int(max_tokens * overlap_ratio)
        
        segments = []
        current_content = text
        segment_id = 0
        
        while len(current_content) > 0:
            current_tokens = self.count_tokens(current_content)
            if current_tokens <= max_tokens:
                segments.append(TextSegment(
                    id=segment_id,
                    content=current_content,
                    token_count=current_tokens
                ))
                break
            
            split_point = self._find_split_point(current_content, max_tokens)
            
            if split_point <= 0:
                segments.append(TextSegment(
                    id=segment_id,
                    content=current_content[:max_tokens],
                    token_count=self.count_tokens(current_content[:max_tokens])
                ))
                current_content = current_content[max_tokens:]
            else:
                segment_content = current_content[:split_point]
                segments.append(TextSegment(
                    id=segment_id,
                    content=segment_content,
                    token_count=self.count_tokens(segment_content)
                ))
                
                overlap_chars = self._tokens_to_chars(overlap_tokens, segment_content)
                next_start = max(split_point - overlap_chars, 0)
                
                current_content = current_content[next_start:]
                segment_id += 1
        
        return segments
    
    def _find_split_point(self, text: str, max_tokens: int) -> int:
        """查找分割点"""
        if len(text) <= max_tokens:
            return len(text)
        
        words = text.split()
        current_tokens = 0
        split_point = 0
        
        for i, word in enumerate(words):
            word_tokens = self.count_tokens(word)
            
            if current_tokens + word_tokens > max_tokens:
                return split_point
            
            current_tokens += word_tokens
            split_point += len(word) + 1
        
        return split_point
    
    def split_by_paragraphs(self, text: str, max_tokens: int = None) -> List[TextSegment]:
        """按段落分割文本"""
        if max_tokens is None:
            max_tokens = self.context_window // 4
        
        paragraphs = text.split('\n\n')
        segments = []
        current_segment = ""
        current_tokens = 0
        segment_id = 0
        
        for paragraph in paragraphs:
            paragraph_tokens = self.count_tokens(paragraph)
            
            if current_tokens + paragraph_tokens <= max_tokens:
                if current_segment:
                    current_segment += '\n\n'
                current_segment += paragraph
                current_tokens += paragraph_tokens
            else:
                if current_segment:
                    segments.append(TextSegment(
                        id=segment_id,
                        content=current_segment,
                        token_count=current_tokens
                    ))
                    segment_id += 1
                
                current_segment = paragraph
                current_tokens = paragraph_tokens
        
        if current_segment:
            segments.append(TextSegment(
                id=segment_id,
                content=current_segment,
                token_count=current_tokens
            ))
        
        return segments

# 使用示例
def text_splitting_example():
    """文本分割示例"""
    splitter = TextSplitter("gpt-4")
    
    long_text = """
    人工智能的发展历程可以追溯到20世纪50年代。当时，计算机科学刚刚起步，科学家们开始探索让机器模拟人类思维的可能性。
    
    1956年的达特茅斯会议被认为是人工智能正式诞生的标志。在这个会议上，约翰·麦卡锡首次提出了"人工智能"这个术语，并组织了一批科学家共同研究如何让机器展示智能行为。
    
    在接下来的几十年里，人工智能经历了多次起伏。20世纪60-70年代的第一次AI热潮带来了许多基础理论和算法，但由于计算能力的限制，进展相对缓慢。
    
    20世纪80年代，专家系统的兴起掀起了第二次AI热潮。专家系统通过编码人类专家的知识来解决问题，在医疗诊断、金融分析等领域取得了显著成果。
    
    然而，90年代初AI再次进入低谷。这次低谷主要是由于专家系统的局限性逐渐显现，以及计算能力仍然不足。许多研究机构和公司减少了AI研究的投入。
    
    21世纪初，随着机器学习算法的进步和计算能力的提升，AI开始重新焕发生机。支持向量机、随机森林等算法在分类和回归任务中表现出色。
    
    2012年深度学习的突破是AI发展的重要里程碑。AlexNet在ImageNet竞赛中的成功展示了深度神经网络的强大能力，此后深度学习在图像识别、自然语言处理等领域取得了革命性进展。
    
    近年来，大语言模型的崛起更是将AI发展推向了新的高度。GPT系列、BERT、T5等模型在各种NLP任务中刷新了记录，展示了AI在理解和生成自然语言方面的惊人能力。
    
    如今，人工智能已经深入到我们生活的方方面面，从智能手机的语音助手到自动驾驶汽车，从医疗诊断到金融分析，AI正在改变着我们的世界。
    
    未来，随着量子计算、神经形态计算等新技术的兴起，人工智能将迎来更加广阔的发展空间。我们可以期待AI在科学发现、疾病治疗、环境保护等领域发挥更大的作用。
    """
    
    print("=== 文本分割测试 ===")
    print(f"原始文本长度: {len(long_text)} 字符")
    print(f"原始Token数: {splitter.count_tokens(long_text)}")
    
    strategies = [
        ("按Token分割", lambda: splitter.split_by_tokens(long_text)),
        ("按段落分割", lambda: splitter.split_by_paragraphs(long_text))
    ]
    
    for strategy_name, split_func in strategies:
        print(f"\n--- {strategy_name} ---")
        segments = split_func()
        
        print(f"分割后片段数: {len(segments)}")
        print(f"平均每个片段Token数: {sum(s.token_count for s in segments)/len(segments):.0f}")
        
        print("\n片段预览:")
        for i, segment in enumerate(segments[:3]):
            print(f"片段 {i+1}: {segment.content[:100]}...")

if __name__ == "__main__":
    text_splitting_example()
​

步骤 3：上下文连贯性维护

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
上下文连贯性维护工具
"""

from typing import List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
import tiktoken

@dataclass
class ContextSummary:
    """上下文摘要"""
    id: int
    summary: str
    key_points: List[str]
    related_previous_ids: List[int]
    
class ContextCoherenceManager:
    """上下文连贯性管理器"""
    
    def __init__(self, model_name: str = "gpt-4"):
        self.model_name = model_name
        self.encoding = tiktoken.encoding_for_model(model_name)
        self.context_window = self._get_context_window()
        self.summaries: List[ContextSummary] = []
    
    def _get_context_window(self) -> int:
        """获取上下文窗口大小"""
        context_limits = {
            "gpt-4": 128000,
            "gpt-4-turbo": 128000,
            "gpt-3.5-turbo": 16384,
            "claude-3": 200000,
            "claude-3.5": 200000,
        }
        return context_limits.get(self.model_name, 8192)
    
    def count_tokens(self, text: str) -> int:
        """计算Token数量"""
        return len(self.encoding.encode(text)) if text else 0
    
    def create_summary(self, content: str, content_id: int, previous_ids: List[int] = None) -> ContextSummary:
        """创建上下文摘要"""
        if previous_ids is None:
            previous_ids = []
        
        key_points = self._extract_key_points(content)
        summary = self._generate_summary(content, key_points)
        
        return ContextSummary(
            id=content_id,
            summary=summary,
            key_points=key_points,
            related_previous_ids=previous_ids
        )
    
    def _extract_key_points(self, text: str) -> List[str]:
        """提取关键点"""
        sentences = text.split('.')
        key_points = []
        
        for sentence in sentences:
            sentence = sentence.strip()
            if len(sentence) > 20:
                important_words = ['重要', '关键', '核心', '主要', '基本', '基础', '重点']
                if any(word in sentence for word in important_words):
                    key_points.append(sentence[:50] + '...' if len(sentence) > 50 else sentence)
        
        return key_points[:5]
    
    def _generate_summary(self, text: str, key_points: List[str]) -> str:
        """生成摘要"""
        if len(text) < 200:
            return text
        
        summary_parts = []
        summary_parts.append("本节要点：")
        
        for i, point in enumerate(key_points[:3]):
            summary_parts.append(f"{i+1}. {point}")
        
        if self.summaries:
            last_summary = self.summaries[-1]
            summary_parts.append(f"相关前文：{last_summary.summary[:50]}...")
        
        return '\n'.join(summary_parts)
    
    def maintain_coherence(self, 
                          conversation: List[Dict[str, str]], 
                          content_id: int) -> Dict[str, Any]:
        """维护上下文连贯性"""
        if not conversation:
            return {"status": "无需维护", "recommendations": []}
        
        total_tokens = sum(self.count_tokens(msg['content']) for msg in conversation)
        recommendations = []
        
        if total_tokens > self.context_window * 0.8:
            recommendations.append("对话接近上下文限制，建议进行摘要或压缩")
        
        current_content = conversation[-1]['content']
        previous_ids = [i for i, msg in enumerate(conversation[:-1])]
        
        summary = self.create_summary(current_content, content_id, previous_ids)
        self.summaries.append(summary)
        
        if len(self.summaries) > 10:
            self.summaries = self.summaries[-5:]
        
        return {
            "status": "已维护",
            "current_summary": summary.summary,
            "total_tokens": total_tokens,
            "recommendations": recommendations,
            "summary_id": content_id
        }

# 使用示例
def coherence_example():
    """上下文连贯性示例"""
    manager = ContextCoherenceManager("gpt-4")
    
    conversation = [
        {"role": "user", "content": "我想了解人工智能的发展历程"},
        {"role": "assistant", "content": "人工智能的发展可以追溯到20世纪50年代。1956年的达特茅斯会议被认为是人工智能正式诞生的标志，约翰·麦卡锡首次提出了'人工智能'这个术语。之后经历了多次起伏，包括60-70年代的第一次AI热潮、80年代的专家系统热潮，以及21世纪深度学习的复兴。近年来，大语言模型的崛起更是将AI发展推向了新的高度。"},
        {"role": "user", "content": "深度学习是如何在AI发展中发挥重要作用的？"},
        {"role": "assistant", "content": "深度学习在AI发展中扮演了关键角色。2012年，AlexNet在ImageNet竞赛中的成功展示了深度神经网络的强大能力。此后，卷积神经网络在图像识别、循环神经网络在序列数据处理、Transformer在自然语言处理等领域都取得了革命性进展。深度学习的核心优势在于能够自动学习特征，不需要人工设计特征工程，这使得AI系统能够处理更复杂的问题。特别是大语言模型的出现，如GPT系列、BERT等，在理解和生成自然语言方面表现出了惊人的能力。"}
    ]
    
    print("=== 上下文连贯性维护测试 ===")
    
    for i, msg in enumerate(conversation):
        print(f"\n轮次 {i+1}: {msg['role']}")
        print(f"内容: {msg['content'][:50]}...")
        
        result = manager.maintain_coherence(conversation[:i+1], i)
        
        print(f"状态: {result['status']}")
        print(f"摘要: {result['current_summary']}")
        
        if result['recommendations']:
            print("建议:")
            for rec in result['recommendations']:
                print(f"  - {rec}")
    
    print(f"\n生成的摘要总数: {len(manager.summaries)}")

if __name__ == "__main__":
    coherence_example()
​

常见问题 FAQ

Q: 如何选择合适的上下文窗口大小？

A: 选择上下文窗口大小时需要考虑：1) 应用场景的需求，如客服对话可能需要较少上下文，而文档分析需要更多；2) API成本，更大的上下文窗口通常意味着更高的成本；3) 模型性能，并非所有模型都支持超大上下文窗口。建议在开发过程中测试不同大小，找到成本和质量的平衡点。

Q: 长文本分割时如何选择重叠大小？

A: 重叠大小通常设置为分割片段的10%-20%。较小的重叠会减少冗余信息，但可能丢失上下文连贯性；较大的重叠会保持更好的连贯性，但会增加处理成本。建议从10%开始测试，根据应用需求调整。对于连续性要求高的任务（如小说翻译），可以设置20%的重叠；对于事实性任务（如文档摘要），10%重叠就足够了。

Q: 如何处理跨文档的上下文关联？

A: 处理跨文档关联的常用策略包括：1) 建立文档索引，记录文档之间的语义关联；2) 使用嵌入向量计算文档相似度；3) 维护文档元数据，记录文档的创建时间、修改历史等；4) 实现增量摘要机制，为新文档自动添加相关前文摘要。关键是要在文档之间建立有意义的连接，而不仅仅是依赖时间顺序。

Q: 上下文窗口管理会增加系统复杂度吗？

A: 是的，上下文窗口管理会增加一定的系统复杂度，但这种复杂度是必要的。为了简化实现，可以考虑：1) 使用现有的库如LangChain、LlamaIndex等；2) 建立标准化的上下文处理流程；3) 实现自动化工具来处理常见的上下文管理任务；4) 在系统架构中预留上下文管理模块的接口。虽然初期投入较大，但长期来看能显著提升应用性能。

Q: 如何监控上下文窗口的使用效率？

A: 监控上下文窗口使用效率需要关注多个指标：1) 上下文利用率，即实际使用的Token数与最大容量的比例；2) 溢出频率，处理超出限制的频率；3) 分割质量，分割后的片段是否保持了语义完整性；4) 响应质量，上下文管理是否影响了回答质量；5) 性能影响，上下文处理带来的额外时间成本。建议建立监控系统，定期分析这些指标并优化处理策略。

本节小结

通过本节的学习，你已经掌握了：

✅ 上下文窗口的基本概念和重要性
✅ 不同模型的上下文窗口限制和特点
✅ 长文本分割和处理的高级策略
✅ 上下文连贯性的维护方法
✅ 多文档上下文管理的最佳实践

这些技术将帮助你在开发大模型应用时有效处理长文本、保持对话连贯性，构建更加智能和高效的AI系统。下一章我们将深入探讨"RAG系统实战"的具体实现。

延伸阅读

• 官方文档：OpenAI GPT-4上下文窗口使用指南（文字描述，不带链接）
• 相关章节：本教程 3.1 节RAG系统实战（文字描述，不带链接）
• 技术博客：LangChain上下文管理最佳实践（文字描述，不带链接）

关键词：大模型应用开发从零到一, 上下文窗口, 长文本处理, 上下文管理, 文档分割, 连贯性维护, Token计算, 上下文优化
难度：进阶
预计阅读：50 分钟