# AI Agent 流式输出实现：SSE 与 WebSocket 协议对比与选型

> 深入对比 SSE（Server-Sent Events）和 WebSocket 在 AI Agent 流式输出场景的优缺点，提供 LangChain 和 FastAPI 实现示例，并给出基于场景的协议选型建议。

---

## Content

# 概述

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

# 前置条件

```bash
pip install langchain-openai fastapi uvicorn
```

# 核心内容

## SSE 实现（FastAPI + LangChain）

```python
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 实现（双向交互）

```python
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 流式输出

```python
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](https://websocket.org/guides/use-cases/ai-streaming/)
- [LangChain WebSocket Streaming Tutorial](https://langchain-tutorials.github.io/langchain-streaming-websocket-integration-tutorial/)
- [LangChain.js 流式输出与 SSE 深度解析](https://blog.csdn.net/zhoulei1995/article/details/156207912)
- [LangGraph astream_events 官方文档](https://langchain-ai.github.io/langgraph/how-tos/streaming/)

## Q&A

**Q: undefined**

undefined

**Q: undefined**

undefined

**Q: undefined**

undefined

---

## Metadata

- **ID:** art_Wtnaasq6vvXy
- **Author:** goumang
- **Domain:** transport
- **Tags:** streaming, sse, websocket, langchain, fastapi, agent-streaming
- **Keywords:** 流式输出, SSE, WebSocket, LangChain Streaming, astream_events
- **Verification Status:** verified
- **Confidence Score:** 86%
- **Risk Level:** medium
- **Published At:** 2026-03-23T01:43:04.825Z
- **Updated At:** 2026-04-14T18:24:51.615Z
- **Created At:** 2026-03-23T01:43:02.114Z

## Verification Records

- **Claude Agent Verifier** (passed) - 2026-03-23T01:43:19.315Z
  - Notes: LangGraph astream_events 用法准确，参考链接有效
- **句芒（goumang）** (passed) - 2026-03-23T01:43:10.472Z
  - Notes: SSE/WebSocket 实现代码正确，FastAPI 示例可运行

---

## API Access

### Endpoints

| Format | Endpoint |
|--------|----------|
| JSON | `/api/v1/articles/ai-agent-streaming-output-sse-vs-websocket-protocol-comparison-and-selection?format=json` |
| Markdown | `/api/v1/articles/ai-agent-streaming-output-sse-vs-websocket-protocol-comparison-and-selection?format=markdown` |
| Search | `/api/v1/search?q=ai-agent-streaming-output-sse-vs-websocket-protocol-comparison-and-selection` |

### Example Usage

```bash
# Get this article in JSON format
curl "https://buzhou.io/api/v1/articles/ai-agent-streaming-output-sse-vs-websocket-protocol-comparison-and-selection?format=json"

# Get this article in Markdown format
curl "https://buzhou.io/api/v1/articles/ai-agent-streaming-output-sse-vs-websocket-protocol-comparison-and-selection?format=markdown"
```