一、问题：用户盯着白屏等 30 秒

非流式：

用户点"发送" → [等待 30s...] → 全文一次出来

流式：

用户点"发送" → W → We → Wel → Welc → Welco → Welcom → Welcome

把 stream=True 打开不难，难的是整个链路：

• SSE 事件解析
• Tool call 流式增量
• 前端消费 + 背压
• 取消生成
• Nginx 代理不缓冲

本文从 FastAPI 后端 到 前端 fetch/React，给完整可运行的实现。

---

二、整体架构

┌─────────────┐   SSE (text/event-stream)    ┌──────────────┐
│   Browser   │ ◄──────────────────────────── │  FastAPI     │
│ (fetch+RS)  │                               │  (Python)    │
└─────────────┘                               └──────┬───────┘
                                                     │ OpenAI SDK
                                                     │ stream=True
                                                     ▼
                                              ┌──────────────┐
                                              │  LLM API     │
                                              └──────────────┘

SSE 事件格式：

data: {"type": "token", "content": "Hello"}
data: {"type": "token", "content": " World"}
data: {"type": "tool_call_start", "name": "search", "call_id": "c1"}
data: {"type": "tool_call_delta", "call_id": "c1", "arguments": "{\"q\""}
data: {"type": "tool_call_end", "call_id": "c1"}
data: {"type": "done", "usage": {...}, "latency_ms": 1234}
data: {"type": "error", "message": "..."}

---

三、后端实现 (FastAPI + SSE)

# streaming_server.py - FastAPI 流式聊天服务
# 启动: pip install fastapi uvicorn openai
#       uvicorn streaming_server:app --reload --port 8000

import json
import asyncio
import time
from typing import Optional, List, Dict, Any, AsyncGenerator

from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from openai import AsyncOpenAI


# ============================================================
# 数据模型
# ============================================================

class ChatMessage(BaseModel):
    role: str
    content: Optional[str] = None


class ChatRequest(BaseModel):
    messages: List[ChatMessage]
    model: str = "gpt-4o-mini"
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    max_tokens: int = Field(default=4096, ge=1)


# ============================================================
# SSE 事件辅助
# ============================================================

class SSE:
    # SSE 事件构建器

    @staticmethod
    def token(content: str) -> str:
        return f"data: {json.dumps({'type': 'token', 'content': content}, ensure_ascii=False)}\n\n"

    @staticmethod
    def thinking(content: str) -> str:
        return f"data: {json.dumps({'type': 'thinking', 'content': content}, ensure_ascii=False)}\n\n"

    @staticmethod
    def tool_call_start(name: str, call_id: str) -> str:
        return f"data: {json.dumps({'type': 'tool_call_start', 'name': name, 'call_id': call_id})}\n\n"

    @staticmethod
    def tool_call_delta(call_id: str, arguments: str) -> str:
        return f"data: {json.dumps({'type': 'tool_call_delta', 'call_id': call_id, 'arguments': arguments})}\n\n"

    @staticmethod
    def tool_call_end(call_id: str) -> str:
        return f"data: {json.dumps({'type': 'tool_call_end', 'call_id': call_id})}\n\n"

    @staticmethod
    def done(usage: Dict, latency_ms: float) -> str:
        return f"data: {json.dumps({'type': 'done', 'usage': usage, 'latency_ms': round(latency_ms, 2)})}\n\n"

    @staticmethod
    def error(message: str, code: str = "unknown") -> str:
        return f"data: {json.dumps({'type': 'error', 'message': message, 'code': code})}\n\n"

    @staticmethod
    def heartbeat() -> str:
        return ": heartbeat\n\n"


# ============================================================
# 流式聊天服务
# ============================================================

