从稀疏性LLM服务系统中提取Tokens：方法与工程实践

前阵子在分析 LLM 推理服务性能时，遇到了一个很实际的问题：为了加速生成，很多服务系统开始利用稀疏性（Sparsity）跳过不必要的计算，但代价是Tokens 的流动路径和统计口径变得非常不透明。你很难搞清楚一个请求到底生成了多少个有效 Tokens、哪些 Tokens 被稀疏策略跳过了、KV Cache 里到底缓存了哪些 Token 的状态。这直接影响成本核算、质量评估和系统调优。围绕这个问题，本文做一次系统化梳理，重点拆解 SparSEEty 这一类思路：如何从利用稀疏性的 LLM 服务系统中提取并理解 Tokens。无论你是做推理优化、LLM 应用开发，还是单纯想深入理解 Tokens 的来龙去脉，这篇内容都值得收藏备用。

1. 背景：为什么稀疏性服务系统让 Tokens 变“难懂”了

1.1 LLM 服务系统中的 Tokens 是如何产生的

在使用大语言模型时，模型并不直接理解人类文字，而是先把文本切分成 Token。Token 可以是一个完整的单词、一个词根、一个标点符号，甚至是半个汉字。比如中文“人工智能”很可能被切分成多个 Token，而不是一个整体。这个切分过程由 Tokenizer 完成。

从服务系统的视角来看，一次请求的生命周期大致是：

1. 用户输入一段 Prompt。
2. Tokenizer 将 Prompt 切分为输入 Tokens。
3. 模型对输入 Tokens 做 Prefill 计算，得到初始的 KV Cache。
4. 模型进入 Decode 阶段，逐个生成输出 Tokens。
5. 每生成一个 Token，就更新一次 KV Cache。
6. 生成结束后，Tokenizers 将 Tokens 解码回文本，返回给用户。

在这个过程中，Tokens 的数量直接决定了：

• 计费金额。
• 延迟和吞吐量。
• GPU 显存占用。
• 缓存命中率。

所以，能够精准提取和分析 Tokens，是服务治理的刚需。

1.2 什么叫做“利用稀疏性的服务系统”

传统 Transformer 在 Decode 阶段，每个 Token 都要和之前所有 Token 做 Attention 计算。假设序列长度是 N，那么计算量是 O(N²) 级别。当序列变长、并发变高时，这个计算量非常昂贵。

稀疏性（Sparsity）在这里指的是：Attention 矩阵中很多位置的权重实际上非常低，对最终结果影响极小。利用稀疏性做优化的服务系统，会主动跳过这些低价值计算，只保留重要的 Attention 关系。

常见的稀疏化思路包括：

• 稀疏注意力（Sparse Attention）：只让每个 Token 关注局部窗口或全局少量特殊 Token。
• KV Cache 剪枝/淘汰：定期删除不重要的历史 KV 状态。
• 动态 Token 剪枝：推理过程中识别不重要 Token，直接丢弃，减少后续计算。
• 投机采样（Speculative Decoding）：用小模型先草拟多个 Tokens，再用大模型一次验证，从而摊薄计算成本。

这些优化手段能以较少的质量损失换取大幅性能提升，但副作用是：你不再能简单地由“输入一次、输出一次”理解 Tokens 的完整路径。

1.3 SparSEEty 解决什么问题

SparSEEty 的核心目标，是从这类稀疏性服务系统中提取完整的 Token 生命周期信息。它要回答几个问题：

• 请求实际生成了哪些 Tokens？
• 哪些 Tokens 在推理过程中被跳过、剪枝或合并了？
• 稀疏策略是否改变了生成结果的 Token 顺序或内容？
• 被剪掉的 Token 对最终回答质量有什么影响？

这个思路的价值在于：性能优化不能以“失去可观测性”为代价。如果你把系统做快了，却无法解释它输出了什么、为什么这么输出，那在生产环境很难放心上线。

2. 关键概念拆解：Tokens、Sparsity、Serving System 的关系

2.1 Token 的三种类型

