部署与成本：上线一个能赚钱的 AI 应用
这一篇把「demo」变成「产品」。
一、部署形态怎么选
| 形态 | 说明 | 成本 | 适用 |
|---|---|---|---|
| 纯 API 调用 | 用云厂商模型 | 按 token | 早期、低频、快速验证 |
| API + 自建小模型 | 简单任务用小模型，难任务路由给大模型 | 混合 | 主流做法 |
| 自建推理服务 | vLLM 部署开源模型 | 固定 + 运维 | 高频、数据敏感 |
| 端侧 | 手机/PC 本地跑量化模型 | 无边际成本 | 隐私敏感、离线 |
分层路由是成本杀手
80% 的简单请求用最便宜的模型，20% 的难请求才升级到大模型。 一个简单的意图分类器（甚至关键词规则）就能分流，成本常降 60%–80%，效果几乎不变。
python
def route_and_answer(question: str) -> str:
if is_simple_lookup(question): # 规则/小分类器
return small_model(question)
if needs_reasoning(question):
return large_model(question)
return medium_model(question)二、服务端的最小骨架
python
# FastAPI + 流式输出 + 超时 + 限流 + 成本记录
import time, uuid, asyncio
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
app = FastAPI()
class ChatIn(BaseModel):
question: str = Field(min_length=1, max_length=4000)
session_id: str | None = None
@app.post("/chat")
async def chat(body: ChatIn):
trace_id = uuid.uuid4().hex[:12]
t0 = time.time()
try:
async with asyncio.timeout(30): # 总超时
stream = await build_stream(body.question, body.session_id)
return StreamingResponse(
wrap_stream(stream, trace_id, t0),
media_type="text/event-stream")
except asyncio.TimeoutError:
raise HTTPException(504, "请求超时，请稍后重试")
async def wrap_stream(stream, trace_id, t0):
usage = {"in": 0, "out": 0}
try:
async for chunk in stream:
usage["out"] += 1
yield f"data: {chunk}\n\n"
except Exception as e:
yield f"data: {json.dumps({'error': '服务异常，已记录'})}\n\n"
log_error(trace_id, e)
finally:
yield "data: [DONE]\n\n"
log_metrics(trace_id, usage, time.time() - t0)上线必做清单
| 类别 | 项 |
|---|---|
| 稳定性 | 超时、重试（指数退避）、熔断、降级到「只回答」 |
| 并发 | 限流（按用户/全局）、队列、背压 |
| 安全 | 密钥走环境变量、输入长度限制、输出脱敏、日志不含 PII |
| 成本 | 每请求 token 上限、日预算告警、缓存 |
| 观测 | trace、错误率、延迟分位、成本看板 |
| 版本 | Prompt / 模型 / 索引都可回滚 |
三、缓存：最便宜的优化
| 层 | 命中条件 | 收益 | 风险 |
|---|---|---|---|
| 精确缓存 | 完全相同的问题 | 高 | 无 |
| 语义缓存 | 相似度 > 阈值 | 很高 | 阈值太低会答错 |
| 前缀缓存 | 相同 system/文档前缀 | 高（TTFT） | 无 |
| 工具结果缓存 | 相同查询参数 | 中 | 数据需新鲜度控制 |
python
def semantic_cache_get(q, threshold=0.95):
v = embed([q])[0]
sims = cache_embs @ v
i = int(np.argmax(sims))
return cache_answers[i] if sims[i] >= threshold else None语义缓存要保守
阈值设低（如 0.85）会把「退款要几天」和「换货要几天」当成同一个问题。 默认从 0.95 起，只在只读、低风险场景开启。
四、成本结构：算清楚才知道优化哪
一次请求的成本 = 输入 token + 输出 token + 检索 + 工具调用 + 基础设施
| 项 | 典型占比 | 优化手段 |
|---|---|---|
| 输入 token（含检索上下文） | 40%–70% | 裁剪上下文、摘要、前缀缓存 |
| 输出 token | 20%–40% | 限制 max_tokens、要求简洁 |
| 检索/重排 | 5%–15% | 减少候选数、跳过重排 |
| 基础设施 | 5%–20% | 复用连接、合并请求 |
输出 token 单价通常是输入的 3–5 倍，所以「限制输出长度」往往是最直接的省钱手段。
python
PRICE_IN, PRICE_OUT = 2.0, 8.0 # 元 / 百万 token（示例，按实际填）
def cost(in_tok, out_tok, n=1):
return n * (in_tok / 1e6 * PRICE_IN + out_tok / 1e6 * PRICE_OUT)五、一次真实的成本优化记录
| 步骤 | 做法 | 每千次成本 | 说明 |
|---|---|---|---|
| 基线 | 大模型 + 全量 8 段上下文 | ¥38.0 | — |
| ① 上下文裁剪 8→4 段 | 重排后只留 Top4 | ¥24.5 | 效果不变 |
| ② 输出限制 800→300 | Prompt 要求简洁 | ¥17.2 | 无损失 |
| ③ 前缀缓存 | system + 政策文档固定前缀 | ¥14.8 | TTFT 降 55% |
| ④ 简单问题路由到小模型（占 62%） | 规则 + 分类器 | ¥6.1 | 效果 -0.4 点 |
这张表就是「AI 工程师」和「会调 API 的人」的差别。
六、延迟优化（按感知排序）
| 手段 | 效果 |
|---|---|
| 流式输出 | 感知 TTFT 从 3s → 0.4s，最有效 |
| 先给「思考中」反馈 | 心理上消除等待焦虑 |
| 并行检索（多路召回同时进行） | 减少 100–300ms |
| 减少重排候选 | 重排是 CPU/GPU 密集 |
| 小模型处理简单请求 | 快 3–10 倍 |
| 预热（避免冷启动） | 首请求不慢 |
七、上线后必须做的事
| 时间 | 动作 |
|---|---|
| 上线 1 小时 | 看错误率、P95 延迟、成本速率 |
| 每天 | 看负反馈 Top10，抽检实际输出 |
| 每周 | 新增回归用例、跑全量评测、成本复盘 |
| 每月 | 模型/Prompt 版本对比、清理无效工具与冗余上下文 |
八、常见生产事故与预防
| 事故 | 原因 | 预防 |
|---|---|---|
| 成本一夜暴涨 10 倍 | 重试风暴 / 循环调用 / max_tokens 未设 | 预算告警 + 单请求上限 + 重试上限 |
| 全站超时 | 上游模型限流，无队列 | 限流 + 熔断 + 排队 + 降级 |
| 答错且无法解释 | 无 trace | 全链路记录 |
| 泄露敏感信息 | 日志打印了原文 | 日志脱敏白名单 |
| Prompt 改坏回不去 | 无版本管理 | Prompt 入 git + 灰度 |
| 索引过期答旧政策 | 无重建流程 | 文档变更触发增量索引 |
一条底线
生产环境的 AI 应用必须有「关掉 AI 也能用」的降级路径。 模型服务不可用时，至少返回：「智能助手暂时不可用，请查看帮助中心 / 转人工」。不要让用户看到 500。
通过标准
下一步
L5 完成。进入 L6 专家进阶。