Claude Prompt长度限制的工程实践：从分块处理到语义压缩

最近在做一个智能客服系统，需要把用户的历史对话记录整理后发给Claude做分析。本来觉得Claude的100K上下文窗口足够大了，没想到实际操作中还是遇到了“prompt is too long”的问题。经过一番折腾，总结出了一套处理长prompt的工程实践方案，在这里分享给大家。

1. 问题背景：为什么100K窗口还不够用？

刚开始我也很疑惑，100K tokens按说能处理大约7.5万汉字，应该够用了吧？但实际项目中，问题比想象中复杂：

• 多轮对话场景：用户和客服的对话记录，加上系统指令、示例对话、格式要求等，很容易就超过限制
• 文档分析任务：需要分析的技术文档、产品说明书等，单篇可能不长，但多篇合并就超了
• 复杂指令场景：包含多个步骤、多个条件的复杂任务描述，prompt本身就很长

更关键的是，Claude的API对单次请求有token数限制（具体数值因模型版本而异），即使总上下文窗口没满，单次请求超限也会失败。这就需要在工程层面做预处理。

2. 技术方案一：智能分块处理

最简单的思路就是把长prompt切成小块，但怎么切才合理？这里有几个关键考虑：

2.1 按token数分块

最基础的方法是按token数均匀分割，但要注意保持语义完整：

from typing import List, Tuple import tiktoken class TokenChunker: def __init__(self, model_name: str = "claude-3"): """初始化tokenizer""" # Claude使用类似GPT的tokenizer self.encoder = tiktoken.get_encoding("cl100k_base") def chunk_by_tokens(self, text: str, max_tokens: int = 4000, overlap_tokens: int = 200) -> List[str]: """ 按token数分块，保持段落完整性 时间复杂度：O(n)，n为文本长度 空间复杂度：O(n) """ # 先按段落分割 paragraphs = text.split('\n\n') chunks = [] current_chunk = [] current_tokens = 0 for para in paragraphs: para_tokens = len(self.encoder.encode(para)) # 如果单个段落就超限，需要进一步分割 if para_tokens > max_tokens: sub_chunks = self._split_large_paragraph(para, max_tokens) for sub_chunk in sub_chunks: if len(current_chunk) > 0 and current_tokens + len(self.encoder.encode(sub_chunk)) > max_tokens: chunks.append('\n\n'.join(current_chunk)) current_chunk = [sub_chunk] current_tokens = len(self.encoder.encode(sub_chunk)) else: current_chunk.append(sub_chunk) current_tokens += len(self.encoder.encode(sub_chunk)) else: if current_tokens + para_tokens > max_tokens: chunks.append('\n\n'.join(current_chunk)) current_chunk = [para] current_tokens = para_tokens else: current_chunk.append(para) current_tokens += para_tokens if current_chunk: chunks.append('\n\n'.join(current_chunk)) # 添加重叠部分，避免上下文断裂 if overlap_tokens > 0 and len(chunks) > 1: chunks = self._add_overlap(chunks, overlap_tokens) return chunks def _split_large_paragraph(self, text: str, max_tokens: int) -> List[str]: """处理超长段落，按句子分割""" sentences = text.replace('。', '。\n').split('\n') chunks = [] current_chunk = [] current_tokens = 0 for sentence in sentences: if not sentence.strip(): continue sentence_tokens = len(self.encoder.encode(sentence)) if current_tokens + sentence_tokens > max_tokens: if current_chunk: chunks.append(''.join(current_chunk)) current_chunk = [sentence] current_tokens = sentence_tokens else: current_chunk.append(sentence) current_tokens += sentence_tokens if current_chunk: chunks.append(''.join(current_chunk)) return chunks def _add_overlap(self, chunks: List[str], overlap_tokens: int) -> List[str]: """为分块添加重叠部分""" overlapped_chunks = [] for i in range(len(chunks)): if i == 0: # 第一个块只添加后向重叠 if len(chunks) > 1: next_start = self._get_overlap_text(chunks[1], overlap_tokens, from_start=True) overlapped_chunks.append(chunks[i] + "\n\n[上下文继续...]\n\n" + next_start) else: overlapped_chunks.append(chunks[i]) elif i == len(chunks) - 1: # 最后一个块只添加前向重叠 prev_end = self._get_overlap_text(chunks[i-1], overlap_tokens, from_start=False) overlapped_chunks.append(prev_end + "\n\n[...接上文]\n\n" + chunks[i]) else: # 中间块添加双向重叠 prev_end = self._get_overlap_text(chunks[i-1], overlap_tokens, from_start=False) next_start = self._get_overlap_text(chunks[i+1], overlap_tokens, from_start=True) overlapped_chunks.append(prev_end + "\n\n[...接上文]\n\n" + chunks[i] + "\n\n[上下文继续...]\n\n" + next_start) return overlapped_chunks def _get_overlap_text(self, text: str, overlap_tokens: int, from_start: bool) -> str: """获取重叠部分的文本""" tokens = self.encoder.encode(text) if from_start: overlap_tokens = tokens[:min(overlap_tokens, len(tokens))] else: overlap_tokens = tokens[-min(overlap_tokens, len(tokens)):] return self.encoder.decode(overlap_tokens)

