跳转至
手写 mini-sglang 阶段三 · 服务化：变成一个服务

API Server 与启动器

最后一块拼图：一个 FastAPI 应用接收 OpenAI 格式的请求，给每个请求分配 uid，发给 tokenizer，再把陆续回来的增量文本以 SSE 流式写回客户端；一个启动器按正确的顺序拉起所有进程。完成这一章，python -m minisgl --model Qwen/Qwen3-0.6B 就是一个可以用任何 OpenAI 客户端访问的推理服务。

自测：能答上来就可以跳过本章

  1. 成百上千个并发的 HTTP 请求共用一个 ZMQ 接收队列，每个请求怎样只拿到属于自己的回复？
  2. SSE（Server-Sent Events）的格式是什么？OpenAI 流式接口的最后一条消息是什么？
  3. 客户端中途断开连接，服务端要做什么？
  4. 启动器为什么用 spawn 而不是 fork 创建子进程？
自测参考答案（先自己答，再展开对照）
  1. 每个请求有唯一的 uid：接收协程收到回复后按 uid 放进对应的回复列表，并设置那个请求的 asyncio.Event；每个请求的处理协程只等待自己的事件、读取自己的列表。
  2. 每条消息是 data: <JSON> 后面跟一个空行；OpenAI 的流式接口每个片段带一个 delta.content，最后一个片段带 finish_reason，然后发送 data: [DONE]。
  3. 发送 AbortMsg 给调度器，释放这个请求占用的请求槽和 KV，不要继续为它生成。
  4. fork 会复制父进程的状态，包括已经初始化的 CUDA 上下文、线程、ZMQ 的上下文，这些在子进程里不能安全使用；spawn 启动一个干净的新解释器，再各自初始化。

本章要写的文件：server/args.py、server/api_server.py、server/launch.py、__main__.py、shell.py。

到这一章为止的文件：73 个（新建 6，修改 0，提前引入 0）
minisgl/
├── __init__.py
├── core.py
├── env.py
├── __main__.py  ← 本章新建
├── shell.py  ← 本章新建
├── attention/  （4 个文件，不变）
├── distributed/  （3 个文件，不变）
├── engine/  （5 个文件，不变）
├── kernel/  （2 个文件，不变）
├── kvcache/  （5 个文件，不变）
├── layers/  （8 个文件，不变）
├── llm/  （2 个文件，不变）
├── message/  （5 个文件，不变）
├── models/  （7 个文件，不变）
├── moe/  （2 个文件，不变）
├── scheduler/  （9 个文件，不变）
├── server/
│   ├── __init__.py  ← 本章新建
│   ├── args.py  ← 本章新建
│   ├── api_server.py  ← 本章新建
│   ├── launch.py  ← 本章新建
├── tokenizer/  （4 个文件，不变）
├── utils/  （8 个文件，不变）
这一步的 main：examples/ch15_server.py——它只用到上面这些文件；python tools/steps.py check 会逐章搭出这棵树、跑这个 main。

一个共享队列，多个等待者

API Server 是一个 asyncio 程序。所有请求的回复都从同一个 ZMQ 队列进来，需要按 uid 分发给各自的处理协程：

