我当时正在为我的团队构建一个内部文档助手。流程大家都懂：一个聊天机器人，从向量数据库中提取关于我们代码库的信息来回答问题，然后发送给大型语言模型。我用 Python 搭建了后端，通过应用程序接口使用了一个不错的模型（感谢 interwestinfo.com 提供可靠的端点），并将所有部分连接起来。很简单，对吧？
接着迎来了第一次真正的测试：有人问了一个需要长篇、深思熟虑才能回答的问题。响应花费了超过 30 秒。用户盯着空白的聊天气泡，不断刷新页面，怀疑应用程序是否崩溃了。这体验可真不怎么样。
我需要随着令牌生成将其流式传输回来，以便用户可以同步阅读。这是经典的“聊天用户界面”模式。但实现它却陷入了一堆半吊子解决方案的泥潭。
我尝试过但无效的方法
1. 轮询
我的第一个想法：发起大型语言模型调用，将部分结果存储在 Redis 中，并让前端每秒轮询一次。这很糟糕。预测端点最终会返回完整响应，所以我需要更改后端以逐个写入令牌。轮询还意味着每条消息大约产生 30 次超文本传输协议请求，感觉很浪费。而且用户界面卡顿——更新是成批出现的，而不是平滑进行的。
2. WebSocket
WebSocket 看起来是显而易见的选择。我编写了一个 FastAPI WebSocket 端点，建立连接，并逐帧流式传输令牌。这确实奏效……除了一个问题：我的部署环境（负载均衡器后面的低预算虚拟专用服务器）具有激进的闲置超时设置。连接会在 60 秒后断开，而重新连接 WebSocket 需要手动处理逻辑。此外，我技术栈中的一半库并不轻易支持 WebSocket——例如，我的身份验证中间件期望的是超文本传输协议请求。
但真正的痛点在于：WebSocket 是双向的。我不需要双向通信。我只需要服务器向客户端推送数据。WebSocket 感觉像是杀鸡用牛刀。
3. 长轮询（坏主意）
是的，我也试过这个。服务器会保持响应打开并刷新数据块。但超文本传输协议/1.1 连接在这方面存在问题，而且我的框架（当时使用的是 Flask）如果不进行猴子补丁就无法优雅地处理它。在经历了两个小时的“连接关闭”错误后，我放弃了。
最终奏效的方法：服务器发送事件
我以前曾使用服务器发送事件来处理实时推文，但从未用于人工智能流式传输。服务器发送事件是一项标准（超文本标记语言 5 的一部分），服务器通过单个长期存在的超文本传输协议连接发送事件流。客户端使用事件源应用程序接口。它是单向的（服务器 → 客户端），这正是我所需要的。
FastAPI 通过流式响应原生支持服务器发送事件。以下是让我的用户体验再次流畅的后端代码：
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio
app = FastAPI()
async def generate_tokens(prompt: str):
# 假设 get_llm_response 是一个生成令牌的异步生成器
# （例如，使用 OpenAI 的流式应用程序接口，参数 `stream=Tr
免责声明：本文内容来自互联网，该文观点不代表本站观点。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容，请到页面底部单击反馈，一经查实，本站将立刻删除。
