手写 mini-sglang
阶段三 · 服务化：变成一个服务
Tokenizer 与增量反分词¶
tokenizer 进程做两件方向相反的事：把 API Server 发来的文本（或对话）变成 token 送给调度器，把调度器发来的一个个新 token 变回文本送给 API Server。前者一行 apply_chat_template 加一行 encode；后者要难得多——一个汉字或 emoji 可能被拆到两三个 token 里，单独解码只会得到"�"；而且解码不是逐 token 可加的。这一章实现 tokenizer 进程，重点是增量反分词。
自测：能答上来就可以跳过本章
- 为什么不能每来一个 token 就
tokenizer.decode([token])发给前端？ - 反分词为什么"不可加"？举一个
decode(a + b) != decode(a) + decode(b)的例子。 - 为什么 tokenizer 和 detokenizer 可以放在同一个进程里？什么时候应该分开？
- EOS token 要不要发给前端？
自测参考答案（先自己答，再展开对照）
- 一个 token 可能只是一个字符的一部分（中文、emoji 的 UTF-8 字节被拆开），单独解码会得到乱码；而且前后 token 之间的空格、合并规则会让单独解码的结果和整体解码不同。
- 字节级 BPE 下，一个汉字的 3 个 UTF-8 字节可能被分到两个 token 里：分别解码得到两段乱码（替换字符），合起来解码才是正确的汉字；类似地，SentencePiece 里的前导空格标记单独解码时会被去掉。
- 分词和反分词都是轻量的 CPU 工作，一个进程就够，还省掉一次进程间通信；请求很多、分词或反分词成为瓶颈时（长提示词、高并发），应该拆成多个进程（但反分词的状态在进程内，拆分要按请求路由）。
- 不要：EOS 只表示生成结束，不是要显示的文本；在服务端判断停止，并作为
finish_reason返回。
本章要写的文件：tokenizer/tokenize.py、tokenizer/detokenize.py、tokenizer/server.py。
到这一章为止的文件：67 个（新建 4，修改 0，提前引入 0）
这一步的 main：minisgl/
├── __init__.py
├── core.py
├── env.py
├── attention/ （4 个文件，不变）
├── distributed/ （3 个文件，不变）
├── engine/ （5 个文件，不变）
├── kernel/ （2 个文件，不变）
├── kvcache/ （5 个文件，不变）
├── layers/ （8 个文件，不变）
├── llm/ （2 个文件，不变）
├── message/ （5 个文件，不变）
├── models/ （7 个文件，不变）
├── moe/ （2 个文件，不变）
├── scheduler/ （9 个文件，不变）
├── tokenizer/
│ ├── __init__.py ← 本章新建
│ ├── tokenize.py ← 本章新建
│ ├── detokenize.py ← 本章新建
│ ├── server.py ← 本章新建
├── utils/ （8 个文件，不变）
examples/ch13_tokenizer.py——它只用到上面这些文件；python tools/steps.py check 会逐章搭出这棵树、跑这个 main。
分词¶
minisgl/tokenizer/tokenize.py · TokenizeManager
class TokenizeManager:
def __init__(self, tokenizer: PreTrainedTokenizerBase) -> None:
self.tokenizer = tokenizer
def tokenize(self, msgs: List[TokenizeMsg]) -> List[torch.Tensor]:
results: List[torch.Tensor] = []
for msg in msgs:
if isinstance(msg.text, list): # chat 消息列表：先套对话模板
prompt = self.tokenizer.apply_chat_template(
msg.text, tokenize=False, add_generation_prompt=True)
else:
prompt = msg.text
ids = self.tokenizer.encode(prompt, return_tensors="pt")
results.append(ids.view(-1).to(torch.int32))
return results
TokenizeMsg.text 可以是纯文本（/generate 接口、或者 OpenAI 请求里的 prompt），也可以是对话消息列表（OpenAI 的 messages）。后者先用模型自带的对话模板展开成一段文本，并加上"助手开始回答"的提示（add_generation_prompt=True）。模板用错是推理服务最常见的效果问题之一：模型会续写用户的话、或者停不下来（见大模型原理手册的分词一章）。
增量反分词¶
字节级 BPE 的词表建立在 UTF-8 字节上：一个不常见的汉字（3 个字节）或 emoji（4 个字节）很可能被拆到两个 token 里。单独解码前一个 token，只能得到不完整的字节，显示为替换字符"�"。另外，SentencePiece 类的分词器会根据上下文决定是否在开头加空格，所以逐个解码再拼接，与整体解码的结果可能不同。
通用的做法（SGLang、vLLM、HF 的 TextStreamer 都类似）是只解码最近的一个窗口，用两次解码之差得到新增文本：
minisgl/tokenizer/detokenize.py · DetokenizeManager.detokenize
每个请求维护几个偏移：
def detokenize(self, msgs: List[DetokenizeMsg]) -> List[str]:
read_ids: List[List[int]] = []
surr_ids: List[List[int]] = []
for msg in msgs:
s = self.decode_map.setdefault(msg.uid, DecodeStatus([], "", 0, 0, 0))
if not (msg.finished and msg.next_token == self.eos_token_id): # EOS 不输出
s.decoded_ids.append(msg.next_token)
read_ids.append(s.decoded_ids[s.surr_offset :])
surr_ids.append(s.decoded_ids[s.surr_offset : s.read_offset])
# 一批请求一起解码，比逐个调用快
read_texts = self.tokenizer.batch_decode(read_ids)
surr_texts = self.tokenizer.batch_decode(surr_ids)
outputs: List[str] = []
for msg, read_str, surr_str in zip(msgs, read_texts, surr_texts, strict=True):
s = self.decode_map[msg.uid]
new_text = read_str[len(surr_str) :]
if new_text and not new_text.endswith("�"): # 新文本完整：推进窗口
output_str = s.decoded_str + new_text
s.decoded_str = output_str
s.surr_offset = s.read_offset
s.read_offset = len(s.decoded_ids)
else: # 不完整：只输出确定的部分，窗口不动
output_str = s.decoded_str + find_printable_text(new_text)
outputs.append(output_str[s.sent_offset :])
s.sent_offset = len(output_str)
if msg.finished:
del self.decode_map[msg.uid]
return outputs
surr_offset到read_offset之间是"已经确认输出、用来提供上下文"的 token；read_offset之后是还没确认的新 token。
每次解码 [surr_offset, 末尾] 和 [surr_offset, read_offset] 两段，前者减去后者就是新增的文本。如果新文本完整（不以"�"结尾），就确认它、把窗口往前推；否则暂时不推进，只输出能确定的部分（find_printable_text：到换行、到中文字符、或到最后一个空格为止），等后续 token 补全字符。sent_offset 记录已经发给前端的字符数，保证每个字符只发一次。
examples/ch13_tokenizer.py
"""第 13 章：对话模板，以及增量反分词如何避免输出半个字符。"""
from minisgl.core import SamplingParams
from minisgl.message import DetokenizeMsg, TokenizeMsg
from minisgl.tokenizer import DetokenizeManager, TokenizeManager
from transformers import AutoTokenizer
tok = AutoTokenizer.from_pretrained("models/Qwen3-0.6B")
ids = TokenizeManager(tok).tokenize([TokenizeMsg(
uid=0, text=[{"role": "user", "content": "你好"}], sampling_params=SamplingParams())])[0]
print("对话模板展开后:", repr(tok.decode(ids)))
text = "鱻䨻🦩 is rare, 推理 naïve" # 前三个字符各被拆成两个 token
out_ids = tok(text).input_ids
dm = DetokenizeManager(tok)
print(f"{'token':>7} {'单独解码':<12} 增量输出")
pieces = []
for i, t in enumerate(out_ids):
piece = dm.detokenize([DetokenizeMsg(uid=1, next_token=t, finished=i == len(out_ids) - 1)])[0]
pieces.append(piece)
print(f"{t:>7} {tok.decode([t])!r:<12} {piece!r}")
print("拼起来:", repr("".join(pieces)), " 与整体解码相同:", "".join(pieces) == tok.decode(out_ids))
运行结果
"鱻"被拆成两个 token，第一个到达时增量输出为空，第二个到达时才输出完整的"鱻"；"推"与前面的空格合在一个 token 里，第一个 token 只输出了能确定的空格。拼起来与整体解码完全相同，前端从来不会收到"�"。
对话模板展开后: '<|im_start|>user\n你好<|im_end|>\n<|im_start|>assistant\n'
token 单独解码 增量输出
100024 '�' ''
119 '�' '鱻'
128237 '�' ''
119 '�' '䨻'
123918 '�' ''
102 '�' '🦩'
374 ' is' ' is'
8848 ' rare' ' rare'
11 ',' ','
46602 ' �' ' '
101 '�' '推'
21887 '理' '理'
94880 ' naï' ' naï'
586 've' 've'
拼起来: '鱻䨻🦩 is rare, 推理 naïve' 与整体解码相同: True
其他细节：
- 一批消息一起
batch_decode，比逐条调用快； - 请求结束（
finished=True）且最后一个 token 是 EOS 时，EOS 不加入解码（它会被解码成<|im_end|>之类的特殊字符串）； - 请求结束后删除它的状态，否则长时间运行会泄漏内存（这就是上一章"问题二"里过期消息的危害）。
tokenizer 进程¶
minisgl/tokenizer/server.py · tokenize_worker
主循环：阻塞地收一条消息，再把已经到达的消息尽量多收一些（最多 @torch.inference_mode()
def tokenize_worker(*, tokenizer_path: str, addr: str, create: bool, backend_addr: str,
frontend_addr: str, local_bs: int, tokenizer_id: int = -1,
ack_queue: mp.Queue | None = None) -> None:
"""tokenizer 进程的主循环。同一个进程既能分词（前端 → 调度器），也能反分词（调度器 → 前端）。"""
send_backend = ZmqPushQueue(backend_addr, create=False, encoder=BaseBackendMsg.encoder)
send_frontend = ZmqPushQueue(frontend_addr, create=False, encoder=BaseFrontendMsg.encoder)
recv_listener = ZmqPullQueue(addr, create=create, decoder=BatchTokenizerMsg.decoder)
tokenizer = load_tokenizer(tokenizer_path)
logger = init_logger(__name__, f"tokenizer_{tokenizer_id}")
tokenize_manager = TokenizeManager(tokenizer)
detokenize_manager = DetokenizeManager(tokenizer)
if ack_queue is not None:
ack_queue.put(f"Tokenize server {tokenizer_id} is ready")
try:
while True:
pending = _unwrap_msg(recv_listener.get())
while len(pending) < local_bs and not recv_listener.empty():
pending.extend(_unwrap_msg(recv_listener.get()))
logger.debug(f"Received {len(pending)} messages")
detok = [m for m in pending if isinstance(m, DetokenizeMsg)]
tok = [m for m in pending if isinstance(m, TokenizeMsg)]
abort = [m for m in pending if isinstance(m, AbortMsg)]
if detok:
texts = detokenize_manager.detokenize(detok)
replies = [UserReply(uid=m.uid, incremental_output=t, finished=m.finished)
for m, t in zip(detok, texts, strict=True)]
send_frontend.put(_pack(replies, BatchFrontendMsg))
if tok:
tensors = tokenize_manager.tokenize(tok)
reqs = [UserMsg(uid=m.uid, input_ids=t, sampling_params=m.sampling_params)
for m, t in zip(tok, tensors, strict=True)]
send_backend.put(_pack(reqs, BatchBackendMsg))
if abort:
send_backend.put(_pack([AbortBackendMsg(uid=m.uid) for m in abort],
BatchBackendMsg))
except KeyboardInterrupt:
pass
local_bs 条），然后按类型分三组处理：DetokenizeMsg 反分词后发给 API Server，TokenizeMsg 分词后发给调度器，AbortMsg 转换成 AbortBackendMsg 转给调度器。多条结果打包成一个 Batch...Msg 发送。
默认（--num-tokenizer 0）只起一个进程，同时负责分词和反分词：API Server 发来的 TokenizeMsg 和调度器发来的 DetokenizeMsg 进入同一个队列。分词和反分词都很快，一个进程足以支撑一张卡的吞吐。当请求很多、提示词很长时，可以用 --num-tokenizer N 起 N 个专门分词的进程，与反分词进程分开。
官方实现
tokenizer/detokenize.py · DetokenizeManager.detokenize（第 70～111 行）（注释写明借鉴自 SGLang）tokenizer/server.py · tokenize_worker（第 30～110 行）- 官方的
load_tokenizer还处理了 Mistral 把对话模板放在单独 JSON 文件里的情况
测试¶
tests/test_ch13_tokenizer.py · test_incremental_detokenize_equals_full_decode
def test_incremental_detokenize_equals_full_decode():
tok = AutoTokenizer.from_pretrained(QWEN3)
text = "推理引擎 😀 streams text: naïve café, 1+1=2.\n第二行结束。"
ids = tok(text).input_ids
dm = DetokenizeManager(tok)
pieces = []
for i, t in enumerate(ids):
pieces += dm.detokenize([DetokenizeMsg(uid=0, next_token=t, finished=i == len(ids) - 1)])
assert "".join(pieces) == tok.decode(ids)
assert all("�" not in p for p in pieces) # 不完整的字符从不发给前端
assert 0 not in dm.decode_map # 结束后清理状态
怎么讲清楚
讲反分词：不能每来一个 token 就单独 decode([token])——反分词不可加，一个汉字或 emoji 可能由几个字节级 token 组成（单独解码是乱码），空格和合并规则也依赖上下文。增量反分词只解码最近的一个窗口，用两次解码结果之差得到新增的文本，末尾是不完整字符（\ufffd）时先等，只输出确定的部分。对话要先套模型自带的对话模板再编码；EOS 不发给前端。tokenizer 和 detokenizer 默认放在同一个进程里（都很轻），请求量大时分词可以多进程并行，但反分词的状态在进程内，只能有一个。
练习¶
find_printable_text在文本以非中文字符结尾时，只输出到最后一个空格。为什么？对日文、韩文有什么影响？- 如果两个 detokenizer 进程同时存在（每个请求的消息可能被分到不同的进程），会出什么问题？
- 给
DetokenizeMsg加上"停止字符串"（stop strings）支持：输出文本中出现指定字符串时结束。应该在 detokenizer 里检查，还是在调度器里检查？
参考答案
- 一个英文单词可能被拆成几个 token，而且下一个 token 可能改变前面部分的解码结果（比如 SentencePiece 的空格处理），在空格处截断最保险。日文假名、韩文谚文不在
_is_chinese_char的范围内，它们会被当作"单词"一直等到空格或换行才输出，流式体验变差。 - 增量反分词的状态（
decode_map）在进程内。同一个请求的 token 被分到两个进程，每个进程都只看到一部分，输出错乱。所以反分词只能有一个进程（或者按 uid 固定路由）。 - 停止字符串是文本层面的，必须在反分词之后检查，所以在 detokenizer 里检查最自然；但检查到之后还要通知调度器中止这个请求、释放资源——需要一条从 detokenizer 到调度器的中止消息。SGLang 0.5.20 就是在调度器进程里解码输出末尾的若干 token 来检查停止字符串的（
schedule_batch.py中Req的check_match_stop_str_prefix等方法）。
小结¶
- 分词：对话先套模板再编码；模板必须用模型自带的。
- 增量反分词只解码最近的窗口，用两次解码之差得到新文本；字符不完整时等待，只输出确定的部分。
- tokenizer 进程按消息类型分组处理，默认分词和反分词共用一个进程；反分词状态在进程内，只能有一个 detokenizer。
发现错误或有更好的写法？在 GitHub 上编辑此页提交问题