minisgl/server/api_server.py · FrontendManager
@dataclass
class FrontendManager:
    config: ServerArgs
    send_tokenizer: ZmqAsyncPushQueue[BaseTokenizerMsg]
    recv_tokenizer: ZmqAsyncPullQueue[BaseFrontendMsg]
    uid_counter: int = 0
    initialized: bool = False
    ack_map: Dict[int, List[UserReply]] = field(default_factory=dict)
    event_map: Dict[int, asyncio.Event] = field(default_factory=dict)

    def new_user(self) -> int:
        uid = self.uid_counter
        self.uid_counter += 1
        self.ack_map[uid] = []
        self.event_map[uid] = asyncio.Event()
        return uid

    async def listen(self) -> None:
        while True:
            for reply in _unwrap(await self.recv_tokenizer.get()):
                if reply.uid not in self.ack_map:  # 已经结束或已取消的请求
                    continue
                self.ack_map[reply.uid].append(reply)
                self.event_map[reply.uid].set()

    async def send_one(self, msg: BaseTokenizerMsg) -> None:
        if not self.initialized:  # 第一次调用时（此时事件循环已经在运行）才能创建后台任务
            asyncio.create_task(self.listen())
            self.initialized = True
        await self.send_tokenizer.put(msg)

    async def wait_for_ack(self, uid: int) -> AsyncIterator[UserReply]:
        event = self.event_map[uid]
        try:
            while True:
                await event.wait()
                event.clear()
                pending, self.ack_map[uid] = self.ack_map[uid], []
                for ack in pending:
                    yield ack
                    if ack.finished:
                        return
        finally:
            self.ack_map.pop(uid, None)
            self.event_map.pop(uid, None)

    async def stream_generate(self, uid: int) -> AsyncIterator[bytes]:
        async for ack in self.wait_for_ack(uid):
            yield f"data: {ack.incremental_output}\n".encode()
        yield b"data: [DONE]\n"

    async def stream_chat_completions(self, uid: int) -> AsyncIterator[bytes]:
        first = True
        async for ack in self.wait_for_ack(uid):
            delta: Dict[str, str] = {}
            if first:
                delta["role"] = "assistant"
                first = False
            if ack.incremental_output:
                delta["content"] = ack.incremental_output
            chunk = {"id": f"chatcmpl-{uid}", "object": "chat.completion.chunk",
                     "choices": [{"delta": delta, "index": 0, "finish_reason": None}]}
            yield f"data: {json.dumps(chunk)}\n\n".encode()
        end = {"id": f"chatcmpl-{uid}", "object": "chat.completion.chunk",
               "choices": [{"delta": {}, "index": 0, "finish_reason": "stop"}]}
        yield f"data: {json.dumps(end)}\n\n".encode()
        yield b"data: [DONE]\n\n"

    async def stream_with_cancellation(self, gen: AsyncIterator[bytes], request: Request,
                                       uid: int) -> AsyncIterator[bytes]:
        """客户端断开连接时，通知后端中止这个请求，释放它占用的 KV 缓存。"""
        try:
            async for chunk in gen:
                if await request.is_disconnected():
                    raise asyncio.CancelledError
                yield chunk
        except asyncio.CancelledError:
            asyncio.create_task(self.abort_user(uid))
            raise

    async def abort_user(self, uid: int) -> None:
        await asyncio.sleep(0.1)
        self.ack_map.pop(uid, None)
        self.event_map.pop(uid, None)
        logger.warning(f"Aborting request {uid}")
        await self.send_one(AbortMsg(uid=uid))

    def shutdown(self) -> None:
        self.send_tokenizer.stop()
        self.recv_tokenizer.stop()
- new_user() 为新请求分配 uid，并创建一个回复列表和一个 asyncio.Event； - 后台协程 listen() 不断从 ZMQ 收 UserReply，追加到对应 uid 的列表里，并 set() 它的事件； - 每个请求的处理协程在 wait_for_ack() 里 await event.wait()，醒来后取走列表里的所有回复逐个 yield，直到遇到 finished； - listen() 在第一次发送时才创建：必须在事件循环已经运行之后，而 FastAPI 应用启动时还没有事件循环。

这是 asyncio 里"一个生产者、多个消费者按键分发"的标准写法。事件被 clear() 之后可能有新回复到达并再次 set()，所以取走回复时是把整个列表换掉，不会漏掉。

OpenAI 兼容接口

minisgl/server/api_server.py · v1_chat_completions
@app.post("/v1/chat/completions")
async def v1_chat_completions(req: OpenAICompletionRequest, request: Request):
    state = get_state()
    if req.messages:
        text: str | List[Dict[str, str]] = [m.model_dump() for m in req.messages]
    else:
        assert req.prompt is not None, "Either 'messages' or 'prompt' must be provided"
        text = req.prompt
    uid = state.new_user()
    await state.send_one(TokenizeMsg(uid=uid, text=text, sampling_params=_sampling_params(req)))
    if req.stream:
        return StreamingResponse(
            state.stream_with_cancellation(state.stream_chat_completions(uid), request, uid),
            media_type="text/event-stream")
    content = ""
    async for ack in state.wait_for_ack(uid):
        content += ack.incremental_output
    return {
        "id": f"chatcmpl-{uid}",
        "object": "chat.completion",
        "created": int(time.time()),
        "model": req.model,
        "choices": [{"index": 0, "message": {"role": "assistant", "content": content},
                     "finish_reason": "stop"}],
    }
