ARTICLE DETAIL

资讯详情

深耕编程入门与网站建设的一线实战洞察。

AI Native Web开发实战:Django流式输出与多模型切换代码包解析

AI Native Web开发实战:Django流式输出与多模型切换代码包解析 简介这份代码包面向希望从架构层面理解并落地AI Native Web产品的开发者尤其适合已掌握TypeScript与Next.js基础、想进一步打通RAG与Prompt工程实践的中级前端或全栈工程师。内容围绕AI作为一等公民的设计思路展开覆盖技术选型、基础骨架搭建、RAG接入与生产环境问题处理等关键环节帮助读者建立从概念到上线的完整认知。资源共3个文件包含1个inscode工程配置、1个html入口页面与1个gitignore忽略规则压缩包约14KB体量轻便便于快速导入与二次开发。目前已有138人学习下载。代码包以清晰模块划分呈现可复用Hook封装、标准化错误处理模板与CI/CD配置脚本读者可据此对照实现向量检索、提示词编排与多租户隔离等细节并参考AI RSC、MCP协议等趋势方向获得可直接投入生产环境验证的工程化起点。1. 从一份 AI Native Web 开发代码包说起它到底能跑出什么如果你最近在找“AI Native Web 开发”的落地代码大概率已经翻到过一堆概念文——讲范式、讲思维、讲未来唯独不给能跑的东西。这份代码包不一样它是一套完整的 Web 应用源码把 AI 能力直接嵌进请求链路里而不是外挂一个聊天窗口。你拿到手就能本地起服务、改配置、接自己的模型接口看到一次完整的“用户输入 → 模型调用 → 流式返回 → 前端渲染”闭环。它适合两类人一是想从传统 CRUD 转向 AI 应用的后端或全栈二是手里有模型但不知道怎么把它做成可交互产品的开发者。代码包本身不绑定云厂商Django 做骨架前端模板直出省掉了前后端分离的联调成本。下面我按“先跑通、再拆解、后避坑”的顺序把这份资源拆开讲清楚。2. 环境准备与首次启动把代码包跑起来要动哪几个文件2.1 依赖清单与 Python 版本选择拿到代码包先别急着pip install -r requirements.txt先看 Python 版本。这类 AI Native 项目通常依赖较新的异步库和 HTTP 客户端Python 3.10 是底线3.11 更稳。我一般会先建虚拟环境避免和系统里其他项目的包打架。代码包里如果有pyproject.toml优先用它没有再用 requirements.txt。常见依赖包括 Django、httpx 或 requests、python-dotenv、以及某个模型 SDK。注意模型 SDK 的版本要和你的模型服务端匹配版本错位是后面 401 和 404 的主要来源。python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install -r requirements.txt # 如果有 pyproject.toml pip install -e .逻辑说明虚拟环境隔离依赖避免全局污染。-e .是以可编辑模式安装当前项目改代码不用重装。参数上如果你在国内网络环境pip 源可以换成清华或阿里云镜像但不要改代码包里的依赖版本号除非你清楚每个库的兼容边界。2.2 配置文件里必须改的三个值代码包一般会带一个.env.example或config/settings_local.py。你需要复制成.env或settings_local.py然后改三个地方模型 API 地址、API Key、以及数据库连接。模型地址不要带多余路径很多新手把/v1/chat/completions直接写进 base_url结果 SDK 又拼了一次变成双路径 404。API Key 不要提交到 Git代码包里如果有.gitignore确认它已经包含.env。# .env 示例 MODEL_BASE_URLhttp://localhost:8000/v1 MODEL_API_KEYsk-your-key-here DATABASE_URLsqlite:///db.sqlite3 DEBUGTrue逻辑说明MODEL_BASE_URL只写到版本号具体端点由代码里的客户端拼接。DATABASE_URL用 SQLite 是为了首次跑通生产再换 PostgreSQL。DEBUGTrue只在本地开线上必须关否则报错页会泄露源码路径。2.3 数据库迁移与静态文件收集Django 项目跑起来之前迁移和静态文件是两道必过坎。迁移不执行登录和会话表不存在页面直接 500。静态文件不收集前端 CSS 和 JS 加载不出来页面裸奔。我一般按这个顺序走makemigrations→migrate→collectstatic。如果代码包里已经带了迁移文件跳过 makemigrations直接 migrate。python manage.py makemigrations python manage.py migrate python manage.py collectstatic --noinput python manage.py runserver 0.0.0.0:8000逻辑说明makemigrations根据模型生成迁移脚本migrate应用到数据库。collectstatic把各 app 的静态文件汇总到 STATIC_ROOT。runserver 0.0.0.0:8000让局域网内其他设备也能访问方便手机端调试流式输出。注意--noinput避免交互确认卡住脚本。3. 拆解 AI Native 请求链路从用户输入到流式渲染的代码走读3.1 视图层如何组织模型调用AI Native Web 开发和传统 Web 最大的区别在视图层传统视图查数据库返回模板AI Native 视图要把用户输入发给模型再把模型输出流式吐给前端。代码包里通常有一个views.py或api/views.py里面用StreamingHttpResponse或EventSourceResponse做流式返回。关键点是不要在视图里同步阻塞等模型全部生成完那样用户会盯着白屏好几秒。正确做法是生成器逐块 yield。# views.py 片段 import json from django.http import StreamingHttpResponse from .services.model_client import ModelClient def chat_stream(request): user_input json.loads(request.body).get(message, ) client ModelClient() def event_stream(): for chunk in client.stream_chat(user_input): # 按 SSE 格式封装 yield fdata: {json.dumps({text: chunk})}\n\n yield data: [DONE]\n\n response StreamingHttpResponse(event_stream(), content_typetext/event-stream) response[Cache-Control] no-cache return response逻辑说明event_stream是一个生成器模型每返回一个 token 就 yield 一次Django 会逐步推给浏览器。content_type必须是text/event-stream否则前端 EventSource 收不到。Cache-Control: no-cache防止中间层缓存流式内容。参数上user_input要做长度截断避免超长 prompt 把模型上下文撑爆。3.2 模型客户端的封装与超时设置代码包里一般会把模型调用单独抽成一个 client 类放在services/目录下。这个类负责拼请求、处理重试、解析流式响应。我见过不少翻车案例都是因为没设超时模型服务卡住时整个 Web 进程被拖死。常见做法是给 httpx 或 requests 设 connect timeout 和 read timeout流式场景 read timeout 要设长一点比如 60 秒但 connect timeout 保持 5 秒。# services/model_client.py 片段 import httpx import os class ModelClient: def __init__(self): self.base_url os.getenv(MODEL_BASE_URL) self.api_key os.getenv(MODEL_API_KEY) self.timeout httpx.Timeout(connect5.0, read60.0, write10.0, pool5.0) def stream_chat(self, message): headers {Authorization: fBearer {self.api_key}} payload {model: default, messages: [{role: user, content: message}], stream: True} with httpx.stream(POST, f{self.base_url}/chat/completions, jsonpayload, headersheaders, timeoutself.timeout) as resp: resp.raise_for_status() for line in resp.iter_lines(): if line.startswith(data: ) and [DONE] not in line: yield line[6:]逻辑说明httpx.Timeout分四个维度控制比单一 timeout 更细。streamTrue告诉模型服务端用流式返回。iter_lines逐行读跳过[DONE]标记。参数上model字段要和你部署的模型名一致不一致会返回 404 或空响应。raise_for_status让 4xx/5xx 直接抛异常方便上层捕获。3.3 前端模板如何消费流式数据代码包的前端通常用原生 EventSource 或 fetch ReadableStream。EventSource 简单但不支持 POST所以如果用户输入较长常见做法是先用 POST 发消息拿到一个 session id再用 EventSource 订阅流。或者直接用 fetch 的 stream reader灵活但代码多一点。下面用 fetch 方式兼容 POST。// static/js/chat.js 片段 async function sendMessage(message) { const resp await fetch(/api/chat/stream/, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ message }) }); const reader resp.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n\n); buffer lines.pop(); for (const line of lines) { if (line.startsWith(data: )) { const data line.slice(6); if (data [DONE]) return; document.getElementById(output).textContent JSON.parse(data).text; } } } }逻辑说明reader.read()逐块读取TextDecoder把字节转字符串。buffer处理跨块截断的 SSE 消息避免 JSON 解析一半报错。lines.pop()把最后不完整的行留到下一轮。参数上stream: true在 decode 时开启流式模式防止多字节字符被截断。4. 避坑与排查这份代码包最容易翻车的五个地方4.1 模型返回 404 或 401现象前端一直转圈后端日志显示HTTPStatusError: 404或401。原因通常是 base_url 多写了/chat/completions或者 API Key 没加载进环境变量。解决检查.env是否被正确读取在 client 初始化时打印一次 base_url 和 key 的前四位。如果 base_url 以/v1结尾代码里再拼/chat/completions是对的如果 base_url 已经包含完整路径就要改代码或改配置二者只留一个。4.2 流式输出在 Nginx 后面变成一次性返回现象本地 runserver 正常流式部署到 Nginx 后用户等很久然后一次性出全文。原因是 Nginx 默认缓冲代理响应。解决在 Nginx 配置里对流式接口关掉缓冲加proxy_buffering off;和X-Accel-Buffering: no响应头。如果用了 CDN也要确认 CDN 不缓冲text/event-stream。4.3 数据库迁移报 no such table现象页面能打开但登录或保存记录时报no such table: auth_user。原因是迁移没执行或者数据库文件路径不对。解决确认DATABASE_URL指向的 sqlite 文件在项目根目录执行python manage.py migrate后看是否生成表。如果用了自定义用户模型要在第一次 migrate 之前就配好AUTH_USER_MODEL中途换模型会非常麻烦。4.4 前端 SSE 连接被浏览器限制现象Chrome 控制台报net::ERR_INCOMPLETE_CHUNKED_ENCODING或者 EventSource 频繁重连。原因是服务端没有正确关闭流或者响应头缺少Connection: keep-alive。解决确保生成器最后 yield 一个[DONE]并正常结束不要无限循环。如果用 EventSource服务端要返回Content-Type: text/event-stream且不能有Content-Length。4.5 模型输出被截断或乱码现象中文输出到一半变成乱码或者最后几个字丢失。原因是流式解码时没有用stream: true多字节字符被切分。解决前端TextDecoder加{ stream: true }后端确保每个 chunk 是完整 UTF-8 序列。如果模型服务端按字节流返回中间层要做缓冲凑齐一个完整字符再发。5. 进阶把代码包改造成可切换多模型的结构5.1 用工厂模式解耦模型客户端跑通之后你大概率想接第二个模型。这时候不要在视图里写 if-else而是把 ModelClient 抽成接口用工厂根据配置返回不同实现。常见做法是定义一个BaseModelClient然后OpenAIClient、LocalClient各自实现stream_chat。配置文件里加一个MODEL_PROVIDER变量工厂读它决定实例化哪个类。这样换模型只改环境变量不动业务代码。# services/factory.py from .openai_client import OpenAIClient from .local_client import LocalClient def get_model_client(): provider os.getenv(MODEL_PROVIDER, openai) if provider openai: return OpenAIClient() elif provider local: return LocalClient() raise ValueError(fUnknown provider: {provider})逻辑说明工厂函数把选择逻辑集中在一处视图只调get_model_client()。参数上MODEL_PROVIDER默认值给一个能跑的避免没配就崩。新增模型时只加一个类和一个分支符合开闭原则。5.2 加一层简单的对话历史管理AI Native Web 应用如果没有历史用户每次都要重新描述上下文体验很差。代码包如果没带历史功能你可以用 Django 的 session 或数据库存最近几轮对话。我一般会在 session 里存一个列表每次请求把最近 10 条消息拼进 prompt。注意 token 上限超出就截断最早的。下面是一个 session 版实现。# views.py 片段 def build_messages(request, new_message): history request.session.get(chat_history, []) history.append({role: user, content: new_message}) # 只保留最近 10 轮防止 token 爆炸 history history[-20:] request.session[chat_history] history return history逻辑说明request.session默认用数据库或缓存存储适合轻量场景。history[-20:]保留最近 20 条消息约 10 轮对话。参数上如果你的模型上下文窗口小把这个数字调低如果窗口大可以调高但要注意响应延迟会随 prompt 长度增加。5.3 验证改造是否成功的三个检查点改完之后别急着上线按这三个点验一遍第一切换MODEL_PROVIDER后服务能正常启动且流式输出不中断第二连续发 15 轮消息观察第 11 轮之后是否还能引用第 1 轮的内容验证历史截断逻辑第三用curl直接打流式接口看返回的 SSE 格式是否每行都以data:开头并以空行结束。我自己的习惯是每次改完模型层都先用 curl 跑一遍再开浏览器因为浏览器会掩盖格式错误。从那以后我每次接新模型都强制走一遍 curl 验证省掉了很多“前端怎么没反应”的排查时间。希望帮到你。本文还有配套的精品资源点击获取
返回列表