流式输出

GPT-6 把「智能界面面向所有人」推出来之后，前端工程师的工作量明显变了：不再是渲染静态页面，而是要接住一个一边想一边输出、中途还要调工具的智能体。Vercel AI SDK 就是为这个场景准备的——它在 Next.js 里用几个钩子，把流式文本、工具调用、消息管理都封好了。 两个核心钩子 useChat 是主力：给它一个后端路由，它负责发消息、接收流式 token、逐字渲染到界面。模型输出不是等好了一次性显示，而是像 ChatGPT 那样一个字一个字蹦出来——这个体验过去要自己处理 SSE 流，现在一行钩子搞定。 useToolExecution（或在后端用 streamText 的工具配置）处理「模型要调工具」这一步：前端能知道模型正在执行哪个工具、拿到什么结果、再接着生成。做「智能体查了数据库 → 把结果画成表 → 再写一段解读」这种界面，就靠它。 后端其实很薄 Next.js 的 Route Handler 里，用 streamText({ model, messages, tools }) 把请求转发给任意 OpenAI 兼容接口，返回流。前端只管渲染，业务逻辑在后端。切模型时只改后端一行。 判断 智能界面时代，前端的护城河从「排版」挪到了「如何优雅地展示一个不确定过程」——流式输出、工具进度、错误回退，这些是新产品的基本盘。Vercel AI SDK 把这些脏活收进钩子，让你把精力放在「这个智能体该长什么样」，而不是流解析。GPT-6 给了界面的脑子，这套工具给它配脸。 补一个工程细节：流式输出虽然体验好，但调试变难——你没法一眼看到完整响应。开发期可以临时关掉流式，让它一次性返回，方便对照日志；上线再打开。另外，工具调用中间态要给用户可见的反馈（「正在查询订单…」），别让用户盯着空白界面以为卡住了，这是智能界面最容易被吐槽的细节。 智能界面的体验细节，往往就藏在流式渲染和中间态反馈这些容易被忽略的地方。 去论坛讨论 这篇文章里的观点，欢迎到硅基AGI论坛讨论。按标题搜索：AI前端

为什么需要流式输出 LLM 生成完整回答可能需要 5-15 秒。用户盯着空白屏幕等待体验极差。流式输出让用户像看打字机一样实时看到生成内容，首 token 延迟降到 200-500ms，体感速度提升 10 倍。 SSE vs WebSocket 对比 维度 SSE (Server-Sent Events) WebSocket 协议 HTTP/1.1 或 HTTP/2 独立协议(ws://) 方向 服务器到客户端(单向) 双向 自动重连 浏览器内置 需手动实现 代理兼容性 好(就是HTTP) 差(需特殊配置) 适合场景 LLM 流式输出(推荐) 实时对话/多Agent通信 实现复杂度 低 中 结论：LLM 流式输出首选 SSE，需要双向通信才用 WebSocket。 SSE 后端实现（FastAPI） from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import httpx, json, asyncio app = FastAPI() LLM_API_URL = "https://api.openai.com/v1/chat/completions" @app.post("/api/chat/stream") async def chat_stream(request: Request): body = await request.json() async def event_generator(): headers = { "Authorization": f"Bearer {body['api_key']}", "Content-Type": "application/json" } payload = { "model": body["model"], "messages": body["messages"], "stream": True, "max_tokens": 4096 } try: async with httpx.AsyncClient(timeout=120) as client: async with client.stream("POST", LLM_API_URL, headers=headers, json=payload) as resp: async for line in resp.aiter_lines(): if not line.startswith("data: "): continue data = line[6:] if data.strip() == "[DONE]": yield "data: " + json.dumps({"type": "done"}) + "\n\n" break chunk = json.loads(data) delta = chunk["choices"][0]["delta"] if "content" in delta: token = delta["content"] yield "data: " + json.dumps({"type": "token", "content": token}) + "\n\n" except httpx.ReadTimeout: yield "data: " + json.dumps({"type": "error", "msg": "LLM timeout"}) + "\n\n" except Exception as e: yield "data: " + json.dumps({"type": "error", "msg": str(e)}) + "\n\n" return StreamingResponse( event_generator(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no", "Connection": "keep-alive"} ) SSE 前端实现 class SSEClient { constructor(url) { this.url = url; this.controller = null; } async stream(messages, onToken, onDone, onError) { this.controller = new AbortController(); try { const resp = await fetch(this.url, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ messages }), signal: this.controller.signal }); const reader = resp.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(); for (const line of lines) { if (!line.startsWith('data: ')) continue; const data = JSON.parse(line.slice(6)); if (data.type === 'token') onToken(data.content); else if (data.type === 'done') onDone(); else if (data.type === 'error') onError(data.msg); } } } catch (e) { if (e.name !== 'AbortError') onError(e.message); } } abort() { this.controller?.abort(); } } // 使用 const client = new SSEClient('/api/chat/stream'); let fullText = ''; await client.stream( [{ role: 'user', content: '解释 RAG 架构' }], (token) => { fullText += token; renderMarkdown(fullText); }, () => console.log('完成'), (err) => console.error('错误:', err) ); WebSocket 方案 需要双向通信（用户中途打断、实时修正）时使用： ...