在服务系统中，Token 至少需要区分为三种类型：

很多系统后台只统计输入和输出两个数字，忽略了第三类。但正是第三类决定了稀疏性优化的实际效果。

2.2 Serving System 的观测难点

普通 LLM 服务系统，比如基于 vLLM、TGI 或自研推理框架搭建的服务，会暴露一些统计指标，例如：

• 请求数。
• 输入 Token 数。
• 输出 Token 数。
• 平均生成速度（Tokens/s）。
• 排队时间。

这些指标对传统密集计算场景是比较准确的。但一旦开启稀疏策略，比如 Attention 稀疏化或 KV Cache 淘汰，问题就出现了：

• 模型内部实际处理的 Token 列表可能短于输入 Token 列表。
• 某些 Token 是“临时生成又被剪掉”的，并未输出给用户，但消耗了算力。
• 稀疏模块可能改变 KV Cache 的组织方式，导致同一 Token 在不同请求中具有不同的“可见性”。

这就是为什么需要一个专门做 Token 提取与分析的系统或方法。

2.3 Sparsity 利用的不同层次

为了准确理解 Token 提取的难度，需要区分稀疏性发生在哪一个层次：

• 模型权重稀疏（Weight Sparsity）：模型中的部分权重为 0，但这不直接影响 Token 数量。它只是让计算更快。
• 注意力稀疏（Attention Sparsity）：Attention 矩阵被稀疏化，Token 之间的关联图变稀疏。可能影响 Tokens 是否参与后续计算。
• 状态稀疏（KV Cache Sparsity）：KV 状态被裁剪，部分历史 Token 状态被移除。这会让模型“遗忘”早期内容，也可能让 Token 提取变得更复杂。
• Token 序列稀疏（Token-level Sparsity）：在序列维度上直接剪掉不重要的 Token，这是最激进的稀疏化方式，也是与 Tokens 提取关系最紧密的类型。

SparSEEty 这类系统，重点关注的通常是后两种，因为它们直接改变 Token 的可见性。

3. 从服务系统中提取 Tokens 的通用方法

在实际工程中，不需要一开始就实现一个完整复杂的系统，可以先掌握几种通用提取方法。下面按照从简单到复杂的顺序梳理。

3.1 方法一：从服务日志与统计接口提取

这是最简单的做法。主流推理服务框架都会在请求结束后记录 Token 统计信息。以 OpenAI 兼容接口为例，响应体里通常会包含 usage 字段：

{ "id": "chatcmpl-123", "object": "chat.completion", "created": 1700000000, "model": "demo-model", "choices": [ { "index": 0, "message": { "role": "assistant", "content": "你好，我是智能助手。" }, "finish_reason": "stop" } ], "usage": { "prompt_tokens": 23, "completion_tokens": 15, "total_tokens": 38 } }

其中：

• prompt_tokens 是输入 Tokens 数。
• completion_tokens 是输出 Tokens 数。
• total_tokens 是两者之和。

这种方式只能拿到总量，拿不到 Token 列表，也拿不到稀疏过程中的中间状态。适用场景是计费和宏观监控。

3.2 方法二：通过 Tokenizer 还原 Token 列表

如果需要知道“具体是哪些 Token”，就需要调用 Tokenizer。以 Hugging Face Transformers 为例：

from transformers import AutoTokenizer tokenizer = AutoTokenizer.from_pretrained("your_model_path") text = "人工智能正在改变世界" tokens = tokenizer.tokenize(text) token_ids = tokenizer.encode(text) print("Tokens:", tokens) print("Token IDs:", token_ids) print("解码还原:", tokenizer.decode(token_ids))

运行后，能看到文本被切分成哪些 Token，以及每个 Token 对应的 ID。这在分析提示词超长、计费争议和内容质量问题时很有用。

但要注意：这个方式只能还原“文本层面”的 Token 切分，无法还原模型推理过程中因稀疏策略而发生的 Token 动态变化。

3.3 方法三：Hook 模型中间层，捕获推理过程 Token 状态