2.2 按语义边界分块

更高级的方法是识别语义边界，比如章节、主题变化等：

import re from dataclasses import dataclass from typing import Optional @dataclass class SemanticChunk: text: str start_pos: int end_pos: int semantic_unit: Optional[str] = None # 如：paragraph, section, chapter class SemanticChunker: def __init__(self): # 定义语义边界模式 self.section_patterns = [ (r'#{1,6}\s+.+', 'heading'), # Markdown标题 (r'第[一二三四五六七八九十]+章', 'chapter'), # 中文章节 (r'\d+\.\d+', 'section'), # 数字章节 (r'[A-Z][A-Z\s]+:', 'label'), # 标签式标题 ] def chunk_by_semantic(self, text: str, max_tokens: int = 4000) -> List[SemanticChunk]: """ 按语义边界分块 算法复杂度分析： 1. 边界检测：O(n)，n为文本长度 2. 分块合并：O(m)，m为边界数量 总复杂度：O(n + m) """ # 检测所有语义边界 boundaries = self._detect_boundaries(text) # 基于边界进行分块 chunks = [] current_start = 0 current_text = "" for i, boundary in enumerate(boundaries): segment = text[current_start:boundary.position] # 检查当前块是否超限 if len(self._estimate_tokens(current_text + segment)) > max_tokens: if current_text: chunks.append(SemanticChunk( text=current_text, start_pos=current_start - len(current_text), end_pos=current_start, semantic_unit=self._identify_unit(current_text) )) current_text = segment current_start = boundary.position - len(segment) else: current_text += segment # 如果边界本身是一个强分割点 if boundary.strength == 'strong': if current_text: chunks.append(SemanticChunk( text=current_text, start_pos=current_start, end_pos=boundary.position, semantic_unit=boundary.unit_type )) current_text = "" current_start = boundary.position # 处理最后一段 if current_text: chunks.append(SemanticChunk( text=current_text, start_pos=current_start, end_pos=len(text), semantic_unit=self._identify_unit(current_text) )) return chunks def _detect_boundaries(self, text: str) -> List['Boundary']: """检测语义边界""" boundaries = [] # 检测预定义模式 for pattern, unit_type in self.section_patterns: for match in re.finditer(pattern, text): boundaries.append(Boundary( position=match.start(), unit_type=unit_type, strength='strong' if unit_type in ['chapter', 'heading'] else 'weak' )) # 按位置排序 boundaries.sort(key=lambda x: x.position) return boundaries def _estimate_tokens(self, text: str) -> int: """估算token数（简化版）""" # 实际项目中应该使用tokenizer return len(text) // 3 # 近似估算

3. 技术方案二：语义压缩

分块虽然简单，但会丢失全局上下文。更好的方案是语义压缩——用更短的文字表达相同的意思。

3.1 使用T5模型进行文本摘要

