流式响应

为什么需要流式响应？ LLM生成一个完整回答可能需要5-30秒。如果等待完整响应再返回，用户体验极差。流式响应（Streaming）让用户看到"逐字打印"的效果，大幅降低感知延迟。 SSE（Server-Sent Events）方案 服务端实现 from fastapi import FastAPI from fastapi.responses import StreamingResponse import json app = FastAPI() @app.post("/chat/stream") async def chat_stream(request: dict): async def event_stream(): # 调用LLM的流式接口 async for chunk in llm.astream( messages=[{"role": "user", "content": request["message"]}] ): data = json.dumps({ "content": chunk.content, "role": "assistant" }, ensure_ascii=False) yield f"data: {data}\n\n" # 发送结束标记 yield f"data: [DONE]\n\n" return StreamingResponse( event_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Nginx不缓冲 } ) 客户端实现 // 浏览器端SSE const eventSource = new EventSource('/chat/stream'); eventSource.onmessage = (event) => { if (event.data === '[DONE]') { eventSource.close(); return; } const data = JSON.parse(event.data); document.getElementById('output').innerHTML += data.content; }; eventSource.onerror = (error) => { console.error('SSE error:', error); eventSource.close(); }; Python客户端 import httpx import json async def stream_chat(url, message): async with httpx.AsyncClient() as client: async with client.stream( "POST", f"{url}/chat/stream", json={"message": message}, timeout=60.0 ) as response: async for line in response.aiter_lines(): if line.startswith("data: "): data = line[6:] if data == "[DONE]": break chunk = json.loads(data) print(chunk["content"], end="", flush=True) WebSocket方案 服务端 from fastapi import FastAPI, WebSocket app = FastAPI() @app.websocket("/ws/chat") async def websocket_chat(ws: WebSocket): await ws.accept() try: while True: message = await ws.receive_text() # 流式生成 async for chunk in llm.astream( messages=[{"role": "user", "content": message}] ): await ws.send_json({ "type": "token", "content": chunk.content }) await ws.send_json({"type": "done"}) except WebSocketDisconnect: print("Client disconnected") 客户端 const ws = new WebSocket('ws://localhost:8000/ws/chat'); ws.onopen = () => { ws.send(JSON.stringify({message: '你好'})); }; ws.onmessage = (event) => { const data = JSON.parse(event.data); if (data.type === 'token') { output.innerHTML += data.content; } else if (data.type === 'done') { console.log('Generation complete'); } }; 高级特性 打字机效果 async def typewriter_stream(text, delay=0.03): """模拟打字机效果的流式输出""" for char in text: yield char await asyncio.sleep(delay) 流式中断 class CancellableStream: def __init__(self): self.cancelled = False def cancel(self): self.cancelled = True async def stream(self, llm, messages): async for chunk in llm.astream(messages): if self.cancelled: break yield chunk 并行流式响应 async def parallel_stream(prompts, llm): """同时流式生成多个响应""" async def stream_one(prompt, index): result = [] async for chunk in llm.astream([{"role": "user", "content": prompt}]): result.append({"index": index, "content": chunk.content}) return result tasks = [stream_one(p, i) for i, p in enumerate(prompts)] results = await asyncio.gather(*tasks) # 交错输出 for i in range(max(len(r) for r in results)): for result in results: if i < len(result): yield result[i] 性能优化 缓冲控制 # Nginx配置：禁用缓冲 location /chat/stream { proxy_pass http://backend; proxy_buffering off; # 关闭代理缓冲 proxy_cache off; # 关闭缓存 chunked_transfer_encoding on; proxy_read_timeout 300s; } Token级vs字符级 # Token级流式（推荐）：每个token一个chunk async for token in llm.astream(messages): yield token # 字符级流式：将token拆分为字符 async for token in llm.astream(messages): for char in token.content: yield char await asyncio.sleep(0.01) # 添加微小延迟 结语 流式响应是LLM应用的标配功能。SSE适合单向流式输出，WebSocket适合需要双向交互的场景。关键点：禁用所有层级的缓冲、正确处理中断、保持连接稳定性。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 Agent 的推理过程往往是漫长的等待——用户盯着加载动画，不知道 Agent 在做什么。流式响应把"等待结果"变成"实时观察思考"，是 Agent 用户体验的关键升级。2026年，三种流式协议各有优劣，选型不当会导致体验降级或工程复杂度爆炸。 一、三种协议对比 核心特性矩阵 特性 SSE WebSocket gRPC Stream 通信方向 服务器→客户端（单向） 双向 双向 底层协议 HTTP/1.1 或 HTTP/2 HTTP 升级 HTTP/2 数据格式 文本（text/event-stream） 文本/二进制 Protobuf（二进制） 自动重连 内置 需手动实现 需手动实现 浏览器支持 原生 EventSource 原生 WebSocket 需 gRPC-Web 代理/CDN兼容 优秀 良好 较差 连接数限制 浏览器6个/域名 无限制 无限制 序列化效率 低（文本） 中 高（Protobuf） 移动端友好 高 中 低 Agent 场景适配分析 Agent 流式需求频谱： 单向输出流 双向交互流 （LLM→用户） （用户↔Agent） │ │ │ ┌───────────┐ │ │ │ SSE 最佳 │ │ │ └───────────┘ │ │ │ │ ┌───────────┐ │ │ │ WebSocket │ │ │ │ 最佳 │ │ │ └───────────┘ │ │ │ │ ┌───────────┐ │ │ │ gRPC 最佳 │ │ │ └───────────┘ │ 简单聊天 ←─────────────→ 复杂多Agent 低延迟 ←─────────────→ 高吞吐 Web前端 ←─────────────→ 微服务后端 二、SSE 实现方案 2.1 服务端 from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio import json app = FastAPI() class AgentStreamEvent: """Agent 流式事件类型""" THINKING = "thinking" # Agent 思考中 TOOL_CALL = "tool_call" # 工具调用 TOOL_RESULT = "tool_result" # 工具结果 CONTENT = "content" # 内容输出 ERROR = "error" # 错误 DONE = "done" # 完成 async def agent_stream_generator( query: str, session_id: str ): """Agent SSE 流式生成器""" try: # 1. 发送思考状态 yield _format_sse(AgentStreamEvent.THINKING, { "message": "正在分析您的请求...", "session_id": session_id }) # 2. Agent 推理（流式 LLM 输出） async for chunk in agent.think_stream(query): if chunk.type == "tool_call": yield _format_sse(AgentStreamEvent.TOOL_CALL, { "tool": chunk.tool_name, "args": chunk.tool_args, "thinking": chunk.reasoning }) # 3. 工具执行 result = await agent.execute_tool(chunk.tool_call) yield _format_sse(AgentStreamEvent.TOOL_RESULT, { "tool": chunk.tool_name, "result": result.summary, "duration_ms": result.duration_ms }) elif chunk.type == "content": yield _format_sse(AgentStreamEvent.CONTENT, { "text": chunk.text, "tokens_so_far": chunk.token_count }) # 4. 完成 yield _format_sse(AgentStreamEvent.DONE, { "session_id": session_id, "total_tokens": agent.total_tokens, "duration_ms": agent.total_duration_ms }) except Exception as e: yield _format_sse(AgentStreamEvent.ERROR, { "message": str(e), "session_id": session_id }) def _format_sse(event_type: str, data: dict) -> str: return f"event: {event_type}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n" @app.post("/api/agent/chat") async def chat(request: ChatRequest): return StreamingResponse( agent_stream_generator(request.query, request.session_id), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no", # Nginx 禁用缓冲 } ) 2.2 客户端 class AgentSSEClient { private eventSource: EventSource | null = null; private reconnectAttempts = 0; private maxReconnects = 3; connect(query: string, sessionId: string) { // 使用 fetch POST + ReadableStream（EventSource 仅支持 GET） fetch('/api/agent/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ query, sessionId }), }).then(response => { const reader = response.body!.getReader(); const decoder = new TextDecoder(); let buffer = ''; const read = () => { reader.read().then(({ done, value }) => { if (done) return; buffer += decoder.decode(value, { stream: true }); const events = buffer.split('\n\n'); buffer = events.pop() || ''; events.forEach(raw => this.handleEvent(raw)); read(); }); }; read(); }); } private handleEvent(raw: string) { const lines = raw.split('\n'); let event = 'message'; let data = ''; lines.forEach(line => { if (line.startsWith('event: ')) event = line.slice(7); if (line.startsWith('data: ')) data = line.slice(6); }); const parsed = JSON.parse(data); switch (event) { case 'thinking': this.onThinking?.(parsed); break; case 'tool_call': this.onToolCall?.(parsed); break; case 'tool_result': this.onToolResult?.(parsed); break; case 'content': this.onContent?.(parsed.text); break; case 'error': this.onError?.(parsed); break; case 'done': this.onDone?.(parsed); break; } } } 三、WebSocket 实现方案 3.1 服务端 from fastapi import WebSocket, WebSocketDisconnect class AgentConnectionManager: """Agent WebSocket 连接管理""" def __init__(self): self.active: dict[str, WebSocket] = {} # session_id → WebSocket self.agent_tasks: dict[str, asyncio.Task] = {} async def connect(self, ws: WebSocket, session_id: str): await ws.accept() self.active[session_id] = ws logger.info(f"WebSocket connected: {session_id}") async def disconnect(self, session_id: str): if session_id in self.active: del self.active[session_id] if session_id in self.agent_tasks: self.agent_tasks[session_id].cancel() del self.agent_tasks[session_id] async def handle_session(self, ws: WebSocket, session_id: str): """处理 WebSocket 会话""" await self.connect(ws, session_id) try: while True: # 接收客户端消息 message = await ws.receive_json() if message["type"] == "chat": # 启动 Agent 任务 task = asyncio.create_task( self._run_agent(ws, session_id, message["content"]) ) self.agent_tasks[session_id] = task elif message["type"] == "interrupt": # 用户中断当前 Agent 执行 if session_id in self.agent_tasks: self.agent_tasks[session_id].cancel() await ws.send_json({ "type": "interrupted", "session_id": session_id }) elif message["type"] == "feedback": # 用户实时反馈（Human-in-the-loop） await self._handle_feedback(session_id, message) except WebSocketDisconnect: await self.disconnect(session_id) async def _run_agent(self, ws: WebSocket, session_id: str, query: str): """运行 Agent 并通过 WebSocket 推送更新""" try: async for event in agent.run_stream(query): await ws.send_json({ "type": event.type, "data": event.data, "timestamp": time.time() }) except asyncio.CancelledError: logger.info(f"Agent task cancelled: {session_id}") except Exception as e: await ws.send_json({ "type": "error", "data": {"message": str(e)} }) manager = AgentConnectionManager() @app.websocket("/ws/agent/{session_id}") async def websocket_endpoint(ws: WebSocket, session_id: str): await manager.handle_session(ws, session_id) 3.2 客户端 class AgentWSClient { private ws: WebSocket | null = null; private messageQueue: string[] = []; connect(sessionId: string) { this.ws = new WebSocket(`wss://api.example.com/ws/agent/${sessionId}`); this.ws.onopen = () => { // 发送排队消息 this.messageQueue.forEach(msg => this.ws!.send(msg)); this.messageQueue = []; }; this.ws.onmessage = (event) => { const data = JSON.parse(event.data); this.handleMessage(data); }; this.ws.onclose = () => { // 自动重连 setTimeout(() => this.connect(sessionId), 3000); }; } sendChat(content: string) { const msg = JSON.stringify({ type: 'chat', content }); if (this.ws?.readyState === WebSocket.OPEN) { this.ws.send(msg); } else { this.messageQueue.push(msg); } } interrupt() { this.ws?.send(JSON.stringify({ type: 'interrupt' })); } } 四、gRPC 流式方案 4.1 Proto 定义 // agent.proto syntax = "proto3"; service AgentService { // 服务端流式：Agent → 客户端 rpc ChatStream(ChatRequest) returns (stream ChatResponse); // 双向流式：支持实时交互 rpc ChatBidirectional(stream ChatMessage) returns (stream ChatResponse); } message ChatRequest { string session_id = 1; string query = 2; map<string, string> metadata = 3; } message ChatMessage { string session_id = 1; MessageType type = 2; // CHAT, INTERRUPT, FEEDBACK string content = 3; } message ChatResponse { ResponseType type = 1; // THINKING, TOOL_CALL, CONTENT, DONE, ERROR string session_id = 2; bytes data = 3; // JSON 编码的事件数据 int64 timestamp = 4; } enum MessageType { CHAT = 0; INTERRUPT = 1; FEEDBACK = 2; } enum ResponseType { THINKING = 0; TOOL_CALL = 1; TOOL_RESULT = 2; CONTENT = 3; DONE = 4; ERROR = 5; } 4.2 服务端实现 import grpc from concurrent import futures class AgentServicer(agent_pb2_grpc.AgentServiceServicer): def ChatStream(self, request, context): """服务端流式：逐条推送 Agent 事件""" try: for event in agent.run(request.query): response = agent_pb2.ChatResponse( type=self._map_event_type(event.type), session_id=request.session_id, data=json.dumps(event.data).encode(), timestamp=int(time.time()) ) yield response except Exception as e: yield agent_pb2.ChatResponse( type=agent_pb2.ERROR, data=json.dumps({"error": str(e)}).encode() ) def ChatBidirectional(self, request_iterator, context): """双向流式：支持中断和实时反馈""" session = None for message in request_iterator: if message.type == agent_pb2.CHAT: # 启动 Agent 执行 for event in agent.run(message.content): if not context.is_active(): break yield agent_pb2.ChatResponse( type=self._map_event_type(event.type), data=json.dumps(event.data).encode() ) elif message.type == agent_pb2.INTERRUPT: agent.interrupt() yield agent_pb2.ChatResponse( type=agent_pb2.DONE, data=json.dumps({"reason": "interrupted"}).encode() ) 五、性能基准测试 测试环境 服务端：4 vCPU / 16GB RAM / Python 3.12 / FastAPI 客户端：1000 并发连接 消息大小：平均 200 bytes / 消息 持续时间：5 分钟 结果对比 指标 SSE WebSocket gRPC Stream 最大并发连接 10,000 50,000 30,000 消息延迟 P50 12ms 8ms 5ms 消息延迟 P95 45ms 25ms 12ms 消息延迟 P99 120ms 60ms 30ms 吞吐量 (msg/s) 50,000 200,000 150,000 内存/连接 32KB 48KB 24KB CPU 利用率 (1k连接) 35% 28% 22% 带宽效率 基准 +15% +40% 六、选型决策树 ┌─────────────────┐ │ 是否需要双向通信？│ └────────┬────────┘ │ ┌──────────────┴──────────────┐ │ 否 │ 是 ▼ ▼ ┌────────────────┐ ┌──────────────────┐ │ 是否是微服务 │ │ 是否是浏览器前端？│ │ 内部通信？ │ └────────┬─────────┘ └───────┬────────┘ │ │ ┌───────┴───────┐ ┌───────┴───────┐ │ 否 │ 是 │ 是 │ 否 ▼ ▼ ▼ ▼ ┌──────────┐ ┌────────────┐ ┌──────┐ ┌────────┐ │ gRPC │ │ WebSocket │ │gRPC │ │ SSE │ │ Stream │ │ │ └──────┘ └────────┘ └──────────┘ └────────────┘ 场景推荐 场景 推荐协议 原因 Web 聊天界面 SSE 原生支持、简单、自动重连 实时协作编辑 WebSocket 双向低延迟 移动 App SSE 移动网络友好、自动重连 微服务间 Agent 通信 gRPC 高效序列化、强类型 多 Agent 系统 gRPC 流式 RPC 适配 Agent 通信 简单 LLM 问答 SSE 单向流足够、实现简单 Human-in-the-loop WebSocket 需要双向交互 大规模推送 SSE CDN 兼容、连接效率高 七、生产环境关键配置 Nginx SSE 配置 location /api/agent/chat { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_buffering off; # 关键：禁用缓冲 proxy_cache off; proxy_read_timeout 300s; # 长连接超时 chunked_transfer_encoding on; } WebSocket 配置 location /ws/agent/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; proxy_read_timeout 86400; # 24小时 proxy_send_timeout 86400; } 八、流式架构 Checklist □ 协议选型基于通信方向和客户端类型 □ SSE 禁用代理缓冲（proxy_buffering off） □ WebSocket 实现心跳和重连机制 □ gRPC 配置 keepalive 和流控 □ 消息序列化使用高效格式（JSON/Protobuf） □ 背压机制防止慢客户端拖垮服务端 □ 连接超时和最大连接数限制 □ 流式错误不中断连接，通过事件传递 □ 客户端实现优雅降级（流式不可用时回退轮询） □ 监控流式连接的延迟和消息丢失率 结语 流式响应是 Agent 从"工具"到"伙伴"的关键体验升级。协议选型没有银弹：SSE 简单可靠适合 Web 场景，WebSocket 灵活双向适合交互场景，gRPC 高效强类型适合微服务场景。理解你的通信模式——是单向输出还是双向交互——然后选择最适合的工具。在 Agent 时代，好的流式架构让用户感觉 Agent 在"思考"，而不是在"卡住"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 用户在发出一个深层推理问题后，等待智能体输出的那几秒钟，往往决定了产品的第一印象。2025 年以来，几乎所有主流的 AGI 智能体产品都标配了"打字机效果"——字符逐字出现，推理过程可见。但这背后的流式响应架构，远比表面上看起来复杂。 智能体流式响应不只是一个 UI 层面的锦上添花，它直接决定了系统的感知延迟、错误恢复能力和可扩展性。本文将围绕 AGI 智能体流式响应架构的设计展开系统讨论。 一、流式响应的核心技术路径 1.1 SSE：轻量级的首选 Server-Sent Events（SSE）是目前智能体流式响应中最广泛使用的传输协议。与 WebSocket 不同，SSE 天然运行在 HTTP 之上，浏览器原生支持 EventSource API，无需复杂的握手和心跳维护。 在典型架构中，LLM 推理引擎以 token 级别粒度产生输出，经由反代层封装为 SSE 事件流，逐帧推送到前端渲染缓冲区。SSE 的优势在于： 部署成本极低：直接复用现有的 HTTP 负载均衡和鉴权中间件 自动重连：浏览器内置断线重连能力 服务端推送天然单向：符合"模型输出→用户"的单一数据流向 但在多智能体协作（Multi-Agent）场景中，SSE 的单向特性成为瓶颈——当我们需要聚合多个子智能体并行产出的流时，SSE 的连接管理复杂度迅速上升。 1.2 WebSocket：双向互动的基石 WebSocket 在需要双向并行数据流的场景中不可替代。典型的例子是 Agent+Tool 的迭代式推理： 用户提问 → Agent 分析 → 调用搜索工具 → 等待结果 → 整合输出 在这个过程中，Agent 需要一边接收工具的返回数据，一边逐步向用户透传推理进度。WebSocket 的全双工特性让"一边等待工具响应、一边输出推理日志"成为可能。 我们团队在实践中采用的是一种分层混合架构： 外层 SSE + 内层 WebSocket：前端流量入口统一使用 SSE，隐藏后端的复杂性与容错逻辑；而在 Agent Engine 内部，各组件之间以 WebSocket 进行实时通信。 这既保持了前端接入的简洁性，又为内部高并发消息交换保留了灵活性。 ...

