stream=True 之后，真正麻烦的是这四件事

data: {"id":"chatcmpl-1","choices":[{"delta":{"role":"assistant"}}]}

data: {"id":"chatcmpl-1","choices":[{"delta":{"content":"你"}}]}

data: {"id":"chatcmpl-1","choices":[{"delta":{"content":"好"}}]}

data: [DONE]

for chunk in resp.iter_bytes():
    for line in chunk.decode("utf-8").split("\n"):   # ← 问题在这一行
        if line.startswith("data:"):
            obj = json.loads(line[5:])                # 偶发 JSONDecodeError

第一次 read:  data: {"id":"chat","choices":[{"delta":{"con
第二次 read:  tent":"你好"}}]}

第一次 read 的字节: ... E4 BD        ← 「你」的前 2 个字节
第二次 read 的字节: A0 ...            ← 「你」的最后 1 个字节

import codecs

decoder = codecs.getincrementaldecoder("utf-8")()
text = decoder.decode(raw_bytes, final=False)

第一次请求的输出：今天天气
连接断开 → 自动重试
第二次请求的输出：今天天气真不错
最终用户看到的：  今天天气今天天气真不错

# -*- coding: utf-8 -*-
"""流式调用：SSE 分帧、增量解码、重试边界。"""
import json
import random
import time

import httpx


def stream_once(url, key, payload, timeout=120.0):
    """单次流式请求，逐段 yield 文本增量。"""
    headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
    buf = ""

    with httpx.Client(timeout=timeout) as client:
        with client.stream("POST", url, json=payload, headers=headers) as resp:
            if resp.status_code >= 400:
                resp.read()  # 错误体是普通 JSON，先读出来再抛
                resp.raise_for_status()

            # iter_text() 内部用增量解码器，跨 chunk 的多字节字符不会被切坏
            for chunk in resp.iter_text():
                # 统一换行，兼容网关把 \n 改写成 \r\n 的情况
                buf += chunk.replace("\r\n", "\n")

                # 关键：先聚合，再按空行切帧；不要直接按行 json.loads
                while "\n\n" in buf:
                    frame, buf = buf.split("\n\n", 1)
                    for line in frame.split("\n"):
                        if not line.startswith("data:"):
                            continue
                        data = line[5:].strip()
                        if data == "[DONE]":
                            return
                        try:
                            obj = json.loads(data)
                        except json.JSONDecodeError:
                            continue  # 畸形帧跳过，不要因此中断整个流
                        choices = obj.get("choices") or []
                        if not choices:
                            continue
                        delta = choices[0].get("delta") or {}
                        text = delta.get("content")
                        if text:
                            yield text


def stream_chat(url, key, payload, max_retries=2, base_delay=0.5):
    """带重试的流式调用：只在尚未产出任何内容时才重试。"""
    for attempt in range(max_retries + 1):
        emitted = False
        try:
            for piece in stream_once(url, key, payload):
                emitted = True
                yield piece
            return
        except httpx.HTTPStatusError as err:
            status = err.response.status_code
            retryable = status == 429 or status >= 500
            if emitted or not retryable or attempt == max_retries:
                raise
            # 429 优先听服务端的 Retry-After
            wait = None
            if status == 429:
                raw = err.response.headers.get("Retry-After")
                if raw and raw.isdigit():
                    wait = float(raw)
            if wait is None:
                wait = base_delay * (2 ** attempt) + random.uniform(0, 0.2)
            time.sleep(wait)
        except httpx.TransportError:
            if emitted or attempt == max_retries:
                raise
            time.sleep(base_delay * (2 ** attempt) + random.uniform(0, 0.2))

import os

payload = {
    "model": "your-model",
    "messages": [{"role": "user", "content": "用三句话介绍什么是 SSE"}],
    "stream": True,
}

# 接口地址和密钥从环境变量读，不要写死在代码里
base_url = os.environ["LLM_BASE_URL"]      # 形如「协议://主机/v1」
api_key = os.environ["LLM_API_KEY"]

t0 = time.perf_counter()
ttft = None
pieces = []

gen = stream_chat(base_url + "/chat/completions", api_key, payload)
try:
    for piece in gen:
        if ttft is None:
            ttft = time.perf_counter() - t0   # 首 token 延迟
        pieces.append(piece)
finally:
    gen.close()   # 关键：中断时主动关闭，否则上游会继续生成、继续计费

total = time.perf_counter() - t0
print(f"首 token 延迟 {ttft:.2f}s，总耗时 {total:.2f}s")
print("".join(pieces))

gen = stream_chat(url, key, payload)
try:
    for piece in gen:
        print(piece, end="", flush=True)
        if user_pressed_stop():
            break
finally:
    gen.close()