class StreamingChatService:
    # 处理普通文本流 + 推理流 + Tool call 流

    def __init__(self, client: AsyncOpenAI):
        self.client = client

    async def stream_chat(
        self, request: ChatRequest,
        cancel_event: Optional[asyncio.Event] = None,
        tools: Optional[List[Dict]] = None,
    ) -> AsyncGenerator[str, None]:
        start = time.time()
        usage = {"prompt_tokens": 0, "completion_tokens": 0}

        try:
            messages = [{"role": m.role, "content": m.content}
                        for m in request.messages]

            stream = await self.client.chat.completions.create(
                model=request.model, messages=messages,
                temperature=request.temperature,
                max_tokens=request.max_tokens, stream=True,
                stream_options={"include_usage": True},
                tools=tools,
            )

            # 追踪 tool call 增量
            tc_buf: Dict[int, Dict] = {}

            async for chunk in stream:
                if cancel_event and cancel_event.is_set():
                    yield SSE.error("Cancelled by user", "cancelled")
                    return

                delta = chunk.choices[0].delta if chunk.choices else None
                if not delta:
                    continue

                # 推理内容 (DeepSeek-R1 thinking)
                if hasattr(delta, "reasoning_content") and delta.reasoning_content:
                    yield SSE.thinking(delta.reasoning_content)
                    continue

                # 普通文本
                if delta.content:
                    yield SSE.token(delta.content)

                # Tool calls
                if delta.tool_calls:
                    for tc in delta.tool_calls:
                        idx = tc.index
                        while len(tc_buf) <= idx:
                            tc_buf[idx] = {"id": "", "name": "", "args": ""}

                        if tc.id:
                            tc_buf[idx]["id"] = tc.id
                        if tc.function:
                            if tc.function.name:
                                tc_buf[idx]["name"] = tc.function.name
                                yield SSE.tool_call_start(
                                    tc.function.name, tc_buf[idx]["id"] or str(idx)
                                )
                            if tc.function.arguments:
                                tc_buf[idx]["args"] += tc.function.arguments
                                yield SSE.tool_call_delta(
                                    tc_buf[idx]["id"] or str(idx),
                                    tc.function.arguments,
                                )

            # 所有 tool calls 结束
            for idx in tc_buf:
                yield SSE.tool_call_end(tc_buf[idx]["id"] or str(idx))

            latency = (time.time() - start) * 1000
            yield SSE.done(usage=usage, latency_ms=latency)

        except asyncio.CancelledError:
            yield SSE.error("Stream cancelled", "cancelled")
        except Exception as e:
            yield SSE.error(str(e), type(e).__name__)


# ============================================================
# FastAPI 应用
# ============================================================

openai_client = AsyncOpenAI(api_key="sk-your-api-key")
chat_service = StreamingChatService(openai_client)
app = FastAPI(title="Streaming Chat API")


@app.get("/health")
async def health():
    return {"status": "ok"}


@app.post("/chat/stream")
async def chat_stream(request: ChatRequest, http_request: Request):
    # 流式端点, 返回 text/event-stream

    async def is_disconnected() -> bool:
        return await http_request.is_disconnected()

    async def generate():
        async for event in chat_service.stream_chat(request):
            if await is_disconnected():
                break
            yield event
            await asyncio.sleep(0)  # 避免前端压力过大

    return StreamingResponse(
        generate(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",        # 禁用 nginx 缓冲
            "Access-Control-Allow-Origin": "*",
        },
    )


@app.post("/chat")
async def chat_non_stream(request: ChatRequest):
    # 非流式端点 (兼容旧版)
    messages = [{"role": m.role, "content": m.content}
                for m in request.messages]
    resp = await openai_client.chat.completions.create(
        model=request.model, messages=messages,
        temperature=request.temperature,
        max_tokens=request.max_tokens,
    )
    return {
        "content": resp.choices[0].message.content,
        "usage": {
            "prompt_tokens": resp.usage.prompt_tokens if resp.usage else 0,
            "completion_tokens": resp.usage.completion_tokens if resp.usage else 0,
        },
    }


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

---

四、Tool Call 流式处理流程

完整交互时序：

User: "今天北京天气怎么样？"
  │
  ▼  [LLM 决定调用 get_weather]