为什么必须流式 AI 生成一个完整回答可能需要 5-30 秒。如果等全部生成完再返回，用户体验极差。流式输出让用户看到"AI 正在打字"，将感知等待时间从 15 秒降到 0.5 秒。 SSE vs WebSocket 维度 SSE WebSocket 协议 HTTP 独立协议 方向 服务端→客户端（单向） 双向 重连 自动 需手动实现 兼容性 所有现代浏览器 所有现代浏览器 复杂度 低 中 适用场景 AI 回复流式输出 实时协作、语音对话 结论：AI 对话场景用 SSE 足够，语音/多模态交互用 WebSocket。 SSE 实现 服务端（Python FastAPI） from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app = FastAPI() @app.post("/api/chat") async def chat(request: dict): async def stream(): # 调用 LLM 流式 API async for chunk in llm.stream( model="gpt-4o", messages=request["messages"] ): # SSE 格式：data: <content>\n\n content = chunk.choices[0].delta.content if content: yield f"data: {json.dumps({'content': content})}\n\n" # 发送结束标记 yield f"data: {json.dumps({'done': True})}\n\n" return StreamingResponse( stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "X-Accel-Buffering": "no", # Nginx 不缓冲 } ) 客户端（JavaScript） class AIChat { async sendMessage(text) { const response = await fetch('/api/chat', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ messages: [...this.history, { role: 'user', content: text }] }) }); 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 }); // 解析 SSE 数据 const lines = buffer.split('\n\n'); buffer = lines.pop(); // 保留不完整的行 for (const line of lines) { if (line.startsWith('data: ')) { const data = JSON.parse(line.slice(6)); if (data.done) { this.onComplete(); } else { this.onToken(data.content); } } } } } onToken(content) { document.getElementById('response').textContent += content; } } 进阶：流式 + 工具调用 当 Agent 需要调用工具时，流式输出变得复杂： ...