- messages（对话）和 prompt（纯文本）二选一，前者由 tokenizer 套用对话模板； - 采样参数原样转给调度器，ignore_eos 是一个扩展字段（压测时常用）； - 流式时返回 text/event-stream：每条回复变成一个 data: {...}\n\n，其中 choices[0].delta.content 是增量文本；最后发一条 finish_reason: "stop" 和 data: [DONE]； - 非流式时收集全部回复拼成一个完整的 JSON。

minisgl/server/api_server.py · FrontendManager.stream_chat_completions
async def stream_chat_completions(self, uid: int) -> AsyncIterator[bytes]:
    first = True
    async for ack in self.wait_for_ack(uid):
        delta: Dict[str, str] = {}
        if first:
            delta["role"] = "assistant"
            first = False
        if ack.incremental_output:
            delta["content"] = ack.incremental_output
        chunk = {"id": f"chatcmpl-{uid}", "object": "chat.completion.chunk",
                 "choices": [{"delta": delta, "index": 0, "finish_reason": None}]}
        yield f"data: {json.dumps(chunk)}\n\n".encode()
    end = {"id": f"chatcmpl-{uid}", "object": "chat.completion.chunk",
           "choices": [{"delta": {}, "index": 0, "finish_reason": "stop"}]}
    yield f"data: {json.dumps(end)}\n\n".encode()
    yield b"data: [DONE]\n\n"

断开连接时中止请求

客户端断开后，如果服务端继续生成，这个请求会一直占用一个请求槽和它的 KV 缓存，直到达到 max_tokens。所以每发一段数据前检查连接是否还在：

minisgl/server/api_server.py · FrontendManager.stream_with_cancellation
async def stream_with_cancellation(self, gen: AsyncIterator[bytes], request: Request,
                                   uid: int) -> AsyncIterator[bytes]:
    """客户端断开连接时，通知后端中止这个请求，释放它占用的 KV 缓存。"""
    try:
        async for chunk in gen:
            if await request.is_disconnected():
                raise asyncio.CancelledError
            yield chunk
    except asyncio.CancelledError:
        asyncio.create_task(self.abort_user(uid))
        raise
检测到断开后，删除这个 uid 的等待状态，发送 AbortMsg。它经过 tokenizer 变成 AbortBackendMsg，调度器从等待队列或 decode 集合里找到这个请求，释放资源（第 11 章讨论了它与重叠调度交织时的细节）。

启动器

minisgl/server/launch.py · start_backend
def start_backend(server_args: ServerArgs) -> None:
    """启动所有后端进程：每个 TP rank 一个调度器进程，一个 detokenizer，若干 tokenizer。"""
    from minisgl.tokenizer import tokenize_worker

    logger = init_logger(__name__, "initializer")
    mp.set_start_method("spawn", force=True)  # CUDA 不支持 fork 出来的子进程
    ack_queue: mp.Queue = mp.Queue()
    world_size = server_args.tp_info.size
    for i in range(world_size):
        mp.Process(target=_run_scheduler,
                   args=(replace(server_args, tp_info=DistributedInfo(i, world_size)), ack_queue),
                   name=f"minisgl-TP{i}-scheduler").start()
    common = dict(tokenizer_path=server_args.model_path, backend_addr=server_args.zmq_backend_addr,
                  frontend_addr=server_args.zmq_frontend_addr, local_bs=1, ack_queue=ack_queue)
    n = server_args.num_tokenizer
    mp.Process(target=tokenize_worker, name="minisgl-detokenizer",
               kwargs=dict(common, addr=server_args.zmq_detokenizer_addr,
                           create=server_args.tokenizer_create_addr, tokenizer_id=n)).start()
    for i in range(n):
        mp.Process(target=tokenize_worker, name=f"minisgl-tokenizer-{i}",
                   kwargs=dict(common, addr=server_args.zmq_tokenizer_addr,
                               create=server_args.tokenizer_create_addr, tokenizer_id=i)).start()
    for _ in range(n + 2):  # rank 0 调度器 + n 个 tokenizer + 1 个 detokenizer
        logger.info(ack_queue.get())