import torch from transformers import T5Tokenizer, T5ForConditionalGeneration from typing import List, Dict import time class TextCompressor: def __init__(self, model_name: str = "t5-small", device: str = "cuda"): """ 初始化压缩模型 参数： - model_name: 模型名称，可选 t5-small, t5-base, t5-large - device: 运行设备，cuda 或 cpu """ self.device = device self.tokenizer = T5Tokenizer.from_pretrained(model_name) self.model = T5ForConditionalGeneration.from_pretrained(model_name).to(device) # 不同模型的性能特征 self.model_configs = { "t5-small": {"max_input": 512, "max_output": 150}, "t5-base": {"max_input": 512, "max_output": 150}, "t5-large": {"max_input": 512, "max_output": 150}, } def compress_text(self, text: str, compression_ratio: float = 0.3) -> str: """ 压缩文本，保留核心语义 时间复杂度：O(n)，n为输入文本长度 GPU内存消耗：与模型大小和批次相关 """ # 预处理文本 preprocessed = self._preprocess_text(text) # 如果文本不长，直接返回 if len(self.tokenizer.encode(preprocessed)) < 100: return text # 构建prompt prompt = f"summarize: {preprocessed}" # 编码输入 inputs = self.tokenizer( prompt, max_length=self.model_configs[self.model.config._name_or_path]["max_input"], truncation=True, return_tensors="pt" ).to(self.device) # 生成摘要 with torch.no_grad(): outputs = self.model.generate( inputs.input_ids, max_length=int(len(self.tokenizer.encode(preprocessed)) * compression_ratio), min_length=50, length_penalty=2.0, num_beams=4, early_stopping=True ) # 解码输出 summary = self.tokenizer.decode(outputs[0], skip_special_tokens=True) return summary def batch_compress(self, texts: List[str], batch_size: int = 8) -> List[str]: """批量压缩文本""" results = [] for i in range(0, len(texts), batch_size): batch = texts[i:i+batch_size] batch_prompts = [f"summarize: {self._preprocess_text(text)}" for text in batch] # 批量编码 inputs = self.tokenizer( batch_prompts, max_length=512, truncation=True, padding=True, return_tensors="pt" ).to(self.device) # 批量生成 with torch.no_grad(): outputs = self.model.generate( inputs.input_ids, attention_mask=inputs.attention_mask, max_length=150, min_length=50, length_penalty=2.0, num_beams=4, early_stopping=True ) # 批量解码 batch_results = [ self.tokenizer.decode(output, skip_special_tokens=True) for output in outputs ] results.extend(batch_results) return results def _preprocess_text(self, text: str) -> str: """文本预处理""" # 移除多余空白 text = ' '.join(text.split()) # 截断过长的文本 tokens = self.tokenizer.encode(text) if len(tokens) > 500: text = self.tokenizer.decode(tokens[:500]) + "..." return text

3.2 性能对比测试

为了帮助大家选择合适的方案，我做了详细的性能测试：

import pandas as pd from dataclasses import dataclass from typing import List @dataclass class PerformanceMetrics: model_name: str device: str compression_ratio: float processing_time: float memory_usage: float rouge_score: float # 评估摘要质量 class CompressionBenchmark: def __init__(self): self.results = [] def run_benchmark(self, test_texts: List[str]): """运行性能测试""" devices = ["cpu", "cuda"] if torch.cuda.is_available() else ["cpu"] models = ["t5-small", "t5-base", "t5-large"] for device in devices: for model_name in models: print(f"Testing {model_name} on {device}...") try: compressor = TextCompressor(model_name, device) # 预热 _ = compressor.compress_text(test_texts[0][:100]) # 正式测试 start_time = time.time() summaries = compressor.batch_compress(test_texts[:10], batch_size=4) end_time = time.time() # 计算指标 avg_time = (end_time - start_time) / len(test_texts[:10]) # 这里简化了ROUGE计算，实际应该使用rouge-score库 rouge_score = self._estimate_quality(test_texts[:10], summaries) self.results.append(PerformanceMetrics( model_name=model_name, device=device, compression_ratio=0.3, processing_time=avg_time, memory_usage=self._get_memory_usage(), rouge_score=rouge_score )) except Exception as e: print(f"Error with {model_name} on {device}: {e}") def get_results_table(self) -> pd.DataFrame: """生成性能对比表格""" df = pd.DataFrame([r.__dict__ for r in self.results]) return df

测试结果对比如下：

从表格可以看出：

1. GPU加速效果明显：相比CPU有5-10倍的提升
2. 模型越大质量越高：t5-large的ROUGE分数最高
3. 内存消耗需注意：大模型需要更多显存

4. 技术方案三：智能缓存策略

很多场景下，prompt中有大量重复内容（比如系统指令、示例对话等），这时候缓存就很有用了。

4.1 基于MD5指纹的缓存系统

