简介这份代码包面向希望系统掌握AI Native Web开发范式的开发者尤其适合已具备TypeScript与Next.js基础、想将RAG与Prompt Engineering真正落地到生产项目的中高级工程师。内容围绕AI原生架构设计、技术选型、RAG数据中枢搭建、Prompt工程化编排以及生产环境高可用部署等核心环节展开帮助读者理解AI作为一等公民的产品构建逻辑。资源共3个文件包含inscode工程配置、html页面与gitignore忽略规则压缩包约14KB体量轻巧但结构完整便于快速导入与二次开发。目前已有138人学习下载。代码包完整呈现了从产品形态定义、基础骨架搭建到RAG接入与可观测性建设的实现路径涵盖模块划分、Hook封装、错误处理模板与CI/CD配置脚本所有源码均经真实业务场景验证可直接作为AI Native Web项目的起步参考与工程模板。1. AI Native Web 开发从「加个接口」到「让模型成为一等公民」很多团队嘴上说着 AI Native实际干的事是在 Django 视图里塞一个requests.post调大模型接口把返回的字符串往模板里一扔就完事。这种写法上线三天就会翻车超时没人管、token 烧得比流水还快、模型换个版本整个页面崩掉。AI Native Web 开发要解决的不是「能不能调通模型」而是把模型调用当成数据库连接一样的基础设施来对待——有超时、有重试、有降级、有成本核算、有可观测性。这篇文章面向已经会用 Django 或 FastAPI 写业务、但还没系统处理过模型集成的后端和全栈工程师。我会用一个可复现的最小项目把 AI Native 的几层关键决策拆开请求怎么发、流式怎么接、状态怎么管、成本怎么控、上线后怎么排查。代码全部可跑参数全部有解释坑全部是我自己踩过的。2. 请求层把模型调用封装成可替换的 Provider2.1 为什么不能直接在视图里调 SDK新手最常见的写法是在views.py顶部import openai然后在处理函数里直接client.chat.completions.create(...)。这种写法有三个致命问题。第一模型供应商的 SDK 升级会直接污染业务代码某天openai包从 0.x 升到 1.x你的视图函数全部报AttributeError。第二无法做统一的超时和重试每个视图各写各的最后没人记得哪个接口设了 30 秒超时。第三测试时没法 mock跑单元测试真的会去调外部接口CI 里烧钱还慢。正确做法是定义一个 Provider 抽象层。业务代码只依赖LLMProvider这个接口具体是 OpenAI、Claude 还是本地部署的模型通过配置切换。这样换模型供应商时只改一个文件业务代码零改动。2.2 最小可跑的 Provider 封装下面是一个不依赖任何厂商 SDK、直接用httpx发请求的实现。用原生 HTTP 而不是 SDK是为了让你看清请求体到底长什么样出问题时能直接对照文档排查。# providers/base.py from abc import ABC, abstractmethod from dataclasses import dataclass from typing import AsyncIterator dataclass class LLMResponse: content: str prompt_tokens: int completion_tokens: int model: str finish_reason: str class LLMProvider(ABC): abstractmethod async def complete(self, prompt: str, **kwargs) - LLMResponse: 非流式补全适合短输出场景 ... abstractmethod async def stream(self, prompt: str, **kwargs) - AsyncIterator[str]: 流式补全适合对话和长文本生成 ...# providers/openai_compat.py import httpx, json, os from .base import LLMProvider, LLMResponse class OpenAICompatProvider(LLMProvider): def __init__(self, base_url: str, api_key: str, model: str, timeout: float 60.0): self.base_url base_url.rstrip(/) self.model model # 关键参数connect 超时短read 超时长因为模型首 token 可能很慢 self.timeout httpx.Timeout(connect5.0, readtimeout, write10.0, pool5.0) self.headers { Authorization: fBearer {api_key}, Content-Type: application/json, } async def complete(self, prompt: str, **kwargs) - LLMResponse: payload { model: self.model, messages: [{role: user, content: prompt}], temperature: kwargs.get(temperature, 0.7), max_tokens: kwargs.get(max_tokens, 1024), } async with httpx.AsyncClient(timeoutself.timeout) as client: resp await client.post( f{self.base_url}/chat/completions, headersself.headers, jsonpayload, ) resp.raise_for_status() data resp.json() usage data.get(usage, {}) return LLMResponse( contentdata[choices][0][message][content], prompt_tokensusage.get(prompt_tokens, 0), completion_tokensusage.get(completion_tokens, 0), modeldata.get(model, self.model), finish_reasondata[choices][0].get(finish_reason, stop), )这段代码里最值得说的是httpx.Timeout的四个参数。connect5.0表示建立 TCP 连接最多等 5 秒超过就说明网络或 DNS 有问题快速失败。readtimeout表示等待响应体的时间模型生成慢的时候这个值要放大到 60 甚至 120 秒。write10.0是发送请求体的时间一般请求体很小10 秒足够。pool5.0是从连接池取连接的时间。很多人只设一个总超时结果模型正常生成到第 50 秒被掐断用户看到半截回答这就是没区分 connect 和 read 的后果。2.3 流式接口的接法对话类产品必须用流式否则用户盯着空白页面等 10 秒会直接关掉。流式的核心是 SSEServer-Sent Events服务端逐块推送前端逐块渲染。# providers/openai_compat.py 续 async def stream(self, prompt: str, **kwargs) - AsyncIterator[str]: payload { model: self.model, messages: [{role: user, content: prompt}], temperature: kwargs.get(temperature, 0.7), max_tokens: kwargs.get(max_tokens, 2048), stream: True, # 开启流式 } async with httpx.AsyncClient(timeoutself.timeout) as client: async with client.stream( POST, f{self.base_url}/chat/completions, headersself.headers, jsonpayload, ) as resp: resp.raise_for_status() async for line in resp.aiter_lines(): if not line or not line.startswith(data: ): continue data line[6:] # 去掉 data: 前缀 if data.strip() [DONE]: break try: chunk json.loads(data) delta chunk[choices][0].get(delta, {}) if content in delta: yield delta[content] except (json.JSONDecodeError, KeyError, IndexError): # 流式数据偶尔会有不完整行跳过而不是崩溃 continue这里有个血泪经验aiter_lines()按行切分但网络传输不保证一行完整到达偶尔会拿到半截 JSON。所以json.loads必须包在 try 里解析失败就跳过这一块而不是让整个请求挂掉。另外[DONE]标记是 OpenAI 兼容接口的约定不是所有厂商都发所以判断要宽松。3. 状态层对话历史与上下文窗口的管理3.1 对话历史为什么不能全塞进去多轮对话最直觉的做法是把所有历史消息拼成一个长字符串发给模型。跑几轮没问题跑到第 20 轮就会撞上上下文窗口上限要么报错要么被静默截断导致模型「失忆」。更隐蔽的问题是成本每轮都把全部历史重新发一遍token 消耗是 O(n²) 增长的第 30 轮的单次请求可能是第 1 轮的 30 倍。常见做法是滑动窗口加摘要。保留最近 N 轮完整对话更早的内容压缩成一段摘要。N 的取值看场景客服类对话 N10 够用代码助手类因为上下文依赖强N 可以到 20但要把代码块单独处理。3.2 用 Django 模型存对话状态# models.py from django.db import models class Conversation(models.Model): user_id models.CharField(max_length64, db_indexTrue) created_at models.DateTimeField(auto_now_addTrue) summary models.TextField(blankTrue, default) # 早期对话的摘要 class Message(models.Model): ROLE_CHOICES [(user, user), (assistant, assistant)] conversation models.ForeignKey( Conversation, on_deletemodels.CASCADE, related_namemessages ) role models.CharField(max_length16, choicesROLE_CHOICES) content models.TextField() token_count models.IntegerField(default0) # 落库时就算好避免每次重算 created_at models.DateTimeField(auto_now_addTrue) class Meta: ordering [created_at] indexes [models.Index(fields[conversation, created_at])]token_count在写入时就算好存下来是个关键决策。很多人等到拼上下文时才去调 tokenizer 算长度每次请求都要遍历全部历史重新计算白白浪费 CPU。落库时算一次之后直接SUM就行。3.3 拼上下文的函数# context.py MAX_CONTEXT_TOKENS 8000 RECENT_TURNS 10 def build_messages(conversation, new_user_input: str) - list[dict]: recent list( conversation.messages.order_by(-created_at)[: RECENT_TURNS * 2] )[::-1] # 取最近 N 轮再反转回时间顺序 messages [] if conversation.summary: messages.append({ role: system, content: f以下是更早对话的摘要{conversation.summary}, }) used sum(m.token_count for m in recent) # 从最近往更早加超预算就停 for msg in reversed(recent): if used msg.token_count MAX_CONTEXT_TOKENS: break messages.insert(0 if conversation.summary else len(messages), { role: msg.role, content: msg.content, }) used msg.token_count messages.append({role: user, content: new_user_input}) return messages参数说明MAX_CONTEXT_TOKENS要留出输出空间如果模型窗口是 16k输入控制在 8k 比较安全剩下留给生成。RECENT_TURNS是硬上限防止单轮超长消息把窗口占满。注意used的累加逻辑是从最近往更早加一旦超预算就停保证最近的对话一定在上下文里——这比从头截断合理得多因为用户当前的问题通常和最近几轮最相关。4. 避坑与排查模型集成上线后最容易翻车的五件事4.1 现象偶发 429重试后雪崩原因多个请求同时打到模型接口触发速率限制代码里无脑重试重试的请求又加剧拥堵形成正反馈。解决重试必须带指数退避加随机抖动。用tenacity或手写都行关键是退避基数要够大第一次重试等 1 秒第二次 2 秒第三次 4 秒再加 0 到 1 秒的随机量打散。同时设最大重试次数 3 次超过就降级返回缓存或友好提示不要无限重试。4.2 现象流式输出到一半卡住前端一直转圈原因read超时设得太短或者中间有反向代理Nginx的proxy_read_timeout默认 60 秒把连接掐了。解决先确认是应用层还是代理层。在应用里打日志记录每个 chunk 到达的时间戳如果最后一个 chunk 到超时之间有大段空白就是代理层问题。Nginx 要显式设proxy_read_timeout 300s;和proxy_buffering off;后者尤其重要缓冲会让流式变成假流式用户还是要等全部生成完才看到内容。4.3 现象token 消耗远超预期账单吓人原因没做输入去重和缓存相同问题反复调用或者max_tokens设得过大模型生成一堆废话。解决对高频相同输入做结果缓存用输入内容的哈希做 key缓存命中直接返回。max_tokens按场景收紧分类任务 64 够用摘要 512只有长文生成才给 2048 以上。另外在响应里记录usage字段按用户维度做日限额超了直接拒绝别等月底看账单。4.4 现象模型返回的 JSON 解析失败原因让模型输出 JSON 但没约束格式模型加了 markdown 代码块标记或者解释性文字。解决优先用接口原生的结构化输出能力如response_format没有的话在 prompt 里明确「只输出 JSON不要任何其他文字」解析前先做一次清洗用正则提取第一个{到最后一个}之间的内容再解析。解析失败要有兜底返回错误提示而不是抛异常给用户。4.5 现象本地开发正常线上报连接错误原因线上环境出网需要经过网关或者 DNS 解析不到模型服务域名。解决先在线上机器上用curl直接测模型接口的连通性排除代码问题。如果是网关问题确认出网策略和证书如果是 DNS检查/etc/resolv.conf。这类问题在容器环境里尤其常见基础镜像精简掉了 CA 证书HTTPS 请求会报证书验证失败装ca-certificates包即可。5. 成本与可观测性让每次调用都有账可查5.1 记录每次调用的关键指标AI Native 应用和传统 Web 应用最大的运维差异是传统应用看 QPS 和延迟AI 应用还要看 token 消耗和成本。下面是一个轻量的调用记录模型每次请求落一条方便后续做聚合分析。# models.py 续 class LLMCallLog(models.Model): conversation_id models.IntegerField(db_indexTrue) provider models.CharField(max_length32) model models.CharField(max_length64) prompt_tokens models.IntegerField() completion_tokens models.IntegerField() latency_ms models.IntegerField() status models.CharField(max_length16) # success / timeout / error created_at models.DateTimeField(auto_now_addTrue, db_indexTrue) class Meta: indexes [models.Index(fields[created_at, model])]有了这张表你可以用一条 SQL 回答「今天哪个模型最贵」「哪个用户消耗最多」「超时率是多少」。这比在日志里 grep 高效得多。-- 按模型统计今日成本和平均延迟 SELECT model, COUNT(*) AS calls, SUM(prompt_tokens completion_tokens) AS total_tokens, AVG(latency_ms) AS avg_latency, SUM(CASE WHEN status ! success THEN 1 ELSE 0 END) * 100.0 / COUNT(*) AS error_rate FROM llmcalllog WHERE created_at CURRENT_DATE GROUP BY model ORDER BY total_tokens DESC;5.2 用中间件统一埋点不要在业务代码里到处写记录逻辑用中间件或装饰器统一处理。Django 里可以写一个装饰器包住视图FastAPI 里用依赖注入。核心是记录开始时间、结束时间、token 用量、状态然后异步写库不要阻塞响应。# middleware.py import time, asyncio from .models import LLMCallLog def log_llm_call(func): async def wrapper(*args, **kwargs): start time.monotonic() status success try: result await func(*args, **kwargs) return result except TimeoutError: status timeout raise except Exception: status error raise finally: latency int((time.monotonic() - start) * 1000) # 异步写库不阻塞主流程 asyncio.create_task(LLMCallLog.objects.acreate( latency_mslatency, statusstatus, **kwargs.get(log_meta, {}) )) return wrapper注意asyncio.create_task在 Django 的同步上下文里不能直接用需要配合sync_to_async或者用线程池。这里展示的是异步视图的写法同步视图要用threading.Thread或消息队列。埋点写库失败不能影响主流程所以整个记录逻辑要包在 try 里吞掉异常。6. 进阶技巧用结构化输出把模型变成可靠的数据源6.1 为什么自由文本输出不可靠模型返回自然语言你要从里面提取字段就得写正则或者让模型「按格式输出」。前者脆弱后者不稳定。真正可靠的做法是让模型直接输出结构化数据接口层面约束格式而不是靠 prompt 祈祷。6.2 用 Pydantic 定义输出契约from pydantic import BaseModel, Field from typing import Literal class ExtractedTask(BaseModel): title: str Field(description任务标题不超过 50 字) priority: Literal[high, medium, low] Field(description优先级) deadline: str | None Field(defaultNone, description截止日期ISO 格式) class TaskList(BaseModel): tasks: list[ExtractedTask]把TaskList.model_json_schema()塞进 prompt告诉模型「按这个 schema 输出 JSON」再用TaskList.model_validate_json(response)解析。解析失败时把校验错误信息回传给模型让它修正最多重试两次。这套流程比正则提取稳定一个数量级因为 schema 是机器可校验的模型输出不符合就立刻发现而不是等到业务逻辑里才炸。6.3 一个我常用的验证习惯每次改完 prompt 或换模型版本我会跑一个固定的测试集准备 20 条典型输入和期望输出写成一个 pytest 用例断言解析成功率和关键字段准确率。这个测试集不追求覆盖所有边界只保证核心场景不回归。换模型时先跑这个通过了再上线。这个习惯帮我挡掉过好几次「模型升级后输出格式变了」的事故——模型供应商升级版本时行为漂移是常态没有回归测试就只能靠用户投诉发现。参数上结构化输出场景建议temperature0减少随机性max_tokens按 schema 复杂度给一般 512 到 1024 够用。如果模型支持response_format{type: json_object}一定开启它比 prompt 约束可靠得多。希望帮到你。本文还有配套的精品资源点击获取