如果服务框架基于 PyTorch 实现，可以在模型推理过程中插入 Hook，逐层捕获输入 Token ID、Attention Mask、以及 KV Cache 的状态。这是 SparSEEty 类系统做深层次提取的基础。

思路如下：

1. 注册 Forward Hook。
2. 在 Hook 中读取当前层的输入 Token ID。
3. 读取 Attention 权重的稀疏模式，识别哪些位置被跳过。
4. 记录 KV Cache 中被保留和未被保留的 Token 索引。
5. 汇总到全局日志。

下面是一个简化示例，演示如何注册 Hook 捕获 Token ID：

import torch from transformers import AutoModelForCausalLM, AutoTokenizer model_name = "your_model_path" tokenizer = AutoTokenizer.from_pretrained(model_name) model = AutoModelForCausalLM.from_pretrained(model_name) captured = {} def make_hook(layer_name): def hook_fn(module, input, output): # 捕获当前层输入的 token ids # 输入形状通常是 (batch, seq_len) hidden_states = input[0] captured[layer_name] = hidden_states.detach().cpu().shape print(f"Layer {layer_name}, hidden shape: {hidden_states.shape}") return hook_fn for name, module in model.named_modules(): if "self_attn" in name or "mlp" in name: module.register_forward_hook(make_hook(name)) prompt = "什么是稀疏性？" inputs = tokenizer(prompt, return_tensors="pt") with torch.no_grad(): outputs = model(**inputs, output_attentions=True) print("捕获结果:") for layer, shape in captured.items(): print(f" {layer}: {shape}")

这段代码不会直接给出最终答案，但它展示了关键思路：在模型内部设置观测点，把 Token 的状态“抓”出来。实际生产环境中，不可能每层都做完整 Hook，因为开销太大，通常是采样分析或者在关键模块上做轻量日志。

3.4 方法四：从推理框架内部导出 Token 轨迹

像 vLLM、TensorRT-LLM 这类高性能推理框架，通常提供了更底层的日志或回调机制。以 vLLM 为例，它支持通过 Logits Processor 或自定义回调获取生成过程中每个 Token 的信息。

伪代码如下：

from vllm import LLM, SamplingParams class TokenTracer: def __init__(self): self.token_history = [] def __call__(self, token_ids, logits): # 每次生成新 token 时调用 self.token_history.append(token_ids[-1]) return logits tracer = TokenTracer() llm = LLM(model="your_model_path") params = SamplingParams( temperature=0.7, max_tokens=256, logits_processors=[tracer] ) result = llm.generate("请介绍一下 LLM", params) print(tracer.token_history)

这个做法参考意义大于直接复制价值，因为不同版本的 vLLM 对 Logits Processor 的接口定义有差异。但它说明了一个趋势：推理框架越来越重视可观测性，允许开发者在 Token 生成的瞬间插入自定义逻辑。

另一种常见做法是在框架的 metrics 中开启 verbose 日志，直接输出每次生成的 Token ID。适合测试环境，不适合高并发生产，因为日志量巨大。

4. 设计一个轻量级 Token 提取分析模块（实战）

这一节我们会动手实现一个轻量级分析模块。它的目标是：对一次请求，同时输出输入 Token 列表、输出 Token 列表、以及被稀疏策略跳过或剪枝的 Token 位置。虽然没法接真实稀疏模型完整复现，但流程和代码结构可以直接迁移。

4.1 项目结构

建议按下面的结构组织代码：

token-extractor/ ├── config.yaml ├── extractor.py ├── analyzer.py ├── log_tracer.py └── requirements.txt

4.2 创建配置

在 config.yaml 中声明模型路径、稀疏策略开关和日志等级：

model: path: "your_model_path" tokenizer_path: "your_model_path" max_new_tokens: 128 sparsity: enabled: true attention_topk: 16 kv_cache_limit: 2048 token_pruning: true logging: level: "INFO" output_file: "token_trace.jsonl"

4.3 实现 Token 提取器

写一个 extractor.py，负责调用模型生成 Token 并记录过程：