import hashlib import json from typing import Any, Optional from datetime import datetime, timedelta import pickle class PromptCache: def __init__(self, max_size: int = 1000, ttl_hours: int = 24): """ 初始化prompt缓存 参数： - max_size: 最大缓存条目数 - ttl_hours: 缓存存活时间（小时） """ self.max_size = max_size self.ttl = timedelta(hours=ttl_hours) self.cache: Dict[str, Dict[str, Any]] = {} self.access_order: List[str] = [] # LRU顺序 def get_fingerprint(self, text: str) -> str: """ 生成文本指纹 算法：MD5哈希 时间复杂度：O(n)，n为文本长度 碰撞概率：极低（1/2^128） """ # 标准化文本（移除空白差异） normalized = ' '.join(text.split()) return hashlib.md5(normalized.encode('utf-8')).hexdigest() def get(self, prompt: str) -> Optional[Any]: """获取缓存结果""" fingerprint = self.get_fingerprint(prompt) if fingerprint not in self.cache: return None cache_entry = self.cache[fingerprint] # 检查是否过期 if datetime.now() - cache_entry['timestamp'] > self.ttl: self._remove(fingerprint) return None # 更新访问顺序（LRU） self.access_order.remove(fingerprint) self.access_order.append(fingerprint) return cache_entry['result'] def set(self, prompt: str, result: Any): """设置缓存""" fingerprint = self.get_fingerprint(prompt) # 如果缓存已满，移除最久未使用的 if len(self.cache) >= self.max_size: lru_key = self.access_order.pop(0) del self.cache[lru_key] # 存储新条目 self.cache[fingerprint] = { 'result': result, 'timestamp': datetime.now(), 'size': len(pickle.dumps(result)) } if fingerprint in self.access_order: self.access_order.remove(fingerprint) self.access_order.append(fingerprint) def get_stats(self) -> Dict[str, Any]: """获取缓存统计信息""" total_size = sum(entry['size'] for entry in self.cache.values()) return { 'total_entries': len(self.cache), 'total_size_bytes': total_size, 'hit_rate': self._calculate_hit_rate(), 'avg_entry_size': total_size / len(self.cache) if self.cache else 0 } def _remove(self, fingerprint: str): """移除缓存条目""" if fingerprint in self.cache: del self.cache[fingerprint] if fingerprint in self.access_order: self.access_order.remove(fingerprint) def _calculate_hit_rate(self) -> float: """计算缓存命中率（简化版）""" # 实际应该记录历史查询 return 0.0

4.2 分层缓存策略

对于不同长度的prompt，可以采用不同的缓存策略：

class HierarchicalCache: def __init__(self): # 短文本缓存（< 100 tokens） self.short_cache = PromptCache(max_size=5000) # 中文本缓存（100-1000 tokens） self.medium_cache = PromptCache(max_size=1000) # 长文本缓存（> 1000 tokens） self.long_cache = PromptCache(max_size=100) # 压缩结果缓存 self.compression_cache = PromptCache(max_size=500) def get_compressed(self, text: str, compressor: TextCompressor) -> str: """获取压缩文本（带缓存）""" fingerprint = self.get_fingerprint(text) # 先查缓存 cached = self.compression_cache.get(text) if cached is not None: return cached # 缓存未命中，执行压缩 compressed = compressor.compress_text(text) # 存入缓存 self.compression_cache.set(text, compressed) return compressed

5. 避坑指南

在实际项目中，我遇到了不少坑，这里总结一下：

5.1 分块导致的上下文断裂问题

问题表现：分块后，模型无法理解跨块的引用关系。

解决方案：

1. 重叠分块：相邻块之间保留200-500个token的重叠
2. 添加连接标记：明确指示上下文关系
3. 维护全局索引：为每个块添加位置信息

def chunk_with_context(text: str, max_tokens: int) -> List[Dict]: chunks = [] sentences = text.split('。') current_chunk = [] current_length = 0 for i, sentence in enumerate(sentences): sentence_length = len(tokenizer.encode(sentence)) if current_length + sentence_length > max_tokens: # 保存当前块 chunk_text = '。'.join(current_chunk) + '。' chunks.append({ 'text': chunk_text, 'start_sentence': len(chunks) * 10, # 虚拟索引 'end_sentence': len(chunks) * 10 + len(current_chunk), 'context_hint': f"第{len(chunks)+1}部分，共{len(sentences)//10+1}部分" }) # 新块包含前一块的最后两句作为上下文 current_chunk = sentences[max(0, i-2):i+1] if i > 0 else [sentence] current_length = sum(len(tokenizer.encode(s)) for s in current_chunk) else: current_chunk.append(sentence) current_length += sentence_length return chunks

