LLM 流式输出与实时交互工程

客户端（浏览器 / App / 终端）
   │  EventSource / fetch(ReadableStream) / WebSocket
   ▼
API 网关（认证、限流、转发）
   ▼
流式服务（应用逻辑、Prompt 组装、工具调用）
   ▼
模型网关（多厂商路由、重试、熔断）
   ▼
模型推理（prefill + decode，逐 token 返回）

async def answer_stream(query: str) -> AsyncIterator[str]:
    """流式接口：逐 token 产出，而非一次性返回完整结果。"""
    async for token in model_stream(query):
        yield token

event: token
data: {"delta": "你", "index": 0}

event: token
data: {"delta": "好", "index": 1}

event: done
data: {}

# sse_server.py — 用 FastAPI 实现 SSE 流式端点
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import json

app = FastAPI()

async def sse_generator(query: str):
    yield 'event: start\ndata: {"role":"assistant"}\n\n'
    async for token in model_stream(query):
        payload = json.dumps({"delta": token}, ensure_ascii=False)
        yield f"event: token\ndata: {payload}\n\n"
    yield "event: done\ndata: {}\n\n"

@app.get("/chat/stream")
async def chat_stream(query: str):
    headers = {
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        "X-Accel-Buffering": "no",   # 禁止 Nginx 缓冲，保证逐条透传
    }
    return StreamingResponse(sse_generator(query),
                             media_type="text/event-stream",
                             headers=headers)

// sse_client.js — 浏览器端增量渲染
const source = new EventSource("/chat/stream?query=" + encodeURIComponent(q));

source.addEventListener("token", (e) => {
  const data = JSON.parse(e.data);
  render.appendDelta(data.delta);        // 追加到光标处
  autoScroll();                          // 保持可见
});

source.addEventListener("done", () => {
  source.close();
  setDoneState();
});

source.onerror = () => {
  // EventSource 默认自动重连；服务端需要保证幂等
  console.log("连接断开，等待自动重连");
};

# incremental_parser.py — 增量 JSON 解析
import json

class IncrementalJsonParser:
    def __init__(self):
        self.buffer = ""

    def feed(self, delta: str):
        """喂入新的字符片段，尝试把累积缓冲区解析为 JSON。"""
        self.buffer += delta
        try:
            return json.loads(self.buffer), True
        except json.JSONDecodeError:
            return None, False   # 不完整，等下一片

    def is_done(self, finish_reason: str) -> bool:
        return finish_reason == "stop"

# render_scheduler.py — 时间片批渲染
import asyncio

class RenderScheduler:
    def __init__(self, interval=0.08):
        self.interval = interval
        self.batch = []

    async def submit(self, token: str):
        self.batch.append(token)

    async def flush_loop(self, on_flush):
        while True:
            await asyncio.sleep(self.interval)
            if not self.batch:
                continue
            buf = "".join(self.batch)
            self.batch.clear()
            on_flush(buf)

# tool_aware_stream.py — 流式工具调用编排
async def agent_stream(user_query: str):
    """Agent 流：遇到工具调用时暂停，执行后恢复。"""
    pending_tool = None
    async for event in model_agent_stream(user_query):
        if event.type == "tool_call":
            yield {"event": "tool_call", "name": event.name,
                   "args": event.arguments}
            pending_tool = event
            break
        yield {"event": "token", "delta": event.delta}

    if pending_tool:
        result = await run_tool(pending_tool.name, pending_tool.arguments)
        yield {"event": "tool_result", "result": result}
        async for event in model_continue_stream(pending_tool, result):
            yield {"event": "token", "delta": event.delta}
    yield {"event": "done"}

def measure_ttft(start_ns: int, first_token_ns: int) -> float:
    """TTFT 度量：从请求发出到第一个 token 上屏（毫秒）。"""
    return (first_token_ns - start_ns) / 1e6

请求 1: [System][历史 H][问题 1]  ← 全价
请求 2: [System][历史 H][问题 2]  ← 前缀 [System][历史 H] 命中缓存
                                     只对新问题 prefill，首 token 更快

# ttft_cache.py — 缓存感知的流式调用
def stream_with_cache(system_prompt, history, query):
    """相同前缀命中缓存，只对新 query 做 prefill。"""
    messages = [{"role": "system", "content": system_prompt}] + history
    messages.append({"role": "user", "content": query})
    return client.chat.completions.create(
        model="gpt-4o", messages=messages, stream=True)

# cancellation.py — 中断的逐级传播
class StreamCancellation:
    def __init__(self):
        self._cancel = asyncio.Event()

    async def cancel(self):
        """用户点击停止 → 置位取消事件。"""
        self._cancel.set()

    async def run(self, async_gen):
        task = asyncio.create_task(self._drain(async_gen))
        try:
            await self._cancel.wait()
        finally:
            task.cancel()
            await close_downstream()   # 关闭 SSE 连接、释放资源

    async def _drain(self, ag):
        async for item in ag:
            pass

async def stream_with_client_disconnect(query):
    try:
        async for token in model_stream(query):
            yield token
    except asyncio.CancelledError:
        await model_client.cancel_last_request()
        raise

// resume_client.js — 基于 index 的续传
let lastIndex = 0;

function reconnectWithResume(query) {
  const es = new EventSource(`/chat/stream?query=${query}&resume_from=${lastIndex}`);
  es.addEventListener("token", (e) => {
    const data = JSON.parse(e.data);
    if (data.index <= lastIndex) return;   // 丢弃重复片段
    render.appendDelta(data.delta);
    lastIndex = data.index;
  });
}

# stream_metrics.py — 流式指标埋点
def trace_stream_span(query, async_gen):
    """为每个流式请求建立 span，逐阶段打点。"""
    span = tracer.start_span("llm.stream", attributes={"query": query})
    span.set_attribute("ttft_ms", measure_ttft())
    token_count = 0
    async for item in async_gen:
        token_count += 1
        span.set_attribute("tokens_seen", token_count)
        yield item
    span.set_attribute("completed", got_done)
    span.set_attribute("finish_reason", finish_reason)
    span.end()

# backpressure.py — 信号量限流 + 队列背压
import asyncio

class StreamGate:
    def __init__(self, max_concurrent=200):
        self.sem = asyncio.Semaphore(max_concurrent)

    async def handle(self, query: str):
        async with self.sem:                 # 超限排队而非丢弃
            async for token in model_stream(query):
                yield token

# nginx.conf — 流式端点禁止缓冲
location /chat/stream {
    proxy_pass http://llm_service;
    proxy_buffering off;            # 关闭缓冲
    proxy_cache off;
    proxy_read_timeout 300s;        # 长连接不超时
    chunked_transfer_encoding on;
}

def test_stream_sequence():
    """断言事件顺序、done 结束、finish_reason 正确。"""
    events = list(collect_events("今天天气怎么样？"))
    kinds = [e["event"] for e in events]
    assert kinds[0] == "start"
    assert kinds[-1] == "done"
    assert all(e["event"] != "done" for e in events[:-1])
    text = "".join(e["delta"] for e in events if e["event"] == "token")
    assert len(text) > 0