启动顺序：

  1. 主进程解析参数，创建 FrontendManager（此时绑定前端的 ZMQ 地址）；
  2. 用 spawn 方式启动每个 TP rank 的调度器进程、一个 detokenizer 进程、N 个 tokenizer 进程。CUDA 不能在 fork 出来的子进程里重新初始化，必须用 spawn；
  3. 每个进程准备好后往一个 multiprocessing.Queue 里放一条"就绪"消息（调度器只有 rank 0 发，且在所有 rank 完成 sync_all_ranks 之后）；主进程收齐 num_tokenizer + 2 条后才继续；
  4. 启动 uvicorn（或者交互式 shell）。

minisgl/server/launch.py · _run_scheduler
def _run_scheduler(args: ServerArgs, ack_queue: mp.Queue) -> None:
    import torch
    from minisgl.scheduler import Scheduler

    with torch.inference_mode():
        scheduler = Scheduler(args)
        scheduler.sync_all_ranks()  # 所有 rank 都初始化完（包括 ZMQ 订阅）再开始
        if args.tp_info.is_primary():
            ack_queue.put("Scheduler is ready")
        if args.silent_output:
            logging.disable(logging.INFO)
        try:
            scheduler.run_forever()
        except KeyboardInterrupt:
            scheduler.shutdown()
交互模式（--shell）不起 HTTP 服务，直接在终端里对话；它把 max_running_req 设为 1、只录制批大小为 1 的 CUDA Graph，因为只有一个用户。

运行

examples/ch15_server.py
"""第 15 章：启动完整的服务，用 OpenAI 客户端访问。"""

import os
import signal
import socket
import subprocess
import sys
import time
from concurrent.futures import ThreadPoolExecutor

import httpx
import psutil
from openai import OpenAI

with socket.socket() as s:
    s.bind(("127.0.0.1", 0))
    port = s.getsockname()[1]
proc = subprocess.Popen(
    [sys.executable, "-m", "minisgl", "--model", "models/Qwen3-0.6B", "--dtype", "float32",
     "--port", str(port), "--num-pages", "4096", "--max-seq-len-override", "1024"],
    stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)
try:
    t = time.time()
    while True:
        try:
            if httpx.get(f"http://127.0.0.1:{port}/v1/models", timeout=1).status_code == 200:
                break
        except httpx.HTTPError:
            time.sleep(0.5)
    children = psutil.Process(proc.pid).children(recursive=True)
    workers = [c for c in children if "multiprocessing-fork" in " ".join(c.cmdline())]
    print(f"服务就绪（约 {time.time() - t:.0f} 秒）：主进程（API Server）+ {len(workers)} 个工作进程"
          f"（调度器、tokenizer）")
    client = OpenAI(base_url=f"http://127.0.0.1:{port}/v1", api_key="none")

    r = client.chat.completions.create(
        model="qwen3", messages=[{"role": "user", "content": "用一句话介绍北京。/no_think"}],
        max_tokens=48, temperature=0)
    print("非流式:", repr(r.choices[0].message.content))

    stream = client.chat.completions.create(
        model="qwen3", messages=[{"role": "user", "content": "Count from 1 to 5. /no_think"}],
        max_tokens=24, temperature=0, stream=True)
    chunks = [c.choices[0].delta.content for c in stream if c.choices[0].delta.content]
    print(f"流式: 收到 {len(chunks)} 个片段:", chunks[:12], "...")

    def ask(i: int) -> float:
        t0 = time.perf_counter()
        client.chat.completions.create(model="qwen3", messages=[{"role": "user", "content": f"{i}+{i}=?"}],
                                       max_tokens=16, temperature=0, extra_body={"ignore_eos": True})
        return time.perf_counter() - t0

    t0 = time.perf_counter()
    with ThreadPoolExecutor(8) as ex:
        lat = list(ex.map(ask, range(8)))
    print(f"8 个并发请求（各 16 个 token）总用时 {time.perf_counter() - t0:.2f}s，单个 {min(lat):.2f}～{max(lat):.2f}s")
finally:
    os.killpg(proc.pid, signal.SIGKILL)
运行结果
服务就绪（约 9 秒）：主进程（API Server）+ 2 个工作进程（调度器、tokenizer）
非流式: '<think>\n\n</think>\n\n北京是中国的首都，位于中国北方，是世界著名的历史文化名城。'
流式: 收到 24 个片段: ['<think>', '\n\n', '</think>', '\n\n', 'Count', ' from', ' ', '1', ' to', ' ', '5', ':'] ...
8 个并发请求（各 16 个 token）总用时 1.32s，单个 1.31～1.32s
- 进程：API Server 主进程，加上一个调度器进程和一个 tokenizer 进程（默认分词、反分词共用）。 - 非流式：Qwen3 的回答先是一段空的 <think></think>（/no_think 关闭了思考），然后是答案。 - 流式：每个 token 一个片段，特殊 token <think> 也作为一个完整片段发出。 - 8 个并发请求被调度器合成一个 batch，总用时与单个请求几乎相同——这就是连续批处理带来的吞吐。