5.2 压缩模型的选择权衡

精度 vs 速度：

• 高精度场景：用t5-large，ROUGE分数高，但速度慢
• 实时场景：用t5-small，速度快，质量可接受
• 平衡选择：t5-base，性价比最高

内存限制：

• GPU内存<4GB：只能用t5-small
• GPU内存8GB：可以用t5-base
• GPU内存>16GB：推荐t5-large

5.3 缓存失效的边界条件

需要刷新缓存的场景：

1. 模型更新：换了新版本的LLM
2. 指令变更：系统prompt有修改
3. 数据漂移：输入数据的分布发生变化
4. 时间过期：缓存时间太长

解决方案：

class SmartCache(PromptCache): def __init__(self, version: str = "1.0"): super().__init__() self.version = version self.data_signature = "" def should_invalidate(self, new_data_sample: str) -> bool: """检查是否需要失效缓存""" # 1. 检查版本 if self.version != CURRENT_MODEL_VERSION: return True # 2. 检查数据分布 new_signature = self._compute_data_signature(new_data_sample) if self._distribution_changed(new_signature): return True # 3. 检查概念漂移 if self._concept_drift_detected(): return True return False def _compute_data_signature(self, text: str) -> str: """计算数据特征签名""" # 简化实现：统计长度分布 lengths = [len(sent) for sent in text.split('。')] avg_len = sum(lengths) / len(lengths) return f"avg_len_{avg_len:.1f}"

6. 完整实现示例

下面是一个整合了所有技术的完整示例：

from typing import List, Dict, Optional, Tuple import asyncio from concurrent.futures import ThreadPoolExecutor class LongPromptProcessor: def __init__(self, chunk_strategy: str = "semantic", compression_model: str = "t5-base", use_cache: bool = True): """ 长prompt处理器 参数： - chunk_strategy: 分块策略，可选 "token" 或 "semantic" - compression_model: 压缩模型 - use_cache: 是否使用缓存 """ self.chunk_strategy = chunk_strategy self.use_cache = use_cache # 初始化组件 if chunk_strategy == "token": self.chunker = TokenChunker() else: self.chunker = SemanticChunker() self.compressor = TextCompressor(model_name=compression_model) if use_cache: self.cache = HierarchicalCache() else: self.cache = None async def process_prompt(self, prompt: str, max_tokens: int = 4000, use_compression: bool = True) -> List[str]: """ 处理长prompt 返回处理后的prompt列表 """ # 1. 检查是否可以直接使用 token_count = len(self.chunker.encoder.encode(prompt)) if token_count <= max_tokens: return [prompt] # 2. 尝试压缩 if use_compression and token_count > max_tokens * 2: compressed = await self._compress_with_cache(prompt) token_count = len(self.chunker.encoder.encode(compressed)) if token_count <= max_tokens: return [compressed] # 3. 分块处理 if self.chunk_strategy == "token": chunks = self.chunker.chunk_by_tokens( prompt if not use_compression else compressed, max_tokens=max_tokens, overlap_tokens=200 ) else: chunks_objs = self.chunker.chunk_by_semantic( prompt if not use_compression else compressed, max_tokens=max_tokens ) chunks = [chunk.text for chunk in chunks_objs] # 4. 添加上下文信息 chunks = self._add_context_info(chunks) return chunks async def _compress_with_cache(self, text: str) -> str: """带缓存的压缩""" if self.cache: cached = self.cache.get_compressed(text, self.compressor) if cached: return cached # 异步执行压缩（避免阻塞） loop = asyncio.get_event_loop() with ThreadPoolExecutor() as pool: compressed = await loop.run_in_executor( pool, self.compressor.compress_text, text ) if self.cache: self.cache.set(text, compressed) return compressed def _add_context_info(self, chunks: List[str]) -> List[str]: """为分块添加上下文信息""" result = [] total = len(chunks) for i, chunk in enumerate(chunks): context_info = f"\n\n[部分 {i+1}/{total}]" if i > 0: context_info += " 接上文..." if i < total - 1: context_info += " 下文继续..." result.append(chunk + context_info) return result

7. 单元测试

为了保证代码质量，一定要写测试：

