用 SSE 流式客户端测量 GroqCloud 首 Token 延迟

GroqCloud 主打低延迟。要量化“快”，至少测量到首 token 时间（TTFT）。 这篇用标准库读 SSE data: 流，实现 stream_chat.py，在 mock 上打印内容并记录 TTFT。

先看清成品

• mock 流式输出「快速回复：测试」，TTFT≈58ms（本机 mock，不代表真实 LPU 延迟）。
• 401 覆盖；2 项 pytest 通过。
• 无效 Key 打真实 API 返回 401 Invalid API Key。

合规提醒：使用自己的 GroqCloud 账号与 API Key；语音内容需你有权处理。模型与价格以 groq.com/pricing 与控制台为准，本文不写价格。遵守各模型许可。

运行说明：没有可用的 Groq 密钥。真实运行过：流式聊天 / Whisper / tool-calling 客户端对本地 mock 的请求，以及对 https://api.groq.com 使用无效密钥得到 HTTP 401（2026-10-23，UTC+8）。未上传真实用户音频，未产生计费。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

94-groqcloud-A-src/
├── stream_chat.py
├── mock_groq.py
└── test_94a.py

第 1 步：流式客户端

#!/usr/bin/env python3
"""Groq-compatible streaming chat; measure time-to-first-token (TTFT)."""
from __future__ import annotations

import argparse
import json
import os
import sys
import time
import urllib.error
import urllib.request


def stream_chat(*, base_url: str, api_key: str, model: str, message: str) -> tuple[str, float, float]:
    url = f"{base_url.rstrip('/')}/openai/v1/chat/completions"
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": message}],
        "stream": True,
    }
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode(),
        headers={
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        },
        method="POST",
    )
    t0 = time.perf_counter()
    ttft = None
    parts: list[str] = []
    try:
        with urllib.request.urlopen(req, timeout=60) as resp:
            while True:
                line = resp.readline()
                if not line:
                    break
                line = line.decode("utf-8", errors="replace").strip()
                if not line.startswith("data:"):
                    continue
                data = line[5:].strip()
                if data == "[DONE]":
                    break
                obj = json.loads(data)
                delta = (obj.get("choices") or [{}])[0].get("delta") or {}
                content = delta.get("content")
                if content:
                    if ttft is None:
                        ttft = time.perf_counter() - t0
                    parts.append(content)
                    print(content, end="", flush=True)
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e
    print()
    total = time.perf_counter() - t0
    return "".join(parts), (ttft or total), total


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("message", nargs="?", default="用一句话说明低延迟为什么重要")
    p.add_argument("--base", default=os.environ.get("GROQ_BASE_URL", "https://api.groq.com"))
    p.add_argument("--key", default=os.environ.get("GROQ_API_KEY", ""))
    p.add_argument("--model", default=os.environ.get("GROQ_MODEL", "llama-3.1-8b-instant"))
    args = p.parse_args(argv)
    if not args.key:
        print("设置 GROQ_API_KEY", file=sys.stderr)
        return 2
    text, ttft, total = stream_chat(base_url=args.base, api_key=args.key, model=args.model, message=args.message)
    print(f"TTFT={ttft*1000:.1f}ms total={total*1000:.1f}ms chars={len(text)}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
from __future__ import annotations
import json, time
from http.server import BaseHTTPRequestHandler, HTTPServer

API_KEY = "gsk-test"


class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        if self.path not in ("/openai/v1/chat/completions", "/v1/chat/completions"):
            self.send_error(404); return
        if self.headers.get("Authorization") != f"Bearer {API_KEY}":
            self.send_response(401); self.end_headers(); self.wfile.write(b'{"error":{"message":"Invalid API Key"}}'); return
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        msg = body["messages"][-1]["content"]
        if body.get("stream"):
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.end_headers()
            time.sleep(0.05)  # simulate TTFT
            for i, ch in enumerate(["快", "速", "回", "复", f"：{msg[:20]}"]):
                chunk = {"id": "chatcmpl-x", "choices": [{"delta": {"content": ch}, "index": 0}]}
                self.wfile.write(f"data: {json.dumps(chunk, ensure_ascii=False)}\n\n".encode())
                self.wfile.flush()
                time.sleep(0.01)
            self.wfile.write(b"data: [DONE]\n\n")
            return
        resp = {"id": "c", "choices": [{"message": {"role": "assistant", "content": f"ok:{msg}"}}],
                "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8},
                "model": body.get("model")}
        self.send_response(200); self.send_header("Content-Type", "application/json"); self.end_headers()
        self.wfile.write(json.dumps(resp, ensure_ascii=False).encode())


if __name__ == "__main__":
    import os
    HTTPServer(("127.0.0.1", int(os.environ.get("PORT", "3095"))), H).serve_forever()

第 3 步：测试

import threading
from http.server import HTTPServer
import mock_groq
from stream_chat import stream_chat

def serve():
    httpd = HTTPServer(("127.0.0.1", 0), mock_groq.H)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    return httpd, httpd.server_address[1]

def test_stream_ttft():
    httpd, port = serve()
    try:
        text, ttft, total = stream_chat(base_url=f"http://127.0.0.1:{port}", api_key=mock_groq.API_KEY,
                                        model="llama-3.1-8b-instant", message="测试")
        assert "测试" in text or "快速" in text
        assert ttft > 0
        assert total >= ttft
    finally:
        httpd.shutdown()

def test_401():
    httpd, port = serve()
    try:
        try:
            stream_chat(base_url=f"http://127.0.0.1:{port}", api_key="bad", model="m", message="x")
            assert False
        except SystemExit as e:
            assert "401" in str(e)
    finally:
        httpd.shutdown()

第 4 步：运行

# 2026-10-23 12:37 UTC+8 mock
快速回复：测试
TTFT=58.0ms total=108.8ms chars=7

# 真实 API + 无效密钥
$ curl ... https://api.groq.com/openai/v1/chat/completions -H 'Authorization: Bearer gsk_invalid' ...
401 {"error":{"message":"Invalid API Key","type":"invalid_request_error","code":"invalid_api_key"}}

$ python -m pytest -q
2 passed

怎么检查结果？

• 能边收边打印；TTFT > 0 且 ≤ total。
• mock 延迟不等于生产 TTFT，上线后用真实 Key 再测。

调试示例

readline 与缓冲。 若用 resp.read() 一次读完就失去流式意义。本客户端按行读 data:。注意某些代理会缓冲 SSE，测量 TTFT 时应尽量直连。

常见问题

模板规则

• 流式按行解析 SSE。
• TTFT 与 total 一并记录。
• 密钥放环境变量。

项目资源

“快”要用 TTFT 说话，不能只看体感。