企业官网建设流程全解析

前四期下来，你的问答机器人已经能跑了：50行检索引擎接上大模型，加了健康监控，还做了重试降级熔断。但上线后你会发现一个扎心的问题：模型生成答案要十几秒，这段时间用户只能盯着光标转圈。转超过5秒，很多人直接关页面。

解决办法就四个字：流式输出。token生成一个吐一个，像打字机一样往外蹦。用户第一秒就看到字在动，体感等待直接清零。OpenAI系的接口基本都支持，开个stream参数就行。

流式输出走的协议叫SSE，Server-Sent Events。说白了就是一条不断开的HTTP连接，服务器分多次往里写数据。格式很死：Content-Type必须是text/event-stream，每条消息以data:开头，消息之间用空行隔开，最后发一条data: [DONE]表示结束。

环境很简单：Python 3.10以上，装个requests就行。为了这篇能直接跑，我先写一个本地的SSE服务器当"假大模型"，它一次吐一个token，客户端负责接。

importjson,threading,timefromhttp.serverimportBaseHTTPRequestHandler,HTTPServerclassSSEHandler(BaseHTTPRequestHandler):defdo_GET(self):self.send_response(200)self.send_header("Content-Type","text/event-stream")self.send_header("Cache-Control","no-cache")self.end_headers()fortokin["大","模型","正在","思考","……"]:body=json.dumps({"choices":[{"delta":{"content":tok}}]})self.wfile.write(f"data:{body}\n\n".encode())self.wfile.flush()time.sleep(0.05)self.wfile.write(b"data: [DONE]\n\n")deflog_message(self,*a):passsrv=HTTPServer(("127.0.0.1",8931),SSEHandler)threading.Thread(target=srv.serve_forever,daemon=True).start()time.sleep(0.3)

客户端是关键。requests的iter_content按字节读，自己攒buffer，遇到连续两个换行就切出一个完整事件。data: [DONE]出现就收尾，中间每个token直接打印。

importrequestsdefstream_chat(url):full=""withrequests.get(url,stream=True,timeout=10)asr:r.raise_for_status()buf=""forrawinr.iter_content(chunk_size=1,decode_unicode=True):ifnotraw:continuebuf+=rawwhile"\n\n"inbuf:event,buf=buf.split("\n\n",1)forlineinevent.splitlines():ifline.startswith("data:"):payload=line[5:].strip()ifpayload=="[DONE]":returnfull tok=json.loads(payload)["choices"][0]["delta"].get("content","")print(tok,end="",flush=True)full+=tokreturnfull text=stream_chat("http://127.0.0.1:8931/stream")print("\n收到全文：",text)srv.shutdown()

跑一遍，终端里"大模型正在思考……"一个字一个字蹦出来，最后收到全文拼接结果。我本地实测就是这个输出，token顺序和内容都对得上。

换成真实大模型接口，改动很小。把requests.get换成post，URL换成/v1/chat/completions，headers里带上Authorization，body里加"stream": True。解析data行的逻辑一字不用改，因为OpenAI系的流式格式就是上面这种。

最后说几个我真实踩过的坑。第一，别用iter_lines偷懒。网络分包可能把一行data切成两半，iter_lines会给你半行，用"\n\n"切事件才保险。第二，stream=True的连接一定要用with包起来，不然连接泄漏，跑一会儿就报too many open files。第三，经过Nginx这类网关时，流可能被缓冲住，表现为半天没字然后一次性全吐出来，网关上加X-Accel-Buffering: no能解决。第四，前端别傻等full字符串，token一到就渲染，这才是流式的意义。

收藏这篇，下次给机器人加流式输出，照着抄半小时搞定，省掉至少两小时的调试。下期Python实战第6期，讲"上下文窗口爆了怎么办"：token计数、滑动窗口截断、多轮对话怎么压缩，关注我不错过。

评论区聊聊：你用过的大模型应用里，哪家的流式输出体验最丝滑？我是Cyrus。