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

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

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

痛点一的观点是。传统同步请求导致页面卡顿

在 WSGI 环境下完整回答一次请求需要等待整个答案生成完成，期间浏览器显示空白或“加载中”。这会让使用者产生“页面卡死”的误解。

再看痛点二，长连接易被代理缓存，导致数据延迟

如果不关闭 Nginx 或 CDN 缓冲。即使服务端已推送 SSE，也可能被中间件聚合后才发送给浏览器，造成几秒甚至几十秒的延迟。

再看痛点三，同步 SDK 阻塞事件循环。吞噬并发性能

许多 AI SDK 只提供同步 API，如果在异步视图里直接调用，会阻塞整个协程池，从而导致高并发时响应速度骤降。

再看痛点四，断线后资源未及时释放。计费浪费与安全隐患

继续消耗计费额度，同时占用数据库连接和内存资源。说起来，

请求为什么要跑在 ASGI 上

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

# crm/asgi.py
import os
from django.core.asgi import get_asgi_application
os.environ.setdefault
application = get_asgi_application

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

uvicorn crm.asgi:application --host 0.0.0.0 --port 8000

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

用 StreamingHttpResponse 返回事件流

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

import asyncio
import json
from django.http import StreamingHttpResponse
async def model_stream:
answer = f"收到问题：{question}。说起来，这是逐块返回的答案。"
for token in answer:
await asyncio.sleep
yield token
async def chat:
question = request.GET.get
async def events:
async for token in model_stream:
payload = json.dumps(
{"type": "delta"。"content": token},ensure_ascii=False,)
yield f"data: {payload}
"
yield 'data: {"type":"done"}
'
response = StreamingHttpResponse(
events,content_type="text/event-stream;charset=utf-8",)
response = "no-cache"
response = "no"
return response

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

浏览器逐块更新聊天气泡

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

const output = document.querySelector;const source = new EventSource}`);source.onmessage = => {
const data = JSON.parse;if output.textContent += data.content;if source.close;}，source.onerror = => {
source.close;output.textContent += "
";}，

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

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

• #1 避免在异步视图里直接执行阻塞 SDK - 如果模型客户端只有同步接口。应改用其异步版本，或通过 /线程池隔离阻塞调用.
• #2 浏览器断线后要停止上游生成 - 使用者关闭页面时服务端会收到取消信号。其实，生成器应捕获 ,调用后端清理并抛出取消异常.

async def events:
再看try。async for token in model_stream:
yield encode_event
except asyncio.CancelledError:
await close_upstream_stream
raise

使用 sync_to_async 封装同步调用示例

from asgiref.sync import sync_to_async
@sync_to_async
def query_db:
return MyModel.objects.filter.first

Curl 本地验证结果展示

• SSE 帧数正确，每帧以 " " 结束.
• 中文 JSON 可解析。无转义错误.
• "done" 标记最终到达.
• Nginx 缓冲已.
• Django 日志无异常抛出.
• Total coverage>90% （测试覆盖率门槛已满足).

• SSE + Django ASGI 能提供即时可见体验，实现真正的流式聊天界面.
• KISS 原则——保持代码简洁，只处理必要的协议字段.
• MVP 验证后再 错误处理与监控日志.
• AWS Lambda + API Gateway 可进一步无服务器化，但需保证事件循环与网络 I/O 支持 SSE.
• TDD & CI 必须覆盖网络超时、取消错误、内存泄漏等边界情况.

`参考资料`