2024 年我在一家做企业知识库的公司里负责长文档问答系统接到的需求很直接客户的合同财报研报动辄上百页他们希望用户上传后直接问模型 这份合同的违约条款是什么 这份财报里研发费用同比增长多少 听起来就是把文档塞进 LLM 让它回答嘛我第一版直接用 GPT-4-128k 把文档拼进 prompt 测了几个 case 效果还行老板看了挺满意可一上线一连串问题就来了第一种最先把我打懵某个客户上传了一份 300 页的财报模型处理一次要 40 秒一次调用 token 消耗 8 万 单次成本超过 1 美元客户一天问 500 次账单直接爆了第二种最难缠某些 case 模型回答得驴唇不对马嘴明明合同里写着违约金 30% 模型回答说 50% 我看 prompt 才发现关键条款被夹在第 80 页和第 81 页中间 而模型在长上下文里出现了 lost in the middle 现象前后都记得中间忘了第三种最离谱客户 A 上传一份合同问完后客户 B 上传另一份合同问问题模型居然把 A 的内容混进了回答里我后来才意识到这是因为我的 KV Cache 复用策略写错了上下文残留没清干净第四种最莫名其妙换成 Claude 200k 后某些场景反而更慢了我才意识到不同模型的长上下文实现差别巨大有的是真长有的是稀疏注意力有的对前缀复用做了优化有的没做第五种最致命有个客户提了一个跨多份文档的问题 比较这五份合同的付款条款 我天真地把 5 份文档全拼进 prompt 模型先是慢得要死然后回答得一塌糊涂前几份的内容几乎被忽略我盯着这一连串问题想了很久才彻底想明白第一版错在一个根本的认知上我以为 长上下文窗口就是把所有内容塞进 prompt 让模型自己处理 可这个认知是错的真正能在生产里用的长文本系统是一个分块加 KV Cache 加检索增强加上下文压缩加跨文档协同的多层架构 任何一层做不好都会让长上下文变成又慢又贵又不准的灾难本文从头梳理为什么 128k 窗口不等于真的能处理 128k 文本分块策略怎么设计 KV Cache 怎么复用 RAG 和长上下文怎么组合上下文压缩有哪些做法以及一些把长文本能力做扎实要避开的工程坑
问题背景:为什么 LLM 长上下文比想象中难用
很多人对 LLM 长上下文的印象是 模型支持 128k 那我直接把文档塞进去就好了 实际工程中这种用法几乎一上线就崩成本飙升延迟拉长准确率不升反降。问题的根源在于:
- 注意力的二次复杂度:Transformer 注意力是 O(n²) 即使做了 FlashAttention 或稀疏化长上下文的推理成本和延迟仍然远高于短上下文。
- lost in the middle 现象:研究表明模型对上下文开头和结尾的注意力远强于中间长上下文里中间部分的信息经常被 忽略 这是模型本身的特性不是 bug。
- KV Cache 的工程复杂度:长上下文的 KV Cache 体积巨大单个请求可能占用数 GB 显存复用策略不对就会爆显存或者跨请求污染。
- token 成本随上下文线性增长:128k token 一次调用可能 1-2 美元高频业务根本扛不住成本。
- 不同模型对长上下文的实现差别巨大:GPT-4-128k Claude 200k Gemini 1M 各自的稀疏注意力前缀缓存 attention sinks 实现都不同性能曲线和准确率特征也都不同。
- 长上下文不等于推理能力:即使模型 看到了 所有信息也不代表它能在所有信息上做正确的推理跨段对比这类任务模型表现仍然很差。
一 上下文窗口与 Token 预算:第一性原理
在做任何架构设计之前必须先理解一个 token 在 LLM 里到底意味着什么。token 不是字也不是词是 BPE 分词后的子词单元中文一个字大约是 1.5-2 个 token 英文一个单词大约是 1-1.3 个 token。所以一份 30 万字的中文文档大约是 50-60 万 token 远超任何模型的窗口。即使是 GPT-4-128k 实际有效输入也只能放约 30-40 万中文字符还得留出输出空间。
def estimate_tokens(text: str, lang: str = 'zh') -> int:
if lang == 'zh':
return int(len(text) * 1.8)
return int(len(text.split()) * 1.3)
class TokenBudget:
def __init__(self, model: str):
self.limits = {
'gpt-4-128k': 128000,
'gpt-4-turbo': 128000,
'claude-3-200k': 200000,
'gemini-1.5-pro': 1000000,
}
self.model = model
self.max_tokens = self.limits.get(model, 8192)
self.reserved_output = 4096
self.reserved_system = 1000
def available_for_context(self) -> int:
return self.max_tokens - self.reserved_output - self.reserved_system
def fit(self, system: str, user: str, docs: list[str]) -> list[str]:
budget = self.available_for_context()
used = estimate_tokens(system) + estimate_tokens(user)
result = []
for doc in docs:
t = estimate_tokens(doc)
if used + t > budget:
break
result.append(doc)
used += t
return result
这个工具函数看似简单却是所有长文本系统的起点。我见过太多团队直接把文档拼进 prompt 结果某次输入超了上限模型返回 context length exceeded 错误用户看到的就是 系统繁忙请稍后再试 这种体验。预算管理必须放在调用前不能依赖模型自己拒绝预留输出空间也必须在预算里减掉否则模型生成到一半被截断。除此之外我们还会在调用前后双重校验防止网络重试时把同一份长 prompt 重复发送导致账单被刷:
import time
import logging
class GuardedLLMClient:
def __init__(self, llm, budget: TokenBudget, cost_per_1k: float = 0.01):
self.llm = llm
self.budget = budget
self.cost_per_1k = cost_per_1k
self.daily_budget_usd = 100.0
self.spent_today = 0.0
self.day_started_at = time.time()
def _reset_if_new_day(self):
if time.time() - self.day_started_at > 86400:
self.spent_today = 0.0
self.day_started_at = time.time()
def call(self, prompt: str, max_output: int = 2048) -> str:
self._reset_if_new_day()
input_tokens = estimate_tokens(prompt)
if input_tokens + max_output > self.budget.max_tokens:
raise ValueError(f'prompt {input_tokens} + output {max_output} 超过窗口')
est_cost = (input_tokens + max_output) / 1000 * self.cost_per_1k
if self.spent_today + est_cost > self.daily_budget_usd:
logging.error(f'当日预算已用尽 spent={self.spent_today:.2f}')
raise RuntimeError('daily budget exceeded')
result = self.llm.complete(prompt, max_tokens=max_output)
self.spent_today += est_cost
return result
二 文档分块策略:分多大 怎么分 重叠多少
分块 chunking 是长文本系统的第一层关键决策。分得太小语义被切碎 检索回来的片段不完整。分得太大单个 chunk 信息密度低相关度差。常见的几种策略各有适用场景:
from typing import List
class FixedSizeChunker:
def __init__(self, size: int = 800, overlap: int = 100):
self.size = size
self.overlap = overlap
def split(self, text: str) -> List[str]:
chunks = []
start = 0
while start < len(text):
end = min(start + self.size, len(text))
chunks.append(text[start:end])
if end == len(text):
break
start = end - self.overlap
return chunks
class SemanticChunker:
def __init__(self, max_size: int = 1000):
self.max_size = max_size
self.separators = ['\n\n', '\n', '。', '!', '?', ';']
def split(self, text: str) -> List[str]:
for sep in self.separators:
if sep in text and len(text) > self.max_size:
parts = text.split(sep)
chunks = []
buf = ''
for p in parts:
if len(buf) + len(p) < self.max_size:
buf += p + sep
else:
if buf:
chunks.append(buf)
buf = p + sep
if buf:
chunks.append(buf)
return chunks
return [text]
class StructuredChunker:
def split_markdown(self, text: str) -> List[dict]:
chunks = []
current_section = {'title': '', 'level': 0, 'content': ''}
for line in text.split('\n'):
if line.startswith('#'):
if current_section['content']:
chunks.append(current_section)
level = len(line) - len(line.lstrip('#'))
current_section = {
'title': line.lstrip('#').strip(),
'level': level,
'content': '',
}
else:
current_section['content'] += line + '\n'
if current_section['content']:
chunks.append(current_section)
return chunks
固定大小分块实现简单适合非结构化文本但容易切断句子。语义分块按句号段落分能保持语义完整但实现复杂。结构化分块利用文档自身的章节标题表格代码块切分对 Markdown PDF 这类有明显结构的文档效果最好。真实工程里我们的做法是先按章节切大块再在大块内按 token 预算二次切并且每个 chunk 都带上父章节的标题作为上下文 这一招对检索准确率有 15-20% 的提升 因为模型能从标题理解片段所在的位置和主题。
关于重叠 overlap 的建议是设为 chunk size 的 10-15%。重叠的作用是当一句话被切断时下一个 chunk 能补全。重叠太大 会增加存储和检索成本 太小则容易丢信息。
三 KV Cache 复用:省时间和钱的关键
KV Cache 是 Transformer 推理的核心优化。当你给模型一段 prompt 模型会把每个 token 的 Key 和 Value 缓存下来。下一个 token 生成时只需要计算新 token 的 Q 与所有历史的 K V 做注意力不用重算前面的 K V。这是单次推理内部的优化。但在多轮对话或同一文档多次提问的场景 跨请求的 KV Cache 复用 是省成本的杀手锏。
import hashlib
from collections import OrderedDict
class PrefixCache:
def __init__(self, max_entries: int = 100):
self.cache = OrderedDict()
self.max_entries = max_entries
def _hash(self, prefix: str) -> str:
return hashlib.sha256(prefix.encode()).hexdigest()
def get(self, prefix: str):
key = self._hash(prefix)
if key in self.cache:
self.cache.move_to_end(key)
return self.cache[key]
return None
def put(self, prefix: str, kv_cache_handle):
key = self._hash(prefix)
if key in self.cache:
self.cache.move_to_end(key)
self.cache[key] = kv_cache_handle
return
if len(self.cache) >= self.max_entries:
self.cache.popitem(last=False)
self.cache[key] = kv_cache_handle
class PromptBuilder:
def __init__(self, cache: PrefixCache):
self.cache = cache
def build(self, system: str, document: str, question: str):
prefix = f'{system}\n\n文档:\n{document}'
kv_handle = self.cache.get(prefix)
if kv_handle is not None:
return {'reuse_kv': True, 'kv_handle': kv_handle, 'new_tokens': question}
return {'reuse_kv': False, 'full_prompt': f'{prefix}\n\n问题:\n{question}'}
OpenAI 和 Anthropic 都在 2024 年推出了 prompt caching 功能。OpenAI 的自动 caching 对 1024 token 以上的前缀 自动复用 缓存的 token 收 50% 的费用且延迟降低 40-80%。Anthropic 的 Claude prompt caching 是显式的需要在 API 调用时标记哪些是 cacheable 缓存的 token 写入收 25% 额外费用读取只收 10%。对于知识库 客服 这种 system prompt + 长文档 + 用户问题 的场景 cached 部分往往占 90% 以上 实际成本能降 80% 延迟能降 70% 这是长文本场景里 ROI 最高的一个优化。
但 KV Cache 复用必须注意:前缀必须完全一致包括空格换行甚至 token 序列 不能在 system prompt 里塞动态变量比如当前时间用户名 否则缓存全部失效。我们早期吃过这个亏因为在 system 里加了一句 当前时间是 xxx 导致 prompt 每次都不同 caching 命中率为 0 成本一分没省。后来把动态变量挪到 user message 末尾命中率立刻回到 95%。
[mermaid]flowchart TD
A[用户提问] --> B{Prompt 前缀
是否命中缓存}
B -->|命中| C[复用 KV Cache
只算新 token]
B -->|未命中| D[全量计算 KV]
C --> E[输出生成]
D --> F[写入 KV Cache]
F --> E
E --> G{是否同文档
多轮问答}
G -->|是| H[保留 KV Cache
等待下一问]
G -->|否| I[释放 KV Cache]
H --> A
四 RAG 与长上下文的组合策略
很多人把 RAG 检索增强 和 长上下文 当成互相替代的方案 但生产里它们其实是互补的。纯 RAG 的问题是 检索可能漏召回 或者把上下文割裂 模型看不到全局。纯长上下文的问题是 慢 贵 lost in the middle。真正能扛的方案是 RAG + 长上下文混合 让 RAG 先做粗筛长上下文做精读。
from dataclasses import dataclass
from typing import List, Optional
@dataclass
class Chunk:
id: str
text: str
embedding: list
metadata: dict
class HybridRetriever:
def __init__(self, embedder, reranker, vector_store, bm25_index):
self.embedder = embedder
self.reranker = reranker
self.vector_store = vector_store
self.bm25 = bm25_index
def retrieve(self, query: str, k: int = 50) -> List[Chunk]:
q_emb = self.embedder.embed(query)
vector_hits = self.vector_store.search(q_emb, top_k=k)
bm25_hits = self.bm25.search(query, top_k=k)
merged = self._rrf_merge(vector_hits, bm25_hits)
reranked = self.reranker.rank(query, merged, top_k=10)
return reranked
def _rrf_merge(self, hits_a, hits_b, k: int = 60):
scores = {}
for rank, hit in enumerate(hits_a):
scores[hit.id] = scores.get(hit.id, 0) + 1 / (k + rank)
for rank, hit in enumerate(hits_b):
scores[hit.id] = scores.get(hit.id, 0) + 1 / (k + rank)
ids = sorted(scores, key=scores.get, reverse=True)
all_hits = {h.id: h for h in hits_a + hits_b}
return [all_hits[i] for i in ids]
class LongContextAnswerer:
def __init__(self, llm, retriever: HybridRetriever, budget: 'TokenBudget'):
self.llm = llm
self.retriever = retriever
self.budget = budget
def answer(self, question: str) -> str:
chunks = self.retriever.retrieve(question, k=50)
fitted = self.budget.fit('你是一位资深分析师', question, [c.text for c in chunks])
context = self._arrange_with_anti_lost_in_middle(fitted, question)
prompt = f'参考资料:\n{context}\n\n问题:{question}\n\n请基于以上资料严谨回答'
return self.llm.complete(prompt)
def _arrange_with_anti_lost_in_middle(self, chunks: list, question: str) -> str:
if len(chunks) <= 2:
return '\n\n'.join(chunks)
head = chunks[:len(chunks) // 2]
tail = chunks[len(chunks) // 2:]
arranged = head + tail[::-1]
return '\n\n---\n\n'.join(arranged)
这里有几个工程细节值得展开。第一是混合检索 BM25 + 向量 用 Reciprocal Rank Fusion 合并 比单一向量检索准确率高 10-15% 因为 BM25 擅长关键词精确匹配 向量擅长语义模糊匹配两者互补。第二是 Reranker 重排序 用 cross-encoder 比如 bge-reranker 对粗排结果做精排能再提升 5-10% 准确率代价是多一次推理。第三是 lost in the middle 的缓解 把最相关的 chunks 放在 prompt 开头和结尾 不重要的放中间 准确率能提升 8% 以上 这是 Stanford 2023 年那篇论文 lost in the middle 的核心发现的实操化。
五 上下文压缩与精炼:让长文本变短
有时候即便分块和 RAG 都做了上下文仍然太长比如客户问 总结这份 500 页财报的所有亮点 RAG 召回 50 个 chunk 拼起来还是 8 万 token。这时就需要上下文压缩。压缩有几种典型做法:
class ContextCompressor:
def __init__(self, llm_small, llm_large):
self.llm_small = llm_small
self.llm_large = llm_large
def compress_by_summary(self, chunks: list[str], query: str) -> str:
summaries = []
for chunk in chunks:
prompt = f'请从以下文本中提取与问题相关的关键信息忽略无关内容\n\n问题:{query}\n\n文本:\n{chunk}\n\n关键信息:'
summary = self.llm_small.complete(prompt, max_tokens=200)
if summary.strip():
summaries.append(summary)
return '\n\n'.join(summaries)
def compress_by_filter(self, chunks: list[str], query: str, threshold: float = 0.6):
scored = []
for c in chunks:
score = self.llm_small.relevance_score(query, c)
if score >= threshold:
scored.append((score, c))
scored.sort(reverse=True)
return [c for _, c in scored]
def map_reduce(self, chunks: list[str], query: str) -> str:
partial = []
for chunk in chunks:
ans = self.llm_small.complete(f'根据这段文本回答 如果不能回答说不知道\n文本:{chunk}\n问题:{query}')
if '不知道' not in ans:
partial.append(ans)
if not partial:
return '在所提供材料中未找到答案'
combined = '\n'.join(partial)
final = self.llm_large.complete(f'以下是多段答案的汇总请综合给出最终回答\n\n{combined}\n\n问题:{query}')
return final
这三种策略各有适用场景。compress_by_summary 用小模型做摘要适合每个 chunk 都有一定价值的场景 比如客户综合提问 缺点是会丢失部分细节。compress_by_filter 用 LLM 做相关性打分只保留高分 chunk 适合检索召回有噪音的场景 干净简单但可能漏关键信息。map_reduce 是经典的 LangChain 方案 适合跨文档跨章节的对比类问题 缺点是 LLM 调用次数多 延迟和成本会上去。生产里我们的做法是按问题类型路由 简单事实问题走 filter 综合分析问题走 summary 跨文档对比走 map_reduce。
六 长文本系统的工程坑:那些文档里学不到的
讲完了原理来说几个真实生产里踩过的坑。第一个坑是 上下文污染 多用户同时使用同一个 LLM 实例时 如果上一个请求的 KV Cache 没清干净 下一个请求可能拿到上一个的残留 这在 vLLM 这类带 prefix cache 的推理框架里特别容易出现 必须严格按用户隔离会话或者用一次性的随机 token 做请求分隔。第二个坑是 token 计数不准 各家分词器对中文 emoji 特殊符号的处理不同你用 tiktoken 估算的 token 数和实际调用的扣费数能差 10-20% 一定要按模型自带的分词器做精确计算否则成本会失控。
import tiktoken
class AccurateTokenCounter:
def __init__(self, model: str):
self.encoder = tiktoken.encoding_for_model(model)
def count(self, text: str) -> int:
return len(self.encoder.encode(text))
def count_messages(self, messages: list[dict]) -> int:
total = 0
per_message_overhead = 4
per_name_overhead = 1
for m in messages:
total += per_message_overhead
for key, value in m.items():
total += self.count(value)
if key == 'name':
total += per_name_overhead
total += 3
return total
class StreamingTruncator:
def __init__(self, max_chunks_in_window: int = 20):
self.max_chunks = max_chunks_in_window
self.window = []
def add(self, chunk: str):
self.window.append(chunk)
if len(self.window) > self.max_chunks:
self.window.pop(0)
def get_context(self) -> str:
return '\n'.join(self.window)
第三个坑是 attention sinks 现象 LLM 对开头几个 token 有异常高的注意力 如果在流式对话场景 滑动窗口直接丢掉开头 模型表现会断崖式下滑 必须保留前面几个 anchor token 这是 MIT 2023 年 StreamingLLM 那篇论文的核心结论。下面是一个保留 anchor 的滑动窗口实现 它把开头几个 token 永远固定住 后面才按滑动窗口去淘汰旧消息:
class AnchoredSlidingWindow:
def __init__(self, anchor_size: int = 4, window_size: int = 2048):
self.anchor_size = anchor_size
self.window_size = window_size
self.anchor_tokens = []
self.window_tokens = []
def feed(self, new_tokens: list[int]):
if len(self.anchor_tokens) < self.anchor_size:
need = self.anchor_size - len(self.anchor_tokens)
self.anchor_tokens.extend(new_tokens[:need])
new_tokens = new_tokens[need:]
self.window_tokens.extend(new_tokens)
overflow = len(self.window_tokens) - self.window_size
if overflow > 0:
self.window_tokens = self.window_tokens[overflow:]
def get_input(self) -> list[int]:
return self.anchor_tokens + self.window_tokens
def reset(self):
self.window_tokens = []
第四个坑是 长上下文场景下输出截断 模型为了省成本会偏向给更短的回答 你必须在 prompt 里显式约束 请给出详尽的回答 不要省略 否则用户体验会变差。第五个坑是 缓存与一致性的矛盾 你为了省成本激进地复用 KV Cache 但用户希望模型 实时 反映最新文档变更 这两者必须做权衡 我们的方案是 对文档级别的更新 直接 invalidate 整个文档的 cache 接受一次冷启动 不能为了省一点钱让用户看到陈旧答案。
关键概念速查
| 概念 | 含义 | 工程价值 |
|---|---|---|
| Context Window | 模型一次能接受的最大 token 数 | 架构上限不能用算法绕过 |
| KV Cache | 注意力 Key Value 的缓存 | 单请求加速跨请求省钱 |
| Prompt Caching | OpenAI/Anthropic 的前缀复用 | 典型场景成本降 80% |
| Lost in the Middle | 模型忽略上下文中间内容 | 把关键信息放头尾 |
| Chunking | 把长文档切成片段 | RAG 检索精度的基础 |
| Overlap | 相邻 chunk 间的重叠字符 | 避免切断关键句子 |
| RRF | Reciprocal Rank Fusion | 合并多路检索结果 |
| Reranker | 对粗排结果做精排 | 准确率再提 5-10% |
| Map-Reduce | 分段回答再汇总 | 跨文档综合提问的方案 |
| Attention Sinks | 开头 token 的高注意力 | 滑动窗口必须保留 anchor |
避坑清单
- 不要把整篇长文档塞进 prompt 即使模型支持 128k 实际有效上下文也只有一半左右且成本爆炸。
- 分块必须保留章节标题作为父级上下文否则模型只看到一段没有出处的文字检索准确率大幅下降。
- 重叠 overlap 设为 chunk 大小的 10-15% 太小切断句子太大浪费存储。
- system prompt 里禁止放任何动态变量 否则 prompt caching 命中率为 0 全部白费。
- 检索一定要混合 BM25 + 向量 不要只用向量 中文场景关键词匹配也很重要。
- 把最相关的 chunks 放 prompt 开头和结尾 中间放次要内容 缓解 lost in the middle。
- token 计数必须用模型自带的分词器估算 自己估算会和真实扣费差 10-20%。
- 多用户场景 KV Cache 必须按会话严格隔离 不能跨用户复用否则可能泄漏隐私。
- 滑动窗口截断时保留开头几个 anchor token 否则模型会因为 attention sinks 缺失而崩溃。
- 文档更新后必须 invalidate 对应缓存 不能为了省一点成本让用户看到陈旧答案。
总结
长上下文这件事很多人的第一直觉是 既然模型支持 128k 那我直接塞进去就好了 这其实是把模型能力和工程能力混为一谈。模型支持 128k 是说它能接受这么多 token 不会报错 但它能不能在这些 token 上做出准确的推理 你愿不愿意为这次调用付 1 美元的成本 用户能不能接受 30 秒的延迟 这些都是另外的问题 而这些问题恰恰是工程要解决的。
从原型到生产 你需要做的事远不止 调用模型 这一件事。你要算 token 预算 设计分块策略 配置 KV Cache 复用 实现混合检索 引入 Reranker 处理 lost in the middle 设计上下文压缩 应对各种边界情况。每一项单独看都不复杂 但它们组合在一起 才是一个能扛流量的长文本系统。少任何一项 都会在某个客户的某次请求里 把你刚才省下的那点成本 连本带利地还回去 而且通常以 服务不可用 的形式还。
我经常用一个比喻来理解长上下文 它有点像高速公路。模型支持 128k 就像高速限速 120 但你能不能开到 120 取决于车况路况天气你的驾驶水平。你不可能因为限速 120 就闭眼踩油门更不能因为你的车跑得到 120 就否定信号灯红绿灯交规的价值。分块 检索 压缩 缓存 这些工程化手段就是长上下文系统的 交通规则 没有它们你就是在裸奔。
这套架构最难的地方在于 它的复杂度在 demo 阶段几乎完全暴露不了。你给老板演示一两个问题模型回答得很漂亮你觉得搞定了 但真正上线你会发现 99% 的复杂度都在 那 1% 的极端 case 里 跨文档对比 模糊提问 海量并发 数据频繁更新 攻击性输入。所以建议任何想做长文本系统的团队 上线前一定要做大量的真实流量回放和压测 千万不要拿你自己设计的两个问题去测就上线 那种系统几乎一定会在第一个月就炸给你看 而你那时连复盘都不知道从哪开始。
—— 别看了 · 2026