🎁个人主页：我滴老baby
🎉欢迎大家点赞👍评论📝收藏⭐文章
🔍系列专栏：AI

等3秒才出结果 vs 逐字蹦出来！流式输出让Agent体验飙升，SSE与WebSocket全攻略

用户等3秒看到完整回答 vs 逐字流式输出，体验天差地别。流式响应是Agent产品的标配。本文从OpenAI流式调用到FastAPI SSE推送再到前端集成，手把手实现完整的流式输出方案。

---

一、流式输出原理

流式输出是提升Agent用户体验最直接有效的手段。传统同步模式下，用户发出请求后需要等待LLM生成完整的回答才能看到任何内容——这个等待时间可能是3-5秒甚至更长。而流式模式下，LLM每生成一个Token就立即推送给前端，用户几乎在发出请求的同时就能看到第一个字，体验提升是质变级的。

1.1 同步 vs 流式对比

下面这张表对比了三种输出方式的差异。对于大多数Agent产品来说，"流式"是标配——它不仅改善了用户体验，还让工具执行的进度可以实时展示：

方式	用户感知	首字节时间	适用场景
同步	等3-5秒才看到	3-5秒	批处理
流式	逐字出现	<100ms	实时对话
Server-Sent	实时推送	<100ms	工具执行

1.2 流式架构

流式输出的架构非常简洁——LLM API开启stream模式后，逐Token返回数据，后端通过SSE或WebSocket推送到前端，前端逐字渲染：

LLM API (stream=True) → SSE/WebSocket → 前端逐字渲染

---

二、Python流式实现

2.1 OpenAI流式调用

下面这段代码展示了最基础的OpenAI流式调用。关键是在创建请求时设置stream=True，然后遍历返回的stream对象，每个chunk包含一个或几个Token。注意要检查delta.content是否存在——流式响应中的某些chunk可能不包含文本内容（如role标记chunk）：

# streaming/openai_stream.py
from openai import OpenAI

client = OpenAI(api_key="your-key")

def stream_chat(prompt: str):
    """流式对话"""
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        stream=True  # 启用流式
    )

    full_response = ""
    for chunk in stream:
        if chunk.choices[0].delta.content:
            token = chunk.choices[0].delta.content
            full_response += token
            print(token, end="", flush=True)

    print()
    return full_response

# 使用
stream_chat("用Python写一个快速排序算法")

2.2 FastAPI SSE推送

在生产环境中，你需要将流式输出通过HTTP推送到前端。SSE（Server-Sent Events）是最常用的方案。下面这段代码使用FastAPI的StreamingResponse实现了SSE推送——它将LLM的每个Token封装为JSON格式的SSE事件，前端通过EventSource API接收：

# streaming/api_stream.py
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI
import json

app = FastAPI()
client = OpenAI()

@app.get("/chat/stream")
async def stream_chat(query: str):
    """SSE流式接口"""
    def generate():
        stream = client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": query}],
            stream=True
        )

        for chunk in stream:
            if chunk.choices[0].delta.content:
                data = json.dumps({
                    "token": chunk.choices[0].delta.content
                })
                yield f"data: {data}\n\n"

        yield "data: [DONE]\n\n"

    return StreamingResponse(
        generate(),
        media_type="text/event-stream"
    )

2.3 LangChain流式

如果你使用LangChain构建Agent，流式输出也很简单。下面这段代码展示了LangChain的流式调用方式，与原生OpenAI SDK类似：

# streaming/langchain_stream.py
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", streaming=True)

for chunk in llm.stream("解释什么是AI Agent"):
    print(chunk.content, end="", flush=True)

2.4 工具调用流式

当Agent在流式输出过程中需要调用工具时，情况会更复杂——你需要同时处理文本输出和工具调用两种事件。下面这段代码实现了一个带工具调用的流式处理器，它会实时推送文本Token、工具开始执行、工具执行结果三种类型的事件：

# streaming/tool_stream.py