import pytest from unittest.mock import Mock, patch class TestLongPromptProcessor: def setup_method(self): self.processor = LongPromptProcessor( chunk_strategy="token", compression_model="t5-small", use_cache=False ) def test_token_chunking(self): """测试token分块""" # 创建长文本 long_text = "测试文本。" * 1000 chunks = self.processor.chunker.chunk_by_tokens(long_text, max_tokens=100) assert len(chunks) > 1 assert all(len(chunk) > 0 for chunk in chunks) # 检查重叠 if len(chunks) > 1: assert "[上下文继续...]" in chunks[0] or "[...接上文]" in chunks[1] def test_compression_quality(self): """测试压缩质量""" original = """ 今天天气很好，阳光明媚，万里无云。我决定去公园散步。 公园里有很多人在锻炼身体，有的在跑步，有的在打太极拳。 我看到一个小孩子在放风筝，风筝飞得很高很高。 我在长椅上坐了一会儿，享受着温暖的阳光和新鲜的空气。 """ compressed = self.processor.compressor.compress_text(original) # 检查压缩比 orig_tokens = len(self.processor.chunker.encoder.encode(original)) comp_tokens = len(self.processor.chunker.encoder.encode(compressed)) assert comp_tokens < orig_tokens * 0.5 # 至少压缩一半 assert len(compressed) > 0 @patch('transformers.T5ForConditionalGeneration.generate') def test_cache_functionality(self, mock_generate): """测试缓存功能""" mock_generate.return_value = [[1, 2, 3]] # 模拟生成结果 processor_with_cache = LongPromptProcessor(use_cache=True) text = "测试文本" # 第一次调用应该执行压缩 result1 = asyncio.run(processor_with_cache._compress_with_cache(text)) # 第二次调用应该从缓存获取 result2 = asyncio.run(processor_with_cache._compress_with_cache(text)) assert result1 == result2 assert mock_generate.call_count == 1 # 应该只调用一次生成 def test_integration(self): """集成测试""" long_prompt = """ 系统指令：你是一个有帮助的助手。 用户历史对话： 用户：你好，我想了解产品A的功能。 助手：产品A具有X、Y、Z功能。 用户：那产品B呢？ 助手：产品B具有P、Q、R功能。 当前问题：请比较产品A和产品B的优缺点。 要求：用表格形式回答，包含功能对比、价格对比、适用场景。 """ * 50 # 重复50次制造长文本 chunks = asyncio.run(self.processor.process_prompt( long_prompt, max_tokens=1000, use_compression=True )) assert len(chunks) >= 1 assert all("[部分" in chunk or len(chunk) <= 1000 for chunk in chunks) if __name__ == "__main__": pytest.main([__file__, "-v"])

8. 数学原理说明

对于感兴趣的读者，这里简单说一下背后的数学原理：

8.1 分块算法的最优解

分块问题可以形式化为：

$$ \min_{chunks} \sum_{i=1}^{n} \text{cost}(chunk_i) $$

其中约束条件为： $$ \text{token_count}(chunk_i) \leq \text{max_tokens} \quad \forall i $$

8.2 压缩模型的损失函数

T5模型使用的损失函数是负对数似然：

$$ \mathcal{L} = -\sum_{t=1}^{T} \log P(y_t | y_{<t}, x; \theta) $$

其中$x$是输入文本，$y$是摘要，$\theta$是模型参数。

9. 性能优化建议

根据我的实践经验，这里给出一些优化建议：

1. 批量处理：尽可能批量处理prompt，减少模型加载开销
2. 异步处理：使用async/await避免阻塞主线程
3. 内存管理：及时清理不需要的缓存和中间结果
4. 监控告警：设置token使用监控，提前预警

10. 开放式问题

在结束之前，我想提出两个值得思考的问题：

1. 动态分块策略：如何根据prompt的语义结构动态调整分块大小？比如对话部分用较小的块，文档部分用较大的块。

2. 增量压缩算法：对于流式输入的长文本，如何实现增量式的实时压缩，而不是等全部输入完毕再压缩？

这两个问题在实际工程中经常遇到，也是我下一步要研究的方向。如果你有好的想法，欢迎一起讨论。

实践体会

经过这几个月的实践，我深刻体会到处理长prompt不是简单的"切一切"就行。需要根据具体场景选择合适的技术组合：

• 客服对话场景：语义分块 + 重叠上下文效果最好
• 文档分析场景：压缩 + 关键信息提取更合适
• 实时交互场景：缓存 + 预压缩能大幅提升响应速度

最重要的是，一定要结合实际业务需求来设计方案，没有银弹。希望我的这些经验对你有所帮助！