长上下文：能力与挑战并存

当模型支持百万token的上下文时，新的问题随之而来：模型真的能有效利用这么长的上下文吗？研究表明，上下文长度和上下文利用效率是两个截然不同的问题。

长上下文的技术难点

注意力衰减

模型对上下文中不同位置信息的关注程度不均匀：

def measure_attention_decay(model, context_length, key_position):
    """测量模型对不同位置信息的关注度"""
    # 在上下文的不同位置放置关键信息
    results = []
    for pos in [0, 0.1, 0.25, 0.5, 0.75, 0.9, 1.0]:
        # pos=0表示开头，pos=1表示结尾
        context = build_context_with_key_at(pos, context_length)
        question = "根据上下文中的关键信息回答..."
        
        accuracy = test_answer_accuracy(model, context, question)
        results.append({"position": pos, "accuracy": accuracy})
    
    return results
# 典型结果：
# position=0.0:  accuracy=85%
# position=0.25: accuracy=52%  ← 中间区域下降
# position=0.5:  accuracy=48%  ← 最低点
# position=0.75: accuracy=55%
# position=1.0:  accuracy=82%

Needle-in-a-Haystack测试

这是评估长上下文能力的标准测试：在大量无关文本中隐藏一句关键信息，测试模型能否找到：

def needle_in_haystack(model, context_length, needle_position):
    """大海捞针测试"""
    # 生成填充文本
    filler = generate_filler_text(context_length - 200)
    
    # 关键信息（针）
    needle = "密码是：Sk7-9mPq"
    
    # 在指定位置插入针
    insert_pos = int(len(filler) * needle_position)
    context = filler[:insert_pos] + needle + filler[insert_pos:]
    
    # 测试模型能否找到
    question = "文档中提到的密码是什么？"
    response = model.generate(context + "\n\n" + question)
    
    return {
        "found": "Sk7-9mPq" in response,
        "context_length": context_length,
        "needle_position": needle_position
    }

def full_evaluation(model):
    """全面评估"""
    results = []
    for length in [1000, 5000, 10000, 50000, 100000, 500000]:
        for pos in [0, 0.1, 0.3, 0.5, 0.7, 0.9, 1.0]:
            result = needle_in_haystack(model, length, pos)
            results.append(result)
    
    # 生成热力图
    return plot_heatmap(results, x="position", y="length", color="found")

多针测试

更复杂的测试在上下文中放置多个关键信息：

def multi_needle_test(model, context_length, n_needles=5):
    """多针测试"""
    filler = generate_filler_text(context_length)
    
    needles = []
    for i in range(n_needles):
        needle = f"事实{i+1}：数值是{random.randint(100,999)}"
        pos = random.uniform(0.1, 0.9)
        insert_pos = int(len(filler) * pos)
        filler = filler[:insert_pos] + needle + filler[insert_pos:]
        needles.append({"content": needle, "position": pos})
    
    # 测试模型能否回忆所有事实
    questions = [f"事实{i+1}的数值是多少？" for i in range(n_needles)]
    
    recall_rate = 0
    for q, needle in zip(questions, needles):
        response = model.generate(filler + "\n\n" + q)
        if needle["content"].split("：")[1] in response:
            recall_rate += 1
    
    return recall_rate / n_needles

长上下文的训练挑战

位置编码外推

模型训练时使用固定长度的位置编码，推理时使用更长上下文时需要外推：

RoPE（旋转位置编码）：

class RoPE:
    def __init__(self, dim, max_position=8192, base=10000):
        self.dim = dim
        self.max_position = max_position
        self.base = base
    
    def encode(self, position):
        """旋转位置编码"""
        freqs = 1.0 / (self.base ** (
            torch.arange(0, self.dim, 2).float() / self.dim
        ))
        angles = position * freqs
        # 复数旋转
        cos = torch.cos(angles)
        sin = torch.sin(angles)
        return cos, sin

# RoPE的外推问题：训练时max_position=4096，
# 推理时需要处理100K+的位置

YaRN（Yet another RoPE extensioN）：

class YaRN:
    """YaRN通过温度缩放和插值实现位置外推"""
    def __init__(self, dim, original_max=8192, extended_max=131072):
        self.original_max = original_max
        self.extended_max = extended_max
        self.scale = extended_max / original_max
        # 温度参数
        self.temperature = 0.1
    
    def adjust_position(self, position):
        """调整位置编码以支持更长上下文"""
        # 分段处理
        if position < self.original_max * 0.3:
            # 低频区域：原始位置
            return position
        elif position < self.original_max * 0.7:
            # 中频区域：插值
            return position * (1 - self.temperature) + position / self.scale * self.temperature
        else:
            # 高频区域：缩放
            return position / self.scale