import json import torch from transformers import AutoModelForCausalLM, AutoTokenizer class TokenExtractor: def __init__(self, config): self.config = config self.tokenizer = AutoTokenizer.from_pretrained(config["model"]["path"]) self.model = AutoModelForCausalLM.from_pretrained( config["model"]["path"], torch_dtype=torch.float16, device_map="auto" ) self.trace = [] def tracer_hook(self, layer_name): def hook_fn(module, input, output): self.trace.append({ "layer": layer_name, "type": "forward", "input_shape": input[0].shape }) return hook_fn def register_hooks(self): for name, module in self.model.named_modules(): if "self_attn" in name: module.register_forward_hook(self.tracer_hook(name)) def extract(self, prompt: str): self.trace = [] self.register_hooks() inputs = self.tokenizer(prompt, return_tensors="pt") prompt_tokens = self.tokenizer.convert_ids_to_tokens(inputs["input_ids"][0]) prompt_token_ids = inputs["input_ids"][0].tolist() with torch.no_grad(): outputs = self.model.generate( **inputs, max_new_tokens=self.config["model"]["max_new_tokens"], output_scores=True, return_dict_in_generate=True ) generated_ids = outputs.sequences[0][inputs["input_ids"].shape[-1]:] output_tokens = self.tokenizer.convert_ids_to_tokens(generated_ids) output_token_ids = generated_ids.tolist() return { "prompt": prompt, "prompt_tokens": prompt_tokens, "prompt_token_ids": prompt_token_ids, "output_tokens": output_tokens, "output_token_ids": output_token_ids, "internal_steps": len(self.trace) } def save_trace(self, data, path): with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(data, ensure_ascii=False) + "\n")

4.4 实现稀疏 Token 分析器

analyzer.py 的功能是：对比稀疏策略开启前后的 Token 序列差异，并标出被跳过或被合并的位置。

from difflib import SequenceMatcher class SparsityAnalyzer: def __init__(self, extractor): self.extractor = extractor def run_comparison(self, prompt): # 基线：关闭稀疏策略的结果 baseline = self._generate(prompt, sparsity=False) # 开启稀疏策略的结果 sparse = self._generate(prompt, sparsity=True) diff = self._compare_tokens( baseline["output_token_ids"], sparse["output_token_ids"] ) return { "baseline_tokens": baseline["output_tokens"], "sparse_tokens": sparse["output_tokens"], "diff_ratio": diff["ratio"], "diff_operations": diff["ops"] } def _generate(self, prompt, sparsity): # 实际环境需要在这里切换模型配置或推理参数 # 这里仅为示例 print(f"Generating with sparsity={sparsity}") return self.extractor.extract(prompt) def _compare_tokens(self, base_ids, sparse_ids): matcher = SequenceMatcher(None, base_ids, sparse_ids) ops = [] for tag, i1, i2, j1, j2 in matcher.get_opcodes(): ops.append({ "tag": tag, "base_slice": base_ids[i1:i2], "sparse_slice": sparse_ids[j1:j2] }) return { "ratio": matcher.ratio(), "ops": ops }

4.5 运行示例

写一个入口脚本，运行提取和分析：

import yaml from extractor import TokenExtractor from analyzer import SparsityAnalyzer with open("config.yaml", "r", encoding="utf-8") as f: config = yaml.safe_load(f) extractor = TokenExtractor(config) analyzer = SparsityAnalyzer(extractor) result = analyzer.run_comparison("大模型稀疏化会丢失信息吗？") print("对比结果:") for op in result["diff_operations"]: print(f" {op['tag']}: base={op['base_slice']} sparse={op['sparse_slice']}")

4.6 预期结果说明

由于实际稀疏模型配置差异很大，这里给出预期分析逻辑：

• 如果 diff_ops 中出现delete，说明稀疏策略删除了一些 Token。
• 如果出现replace，说明部分 Token 被替换或合并。
• 如果 ratio 接近 1.0，说明稀疏策略对输出 Token 序列影响很小。

