企业官网建设流程全解析

去年年底有一个 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

然后在一个空目录里初始化虚拟环境并安装 Django：

uv venv .venv source .venv/bin/activate uv pip install django requests openai django-cors-headers

这里简单解释一下为什么装这些包：

• django：Web 框架本体。
• 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系列为主力，常见的有：

实际项目里我用的是qwen-plus，原因是它在我这个问答场景下响应速度、生成质量都比较令人满意，成本也稳定。

3.2 选哪种调用方式：OpenAI 兼容模式还是 DashScope SDK

百炼平台提供了两类调用方式，一开始很多人会纠结。我分别用过之后，给一个比较直观的对比：

我最终选择了 OpenAI 兼容模式，因为openai库对stream=True参数的处理已经非常成熟，网络上可参考的代码也多。等会后端实现部分，我也会用这种模式作为主要示例。

3.3stream=True时百炼到底返回什么

先搞清楚返回内容结构，对接时就不会懵。调用百炼兼容版接口时，加上stream=True，服务端会通过 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": f"Bearer {settings.DASHSCOPE_API_KEY}", "Content-Type": "application/json", } payload = { "model": settings.DASHSCOPE_MODEL, "messages": messages, "stream": True, } try: with requests.post( url, json=payload, headers=headers, stream=True, timeout=(10, 300), # 连接超时10秒，读取超时300秒 ) as resp: resp.raise_for_status() for line in resp.iter_lines(decode_unicode=True): 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 f"data: {json.dumps({'content': content}, ensure_ascii=False)}\n\n" except requests.exceptions.ConnectionError: # 客户端断开，或者百炼连接异常，都要正常退出生成器 yield f"data: {json.dumps({'error': '连接中断'}, ensure_ascii=False)}\n\n" except Exception as exc: yield f"data: {json.dumps({'error': str(exc)}, ensure_ascii=False)}\n\n"

这里有几个细节值得单独说明。

resp.iter_lines(decode_unicode=True)是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"}, status=400) user_message = body.get("message", "").strip() if not user_message: return JsonResponse({"error": "message is required"}, status=400) messages = [ {"role": "system", "content": "你是一个乐于解答问题的 AI 助手。"}, {"role": "user", "content": user_message}, ] response = StreamingHttpResponse( generate_stream_response(messages), content_type="text/event-stream", ) response["Cache-Control"] = "no-cache" response["X-Accel-Buffering"] = "no" response["Connection"] = "keep-alive" return response

给这个视图配置一下路由，chat/urls.py：

from django.urls import path from . import views urlpatterns = [ path("api/chat/stream/", views.chat_stream, name="chat_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: no，Nginx 会读取这个头并忽略缓冲设置。两种方式可以同时使用，前端效果更稳。

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 的流式响应原封不动地转发给前端，同时处理好代理缓冲、前端渲染和连接生命周期。希望这篇实战记录能帮你少踩一些坑，顺利做出有“智能感”的产品体验。