SSE: {"type": "tool_call_start", "name": "get_weather", "call_id": "c1"}
SSE: {"type": "tool_call_delta", "call_id": "c1", "arguments": "{\"city\""}
SSE: {"type": "tool_call_delta", "call_id": "c1", "arguments": ": \"北京\"}"}
SSE: {"type": "tool_call_end", "call_id": "c1"}
  │
  ▼  [后端执行 get_weather("北京"), 结果注入 messages, 再次调 LLM]
  │
SSE: {"type": "token", "content": "北京"}
SSE: {"type": "token", "content": "今天"}
SSE: {"type": "token", "content": "晴"}
...
SSE: {"type": "done", ...}

处理多轮 tool call 的后端代码：

async def stream_with_tools(self, request: ChatRequest,
                             tools: List[Dict],
                             executor,  # async func(name, args) -> dict
                             cancel_event=None) -> AsyncGenerator[str, None]:
    # 自动循环: LLM -> tool call -> 执行工具 -> 注入结果 -> LLM

    messages = [{"role": m.role, "content": m.content}
                for m in request.messages]
    max_rounds = 5

    for _ in range(max_rounds):
        tc_collected: List[Dict] = []  # 本轮收集的 tool calls
        tc_current: Dict = {}

        stream = await self.client.chat.completions.create(
            model=request.model, messages=messages,
            tools=tools, stream=True,
        )

        async for chunk in stream:
            if cancel_event and cancel_event.is_set():
                yield SSE.error("Cancelled")
                return

            delta = chunk.choices[0].delta if chunk.choices else None
            if not delta:
                continue

            if delta.content:
                yield SSE.token(delta.content)

            if delta.tool_calls:
                for tc in delta.tool_calls:
                    idx = tc.index
                    while len(tc_collected) <= idx:
                        tc_collected.append({
                            "id": "", "type": "function",
                            "function": {"name": "", "arguments": ""}
                        })
                    if tc.id:
                        tc_collected[idx]["id"] = tc.id
                    if tc.function:
                        if tc.function.name:
                            tc_collected[idx]["function"]["name"] = tc.function.name
                        if tc.function.arguments:
                            tc_collected[idx]["function"]["arguments"] += tc.function.arguments
                            yield SSE.tool_call_delta(tc.id or str(idx), tc.function.arguments)

        # 无 tool call, 结束
        if not tc_collected:
            break

        # 注入 assistant message (含 tool_calls)
        messages.append({
            "role": "assistant", "content": None,
            "tool_calls": tc_collected,
        })

        # 执行工具并注入结果
        for tc in tc_collected:
            try:
                args = json.loads(tc["function"]["arguments"])
                result = await executor(tc["function"]["name"], args)
            except Exception as e:
                result = {"error": str(e)}

            messages.append({
                "role": "tool",
                "tool_call_id": tc["id"],
                "content": json.dumps(result, ensure_ascii=False),
            })

    yield SSE.done({}, 0)

---

五、前端消费

5.1 fetch + ReadableStream（推荐）

// stream-client.js - 流式聊天前端客户端
// 用法:
//   const stream = streamChat(messages, { onToken, onDone, onError });
//   stream.abort();  // 取消

