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)applicationget_asgi_application()启动时不要再把生产流式服务指向 WSGIuvicorn crm.asgi:application--host127.0.0.1--port8000本地 RuyiDjangoCRM 已经用同一个 ASGI 入口承载长连接通知流并在响应中关闭 Nginx 缓冲。这比单独写一个“能输出几段文字”的演示更接近真实部署。用 StreamingHttpResponse 返回事件流下面先用异步生成器模拟模型逐词输出。接入任意支持流式响应的大模型时只需要替换model_stream()Django 这一层保持不变。importasyncioimportjsonfromdjango.httpimportStreamingHttpResponseasyncdefmodel_stream(question:str):answerf收到问题{question}。这是逐块返回的答案。fortokeninanswer:awaitasyncio.sleep(0.03)yieldtokenasyncdefchat(request):questionrequest.GET.get(q,请介绍 Django ASGI)asyncdefevents():asyncfortokeninmodel_stream(question):payloadjson.dumps({type:delta,content:token},ensure_asciiFalse,)yieldfdata:{payload}\n\nyielddata: {type:done}\n\nresponseStreamingHttpResponse(events(),content_typetext/event-stream; charsetutf-8,)response[Cache-Control]no-cacheresponse[X-Accel-Buffering]noreturnresponse每个事件必须以两个换行结束。不要把模型文本直接拼成未经编码的 JSON否则引号、换行或反斜杠很容易破坏事件格式。浏览器逐块更新聊天气泡如果接口使用 GET浏览器原生EventSource足够简单constoutputdocument.querySelector(#answer);constsourcenewEventSource(/api/chat/?q${encodeURIComponent(question)});source.onmessage(event){constdataJSON.parse(event.data);if(data.typedelta)output.textContentdata.content;if(data.typedone)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 部署官方文档