这份分析结果可以进一步用于质量评估。比如，你可以在输出 Token 层面做困惑度对比、语义相似度对比、或者人工评估。

5. 常见问题与排查思路

5.1 提取到的 Token 数量与服务端统计不一致

在实际项目中，凡是涉及 Tokens 计费的场景，都要保证“统计口径一致”。最稳妥的做法是以服务端记录的 usage 为准，本地提取的结果只用于质量分析。

5.2 Hook 提取导致推理变慢

5.3 稀疏策略开启后输出 Token 质量明显下降

5.4 Tokenizer 编解码结果与模型内部不一致

排查这类问题，可以先用一个固定 Prompt 做单轮测试，逐步增加历史消息，找到边界条件。

6. 最佳实践与工程建议

6.1 对 Token 的统计做分级处理

生产环境不推荐把所有 Token 明细都落盘。更好的做法是分级：

• 一级：只记录输入 Token 数、输出 Token 数、平均生成速度，用于监控。
• 二级：对部分采样请求记录 Token 列表，用于质量问题回溯。
• 三级：对特定标记请求记录完整稀疏轨迹，用于研发调优。

这样既控制开销，又能保留排错能力。

6.2 始终保留稀疏策略的开关和版本标识

在做 Token 提取和分析时，除了记录 Token 本身，还要记录：

• 模型版本。
• 推理框架版本。
• 稀疏策略类型。
• 稀疏参数（如 topk 大小、窗口大小、淘汰阈值）。

没有这些信息，Token 序列的对比分析会失去意义。建议在每条日志里带上这些字段。

6.3 不要在生成路径上做昂贵的 Python Hook

如果追求性能，不要在逐 Token 生成路径上做 Python 层 Hook。推荐方式：

• 把 Token 提取器编译成 C++/CUDA 扩展。
• 使用推理框架自带的 metrics 接口。
• 按概率采样，而不是全量收集。

6.4 安全与合规红线

提取 Token 日志时，必须注意隐私合规风险。用户输入的 Prompt 和模型输出都可能包含敏感信息。建议：

• 对日志中的文本做脱敏处理。
• 只保留 Token ID，不保留原始文本。
• 设置日志访问权限，遵循最小权限原则。
• 生产环境日志保留周期要明确。

6.5 将 Token 提取与评估流程打通

Token 提取本身不是目的，最终要为评估服务。建议形成闭环：

1. 提取 Token 轨迹。
2. 对比基线与稀疏策略结果。
3. 在 Token 序列上做质量评估（精确匹配、ROUGE、BERTScore 等）。
4. 根据评估结果调整稀疏参数。
5. 再次提取和验证。

这个闭环投入不大，但对系统长期演进非常关键。

7. 总结与后续学习路线

这篇文章从 LLM 服务系统的实际痛点出发，介绍了为什么稀疏性优化会让 Tokens 变得难以提取和分析，然后整理了四种从服务系统中提取 Tokens 的通用方法：从统计接口、从 Tokenizer、从模型 Hook、从推理框架回调。

文中给出了一套轻量级 Token 提取分析模块的设计代码，涵盖配置、提取、对比、日志四个部分。最后补充了常见问题排查思路和工程实践建议。

如果想继续深挖，可以按下面的路线学习：

• 先熟悉自己使用的推理框架的官方日志和 metrics 接口。
• 再深入研究 Attention 机制，理解 KV Cache 的组织方式。
• 接着了解稀疏注意力论文与开源实现，例如各种 Top-K Attention 和滑窗 Attention 方案。
• 然后尝试编写一个小型评估脚本，对比稀疏开启前后的 Token 序列语义相似度。
• 最后再考虑设计一个适合自己业务的 Token 可观测系统，重点关注性能开销与隐私合规。

Tokens 是 LLM 系统的“货币”，理解它的流动和变化，是做大模型应用开发绕不开的基本功。希望这篇文章能帮你把稀疏性服务系统这块黑盒子打开一条缝，让你在做性能优化时，不只看到速度提升了，还能看清 Tokens 到底经历了什么。