async def stream_with_tools(query: str):
    """带工具调用的流式输出"""
    messages = [{"role": "user", "content": query}]

    # 流式调用LLM
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=[...],
        stream=True
    )

    tool_calls = {}
    current_content = ""

    for chunk in stream:
        delta = chunk.choices[0].delta

        # 文本输出
        if delta.content:
            yield {"type": "text", "content": delta.content}

        # 工具调用
        if delta.tool_calls:
            for tc in delta.tool_calls:
                idx = tc.index
                if idx not in tool_calls:
                    tool_calls[idx] = {
                        "name": "", "arguments": ""
                    }
                if tc.function.name:
                    tool_calls[idx]["name"] = tc.function.name
                    yield {"type": "tool_start",
                           "tool": tc.function.name}
                if tc.function.arguments:
                    tool_calls[idx]["arguments"] += tc.function.arguments

    # 执行收集到的工具调用
    for idx, tc in tool_calls.items():
        yield {"type": "tool_executing", "tool": tc["name"]}
        result = execute_tool(tc["name"], tc["arguments"])
        yield {"type": "tool_result", "tool": tc["name"],
               "result": result}

---

三、前端集成

3.1 Gradio流式

Gradio是对Agent开发者最友好的流式UI方案。下面这段代码只需几行就能实现一个流式对话界面，yield response的写法让Gradio自动处理流式渲染：

# streaming/gradio_stream.py
import gradio as gr
from openai import OpenAI

client = OpenAI()

def stream_chat(message, history):
    messages = [{"role": "system", "content": "你是AI助手"}]
    for user, assistant in history:
        messages.append({"role": "user", "content": user})
        messages.append({"role": "assistant", "content": assistant})
    messages.append({"role": "user", "content": message})

    stream = client.chat.completions.create(
        model="gpt-4o", messages=messages, stream=True
    )

    response = ""
    for chunk in stream:
        if chunk.choices[0].delta.content:
            response += chunk.choices[0].delta.content
            yield response

demo = gr.ChatInterface(stream_chat, title="流式对话Agent")
demo.launch()

3.2 前端JavaScript消费SSE

如果你使用自定义前端，JavaScript的EventSource API可以直接消费SSE流。下面这段代码展示了最基本的前端SSE消费逻辑：

const eventSource = new EventSource("/chat/stream?query=你好");

eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    if (data.token) {
        document.getElementById("output").textContent += data.token;
    }
};

eventSource.onerror = () => eventSource.close();

---

四、流式性能优化

4.1 优化策略

流式输出的核心体验指标是"首字节延迟"——用户从发出请求到看到第一个字的时间。下面这张表列出了四种优化策略及其效果。其中"Token批处理"是最实用的——每次累积几个Token再推送，可以减少网络开销而不影响用户体验：

策略	说明	效果
减少首字节延迟	使用小模型先回复	体验提升明显
Token批处理	累积几个token再推送	减少网络开销
缓存前缀	相同前缀不重复计算	减少计算量
连接复用	保持长连接	减少握手

4.2 各框架流式支持

不同的技术栈对流式输出的支持程度不同。下面这张表对比了四种常用框架的流式能力。如果你需要快速搭建流式Agent，Gradio是最省事的选择；如果需要更精细的控制，FastAPI+SSE是最佳方案：

框架	流式API	工具流式	SSE	WebSocket
OpenAI	✅	✅	需自建	需自建
LangChain	✅	部分	需自建	需自建
FastAPI	需封装	需封装	✅	✅
Gradio	✅	✅	✅	✅

---

总结

流式输出是Agent产品的用户体验标配，没有流式的Agent就像没有动画的App——功能虽然一样，但体验差了几个档次。以下是本文的核心要点：

1. SSE是最常用的流式推送方式——基于HTTP协议，实现简单，兼容性好，适合大多数Agent场景。

2. 工具调用也可以流式——Agent在调用工具时，可以实时向用户展示"正在搜索…"、"正在计算…"等进度信息，而不是让用户在空白屏幕前等待。

3. FastAPI + Gradio是最简单的流式方案——FastAPI处理后端SSE推送，Gradio处理前端渲染，几行代码就能实现完整的流式体验。

4. 首字节延迟是核心体验指标——尽量控制在100ms以内。如果LLM响应慢，可以考虑先用小模型生成一个"正在思考…"的开头。

下一篇预告：《用Python构建自动化测试智能体》