async function streamChat(messages, callbacks) {
    const { onToken, onThinking, onToolCallStart,
            onToolCallDelta, onToolCallEnd, onDone, onError } = callbacks;

    const controller = new AbortController();
    const startTime = performance.now();

    try {
        const response = await fetch("http://localhost:8000/chat/stream", {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({ messages, model: "gpt-4o-mini" }),
            signal: controller.signal,
        });

        if (!response.ok) {
            throw new Error("HTTP " + response.status);
        }

        const reader = response.body.getReader();
        const decoder = new TextDecoder();
        let buffer = "";

        while (true) {
            const { done, value } = await reader.read();
            if (done) break;

            buffer += decoder.decode(value, { stream: true });
            const lines = buffer.split("\n");
            buffer = lines.pop() || "";  // 不完整的行留在 buffer

            for (const line of lines) {
                if (!line.startsWith("data: ")) continue;

                try {
                    const data = JSON.parse(line.slice(6));

                    switch (data.type) {
                        case "token":
                            onToken && onToken(data.content);
                            break;
                        case "thinking":
                            onThinking && onThinking(data.content);
                            break;
                        case "tool_call_start":
                            onToolCallStart && onToolCallStart(data.name, data.call_id);
                            break;
                        case "tool_call_delta":
                            onToolCallDelta && onToolCallDelta(data.call_id, data.arguments);
                            break;
                        case "tool_call_end":
                            onToolCallEnd && onToolCallEnd(data.call_id);
                            break;
                        case "done":
                            const elapsed = ((performance.now() - startTime) / 1000).toFixed(2);
                            onDone && onDone({ ...data, elapsed_seconds: elapsed });
                            return;
                        case "error":
                            onError && onError(data);
                            return;
                    }
                } catch (e) {
                    // 忽略 JSON 解析错误
                }
            }
        }
    } catch (err) {
        if (err.name === "AbortError") {
            onError && onError({ type: "error", message: "Cancelled", code: "cancelled" });
        } else {
            onError && onError({ type: "error", message: err.message, code: "network" });
        }
    }

    return { abort: () => controller.abort() };
}


// --- 使用示例 ---
const active = streamChat(
    [{ role: "user", content: "Explain quantum computing" }],
    {
        onToken: (t) => {
            document.getElementById("output").textContent += t;
        },
        onDone: (r) => console.log("Done!", r.usage),
        onError: (e) => console.error("Error:", e),
    }
);
// active.abort(); // 取消

5.2 React Hook 封装

// useStreamingChat.ts
import { useState, useRef, useCallback } from "react";

interface Message {
    role: "user" | "assistant";
    content: string;
}

export function useStreamingChat() {
    const [messages, setMessages] = useState<Message[]>([]);
    const [isStreaming, setIsStreaming] = useState(false);
    const abortRef = useRef<{ abort: () => void } | null>(null);

    const sendMessage = useCallback(async (content: string) => {
        const userMsg: Message = { role: "user", content };
        const assistantMsg: Message = { role: "assistant", content: "" };

        setMessages(prev => [...prev, userMsg, assistantMsg]);
        setIsStreaming(true);

        abortRef.current = await streamChat(
            [...messages, userMsg].map(m => ({
                role: m.role, content: m.content
            })),
            {
                onToken: (token) => {
                    setMessages(prev => {
                        const updated = [...prev];
                        const last = updated[updated.length - 1];
                        if (last.role === "assistant") {
                            last.content += token;
                        }
                        return [...updated];
                    });
                },
                onDone: () => setIsStreaming(false),
                onError: (err) => {
                    setIsStreaming(false);
                    console.error(err);
                },
            }
        );
    }, [messages]);

    const cancelStream = useCallback(() => {
        abortRef.current?.abort();
        setIsStreaming(false);
    }, []);

    return { messages, isStreaming, sendMessage, cancelStream };
}

---

六、Nginx 反代配置

# 关键: 禁用缓冲, 否则 SSE 变成"一次性输出"
location /chat/stream {
    proxy_pass http://backend:8000;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_buffering off;           # 核心: 关闭缓冲
    proxy_cache off;
    chunked_transfer_encoding on;
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
}

---

七、性能优化清单

优化项	方案	效果
Nginx 缓冲	proxy_buffering off	避免缓存 SSE 数据
连接池	HTTP/2 多路复用	减少连接开销
前排背压	buffer > 阈值暂停 read	避免浏览器 OOM
心跳保活	每 15s : heartbeat	防代理超时断连
DOM 批量渲染	requestAnimationFrame 批量更新	减少回流

---

八、总结

流式输出全链路：

1. 后端 — FastAPI StreamingResponse + SSE + AsyncOpenAI(stream=True)
2. 前端 — fetch + ReadableStream（比 EventSource 更灵活）
3. Tool call 流 — 增量解析 + 执行 + 注入 → 再流式输出
4. 取消 — AbortController(前端) + asyncio.Event(后端)
5. 部署 — Nginx proxy_buffering off