训练数据构造

长上下文能力需要长序列训练数据：

class LongContextDataBuilder:
    def build_training_data(self):
        data = []
        
        # 1. 长文档
        long_docs = collect_long_documents(min_length=10000)
        data.extend(long_docs)
        
        # 2. 多文档拼接
        multi_docs = self._concat_related_docs(target_length=50000)
        data.extend(multi_docs)
        
        # 3. 长对话
        long_conversations = collect_long_conversations(min_turns=50)
        data.extend(long_conversations)
        
        # 4. 代码仓库级
        repo_level = self._build_repo_context(target_length=100000)
        data.extend(repo_level)
        
        return data
    
    def _concat_related_docs(self, target_length):
        """拼接相关文档构造长上下文"""
        docs = []
        current_length = 0
        
        while current_length < target_length:
            # 检索与已有内容相关的文档
            doc = retrieve_related_doc(docs)
            docs.append(doc)
            current_length += len(doc)
        
        return "\n\n".join(docs)

长上下文的工程优化

分块注意力

class ChunkedAttention:
    """分块注意力：减少长序列的注意力计算复杂度"""
    def __init__(self, chunk_size=4096):
        self.chunk_size = chunk_size
    
    def attention(self, query, key, value):
        B, T, D = query.shape
        
        # 分块计算
        chunks = T // self.chunk_size
        outputs = []
        
        for i in range(chunks):
            start = i * self.chunk_size
            end = start + self.chunk_size
            
            # 块内注意力
            q_chunk = query[:, start:end]
            k_chunk = key[:, start:end]
            v_chunk = value[:, start:end]
            
            # 块内attention + 块间attention
            local_attn = self._local_attention(q_chunk, k_chunk, v_chunk)
            global_attn = self._global_attention(q_chunk, key, value)
            
            outputs.append(local_attn + global_attn)
        
        return torch.cat(outputs, dim=1)

KV Cache压缩

class KVCacheCompressor:
    """长上下文下KV Cache的压缩"""
    def __init__(self, budget=8192):
        self.budget = budget  # 保留的token数
    
    def compress(self, kv_cache, attention_scores):
        # 1. 重要性评分
        importance = attention_scores.sum(dim=-1)  # [T]
        
        # 2. 保留重要token
        if len(importance) > self.budget:
            # 保留Top-K + 最近N个
            topk_idx = importance.topk(self.budget // 2).indices
            recent_idx = torch.arange(
                len(importance) - self.budget // 2, len(importance)
            )
            keep_idx = torch.cat([topk_idx, recent_idx]).unique()
            
            kv_cache = kv_cache[keep_idx]
        
        return kv_cache

检索增强长上下文

class RetrievalAugmentedLongContext:
    """结合检索的长上下文处理"""
    def __init__(self, embedder, vector_store, llm):
        self.embedder = embedder
        self.store = vector_store
        self.llm = llm
    
    def process_long_document(self, doc, query):
        """处理超长文档"""
        # 1. 分块
        chunks = split_into_chunks(doc, size=1000, overlap=200)
        
        # 2. 向量化存储
        embeddings = [self.embedder.encode(c) for c in chunks]
        self.store.add(chunks, embeddings)
        
        # 3. 检索相关块
        query_emb = self.embedder.encode(query)
        relevant = self.store.search(query_emb, top_k=10)
        
        # 4. 按原始顺序排列
        relevant.sort(key=lambda x: x["original_index"])
        
        # 5. 构建压缩上下文
        context = "\n...".join([r["text"] for r in relevant])
        
        return self.llm.generate(query, context=context)

长上下文 vs RAG

维度长上下文RAG
信息完整性高（看全文）中（看片段）
准确性依赖模型利用率依赖检索质量
延迟高（长序列推理慢）低（短序列推理快）
成本高（token多）低（token少）
可扩展性受窗口限制可扩展到无限文档
适合场景单个长文档多文档检索

最佳实践：长上下文处理核心文档，RAG扩展到更多文档。

结语

长上下文能力的提升不是线性的——从8K到32K是量变，从128K到1M是质变。质变不仅是技术挑战，更是应用范式的转变。当模型能一次"读完"一整本书、一整个代码仓库时，新的应用场景将自然涌现。但前提是我们解决了长上下文下的信息利用效率问题——看得到不等于看得到关键。