大模型上下文窗口管理技巧突破长度限制的艺术前言大模型的上下文窗口Context Window是指模型能够处理的最大输入长度。目前主流模型的上下文窗口从 4K 到 128K 不等GPT-4 Turbo 甚至达到了 128K tokens。然而随着应用场景的复杂化我们经常需要处理超过这个限制的文本。在实际项目中我曾遇到需要让模型分析长篇技术文档、处理长对话历史、或者对整本书进行摘要的场景。这些需求都涉及上下文窗口的管理问题。今天分享一些我在实践中总结的技巧。上下文窗口的基础知识Token 的概念Token 是模型处理文本的基本单位。对于英文一个 token 大约等于 4 个字符或 0.75 个单词。对于中文通常 1-2 个汉字等于一个 token。import tiktoken # 使用 tiktoken 计算 token 数 enc tiktoken.get_encoding(cl100k_base) # GPT-4 使用 text Hello, world! tokens enc.encode(text) print(fText: {text}) print(fTokens: {tokens}) print(fToken count: {len(tokens)})上下文窗口的组成在对话场景中上下文窗口通常包含┌────────────────────────────────────────────────────────────┐ │ 上下文窗口 │ ├────────────────────────────────────────────────────────────┤ │ 系统提示 │ 用户历史 │ 助手历史 │ 当前查询 │ 可用长度 │ │ │ (System) │ (User) │ (Assistant) │ (Query) │ for Reply │ └────────────────────────────────────────────────────────────┘系统提示和对话历史会占用上下文空间我们需要管理好这个空间确保有足够的空间生成回复。分块策略固定窗口分块最简单的方法是将文本按固定大小分割class FixedSizeChunker: def __init__(self, chunk_size: int, overlap: int): self.chunk_size chunk_size self.overlap overlap def chunk(self, text: str) - List[str]: 固定大小分块 words text.split() chunks [] for i in range(0, len(words), self.chunk_size - self.overlap): chunk .join(words[i:i self.chunk_size]) chunks.append(chunk) return chunks语义分块固定分块可能把语义相关的句子切断。语义分块尝试在句子边界处分割import re class SemanticChunker: def __init__(self, max_tokens: int): self.max_tokens max_tokens def chunk(self, text: str) - List[str]: 按语义边界分块 # 按段落分割 paragraphs text.split(\n) chunks [] current_chunk [] current_size 0 for para in paragraphs: para_tokens self._count_tokens(para) if current_size para_tokens self.max_tokens: # 保存当前 chunk if current_chunk: chunks.append(\n.join(current_chunk)) # 如果段落本身太大按句子分割 if para_tokens self.max_tokens: sub_chunks self._split_long_paragraph(para) chunks.extend(sub_chunks) current_chunk [] current_size 0 else: current_chunk [para] current_size para_tokens else: current_chunk.append(para) current_size para_tokens # 添加最后一个 chunk if current_chunk: chunks.append(\n.join(current_chunk)) return chunks def _count_tokens(self, text: str) - int: import tiktoken enc tiktoken.get_encoding(cl100k_base) return len(enc.encode(text)) def _split_long_paragraph(self, para: str) - List[str]: 分割长段落 sentences re.split(r([。.!?]), para) chunks [] current current_size 0 for i in range(0, len(sentences) - 1, 2): sentence sentences[i] sentences[i 1] sentence_size self._count_tokens(sentence) if current_size sentence_size self.max_tokens: if current: chunks.append(current) current sentence current_size sentence_size else: current sentence current_size sentence_size if current: chunks.append(current) return chunks递归分块class RecursiveChunker: 使用多级分隔符的递归分块 def __init__(self, separators: List[str], max_tokens: int): self.separators separators self.max_tokens max_tokens def chunk(self, text: str) - List[str]: 递归分割直到满足大小要求 return self._chunk_recursive(text, 0) def _chunk_recursive(self, text: str, separator_idx: int) - List[str]: if separator_idx len(self.separators): # 最后手段直接按大小截断 return self._fixed_chunk(text) separator self.separators[separator_idx] parts text.split(separator) result [] current for part in parts: test current separator part if current else part test_size self._count_tokens(test) if test_size self.max_tokens: current test else: if current: result.append(current) # 递归处理超出大小的部分 if self._count_tokens(part) self.max_tokens: sub_chunks self._chunk_recursive(part, separator_idx 1) result.extend(sub_chunks[:-1]) current sub_chunks[-1] if sub_chunks else else: current part if current: result.append(current) return result def _fixed_chunk(self, text: str) - List[str]: enc tiktoken.get_encoding(cl100k_base) tokens enc.encode(text) chunks [] for i in range(0, len(tokens), self.max_tokens): chunk_tokens tokens[i:i self.max_tokens] chunks.append(enc.decode(chunk_tokens)) return chunks def _count_tokens(self, text: str) - int: enc tiktoken.get_encoding(cl100k_base) return len(enc.encode(text))对话历史管理总结压缩class ConversationManager: 对话历史管理器 def __init__( self, llm, max_tokens: int 3000, summary_tokens: int 500 ): self.llm llm self.max_tokens max_tokens self.summary_tokens summary_tokens self.messages [] def add_message(self, role: str, content: str): 添加消息 self.messages.append({role: role, content: content}) self._maybe_summarize() def _maybe_summarize(self): 检查是否需要总结历史 total_tokens sum( self._count_tokens(m[content]) for m in self.messages ) if total_tokens self.max_tokens: self._summarize_old_messages() def _summarize_old_messages(self): 总结旧消息 # 保留最近的消息 recent_messages self.messages[-4:] # 最近 2 轮对话 # 将早期消息合并为总结 old_messages self.messages[:-4] if not old_messages: return summary_prompt f请总结以下对话的要点 {chr(10).join([f{m[role]}: {m[content]} for m in old_messages])} 总结简洁明了保留关键信息 summary self.llm.generate(summary_prompt) # 用总结替换旧消息 self.messages [ {role: system, content: f早期对话摘要{summary}} ] recent_messages def get_context(self) - List[dict]: 获取当前上下文 return self.messages def _count_tokens(self, text: str) - int: enc tiktoken.get_encoding(cl100k_base) return len(enc.encode(text))分层记忆class HierarchicalMemory: 分层记忆系统 def __init__(self, llm, token_limits: dict): self.llm llm self.token_limits token_limits # 三层记忆 self.working_memory [] # 当前对话 self.episodic_memory [] # 最近会话 self.semantic_memory # 长期知识 def add_working(self, role: str, content: str): 添加到工作记忆 self.working_memory.append({role: role, content: content}) self._condense_working() def _condense_working(self): 压缩工作记忆 tokens self._count_tokens( \n.join([m[content] for m in self.working_memory]) ) if tokens self.token_limits[working]: # 保留最近的压缩更早的 recent self.working_memory[-2:] older self.working_memory[:-2] if older: summary self._summarize(older) self.episodic_memory.append({ type: episode, content: summary }) self.working_memory recent def get_context(self) - str: 构建完整上下文 parts [] # 语义记忆 if self.semantic_memory: parts.append(f【背景知识】\n{self.semantic_memory}) # 情景记忆 if self.episodic_memory: episodes \n.join([ f- {ep[content]} for ep in self.episodic_memory[-3:] ]) parts.append(f【近期会话】\n{episodes}) # 工作记忆 if self.working_memory: dialogue \n.join([ f{m[role]}: {m[content]} for m in self.working_memory ]) parts.append(f【当前对话】\n{dialogue}) return \n\n.join(parts)长文本处理策略Map-Reduce处理长文本的经典方法class MapReduceProcessor: Map-Reduce 文本处理 def __init__(self, llm, chunk_size: int 3000): self.llm llm self.chunk_size chunk_size def process(self, task: str, document: str) - str: Map-Reduce 处理 # 1. Map将文档分块每块独立处理 chunks self._chunk(document) chunk_summaries [] for i, chunk in enumerate(chunks): summary self._map_step(task, chunk, i 1, len(chunks)) chunk_summaries.append(summary) # 2. Reduce合并所有结果 final_result self._reduce_step(task, chunk_summaries) return final_result def _chunk(self, text: str) - List[str]: 分块 enc tiktoken.get_encoding(cl100k_base) tokens enc.encode(text) chunks [] for i in range(0, len(tokens), self.chunk_size): chunk_tokens tokens[i:i self.chunk_size] chunks.append(enc.decode(chunk_tokens)) return chunks def _map_step(self, task: str, chunk: str, idx: int, total: int) - str: Map 步骤处理单个 chunk prompt f这是关于{task}的任务。 文档的第 {idx}/{total} 部分内容如下 {chunk} 请提取与任务相关的信息用简洁的语言描述 return self.llm.generate(prompt) def _reduce_step(self, task: str, summaries: List[str]) - str: Reduce 步骤合并所有结果 combined \n\n.join( f第 {i1} 部分{s} for i, s in enumerate(summaries) ) prompt f任务{task} 以下是文档各部分的处理结果 {combined} 请综合以上信息给出完整的回答 return self.llm.generate(prompt)Sliding Window Attention模拟 Transformer 的滑动窗口机制class SlidingWindowProcessor: 滑动窗口处理 def __init__(self, llm, window_size: int 3000, overlap: int 500): self.llm llm self.window_size window_size self.overlap overlap def process(self, task: str, document: str) - str: 滑动窗口处理 enc tiktoken.get_encoding(cl100k_base) tokens enc.encode(document) results [] step self.window_size - self.overlap for i in range(0, len(tokens), step): window_tokens tokens[i:i self.window_size] window_text enc.decode(window_tokens) # 提取相关信息 result self._process_window(task, window_text) results.append(result) if i self.window_size len(tokens): break # 合并结果 return self._merge_results(task, results) def _process_window(self, task: str, window: str) - str: prompt f任务{task} 内容窗口 {window} 提取与任务相关的所有信息 return self.llm.generate(prompt)位置编码与外推什么是外推外推Extrapolation是指模型处理超过训练长度序列的能力。例如如果模型在 2048 tokens 上训练理论上就不能处理 4096 tokens 的输入。RoPE 旋转位置编码RoPE 是现代 LLM 广泛使用的位置编码方案def precompute_rope_params(head_dim, max_seq_len, theta10000.0): 预计算 RoPE 参数 inv_freq 1.0 / (theta ** (torch.arange(0, head_dim, 2).float() / head_dim)) t torch.arange(max_seq_len) freqs torch.einsum(i,j-ij, t, inv_freq) emb torch.cat((freqs, freqs), dim-1) return torch.cos(emb), torch.sin(emb) def apply_rope(x, cos, sin): 应用 RoPE x1, x2 x[..., : x.shape[-1] // 2], x[..., x.shape[-1] // 2 :] return torch.cat([ x1 * cos - x2 * sin, x1 * sin x2 * cos ], dim-1)位置插值位置插值Position Interpolation是一种扩展上下文窗口的方法def interpolate_positions(freqs, scale_factor): 位置插值 # 将位置索引缩放使新位置映射到原始范围 scaled_freqs freqs / scale_factor return scaled_freqs总结上下文窗口管理是大模型应用中的重要课题。通过合理的分块策略、历史管理、压缩技术我们可以有效突破模型的长度限制。关键要点选择合适的分块策略固定、语义、递归对话历史要及时压缩和总结长文档使用 Map-Reduce 等分解策略关注位置编码的外推能力希望这些技巧对大家的项目有所帮助。