去年年底有一个 Django 项目需要接入大模型问答功能需求非常直观用户在聊天窗口输入问题回答要一个字一个字地“蹦”出来而不是等十几秒后一次性拿到整段结果。如果只是简单地在后端用requests.post同步调用阿里云百炼大模型接口再把完整结果包装成一个 JSON 返回给前端体验会非常糟糕——网络稍有波动用户就会面对一个转圈几秒钟然后突然出现一整屏文字的结果。更麻烦的是一旦生成时间过长网关层直接断开连接前端什么都拿不到。所以方案从一开始就确定了走流式输出。这篇文章把我在 Django 中接入阿里云百炼大模型并实现流式输出的完整过程整理出来重点解决 SSE 协议对接、StreamingHttpResponse链路、前端 Markdown 实时渲染以及部署到 Nginx 后流式响应被缓冲卡住这几个核心问题。适合正在用 Django 做 Web 开发、想接大模型流式输出的读者参考。1. 为什么大模型接口天然就适合走流式输出聊实现之前先搞明白一个容易被忽略的问题大模型返回的内容本身是“逐 token 生成”的而不是一次性生成完毕。我们要做的只是把这种生成节奏原封不动地搬到用户浏览器上。1.1 从用户体感看普通接口和流式接口的差距大模型生成一段 500 字的内容算力消耗通常需要 3 到 10 秒。如果采用传统模式后端调用百炼接口时必须等到所有 token 全部生成完才能拿到完整文本并返回给前端。也就是说用户在这几秒钟内只能对着一个加载动画发呆。流式输出则是后端收到百炼的第一个增量数据块后立刻通过 HTTP 响应推给前端前端再用 JavaScript 把不断到达的数据块拼接、渲染。用户看到的就是打字机效果从按下回车到第一个字出现在屏幕上通常只需要几百毫秒。这种体验上的差距在问答、客服、写作辅助等场景里几乎是决定性的。也是为什么大模型 Chat 类产品几乎都使用流式输出。1.2 SSE 协议是流式输出的“地基”很多人会把 SSE 和 WebSocket 混在一起前端同事也经常问“是不是要用 WebSocket”其实大模型流式输出绝大多数用 SSE 就够了。SSE 全称 Server-Sent Events是建立在 HTTP 之上的一个轻量级协议。服务端在响应头里声明Content-Type: text/event-stream然后可以持续向客户端推送多行文本。每一条消息的基本格式是data: 这里是内容 data: 后续内容空行表示一条消息结束。大模型平台返回流式数据时会不断推送这样的分块最后用一个data: [DONE]标记结束。SSE 相比 WebSocket 的优势在于实现简单不需要额外握手协议普通 HTTP 请求就能承载。服务端断开连接后浏览器还可以自动重连这对大模型流式输出场景来说省去了很多手写心跳逻辑的麻烦。1.3 在这条链路上Django 的角色是什么Django 在数据建模、ORM、Admin 后台方面极其成熟但在很多人印象里“对长连接支持不好”。实际上 Django 从 3.2 开始完善了 ASGI 支持而且它的同步StreamingHttpResponse从 1.5 时代就有了做 SSE 转发并不存在什么无法逾越的障碍。我们的做法很简单Django 后端作为百炼 API 的“中间人”接收前端请求再带着 API Key 向百炼发起流式请求拿到增量内容后逐块写到StreamingHttpResponse里。前端不会直接接触百炼 API Key密钥安全也能控住。2. 项目初始化虚拟环境、工程骨架与跨域准备流式输出不只是一段后端代码它需要一个完整的 Django 工程来承载。这里我按实际项目中的习惯用 uv 管理虚拟环境和依赖而不是直接铺开一整串 pip 命令。2.1 用 uv 快速搭建 Python 3.11 Django 环境最近 uv 在 Python 圈子里讨论度很高核心优势是“快”。它会缓存所有已下载的发行包新建虚拟环境、安装依赖的速度比传统pip快一个数量级。如果你还没有安装 uv可以先安装curl -LsSf https://astral.sh/uv/install.sh | sh然后在一个空目录里初始化虚拟环境并安装 Djangouv venv .venv source .venv/bin/activate uv pip install django requests openai django-cors-headers这里简单解释一下为什么装这些包djangoWeb 框架本体。requests后端向百炼发起流式 HTTP 请求。也可以用httpx但我这里用requests是因为它的iter_lines对流式响应支持很好代码量小。openai如果你选择百炼平台的 OpenAI 兼容模式这个库可以帮你规范化接口调用省去手写请求体。django-cors-headers前后端分离开发时必装避免浏览器的跨域限制拦截请求。2.2 创建 Django 工程和 chat 应用我用一个干净的工程骨架作为演示django-admin startproject config . python manage.py startapp chat创建好之后把chat注册到config/settings.py的INSTALLED_APPS同时加上跨域相关配置INSTALLED_APPS [ django.contrib.admin, django.contrib.auth, django.contrib.contenttypes, django.contrib.sessions, django.contrib.messages, django.contrib.staticfiles, corsheaders, chat, ] MIDDLEWARE [ django.middleware.security.SecurityMiddleware, corsheaders.middleware.CorsMiddleware, # 注意放在 CommonMiddleware 前面 django.contrib.sessions.middleware.SessionMiddleware, django.middleware.common.CommonMiddleware, django.middleware.csrf.CsrfViewMiddleware, django.contrib.auth.middleware.AuthenticationMiddleware, django.contrib.messages.middleware.MessageMiddleware, django.middleware.clickjacking.XFrameOptionsMiddleware, ] CORS_ALLOW_ALL_ORIGINS True # 开发环境先放开生产环境请用白名单ALLOWED_HOSTS建议也先配置好比如开发时用ALLOWED_HOSTS [*]生产环境再收敛为具体域名避免随意跨域访问。2.3 把百炼 API Key 写进配置文件不要把 Key 硬编码在views.py里也不要放在代码仓库。我一般习惯放到环境变量然后在settings.py里读取import os DASHSCOPE_API_KEY os.getenv(DASHSCOPE_API_KEY, ) DASHSCOPE_MODEL os.getenv(DASHSCOPE_MODEL, qwen-plus)开发时可以在.env文件里维护用django-dotenv或者直接export到 shell都很常见。关键是让“配置”和“代码”分离方便后续部署。3. 接通百炼前先理解 API 的三种关键要素很多新手接入百炼时容易卡在 API 报文结构上。官方文档虽然全面但信息密度太大。我这里只提炼出必须搞清楚的三个点API Key、模型 ID、调用方式。3.1 API Key 和模型 ID 去哪里拿在阿里云控制台搜索“百炼”进入大模型服务平台后在右上角或API-KEY 管理页面可以创建新的 API Key。创建后复制出来保存好它等同于你调用大模型接口的“密码”。模型 ID 则决定你实际使用哪个大模型。百炼平台上以qwen系列为主力常见的有模型 ID定位qwen-plus通用对话平衡性能和成本qwen-turbo快速响应适合对延迟敏感的场景qwen-max效果最好适合复杂任务qwen-long长文本场景支持更大的上下文实际项目里我用的是qwen-plus原因是它在我这个问答场景下响应速度、生成质量都比较令人满意成本也稳定。3.2 选哪种调用方式OpenAI 兼容模式还是 DashScope SDK百炼平台提供了两类调用方式一开始很多人会纠结。我分别用过之后给一个比较直观的对比对比维度DashScope 官方 SDKOpenAI 兼容模式依赖dashscopeopenai 或直接 requests代码风格阿里云自有的 API 风格和 OpenAI SDK 完全一致学习成本第一次接触需要看文档熟悉 OpenAI 生态的人几乎零成本后续可迁移性锁定阿里云换其他 OpenAI 兼容平台时基本不用大改我最终选择了 OpenAI 兼容模式因为openai库对streamTrue参数的处理已经非常成熟网络上可参考的代码也多。等会后端实现部分我也会用这种模式作为主要示例。3.3streamTrue时百炼到底返回什么先搞清楚返回内容结构对接时就不会懵。调用百炼兼容版接口时加上streamTrue服务端会通过 SSE 连续推送片段每个片段的格式类似{choices:[{delta:{content:你好},index:0}]}再下一片可能是{choices:[{delta:{content:很高兴},index:0}]}注意这里拿到的字段是choices[0].delta.content而不是一次性返回的choices[0].message.content。很多人在对接流式输出时容易看错字段导致前端页面一直是空白。4. 后端核心DjangoStreamingHttpResponse链路实现把这层接口想清楚标题里说的“流式输出实战”就完成了大半。4.1 用生成器包装百炼的流式响应Django 的StreamingHttpResponse接受一个迭代器或生成器它不会等到生成器完全结束才返回而是“生成一块、发送一块”。这是实现流式输出的关键。我新建了chat/views.py核心代码如下import json import requests from django.http import StreamingHttpResponse, JsonResponse from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_http_methods from django.conf import settings def generate_stream_response(messages): 调用百炼 OpenAI 兼容接口把流式增量包装成 SSE 数据。 url https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions headers { Authorization: fBearer {settings.DASHSCOPE_API_KEY}, Content-Type: application/json, } payload { model: settings.DASHSCOPE_MODEL, messages: messages, stream: True, } try: with requests.post( url, jsonpayload, headersheaders, streamTrue, timeout(10, 300), # 连接超时10秒读取超时300秒 ) as resp: resp.raise_for_status() for line in resp.iter_lines(decode_unicodeTrue): if not line: continue if not line.startswith(data:): continue data_str line[len(data:):].strip() if data_str [DONE]: break json_data json.loads(data_str) choices json_data.get(choices, []) if not choices: continue delta choices[0].get(delta, {}) content delta.get(content, ) if content: # 这里重新包装成前端好解析的 SSE 格式 yield fdata: {json.dumps({content: content}, ensure_asciiFalse)}\n\n except requests.exceptions.ConnectionError: # 客户端断开或者百炼连接异常都要正常退出生成器 yield fdata: {json.dumps({error: 连接中断}, ensure_asciiFalse)}\n\n except Exception as exc: yield fdata: {json.dumps({error: str(exc)}, ensure_asciiFalse)}\n\n这里有几个细节值得单独说明。resp.iter_lines(decode_unicodeTrue)是requests库提供的流式行迭代方法它会自动按换行符分割数据。百炼返回的 SSE 每一行基本都是一条data:JSON所以直接用行迭代就能拿到增量内容。timeout(10, 300)的写法很关键。第一个数字是连接超时第二个是读超时。大模型生成一句长文本可能超过 30 秒不能只设一个 3 秒超时否则会误报超时。但也不能完全不设否则百炼服务异常时后端服务会被长时间挂住。4.2 视图函数返回StreamingHttpResponse有了生成器视图层就非常简单了csrf_exempt require_http_methods([POST]) def chat_stream(request): try: body json.loads(request.body) except json.JSONDecodeError: return JsonResponse({error: invalid json}, status400) user_message body.get(message, ).strip() if not user_message: return JsonResponse({error: message is required}, status400) messages [ {role: system, content: 你是一个乐于解答问题的 AI 助手。}, {role: user, content: user_message}, ] response StreamingHttpResponse( generate_stream_response(messages), content_typetext/event-stream, ) response[Cache-Control] no-cache response[X-Accel-Buffering] no response[Connection] keep-alive return response给这个视图配置一下路由chat/urls.pyfrom django.urls import path from . import views urlpatterns [ path(api/chat/stream/, views.chat_stream, namechat_stream), ]然后在工程的根 URLconf 里 include 一下from django.urls import path, include urlpatterns [ path(admin/, admin.site.urls), path(, include(chat.urls)), ]开发时启动python manage.py runserver用 curl 验证效果curl -N -X POST http://127.0.0.1:8000/api/chat/stream/ \ -H Content-Type: application/json \ -d {message: 请用三句话介绍杭州}-N参数让 curl 不要缓冲输出这样可以看到内容逐渐刷出来而不是一次性打印。测试成功后再接前端可以大大减少排查复杂度。4.3 为什么要格外重视Cache-Control和X-Accel-Buffering很多人在本地跑通后一部署到服务器上就发现流式输出变成了一次性返回。原因大多数是这两个响应头没处理好。Cache-Control: no-cache是告诉浏览器不要对接口响应做缓存确保每次请求都实时连接。X-Accel-Buffering: no则是给 Nginx 这类反向代理看的。如果 Nginx 默认开启了缓冲它会等后端把整段响应都发送完再一次性转发给浏览器流式效果当然就没了。这个头字段在后端响应里直接设置Nginx 会尊重它从而对该请求关闭缓冲。5. 前端消费流fetch 解析、SSE 处理与 Markdown 渲染后端把流推到浏览器端后前端还要会“读”这个流。我尽量用原生 JavaScript 实现核心这样不依赖特定框架你在 Vue、React 里都能轻松迁移。5.1 用 fetch 的 ReadableStream 逐行解析 SSE不能直接用response.json()去读取接口因为这是一个流式响应。需要借助response.body.getReader()读取字节流然后手动按行切割。我用接近生产环境的写法给出示例async function startChatStream() { const resp await fetch(/api/chat/stream/, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ message: userInput.value }), }); if (!resp.ok) { console.error(request failed); return; } const reader resp.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n); buffer lines.pop(); for (const line of lines) { const trimmed line.trim(); if (!trimmed.startsWith(data:)) continue; const payload trimmed.slice(5).trim(); if (payload [DONE]) return; try { const jsonData JSON.parse(payload); if (jsonData.content) { assistantOutput.textContent jsonData.content; } } catch (e) { console.warn(parse error:, e); } } } }注意这里我用textContent 而不是innerHTML 目的就是避免把模型输出当作 HTML 解析导致 XSS。如果是纯文本展示这个写法安全又简单。5.2 处理 Markdown 渲染不要每次更新都全量重渲染百炼模型输出经常带 Markdown比如代码块、标题、列表。如果直接把原始文本展示给用户阅读体验很差。网上很多教程会直接建议innerHTML marked.parse(allText)这在小段文本上没问题但在流式输出里会引发明显的性能问题。模型每返回一个增量 token前端都要把整个历史文本重新组装成一个 DOM 片段并替换内容一旦超过几千字页面会明显卡顿。我的处理方式是加一个“节流渲染”let rendering false; let pendingRender false; function scheduleRender(text) { markdownContainer.textContent text; if (!rendering) { rendering true; requestAnimationFrame(() { markdownContainer.innerHTML markdownRenderer.render(text); rendering false; if (pendingRender) { pendingRender false; scheduleRender(text); } }); } else { pendingRender true; } }思路很简单流式内容到达时先更新一个纯文本缓冲然后用requestAnimationFrame控制渲染帧率只在浏览器下一帧的时候执行一次真正的 Markdown 全量渲染。这样既能保持打字机动画效果又不会让浏览器因为频繁操作 DOM 而卡死。如果用 Vue可以配合computed属性延迟计算用 React 则可以把 Markdown 渲染放到useDeferredValue或useMemo里。核心思想一致降低渲染频率保证主线程不被流式更新阻塞。5.3 用户中断请求时后端生成器也必须停下来SSE 如果只是单向给前端推数据看起来很简单但用户可能中途关闭页面或点击“停止生成”。这时如果后端还在继续调用百炼等于白白消耗 token 费用。前端通过AbortController可以中止请求const controller new AbortController(); fetch(/api/chat/stream/, { signal: controller.signal }); // 点击停止时 controller.abort();当浏览器断开连接后Django 后端向StreamingHttpResponse写入内容时会抛出ConnectionError或BrokenPipeError。所以生成器里的异常捕获很重要捕获到后立刻break退出不再从百炼读取后续数据。这也是我上面代码里加上try/except的原因。如果你用的是 ASGI 和异步视图还需要注意取消异步任务的协程否则即使客户端断开后台任务也可能继续跑一段时间。对于纯同步视图生成器退出就代表请求结束处理逻辑会简单许多。6. 生产部署宝塔、Nginx 与流式响应冲突的排雷本地开发时runserver一切正常一上宝塔面板部署发现流式输出全变成了“一二十分钟后一次性出现”这种情况我至少见过三次。问题不在 Django而在代理层。6.1 现象定位Nginx 缓冲是最大“真凶”请求链路变成了浏览器 - Nginx - Gunicorn/Uvicorn - Django - 百炼。Nginx 默认会对上游应用响应开启缓冲它会把后端返回的内容攒起来直到攒满一定大小再发给浏览器。对于大模型流式响应后端生成一段、Nginx 攒一段前端自然要等很久。最直接的解决方式是在对应的location里关掉缓冲location /api/ { proxy_pass http://127.0.0.1:8000; proxy_http_version 1.1; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; set $x_accel_buffering no; }这里proxy_buffering off让 Nginx 收到上游数据后立即转发给客户端。proxy_read_timeout 300s也要调大避免后端长时间不产生数据时 Nginx 主动断开。如果不想每个站点都改 Nginx 配置也可以只在 Django 响应里设置X-Accel-Buffering: noNginx 会读取这个头并忽略缓冲设置。两种方式可以同时使用前端效果更稳。6.2 Gunicorn 还是 Uvicorn流式接口怎么选 worker 类型Django 传统部署常使用 Gunicorn。如果你的 Django 项目还是同步视图那么 Gunicorn 的syncworker 在理论上也可以处理流式响应但有一个前提每个连接会一直占用一个 worker。当并发用户数上来后worker 很容易被打满。更常见的选择是给 Gunicorn 添加--threads 4甚至更多线程让一个 worker 能同时处理多个请求。示例gunicorn config.wsgi:application -w 2 --threads 4 -b 127.0.0.1:8000 --timeout 300如果你想把 Django 的异步能力用起来改用 Uvicorn 直接运行 ASGI 应用也很好uvicorn config.asgi:application --host 127.0.0.1 --port 8000不过需要注意目前很多生产项目还是用 Gunicorn 管理 worker 进程、用 Uvicorn worker 跑 ASGI。这是另一个话题了这里只提醒一点流式接口尽量保证 worker 数量不要卡得太死否则用户一多前面看到的“打字机”就变成“卡带机”了。6.3 长连接和数据库连接不要互相拖累在 Django 中普通请求结束时会自动关闭数据库连接但StreamingHttpResponse是一个长连接。生成器持续向百炼读取数据并写入响应的过程中请求并没有结束。如果你在生成器里使用了 ORM 操作或者打开了一个数据库查询那么这个连接会被占住直到整个 SSE 流结束。这在下游并发访问时会放大成数据库连接池耗尽的问题。我的建议是不要在StreamingHttpResponse生成器内部做数据库查询。所有需要从数据库拿的数据在进入生成器之前就提前查好或者完整放进内存避免长连接占用数据库连接。如果确实需要动态查询请使用close_old_connections()及时处理。6.4 API Key 安全永远不要让前端直连百炼有的项目图省事让浏览器直接带着 API Key 请求百炼接口这是绝对不可取的。API Key 放在前端代码里用户通过浏览器的开发者工具一抓就能看到等于把大模型调用额度完全暴露出去。正确做法是像本文一样由 Django 后端作为代理保存 Key前端只访问你自己的域名接口。同时在后端加一层简单的访问控制比如登录校验、请求频率限制防止接口被恶意刷量。如果你用的不是长期有效的 API Key而是短期 Token那还要考虑 Token 刷新机制。但百炼目前主流的 API Key 模式已经够用关键是不要落到前端。最后再分享一点实战细节这个方案跑通之后有几个细节对我后来帮助很大。第一调试流式接口时一定要先脱离前端用curl -N验证后端输出。我见过很多同事一上来就写前端结果发现页面不显示最后绕了一圈才发现是后端压根没返回流白白浪费大量时间。第二别忽视响应头的兼容性。前端如果用的是EventSource它只能发起 GET 请求所以许多实现会改成 POST fetch。我这里用的是 fetch ReadableStream既能传 body又能处理 Error 状态相对更灵活。第三根据自己的业务调节“系统提示词”。流式输出玩得再花如果系统提示词写得模棱两可用户拿到手的答案质量也会很差。建议给百炼模型设计结构化的 system 内容这会直接影响最终回答的稳定性和你的业务匹配度。Django 接入阿里云百炼并不复杂核心是把流式请求的每一块增量内容用 Django 的流式响应原封不动地转发给前端同时处理好代理缓冲、前端渲染和连接生命周期。希望这篇实战记录能帮你少踩一些坑顺利做出有“智能感”的产品体验。