核心问题：模型训练时只用了4K长度的数据，为什么推理时能处理128K甚至更长的文本？KV Cache是什么？缓存命中又是什么意思？
目录
- 第一章：上下文是什么——建立直觉
- 第二章：位置编码——让模型理解"顺序"
- 第三章：上下文长度为什么有限制
- 第四章：上下文扩展技术——让模型"看得更远"
- 第五章：KV Cache——推理加速的核心机制
- 第六章：KV Cache缓存命中与优化
- 第七章：长上下文推理优化技术
- 第八章：实战与工程实践
- 附录：术语表与参考资料
第一章：上下文是什么——建立直觉
1.1 什么是上下文长度？
上下文长度(Context Length) = 模型一次能处理的最大token数量
类比理解：
上下文长度 ≈ 你的"短期记忆容量"
– 你读一段话（100个token）：轻松理解 ✓
– 你读一本书（10万个token）：需要记笔记，否则记不住前面的内容
大模型也一样：
– 上下文长度4K：一次能处理约3000个中文字
– 上下文长度128K：一次能处理约10万个中文字
– 上下文长度1M：一次能处理约75万个中文字
1.2 为什么上下文长度很重要？
场景1：长文档理解
用户：请帮我总结这份100页的报告
如果上下文只有4K → 放不下100页 → 无法完成
如果上下文有128K → 可以一次性处理 → 完美完成
场景2：多轮对话
用户和AI聊了100轮对话
每轮约200 token → 总计2万token
如果上下文只有4K → 聊到第20轮就"忘了"前面的内容
如果上下文有128K → 可以记住全部对话
场景3：代码理解
用户：请帮我review这个2000行的代码文件
2000行 ≈ 1.5万token
如果上下文只有4K → 放不下完整代码
如果上下文有128K → 轻松处理
1.3 主流模型的上下文长度（2026年）
┌───────────────────┬──────────────┬────────────────┐
│ 模型 │ 上下文长度 │ 约等于 │
├───────────────────┼──────────────┼────────────────┤
│ GPT-4 Turbo │ 128K tokens │ ~10万中文字 │
│ GPT-4o │ 128K tokens │ ~10万中文字 │
│ Claude 3.5 Sonnet │ 200K tokens │ ~15万中文字 │
│ Claude Opus 4 │ 200K tokens │ ~15万中文字 │
│ Gemini 1.5 Pro │ 2M tokens │ ~150万中文字 │
│ Qwen2.5-7B │ 128K tokens │ ~10万中文字 │
│ LLaMA 3.1 70B │ 128K tokens │ ~10万中文字 │
│ DeepSeek-V3 │ 128K tokens │ ~10万中文字 │
│ Mistral Large │ 128K tokens │ ~10万中文字 │
└───────────────────┴──────────────┴────────────────┘
第二章：位置编码——让模型理解"顺序"
2.1 为什么需要位置编码？
Transformer的核心：Self-Attention
Self-Attention的本质：计算每个token与其他token的相关性
问题：
Self-Attention是"位置无关"的
"我爱你" 和 "你爱我" 的Attention计算完全相同！
模型不知道哪个词在前、哪个词在后
解决方案：位置编码
给每个token注入位置信息
模型就能区分"第1个token"和"第5个token"
2.2 绝对位置编码（原始Transformer）
原始Transformer使用正弦位置编码：
PE(pos, 2i) = sin(pos / 10000^(2i/d))
PE(pos, 2i+1) = cos(pos / 10000^(2i/d))
其中 pos 是位置，i 是维度索引，d 是嵌入维度
token embedding = word_embedding + position_embedding
问题：
训练时最大位置 = 4096（假设训练长度4K）
推理时如果位置 = 8192（超过训练长度）
→ 位置编码是模型从未见过的 → 泛化能力差
2.3 RoPE（旋转位置编码）——当前主流
RoPE = Rotary Position Embedding = 旋转位置编码
几乎所有主流大模型都使用RoPE（Qwen、LLaMA、DeepSeek等）
核心思想：
不是把位置信息"加"到token embedding上
而是把位置信息"旋转"到注意力计算中
对于位置m的token：
q_m = R(m) × q (R是旋转矩阵)
k_n = R(n) × k
注意力分数：
q_m · k_n = (R(m)×q) · (R(n)×k) = q · R(m-n)×k
关键洞察：
注意力分数只取决于相对位置 (m-n)，而不是绝对位置m和n
→ 模型学习的是"两个token之间的距离关系"
RoPE的直觉理解：
想象每个token有一个"方向"（向量）
RoPE根据token的位置，把这个向量旋转一个角度
位置0: 旋转0°
位置1: 旋转θ
位置2: 旋转2θ
位置N: 旋转Nθ
两个token的注意力 = 它们向量的点积
点积取决于两个向量的相对角度 → 取决于相对位置
这就是为什么RoPE能泛化到更长的上下文：
即使位置N超过了训练时的最大位置
只要相对位置在训练范围内，模型就能正常工作
2.4 ALiBi（Attention with Linear Biases）
ALiBi = 带线性偏置的注意力
部分模型使用（如BLOOM）
核心思想：
直接在注意力分数上加一个与距离成正比的惩罚
距离越远，惩罚越大
Attention(q, k) = q·k^T / √d – m × |i-j|
其中 m 是一个超参数，|i-j| 是两个token的距离
效果：
– 距离近的token注意力更强（局部信息优先）
– 自然地泛化到更长序列（线性惩罚外推性好）
2.5 位置编码对比
┌────────────┬──────────────┬──────────────┬──────────────┐
│ 方法 │ 代表模型 │ 外推能力 │ 训练效率 │
├────────────┼──────────────┼──────────────┼──────────────┤
│ 正弦编码 │ 原始Transformer│ 差 │ 高 │
│ 可学习编码 │ BERT, GPT-2 │ 差 │ 高 │
│ RoPE │ Qwen, LLaMA │ 中(需扩展) │ 高 │
│ ALiBi │ BLOOM │ 好 │ 高 │
│ YaRN │ 长上下文扩展 │ 最好 │ 中 │
└────────────┴──────────────┴──────────────┴──────────────┘
第三章：上下文长度为什么有限制
3.1 限制1：训练数据长度
模型训练时使用固定的最大序列长度
例如：Qwen2.5-7B训练时最大长度 = 4096 token
这意味着：
位置编码只在 [0, 4096] 范围内训练过
位置 = 4097 的token，模型从未见过
→ 直接使用更长序列，位置编码会"出错"
3.2 限制2：注意力计算复杂度
标准Self-Attention的计算量：
O(n² × d)
其中 n = 序列长度，d = 隐藏维度
n = 4K: 计算量 = 16M × d
n = 128K: 计算量 = 16G × d (1000倍！)
n = 1M: 计算量 = 1T × d (62500倍！)
→ 上下文越长，计算越慢，显存需求越大
3.3 限制3：KV Cache显存
KV Cache存储所有历史token的Key和Value
显存占用 = 2 × n_layers × n_heads × seq_len × head_dim × bytes
以Qwen2.5-7B为例（32层，32头，head_dim=128）：
4K长度: 2 × 32 × 32 × 4096 × 128 × 2B = 2GB
128K长度: 2 × 32 × 32 × 131072 × 128 × 2B = 64GB ← 单张A100放不下！
1M长度: 2 × 32 × 32 × 1048576 × 128 × 2B = 512GB
→ 更长上下文需要更多显存
3.4 限制4："迷失在中间"问题
即使模型支持长上下文，也不意味着它能有效利用所有信息
Lost in the Middle现象：
模型对序列开头和结尾的信息记忆较好
对中间部分的信息记忆较差
[开头信息] ………… [中间信息] ………… [结尾信息]
记忆好 记忆差 记忆好
这是注意力机制的固有特性，通过长上下文训练可以缓解
第四章：上下文扩展技术——让模型"看得更远"
4.1 为什么能扩展？——核心原理
问题：模型训练时只用了4K长度，为什么推理时能处理128K？
答案：位置编码的"外推"和"插值"技术
关键洞察：
RoPE编码的核心是"相对位置"
如果我们能把长序列的相对位置"压缩"到训练范围内
模型就能处理更长的序列
类比：
训练时见过的最大距离是4K
现在需要处理128K的序列
→ 把128K的位置"缩放/映射"到4K范围内
→ 模型就能"理解"这些位置了
4.2 位置插值 (Position Interpolation, PI)
Meta提出的最基础的扩展方法
核心思想：
不改变位置编码的结构
直接把位置值缩小
原始：位置 pos ∈ [0, L_new]
插值后：pos × (L_train / L_new) ∈ [0, L_train]
示例：
训练长度 L_train = 4096
目标长度 L_new = 16384
缩放因子 = 4096 / 16384 = 0.25
位置 8192 → 8192 × 0.25 = 2048（在训练范围内）
优点：简单有效
缺点：所有位置被"压缩"，近距离的分辨率降低
4.3 NTK-Aware Scaling
核心改进：
不是均匀缩放所有位置
而是根据频率不同，采用不同的缩放策略
RoPE的不同维度有不同的"频率"：
– 低频维度：编码远距离关系
– 高频维度：编码近距离关系
NTK-Aware Scaling：
– 高频维度不缩放（保持近距离精度）
– 低频维度缩放（扩展远距离范围）
这样既保持了近距离的精度，又扩展了远距离的范围
4.4 YaRN (Yet another RoPE extensioN)
目前效果最好的RoPE扩展方法
核心创新：
1. NTK-aware插值（不同频率不同缩放）
2. 注意力温度缩放（调整注意力分布）
3. 分段处理（近距离/远距离不同策略）
效果：
– 训练4K → 推理128K，精度损失极小
– 不需要大量长文本数据微调
– 或者只需少量长文本数据就能很好扩展
YaRN被Qwen2.5、DeepSeek-V3等主流模型采用
4.5 动态NTK (Dynamic NTK)
核心思想：
不使用固定的缩放因子
而是根据当前输入的实际长度动态调整
如果输入长度 < 训练长度 → 不缩放
如果输入长度 > 训练长度 → 按比例缩放
优点：
– 短文本完全不受影响
– 长文本自动适配
– 无需重新训练
4.6 LongRoPE
微软提出的方法，支持超长上下文扩展（2048K）
核心创新：
1. 搜索最优的缩放因子（不是简单公式计算）
2. 渐进式扩展（先扩展到32K，再扩展到128K，…）
3. 短文本也保持精度
效果：训练2K → 推理2048K
4.7 长上下文训练策略
除了位置编码扩展，还需要长文本训练：
策略1：渐进式训练
训练阶段1：4K长度 × 大量数据
训练阶段2：32K长度 × 长文本数据
训练阶段3：128K长度 × 超长文本数据
策略2：ABF (Adjusted Base Frequency)
调整RoPE的base frequency
从10000增加到更大的值（如500000）
让位置编码覆盖更大范围
策略3：长短混合训练
同时使用短文本和长文本训练
短文本：保持基础能力
长文本：提升长上下文能力
第五章：KV Cache——推理加速的核心机制
5.1 为什么需要KV Cache？
问题：Transformer推理时，每生成一个新token都要重新计算所有历史token的Key和Value
生成第1个token：计算1个token的KV
生成第2个token：计算2个token的KV（重复计算第1个！）
生成第3个token：计算3个token的KV（重复计算第1、2个！）
…
生成第N个token：计算N个token的KV（重复计算所有历史！）
总计算量 = 1 + 2 + 3 + … + N = N(N+1)/2 = O(N²)
大量重复计算！
解决方案：KV Cache
缓存已经计算过的Key和Value
每次只计算新token的KV，然后和缓存拼接
总计算量 = 1 + 1 + 1 + … + 1 = N = O(N)
从O(N²)降到O(N)！
5.2 KV Cache的工作原理
┌─────────────────────────────────────────────────────────────┐
│ KV Cache 工作原理 │
│ │
│ ═══ 生成第1个token ═══ │
│ │
│ 输入: [token_0] │
│ 计算: K₀ = W_K × x₀, V₀ = W_V × x₀ │
│ 缓存: K_cache = [K₀], V_cache = [V₀] │
│ 输出: token_1 │
│ │
│ ═══ 生成第2个token ═══ │
│ │
│ 输入: [token_1]（只需要新token！） │
│ 计算: K₁ = W_K × x₁, V₁ = W_V × x₁ │
│ 缓存: K_cache = [K₀, K₁], V_cache = [V₀, V₁] │
│ Attention: Q₁ 与 [K₀, K₁] 做注意力 │
│ 输出: token_2 │
│ │
│ ═══ 生成第N个token ═══ │
│ │
│ 输入: [token_N]（只需要新token！） │
│ 计算: K_N, V_N │
│ 缓存: K_cache = [K₀…K_N], V_cache = [V₀…V_N] │
│ Attention: Q_N 与所有缓存的K做注意力 │
│ 输出: token_N+1 │
│ │
│ 关键：历史token的K和V不需要重新计算！ │
└─────────────────────────────────────────────────────────────┘
5.3 KV Cache的显存计算
KV Cache的显存占用：
单个token的KV显存 = 2 × n_layers × n_heads × head_dim × bytes
其中：
– 2: Key和Value各一份
– n_layers: Transformer层数
– n_heads: 注意力头数
– head_dim: 每个头的维度
– bytes: 数据类型字节数（FP16=2字节）
以Qwen2.5-7B为例：
n_layers=28, n_heads=28, head_dim=128
单token KV显存 = 2 × 28 × 28 × 128 × 2B = 401,408B ≈ 0.4MB
不同上下文长度的KV Cache：
4K tokens: 0.4MB × 4096 = 1.6 GB
16K tokens: 0.4MB × 16384 = 6.4 GB
128K tokens: 0.4MB × 131072 = 51.2 GB ← 超过单张4090显存！
1M tokens: 0.4MB × 1048576 = 400 GB ← 需要多卡分片
5.4 Prefill阶段 vs Decode阶段
大模型推理分为两个阶段：
┌─────────────────────────────────────────────────────────────┐
│ Prefill阶段（预填充） │
│ │
│ 输入: 用户的完整提示（如1000个token） │
│ 处理: 一次性计算所有token的KV，填充KV Cache │
│ 特点: 并行计算，GPU利用率高，速度快 │
│ 输出: KV Cache + 第一个生成的token │
└─────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────┐
│ Decode阶段（解码/生成） │
│ │
│ 输入: 上一步生成的1个token │
│ 处理: 计算新token的KV，与缓存拼接，生成下一个token │
│ 特点: 串行计算，GPU利用率低（每次只处理1个token） │
│ 输出: 一个新的token │
│ │
│ 重复直到生成结束标记 <eos> │
└─────────────────────────────────────────────────────────────┘
Prefill: 1次前向传播，处理所有输入token（高并行）
Decode: N次前向传播，每次只处理1个token（低并行，但有KV Cache加速）
第六章：KV Cache缓存命中与优化
6.1 什么是缓存命中？
缓存命中(Cache Hit) = KV Cache中已经有需要的token的KV，不需要重新计算
类比理解：
你在图书馆找书
– 缓存命中：书就在你桌上 → 直接看（快）
– 缓存未命中：书在书架上 → 需要去取（慢）
在大模型推理中的含义：
场景1：多轮对话
第1轮：用户: "你好" → Prefill计算"你好"的KV → 缓存
第2轮：用户: "你好，请介绍一下自己"
→ "你好"的KV已经在缓存中 → 命中！
→ 只需要计算"请介绍一下自己"的KV → 节省计算
场景2：系统提示复用
系统提示: "你是一个有帮助的AI助手。"
每次用户提问时，系统提示不需要重新计算
→ 直接使用缓存的KV → 命中！
6.2 Prefix Caching（前缀缓存）
核心思想：
如果两个请求有相同的前缀
它们的KV Cache可以共享
请求A: [系统提示] + [用户问题A]
请求B: [系统提示] + [用户问题B]
系统提示的KV Cache可以复用！
不使用Prefix Caching：
请求A: 计算全部KV → 10ms
请求B: 计算全部KV → 10ms
总计: 20ms
使用Prefix Caching：
请求A: 计算全部KV → 10ms，缓存系统提示部分
请求B: 命中系统提示缓存 + 只计算用户问题B → 5ms
总计: 15ms（节省25%）
6.3 vLLM中的PagedAttention
vLLM是目前最流行的大模型推理框架
PagedAttention是其核心优化技术
传统KV Cache的问题：
每个请求需要连续的显存空间
但生成长度未知 → 需要预分配最大长度 → 浪费显存
PagedAttention的解决方案：
像操作系统的虚拟内存一样管理KV Cache
将KV Cache分成固定大小的"页"(Page)
按需分配，不需要连续内存
┌────────────────────────────────────────────┐
│ 物理显存 │
│ ┌───┐ ┌───┐ ┌───┐ ┌───┐ ┌───┐ ┌───┐ │
│ │ P0│ │ P1│ │ P2│ │ P3│ │ P4│ │ P5│ … │
│ └───┘ └───┘ └───┘ └───┘ └───┘ └───┘ │
│ ↑ ↑ ↑ │
│ │ │ │ │
│ ┌──┴──┐ ┌──┴──┐ ┌──┴──┐ │
│ │请求A │ │请求B │ │请求C │ (逻辑页) │
│ │[P0] │ │[P1] │ │[P2] │ │
│ │[P3] │ │[P4] │ │ │ │
│ └─────┘ └─────┘ └─────┘ │
└────────────────────────────────────────────┘
优势：
– 按需分配，不浪费显存
– 支持Prefix Caching（相同前缀共享物理页）
– 支持并行请求
– 显存利用率从60-80%提升到95%+
6.4 Prefix Caching的实现原理
vLLM/SGLang中的Prefix Caching：
1. 计算每个token块的哈希值
hash = hash(token_ids, prefix_hash)
2. 缓存查找
如果哈希值在缓存中 → 命中！直接使用
如果不在 → 未命中，重新计算并缓存
3. 哈希计算示例
token序列: [系统提示1, 系统提示2, 用户问题]
block_hash[0] = hash([系统提示1])
block_hash[1] = hash([系统提示2], block_hash[0])
block_hash[2] = hash([用户问题], block_hash[1])
如果另一个请求的前缀相同:
[系统提示1, 系统提示2, 另一个问题]
block_hash[0] 和 block_hash[1] 命中！
只需要计算 block_hash[2]（新问题的部分）
缓存命中率取决于：
– 系统提示是否相同（通常相同 → 高命中率）
– 多轮对话的前缀（之前的对话历史 → 命中）
– 共享模板（如few-shot示例 → 命中）
6.5 Prompt Cache vs KV Cache vs Prefix Cache
三个容易混淆的概念：
KV Cache:
存储已计算token的Key和Value矩阵
用于避免重复计算
每次推理都会使用
Prompt Cache:
缓存用户提示的KV Cache
在多轮对话中复用
每轮对话的共同前缀不需要重新计算
Prefix Cache:
更通用的缓存机制
不仅限于同一用户的多轮对话
不同用户如果共享相同的系统提示，也能复用
用于高并发推理服务
6.6 缓存失效与更新
什么时候缓存会失效？
1. 模型更新
模型参数变了 → 所有KV Cache失效 → 需要重新计算
2. Tokenizer变化
文本tokenize的结果变了 → 缓存失效
3. 缓存空间不足
显存满了 → 需要淘汰旧缓存
策略：LRU（最近最少使用）、FIFO（先进先出）
4. 对话历史变化
用户编辑了之前的对话 → 该位置之后的缓存全部失效
第七章：长上下文推理优化技术
7.1 GQA (Grouped Query Attention)
标准Multi-Head Attention (MHA):
每个注意力头都有独立的Q、K、V
→ KV Cache很大
GQA (Grouped Query Attention):
多个Query头共享一组K和V
→ KV Cache更小
MHA: Q₁ Q₂ Q₃ Q₄ K₁ K₂ K₃ K₄ V₁ V₂ V₃ V₄ (12份参数)
GQA: Q₁ Q₂ Q₃ Q₄ K₁₂ K₁₂ V₁₂ V₁₂ (8份参数，K/V共享)
Qwen2.5、LLaMA 3等模型都使用GQA
KV Cache减少约50-75%
7.2 MQA (Multi-Query Attention)
MQA是GQA的极端情况：
所有Query头共享同一个K和V
MHA: Q₁ Q₂ … Q_h K₁ K₂ … K_h V₁ V₂ … V_h
MQA: Q₁ Q₂ … Q_h K₁ V₁
KV Cache减少到 1/h（h是头数）
但可能损失精度
7.3 Flash Attention
Flash Attention是注意力计算的优化（不改变结果，只加速计算）
核心优化：
1. 分块计算(Blocking): 将注意力矩阵分成小块
2. 重计算(Recomputation): 反向传播时重新计算，不存储中间结果
3. 内存高效: 避免存储完整的 n×n 注意力矩阵
标准Attention: 需要 O(n²) 显存存储注意力矩阵
Flash Attention: 只需要 O(n) 显存
效果：
– 显存使用大幅减少
– 计算速度提升2-4倍
– 支持更长的上下文
7.4 Sliding Window Attention
核心思想：
每个token只关注附近的W个token
不是关注所有历史token
标准Attention: token关注所有n个历史token
Sliding Window: token只关注最近W个token
计算量: O(n²) → O(n×W)
缺点：可能丢失远距离信息
解决：多层叠加，信息逐步传播
7.5 Ring Attention
用于分布式长上下文推理
核心思想：
将长序列分配到多个GPU上
每个GPU处理一段KV Cache
通过环形通信聚合注意力结果
GPU0: [token 0-32K]
GPU1: [token 32K-64K]
GPU2: [token 64K-96K]
GPU3: [token 96K-128K]
每个GPU计算自己那段的注意力
通过Ring通信交换KV数据
→ 支持超长上下文（理论无限制）
第八章：实战与工程实践
8.1 选择合适的上下文长度
不需要总是使用最大上下文长度！
短对话（<4K）→ 使用默认上下文，速度最快
中等文档（4-32K）→ 标准上下文
长文档（32-128K）→ 长上下文模式，注意显存
超长文档（>128K）→ 考虑文档分块、RAG等策略
实际影响：
– 上下文越长，Prefill越慢
– 上下文越长，KV Cache占用越大
– 上下文越长，推理成本越高（按token计费）
8.2 vLLM中的KV Cache配置
from vllm import LLM, SamplingParams
# 创建LLM实例，配置KV Cache
llm = LLM(
model="Qwen/Qwen2.5-7B-Instruct",
max_model_len=32768, # 最大上下文长度
gpu_memory_utilization=0.9, # GPU显存使用率
enable_prefix_caching=True, # 开启Prefix Caching
tensor_parallel_size=1, # 张量并行数
)
# 生成
params = SamplingParams(
temperature=0.7,
max_tokens=1024,
top_p=0.9
)
# 第一次请求：Prefill + Cache
result = llm.generate(["你是一个有帮助的AI助手。请介绍一下Python。"], params)
# 第二次请求：Prefix Cache命中（系统提示相同）
result = llm.generate(["你是一个有帮助的AI助手。请介绍一下Java。"], params)
8.3 监控KV Cache使用
# vLLM中查看KV Cache状态
import requests
# 获取指标
metrics = requests.get("http://localhost:8000/metrics").text
# 关键指标：
# – vllm:gpu_cache_usage_perc: KV Cache显存使用百分比
# – vllm:num_requests_running: 正在运行的请求数
# – vllm:num_requests_waiting: 等待的请求数
# – vllm:prompt_tokens_per_request: 平均prompt长度
# – vllm:generation_tokens_per_request: 平均生成长度
8.4 长上下文处理最佳实践
1. 文档预处理
– 去除无关内容（页眉页脚、重复段落）
– 结构化分块（按章节、段落）
– 保留关键信息
2. RAG + 长上下文
– 先用RAG检索相关片段
– 将检索结果放入上下文
– 比纯长上下文更高效
3. 压缩上下文
– 对历史对话做摘要
– 只保留关键信息
– 减少token数量
4. 分层处理
– 第一层：快速筛选（用小模型/短上下文）
– 第二层：深入分析（用大模型/长上下文）
8.5 完整实战示例
示例1：构建一个支持长文档的问答系统
"""
实战：基于vLLM的长文档问答系统
演示：上下文管理 + Prefix Caching + 流式输出
"""
from vllm import LLM, SamplingParams
import tiktoken
class LongDocQA:
def __init__(self, model_path="Qwen/Qwen2.5-7B-Instruct", max_context=32768):
# 初始化vLLM引擎，开启Prefix Caching
self.llm = LLM(
model=model_path,
max_model_len=max_context,
gpu_memory_utilization=0.9,
enable_prefix_caching=True, # 关键：开启前缀缓存
)
self.tokenizer = tiktoken.get_encoding("cl100k_base")
self.max_context = max_context
self.system_prompt = "你是一个专业的文档分析助手。基于提供的文档内容回答问题。请用中文回答。"
self.chat_history = [] # 多轮对话历史
self.document_text = "" # 当前加载的文档
def load_document(self, file_path: str):
"""加载文档并检查token数"""
with open(file_path, 'r', encoding='utf-8') as f:
self.document_text = f.read()
doc_tokens = len(self.tokenizer.encode(self.document_text))
sys_tokens = len(self.tokenizer.encode(self.system_prompt))
# 预留生成空间
available = self.max_context – sys_tokens – 2048
if doc_tokens > available:
print(f"警告：文档有{doc_tokens}个token，超过上下文限制{available}")
print("进行智能截断…")
self.document_text = self._smart_truncate(self.document_text, available)
else:
print(f"文档加载成功：{doc_tokens}个token，上下文空间充足")
def _smart_truncate(self, text: str, max_tokens: int) –> str:
"""智能截断：保留开头和结尾，截断中间部分"""
tokens = self.tokenizer.encode(text)
if len(tokens) <= max_tokens:
return text
# 保留前60%和后40%
head_len = int(max_tokens * 0.6)
tail_len = max_tokens – head_len
head = self.tokenizer.decode(tokens[:head_len])
tail = self.tokenizer.decode(tokens[–tail_len:])
return head + "\\n\\n[…中间部分已省略…]\\n\\n" + tail
def ask(self, question: str) –> str:
"""提问并获取回答"""
# 构建prompt
context = f"文档内容：\\n{self.document_text}\\n\\n"
# 添加历史对话
history_text = ""
for q, a in self.chat_history[–5:]: # 只保留最近5轮
history_text += f"用户：{q}\\n助手：{a}\\n"
user_message = context + history_text + f"用户：{question}\\n助手："
messages = [
{"role": "system", "content": self.system_prompt},
{"role": "user", "content": user_message}
]
# 生成
params = SamplingParams(
temperature=0.3,
max_tokens=2048,
top_p=0.9,
stop=["用户："] # 遇到下一个"用户："停止
)
outputs = self.llm.generate(
[self.llm.get_tokenizer().apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)],
params
)
answer = outputs[0].outputs[0].text.strip()
# 保存对话历史
self.chat_history.append((question, answer))
return answer
# 使用示例
if __name__ == "__main__":
qa = LongDocQA(max_context=32768)
qa.load_document("水泥工艺技术手册.txt")
print("\\n=== 文档问答系统 ===")
print("输入 'quit' 退出\\n")
while True:
question = input("你的问题：").strip()
if question.lower() == 'quit':
break
answer = qa.ask(question)
print(f"\\n回答：{answer}\\n")
这个示例展示了：
示例2：监控KV Cache命中率和显存使用
"""
实战：KV Cache监控与优化
帮助你理解和调优推理性能
"""
import time
import requests
from vllm import LLM, SamplingParams
def benchmark_prefix_caching():
"""
对比开启/关闭Prefix Caching的性能差异
"""
system_prompt = "你是一个专业的AI助手，擅长回答技术问题。请用中文详细回答。" * 5
# 模拟较长的系统提示（约500 token）
questions = [
"什么是机器学习？",
"什么是深度学习？",
"什么是强化学习？",
"什么是自然语言处理？",
"什么是计算机视觉？",
]
# ===== 测试1：不开启Prefix Caching =====
print("=" * 60)
print("测试1：不开启Prefix Caching")
print("=" * 60)
llm_no_cache = LLM(
model="Qwen/Qwen2.5-7B-Instruct",
max_model_len=8192,
gpu_memory_utilization=0.85,
enable_prefix_caching=False, # 关闭
)
params = SamplingParams(temperature=0.7, max_tokens=256)
times_no_cache = []
for q in questions:
prompt = f"{system_prompt}\\n\\n用户：{q}\\n助手："
start = time.time()
llm_no_cache.generate([prompt], params)
elapsed = time.time() – start
times_no_cache.append(elapsed)
print(f" 问题: {q[:20]}… 耗时: {elapsed:.3f}s")
avg_no_cache = sum(times_no_cache) / len(times_no_cache)
print(f" 平均耗时: {avg_no_cache:.3f}s")
# ===== 测试2：开启Prefix Caching =====
print(f"\\n{'='*60}")
print("测试2：开启Prefix Caching")
print("=" * 60)
llm_with_cache = LLM(
model="Qwen/Qwen2.5-7B-Instruct",
max_model_len=8192,
gpu_memory_utilization=0.85,
enable_prefix_caching=True, # 开启
)
times_with_cache = []
for q in questions:
prompt = f"{system_prompt}\\n\\n用户：{q}\\n助手："
start = time.time()
llm_with_cache.generate([prompt], params)
elapsed = time.time() – start
times_with_cache.append(elapsed)
print(f" 问题: {q[:20]}… 耗时: {elapsed:.3f}s")
avg_with_cache = sum(times_with_cache) / len(times_with_cache)
print(f" 平均耗时: {avg_with_cache:.3f}s")
# ===== 结果对比 =====
print(f"\\n{'='*60}")
print("结果对比")
print("=" * 60)
speedup = avg_no_cache / avg_with_cache if avg_with_cache > 0 else 0
print(f" 无缓存平均耗时: {avg_no_cache:.3f}s")
print(f" 有缓存平均耗时: {avg_with_cache:.3f}s")
print(f" 加速比: {speedup:.2f}x")
print(f"\\n 说明：第一个问题两者速度相近（缓存还没建立）")
print(f" 后续问题有缓存更快（系统提示的KV被复用）")
if __name__ == "__main__":
benchmark_prefix_caching()
示例3：处理超长文档——分块策略
"""
实战：超长文档处理策略
当文档超过上下文限制时的解决方案
"""
import numpy as np
from typing import List
class SmartChunker:
"""
智能文档分块器
策略：语义分块 + 摘要压缩
"""
def __init__(self, max_chunk_tokens=4096, overlap_tokens=200):
self.max_chunk_tokens = max_chunk_tokens
self.overlap_tokens = overlap_tokens
def chunk_by_sections(self, document: str) –> List[str]:
"""按章节分块"""
# 尝试按标题分割
sections = []
current_section = ""
for line in document.split('\\n'):
if line.startswith('#') and current_section:
sections.append(current_section)
current_section = line + '\\n'
else:
current_section += line
if current_section:
sections.append(current_section)
# 合并过短的章节
merged = []
buffer = ""
for section in sections:
if len(buffer) + len(section) < self.max_chunk_tokens * 3: # 粗略估计
buffer += "\\n" + section
else:
if buffer:
merged.append(buffer)
buffer = section
if buffer:
merged.append(buffer)
return merged
def chunk_sliding_window(self, document: str) –> List[str]:
"""滑动窗口分块（带重叠）"""
chars_per_token = 4 # 中文约2-4字符/token
max_chars = self.max_chunk_tokens * chars_per_token
overlap_chars = self.overlap_tokens * chars_per_token
chunks = []
start = 0
while start < len(document):
end = min(start + max_chars, len(document))
chunks.append(document[start:end])
start = end – overlap_chars # 重叠
return chunks
def hierarchical_summary(self, chunks: List[str], llm) –> str:
"""
层次化摘要：先摘要每个块，再综合摘要
解决超长文档完全超出上下文的问题
"""
# 第一层：每个块生成摘要
chunk_summaries = []
for i, chunk in enumerate(chunks):
prompt = f"请用100字总结以下内容的核心要点：\\n\\n{chunk}"
summary = llm.generate([prompt], max_tokens=200)
chunk_summaries.append(summary)
print(f" 已摘要: 块{i+1}/{len(chunks)}")
# 第二层：综合所有摘要
all_summaries = "\\n\\n".join(
f"[第{i+1}部分] {s}" for i, s in enumerate(chunk_summaries)
)
final_prompt = f"基于以下各部分摘要，给出完整的总结：\\n\\n{all_summaries}"
final_summary = llm.generate([final_prompt], max_tokens=1000)
return final_summary
# 使用示例
def demo_long_doc_processing():
"""演示超长文档处理"""
# 模拟一个超长文档
with open("长文档.txt", "r", encoding="utf-8") as f:
document = f.read()
print(f"文档长度: {len(document)} 字符")
chunker = SmartChunker(max_chunk_tokens=4096, overlap_tokens=200)
# 策略1：按章节分块
print("\\n策略1：按章节分块")
sections = chunker.chunk_by_sections(document)
print(f" 分成 {len(sections)} 个章节")
for i, s in enumerate(sections):
print(f" 第{i+1}章: {len(s)} 字符")
# 策略2：滑动窗口分块
print("\\n策略2：滑动窗口分块")
windows = chunker.chunk_sliding_window(document)
print(f" 分成 {len(windows)} 个窗口（有重叠）")
# 策略3：层次化摘要（需要LLM）
# summary = chunker.hierarchical_summary(sections, llm)
if __name__ == "__main__":
demo_long_doc_processing()
示例4：多轮对话中的KV Cache管理
"""
实战：多轮对话的上下文管理
演示如何有效管理对话历史，避免超出上下文限制
"""
class ConversationManager:
"""
对话管理器
智能管理对话历史，优化KV Cache使用
"""
def __init__(self, max_context_tokens=8192, reserve_output=2048):
self.max_context = max_context_tokens
self.reserve = reserve_output
self.system_prompt = "你是一个有帮助的AI助手。"
self.messages = [] # 完整对话历史
self.token_counts = [] # 每轮对话的token数
def add_message(self, role: str, content: str):
"""添加消息"""
self.messages.append({"role": role, "content": content})
# 粗略估计token数（中文约1.5字/token）
tokens = len(content) * 2 // 3
self.token_counts.append(tokens)
def get_context_budget(self) –> dict:
"""计算上下文预算"""
system_tokens = len(self.system_prompt) * 2 // 3
total_history = sum(self.token_counts)
available = self.max_context – system_tokens – self.reserve
return {
'system': system_tokens,
'history': total_history,
'available': available,
'remaining': available – total_history,
'usage_pct': total_history / available * 100
}
def should_summarize(self) –> bool:
"""是否需要压缩历史"""
budget = self.get_context_budget()
return budget['usage_pct'] > 80 # 超过80%时压缩
def compress_history(self, llm=None):
"""
压缩对话历史
策略：保留最近N轮，摘要前面的对话
"""
if not self.should_summarize():
return
print(" 上下文使用超过80%，开始压缩历史…")
budget = self.get_context_budget()
# 保留最近3轮完整对话
keep_recent = 3
recent = self.messages[–keep_recent * 2:] # user + assistant各一轮
old = self.messages[:–keep_recent * 2]
if old:
# 对旧对话生成摘要
old_text = "\\n".join(
f"{m['role']}: {m['content'][:100]}…"
for m in old
)
if llm:
summary_prompt = f"用50字总结这段对话的核心内容：\\n{old_text}"
summary = llm.generate([summary_prompt], max_tokens=100)
else:
summary = f"[对话摘要：包含{len(old)}条历史消息]"
# 替换为摘要 + 最近对话
self.messages = [
{"role": "system", "content": f"{self.system_prompt}\\n\\n历史对话摘要：{summary}"}
] + recent
print(f" 压缩完成：{len(old)}条消息 → 1条摘要 + {len(recent)}条最近消息")
def build_prompt(self) –> str:
"""构建最终的prompt（供vLLM使用）"""
budget = self.get_context_budget()
if budget['remaining'] < 0:
# 紧急截断：只保留系统提示和最近1轮
messages = [self.messages[0]] + self.messages[–2:]
else:
messages = self.messages
return messages
# 使用示例
def demo_conversation():
"""演示多轮对话管理"""
manager = ConversationManager(max_context_tokens=4096, reserve_output=1024)
manager.add_message("system", manager.system_prompt)
# 模拟多轮对话
conversations = [
("user", "什么是强化学习？"),
("assistant", "强化学习是机器学习的一个分支，通过与环境交互来学习最优策略…"),
("user", "它和监督学习有什么区别？"),
("assistant", "监督学习使用标注数据训练，而强化学习通过试错和奖励信号学习…"),
("user", "那Q-Learning是什么？"),
("assistant", "Q-Learning是一种基于价值的强化学习算法，通过学习Q函数来选择最优动作…"),
("user", "DQN又是怎么改进Q-Learning的？"),
("assistant", "DQN使用深度神经网络来近似Q函数，引入了经验回放和目标网络…"),
("user", "PPO算法呢？它和DQN有什么不同？"),
("assistant", "PPO是基于策略的算法，直接优化策略而非价值函数…"),
]
for role, content in conversations:
manager.add_message(role, content)
# 检查是否需要压缩
budget = manager.get_context_budget()
print(f"轮次{len([m for m in manager.messages if m['role']=='user'])}: "
f"上下文使用 {budget['usage_pct']:.1f}%")
if manager.should_summarize():
manager.compress_history()
# 构建最终prompt
final_messages = manager.build_prompt()
print(f"\\n最终消息数: {len(final_messages)}")
for m in final_messages:
print(f" [{m['role']}] {m['content'][:50]}…")
if __name__ == "__main__":
demo_conversation()
示例5：生产环境的上下文优化配置
"""
实战：vLLM生产环境配置模板
针对不同场景的优化配置
"""
# ===== 场景1：聊天机器人（中等上下文，高并发）=====
chat_config = {
"model": "Qwen/Qwen2.5-7B-Instruct",
"max_model_len": 8192,
"gpu_memory_utilization": 0.90,
"enable_prefix_caching": True, # 系统提示复用
"tensor_parallel_size": 1,
"max_num_seqs": 64, # 最大并发请求数
"max_num_batched_tokens": 32768, # 单batch最大token数
"block_size": 16, # KV Cache页大小
}
# ===== 场景2：文档分析（长上下文，低并发）=====
doc_analysis_config = {
"model": "Qwen/Qwen2.5-7B-Instruct",
"max_model_len": 32768, # 更长的上下文
"gpu_memory_utilization": 0.95, # 更多显存给KV Cache
"enable_prefix_caching": True,
"tensor_parallel_size": 1,
"max_num_seqs": 8, # 并发少但每个请求很长
"max_num_batched_tokens": 65536,
"block_size": 16,
}
# ===== 场景3：代码生成（中等上下文，中等并发）=====
code_gen_config = {
"model": "Qwen/Qwen2.5-Coder-7B-Instruct",
"max_model_len": 16384,
"gpu_memory_utilization": 0.88,
"enable_prefix_caching": True,
"tensor_parallel_size": 1,
"max_num_seqs": 32,
"max_num_batched_tokens": 32768,
"block_size": 16,
"enforce_eager": True, # 代码生成需要精确结果
}
# ===== 场景4：多卡部署（大模型，高并发）=====
multi_gpu_config = {
"model": "Qwen/Qwen2.5-72B-Instruct",
"max_model_len": 32768,
"gpu_memory_utilization": 0.90,
"enable_prefix_caching": True,
"tensor_parallel_size": 4, # 4卡张量并行
"pipeline_parallel_size": 1,
"max_num_seqs": 128,
"max_num_batched_tokens": 131072,
"block_size": 16,
}
print("=== 各场景配置说明 ===")
print("""
场景选择指南：
聊天机器人 → chat_config（8K上下文，高并发）
文档分析 → doc_analysis_config（32K上下文，低并发）
代码生成 → code_gen_config（16K上下文，精确输出）
大模型 → multi_gpu_config（多卡并行，长上下文）
关键参数说明：
max_model_len: 最大上下文长度（越长显存占用越大）
gpu_memory_utilization: GPU显存使用率（留余量给系统）
enable_prefix_caching: 开启前缀缓存（强烈建议开启）
max_num_seqs: 最大并发数（越多显存越分散）
block_size: KV Cache页大小（16通常最优）
""")
附录：术语表与参考资料
术语表
上下文长度(Context Length): 模型一次能处理的最大token数
位置编码(Position Encoding): 注入token位置信息的技术
RoPE: Rotary Position Embedding，旋转位置编码
ALiBi: Attention with Linear Biases
位置插值(Position Interpolation): 缩放位置值到训练范围
NTK-Aware: 按频率不同采用不同缩放策略
YaRN: Yet another RoPE extensioN
KV Cache: 缓存已计算token的Key和Value矩阵
Prefill: 预填充阶段，一次性处理所有输入token
Decode: 解码阶段，逐个生成新token
Prefix Caching: 相同前缀的KV Cache共享
PagedAttention: vLLM的分页KV Cache管理
GQA: Grouped Query Attention，分组查询注意力
MQA: Multi-Query Attention，多查询注意力
Flash Attention: 内存高效的注意力计算优化
Ring Attention: 分布式长上下文推理
缓存命中(Cache Hit): KV Cache中有需要的数据
缓存未命中(Cache Miss): KV Cache中没有需要的数据
参考论文
1. RoPE: Su et al., "RoFormer: Enhanced Transformer with Rotary Position Embedding", 2021
2. ALiBi: Press et al., "Train Short, Test Long: Attention with Linear Biases", ICLR 2022
3. Position Interpolation: Chen et al., "Extending Context Window of Large Language Models via Position Interpolation", 2023
4. YaRN: Peng et al., "YaRN: Efficient Context Window Extension of Large Language Models", 2023
5. LongRoPE: Ding et al., "LongRoPE: Extending LLM Context Window Beyond 2 Million Tokens", 2024
6. Flash Attention: Dao et al., "FlashAttention: Fast and Memory-Efficient Exact Attention", NeurIPS 2022
7. Flash Attention 2: Dao, "FlashAttention-2: Faster Attention with Better Parallelism", 2023
8. GQA: Ainslie et al., "GQA: Training Generalized Multi-Query Transformer Models", 2023
9. PagedAttention: Kwon et al., "Efficient Memory Management for Large Language Model Serving with PagedAttention", SOSP 2023
10. Ring Attention: Liu et al., "Ring Attention with Blockwise Transformers for Near-Infinite Context", 2023
