Django 接入 AI 大模型实战：从零做一个流式聊天网站

Django 接入 AI 大模型实战：从零做一个流式聊天网站

普通聊天接口要等大模型把整段答案生成完，浏览器才看到内容。模型只要思考十几秒，用户就会怀疑页面卡死。流式聊天解决的不是“模型变快”，而是让已经生成的内容立即可见。

我用 Django 6.0.7 做了一个最小实验，并对照本地 RuyiDjangoCRM 的真实 ASGI 与通知流实现检查关键边界。结论很明确：Django 可以直接承担 AI 流式接口，但生产环境必须使用 ASGI，并认真处理响应格式、网关缓冲、断线清理和阻塞调用。

请求为什么要跑在 ASGI 上

WSGI 擅长一次请求、一次完整响应。流式聊天会长时间保持连接，更适合 ASGI 的异步执行模型。项目入口仍然很简单：

# crm/asgi.pyimportosfromdjango.core.asgiimportget_asgi_application os.environ.setdefault("DJANGO_SETTINGS_MODULE","crm.settings")application=get_asgi_application()

启动时不要再把生产流式服务指向 WSGI：

uvicorn crm.asgi:application--host127.0.0.1--port8000

本地 RuyiDjangoCRM 已经用同一个 ASGI 入口承载长连接通知流，并在响应中关闭 Nginx 缓冲。这比单独写一个“能输出几段文字”的演示更接近真实部署。

用 StreamingHttpResponse 返回事件流

下面先用异步生成器模拟模型逐词输出。接入任意支持流式响应的大模型时，只需要替换model_stream()，Django 这一层保持不变。

importasyncioimportjsonfromdjango.httpimportStreamingHttpResponseasyncdefmodel_stream(question:str):answer=f"收到问题：{question}。这是逐块返回的答案。"fortokeninanswer:awaitasyncio.sleep(0.03)yieldtokenasyncdefchat(request):question=request.GET.get("q","请介绍 Django ASGI")asyncdefevents():asyncfortokeninmodel_stream(question):payload=json.dumps({"type":"delta","content":token},ensure_ascii=False,)yieldf"data:{payload}\n\n"yield'data: {"type":"done"}\n\n'response=StreamingHttpResponse(events(),content_type="text/event-stream; charset=utf-8",)response["Cache-Control"]="no-cache"response["X-Accel-Buffering"]="no"returnresponse

每个事件必须以两个换行结束。不要把模型文本直接拼成未经编码的 JSON，否则引号、换行或反斜杠很容易破坏事件格式。

浏览器逐块更新聊天气泡

如果接口使用 GET，浏览器原生EventSource足够简单：

constoutput=document.querySelector("#answer");constsource=newEventSource(`/api/chat/?q=${encodeURIComponent(question)}`);source.onmessage=(event)=>{constdata=JSON.parse(event.data);if(data.type==="delta")output.textContent+=data.content;if(data.type==="done")source.close();};source.onerror=()=>{source.close();output.textContent+="\n[连接已中断，请重试]";};

聊天问题通常包含较长上下文，实际项目更常用fetch()发 POST，再读取response.body。无论选哪一种，协议都应该显式区分delta、error和done，而不是靠空字符串猜测状态。

四个最容易漏掉的工程边界

第一，不要在异步视图里直接执行阻塞 SDK。如果模型客户端只有同步接口，应改用其异步版本，或通过sync_to_async()/ 线程池隔离阻塞调用。

第二，浏览器断线后要停止上游生成。用户关闭页面时，服务端会收到取消信号。生成器应捕获asyncio.CancelledError，关闭模型流并释放计费连接，然后继续抛出取消异常。

asyncdefevents():try:asyncfortokeninmodel_stream(question):yieldencode_event("delta",token)exceptasyncio.CancelledError:awaitclose_upstream_stream()raise

第三，转发网关不能攒够一批再吐给浏览器。X-Accel-Buffering: no是应用侧信号，Nginx、CDN 和入口网关仍要逐层验证超时与缓冲配置。

第四，异步视图不代表 ORM 查询自动异步。Django 提供aget()、afirst()、aiterator()等异步方法；事务边界复杂时，可以把完整同步数据库函数交给sync_to_async()，不要把零散同步查询塞进事件循环。

本地验证结果

最小实验把答案拆成 6 个 SSE 帧，并验证每一帧都以\n\n结束、中文 JSON 可解析、最终存在done事件。RuyiDjangoCRM 的搜索与 SSE 定向测试共 14 项全部通过；项目测试命令只因定向运行导致全仓覆盖率 38% 低于 50% 门槛而返回非零，测试本身没有失败。

结论

流式聊天真正的完成标准不是“终端能看到几个 token”，而是浏览器能即时显示、连接中断能释放资源、网关不会缓冲、数据库操作不会堵塞事件循环。做到这四点，Django 才算真正接住了 AI 大模型的流式输出。

参考资料

• Django 异步支持官方文档
• Django ASGI 部署官方文档