官方实现

官方的 /v1/chat/completions 非流式响应里 usage 字段恒为 0；我们直接省略了它。官方交互模式在退出时用 psutil 杀掉所有子进程，我们沿用了这个做法。

测试

tests/test_ch15_server.py · test_client_disconnect_aborts_request
def test_client_disconnect_aborts_request(server):
    """客户端中途断开：服务端中止请求并释放资源，之后的请求照常工作。"""
    port, log_path = server
    with httpx.stream("POST", f"http://127.0.0.1:{port}/v1/chat/completions",
                      json={"model": "m", "prompt": "Tell me a long story.", "max_tokens": 500,
                            "stream": True, "ignore_eos": True}, timeout=60) as r:
        for i, _ in enumerate(r.iter_lines()):
            if i >= 3:
                break  # 读了几段就断开
    time.sleep(1.0)
    resp = httpx.post(f"http://127.0.0.1:{port}/v1/chat/completions",
                      json={"model": "m", "prompt": "1 + 1 =", "max_tokens": 4}, timeout=60)
    assert resp.status_code == 200
    assert "Aborting request" in log_path.read_text()
tests/test_ch15_server.py 另外两个测试：非流式输出与 Hugging Face 的贪心结果逐字相同；流式与并发请求都能正确返回。

怎么讲清楚

讲 API Server：成百上千个 HTTP 请求共用一个 ZMQ 接收队列，用"按 uid 分发的回复列表 + 每个请求一个 asyncio.Event"——后台任务收到回复后放进对应请求的列表并唤醒它。OpenAI 流式接口是 SSE：每个片段一行 data: {...}，内容在 delta.content 里，最后一条带 finish_reason，再发 data: [DONE]。客户端中途断开时要发中止消息，释放请求槽和 KV，否则引擎还在为没人要的请求生成。启动器用 spawn 而不是 fork 创建子进程：fork 会复制父进程里已经初始化的 CUDA、线程和锁的状态，不安全。

练习

  1. 在非流式响应里加上正确的 usage（提示词 token 数、生成 token 数）。这些数字在哪个进程里知道？需要在哪些消息里加字段？
  2. 实现 /v1/completions（非对话的补全接口），它和 /v1/chat/completions 的请求、响应格式有什么不同？
  3. 现在 uid 由 API Server 单调递增分配。如果将来要起多个 API Server 进程共同对外服务，uid 怎么分配才不会冲突？
参考答案
  1. 提示词长度在 tokenizer 里知道，生成数量在调度器里知道。可以在 UserReply 里加 prompt_tokens、completion_tokens（由 detokenizer 在结束时填入：它能数出收到了几个 token，提示词长度需要 tokenizer 在分词时记下、或者由调度器随最后一条 DetokenizeMsg 带回）。
  2. 请求用 prompt 字段而不是 messages，不套对话模板；响应里是 choices[0].text 而不是 message.content，流式时每个片段是 choices[0].text，object 是 text_completion。
  3. 让每个 API Server 使用不同的前缀或步长，例如 uid = 进程编号 + 进程数 × 本地计数；或者使用 UUID。detokenizer 的回复也要能路由回对应的 API Server（每个 API Server 一个接收地址，消息里带上来源）。

小结

  • API Server 用"按 uid 分发的回复列表 + asyncio.Event"让所有请求共用一个 ZMQ 接收队列。
  • OpenAI 流式接口：SSE，每个片段一个 delta.content，最后是 finish_reason 和 [DONE]。
  • 客户端断开时发送 AbortMsg，释放请求占用的请求槽和 KV。
  • 启动器用 spawn 拉起调度器和 tokenizer 进程，等所有进程就绪再启动 HTTP 服务。

本章练习题（1 道）

在浏览器里直接写代码、跑测试；也可以在本地用 python practice/judge.py 判题。

发现错误或有更好的写法？在 GitHub 上编辑此页提交问题