概述

AI Agent 流式输出解决大模型"思考时间"带来的用户等待问题。SSE 适合单向 Token 流；WebSocket 适合双向交互式 Agent。

前置条件

pip install langchain-openai fastapi uvicorn

核心内容

SSE 实现（FastAPI + LangChain）

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI

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

@app.get("/stream")
async def stream_response(query: str):
    async def generate():
        async for chunk in llm.astream(query):
            token = chunk.content
            if token:
                yield f"data: {token}\n\n"  # SSE 格式
        yield "data: [DONE]\n\n"
    return StreamingResponse(generate(), media_type="text/event-stream",
                             headers={"Cache-Control": "no-cache"})

WebSocket 实现（双向交互）

from fastapi import WebSocket
from langchain_openai import ChatOpenAI

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

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    while True:
        query = await ws.receive_text()
        async for chunk in llm.astream(query):
            if chunk.content:
                await ws.send_text(chunk.content)
        await ws.send_text("[DONE]")

LangGraph Agent 流式输出

from langgraph.graph import StateGraph, END

# astream_events 获取完整事件流
async for event in app.astream_events(inputs, version="v2"):
    kind = event["event"]
    if kind == "on_chat_model_stream":
        chunk = event["data"]["chunk"]
        if chunk.content:
            print(chunk.content, end="", flush=True)
    elif kind == "on_tool_start":
        print(f"\n[调用工具: {event['name']}]")
    elif kind == "on_tool_end":
        print(f"[工具结束: {event['data']['output'][:50]}]")

协议对比

特性	SSE	WebSocket
方向	服务端→客户端	双向
协议	HTTP/1.1	WS
重连	自动	需手动
代理兼容	好	可能被拦截
适用场景	Token 流输出	多轮交互 Agent

选型建议

• SSE：聊天机器人 Token 流、内容生成流——简单可靠
• WebSocket：交互式 Agent（用户可中断/追问）、实时协作
• 长连接 Durable Sessions（2026 新趋势）：支持 Agent 暂停恢复

参考资料

• AI Token Streaming: From SSE to Durable Sessions
• LangChain WebSocket Streaming Tutorial
• LangChain.js 流式输出与 SSE 深度解析
• LangGraph astream_events 官方文档