AI 应用工程

一、工程案例：把一个 Demo 级聊天功能做成生产级

场景：有一个 15 行的 FastAPI 聊天 Demo，直接调用某个模型 API。目标是把它变成一个能上生产的服务：支持流式输出、限流、配额、降级、成本归集、可观测，应对 1000 并发、抵御脚本刷 Token、在供应商故障时自动切换、账单可查且可控。

为什么必须做：Demo 只关心「能不能跑」，生产关心「能不能一直好用」。一个生产级 AI 应用，业务逻辑代码可能只占 20%，工程代码占 80%。

交付物（后文代码按此结构落位，缺一个都跑不起来）：

ai-service/
├── app/
│   ├── __init__.py
│   ├── main.py          # ①③ 路由、流式入口
│   ├── guard.py         # ⑥ 降级、限流、配额、预算
│   ├── cache.py         # ④ 精确缓存 + 语义缓存
│   ├── safety.py        # ⑦ 审核与质量兜底
│   └── usage.py         # ⑥ 记账落库
├── config.yaml          # ⑤ 网关配置
├── migrations/001_init.sql
├── .dockerignore        # 排除 .env 与缓存
├── Dockerfile
├── requirements.txt
└── docker-compose.yml   # ⑧

动手前确认：Python 3.11+、Docker Compose v2、Redis 7、PostgreSQL 15 可用；有一个支持 OpenAI 兼容接口的模型账号。本文的模型名、单价、限流额度都是示例，以厂商当前文档为准。

1.1 主线（按顺序执行）

下面的代码按改造顺序展示，实际按 ⑧ 的目录拆到 app/main.py、app/guard.py、app/cache.py、app/safety.py、app/usage.py；跨文件调用补上对应 import 即可。同一段里的函数允许前向引用（Python 在调用时才解析名字）。

# ① 基线 Demo（改造前）——先证明「慢」是可复现的
from fastapi import FastAPI
from pydantic import BaseModel
from openai import OpenAI

app = FastAPI()
client = OpenAI(api_key="sk-xxx")

class ChatRequest(BaseModel):
    question: str
    user_id: str = "u_demo"
    model: str = "fast"          # 逻辑模型名，实际后端由 ⑤ 的网关决定

@app.post("/ask")
def ask(req: ChatRequest):
    resp = client.chat.completions.create(
        model="gpt-4o", messages=[{"role": "user", "content": req.question}])
    return {"answer": resp.choices[0].message.content}

# ② 基线测量（服务先起来：uvicorn app.main:app --port 8080）
for i in $(seq 5); do
  curl -s -o /dev/null -w "总耗时: %{time_total}s\n" \
    -X POST "http://localhost:8080/ask" \
    -H "Content-Type: application/json" \
    -d '{"question":"介绍一下公司发展历史"}'
done
# 示例输出（实际值取决于模型与网络）：总耗时: 8.21s / 7.94s / 8.33s ...

改造清单：① 流式输出 → ② 精确缓存 → ③ 语义缓存 → ④ 多模型路由 → ⑤ 降级链 → ⑥ 限流配额 → ⑦ 成本统计 → ⑧ 可观测追踪 → ⑨ 内容审核 → ⑩ 质量兜底。

请求进来后按下面的顺序判定，任何一步失败都要在花 Token 之前返回：

# ③ 流式输出（FastAPI + SSE）：限流 → 审核 → 配额 → 缓存 → 生成
import asyncio, json, os
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse, JSONResponse
from openai import AsyncOpenAI

app = FastAPI()
client = AsyncOpenAI(base_url=os.getenv("GATEWAY_URL", "http://localhost:4000"),
                     api_key=os.getenv("GATEWAY_KEY", "sk-gateway"))  # 替换 ① 的 client，走 ⑤ 网关

@app.post("/chat/stream")
async def chat_stream(req: ChatRequest, request: Request):
    if req.model not in MODEL_CHAIN:                              # ⑥：逻辑模型白名单
        return JSONResponse({"error": "unknown_model"}, status_code=400)
    ok, _ = is_allowed(f"user:{req.user_id}", 20, 60)            # ⑥：20 次/分钟
    if not ok:
        return JSONResponse({"error": "rate_limited", "retry_after": 60},
                            status_code=429, headers={"Retry-After": "60"})
    if not (await asyncio.to_thread(moderate_input, req.question))["allowed"]:   # ⑦
        return JSONResponse({"error": "blocked", "blocked": True})
    est = estimate_tokens(req.question) + 512                     # ⑥：预留输出额度
    ok, remain = reserve_quota(req.user_id, est)
    if not ok:
        return JSONResponse({"error": "quota_exceeded", "remaining": remain}, status_code=403)
    within_budget, spent = check_budget(0.01)                     # ⑥：全局日预算
    if not within_budget:
        settle_quota(req.user_id, est, 0)                         # 退回预扣
        return JSONResponse({"error": "budget_exceeded", "spent": round(spent, 2)},
                            status_code=503)

    cached = get_cached(req.question, req.model, 0, PROMPT_VERSION) \
        or await find_similar(req.question, req.model, PROMPT_VERSION)
    if cached:                                                    # 缓存命中：退回预扣，不计费
        settle_quota(req.user_id, est, 0)
        await log_usage(req.user_id, req.model, None, from_cache=True)  # 记一行 0 成本
        body = {"delta": cached["answer"], "from_cache": True,
                "similarity": cached.get("similarity")}
        async def cached_stream():
            yield f"data: {json.dumps(body, ensure_ascii=False)}\n\n"
            yield ("event: done\ndata: " + json.dumps(
                {"done": True, "from_cache": True, "model_used": req.model}) + "\n\n")
        return StreamingResponse(cached_stream(), media_type="text/event-stream",
            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

    async def event_generator():
        buffer, released, usage, ok = "", False, None, True
        parts = []
        try:
            stream = await client.chat.completions.create(
                model=req.model, messages=[{"role": "user", "content": req.question}],
                temperature=0, stream=True, stream_options={"include_usage": True})
            async for chunk in stream:
                if await request.is_disconnected():
                    break
                if chunk.usage:
                    usage = chunk.usage
                delta = chunk.choices[0].delta.content if chunk.choices else None
                if not delta:
                    continue
                parts.append(delta)
                if _is_degenerate("".join(parts)[-200:]):        # 边生成边查复读
                    ok = False
                    metrics.incr("quality_issue.degenerate_repetition")
                    yield 'event: fallback\ndata: {"reason":"degenerate_repetition"}\n\n'
                    return
                if not released:
                    buffer += delta
                    if len(buffer) < 120:        # 缓冲前 N 字，审核通过才放行
                        continue
                    if not (await asyncio.to_thread(moderate_output, buffer))["allowed"]:
                        ok = False
                        yield 'event: blocked\ndata: {"reason":"output_moderation"}\n\n'
                        return
                    released = True
                    yield f"data: {json.dumps({'delta': buffer}, ensure_ascii=False)}\n\n"
                    buffer = ""
                else:
                    yield f"data: {json.dumps({'delta': delta}, ensure_ascii=False)}\n\n"
            if buffer and not released:
                yield f"data: {json.dumps({'delta': buffer}, ensure_ascii=False)}\n\n"
            if ok:
                answer = "".join(parts)
                set_cached(req.question, req.model, 0, PROMPT_VERSION, answer,
                           {"total_tokens": usage.total_tokens if usage else est})
                asyncio.create_task(remember_semantic(req.question, answer, req.model, PROMPT_VERSION))
            yield ("event: done\ndata: " + json.dumps(
                {"done": True, "from_cache": False, "model_used": req.model}) + "\n\n")
        except Exception as e:
            ok = False
            yield f"event: error\ndata: {json.dumps({'error': type(e).__name__})}\n\n"
        finally:
            actual = usage.prompt_tokens + usage.completion_tokens if usage else est
            settle_quota(req.user_id, est, actual)                   # 多退少补
            await log_usage(req.user_id, req.model, usage, from_cache=False,
                            est_tokens=actual)
    return StreamingResponse(event_generator(), media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})

三个容易踩的细节：data 里放 JSON 而不是裸文本，否则回答中的换行会把一条消息拆成多条；stream_options 不是所有厂商都支持，以厂商当前文档为准，不支持时用估算值结算；finally 保证断连也结算，否则配额只增不减。断连时 usage 可能为空，成本要按预估值记，不要记成 0，否则账会少算。

# ④ 精确缓存 + 语义缓存（Redis）
import hashlib, json, os, re, redis, numpy as np

r = redis.Redis.from_url(os.getenv("REDIS_URL", "redis://localhost:6379/0"))
THRESHOLD = 0.92
TTL = 86400
PROMPT_VERSION = "chat_v3"        # 与 prompt_template.version 对应

def cache_key(q, model, temp, prompt_version):
    norm = " ".join(q.strip().lower().split())
    raw = f"{norm}|{model}|{temp}|{prompt_version}"      # Prompt 变了必须换 key
    return "ai:exact:" + hashlib.sha256(raw.encode()).hexdigest()

def get_cached(q, model, temp, pv):
    data = r.get(cache_key(q, model, temp, pv))
    return json.loads(data) if data else None

def set_cached(q, model, temp, pv, answer, usage, ttl=TTL):
    r.setex(cache_key(q, model, temp, pv), ttl,
            json.dumps({"answer": answer, "usage": usage}, ensure_ascii=False))

async def embed(text):
    resp = await client.embeddings.create(model="bge-m3", input=text)  # 模型名以网关配置为准
    v = np.asarray(resp.data[0].embedding, dtype=np.float32)
    return v / (np.linalg.norm(v) + 1e-9)

def b2v(blob):
    return np.frombuffer(blob, dtype=np.float32)

def entities(text):
    return set(re.findall(r"\d+(?:\.\d+)?|https?://\S+", text))

def same_entities(a, b):
    return entities(a) == entities(b)

async def find_similar(q, model, pv):
    qv = await embed(q)
    best = (-1.0, None)
    for k in r.scan_iter(match="ai:sem:*", count=500):   # 量大时改用向量索引
        score = float(np.dot(qv, b2v(r.hget(k, "vec"))))
        if score > best[0]:
            best = (score, k)
    if best[0] >= THRESHOLD and best[1]:
        payload = json.loads(r.hget(best[1], "payload"))
        same_scope = payload.get("model") == model and payload.get("prompt_version") == pv
        if same_scope and same_entities(q, payload["question"]):  # 实体必须一致
            return {**payload, "similarity": round(best[0], 4), "from_cache": True}
    return None

async def remember_semantic(q, answer, model, pv):
    qv = await embed(q)
    key = f"ai:sem:{hashlib.sha1(q.encode()).hexdigest()[:16]}"
    pipe = r.pipeline()
    pipe.hset(key, mapping={
        "vec": qv.astype(np.float32).tobytes(),
        "payload": json.dumps({"question": q, "answer": answer,
                               "model": model, "prompt_version": pv},
                              ensure_ascii=False)})
    pipe.expire(key, TTL)
    pipe.execute()

scan_iter 是全量扫描，只在缓存条数少时可用；超过几万条就换成向量库（按 ANN 检索，再回表校验实体）。语义缓存命中后同样不扣配额。Embedding 与写入用后台任务，不能卡住流式响应的收尾。

# ⑤ 模型网关配置（LiteLLM，字段与默认行为以厂商当前文档为准）
model_list:
  - model_name: smart
    litellm_params:
      model: openai/gpt-4o-mini
      api_key: os.environ/OPENAI_API_KEY
    model_info:
      input_cost_per_token: 0.00000015   # 示例价，美元/token，以厂商当前文档为准
      output_cost_per_token: 0.0000006
  - model_name: fast
    litellm_params:
      model: deepseek/deepseek-chat
      api_key: os.environ/DEEPSEEK_API_KEY
  - model_name: local
    litellm_params:
      model: openai/qwen                 # 自建 vLLM 走 OpenAI 兼容协议
      api_base: http://localhost:8000/v1
      api_key: sk-internal
router_settings:
  fallbacks: [{"smart": ["fast", "local"]}]   # ★ 降级链
  num_retries: 2
  timeout: 60
general_settings:
  master_key: sk-gateway            # 示例；生产用环境变量注入
  database_url: postgresql://user:pass@postgres:5432/litellm

只在网关或只在应用层做降级，不要两层都配：两层都配会变成「网关重试 3 次 × 应用重试 3 次」，故障时把下游打得更惨，账单也会翻倍。本文后面 ⑥ 的应用层降级链用于演示逻辑，生产上二选一。

local 需要另起一个 vLLM（OpenAI 兼容端口），本文 compose 里没有它；没部署时先从 MODEL_CHAIN 和 fallbacks 里删掉，否则每次降级都会多等一次连接失败。

# ⑥ 降级链 + 限流 + 配额 + 成本统计
import asyncio, hashlib, os, time, uuid
from datetime import date
from decimal import Decimal

import httpx, psycopg

# 示例价（元/千 token）；上线前用厂商当前文档或网关返回值覆盖
PRICES = {"smart": {"in": Decimal("0.0011"), "out": Decimal("0.0043")},
          "fast":  {"in": Decimal("0.0005"), "out": Decimal("0.0015")},
          "local": {"in": Decimal("0"),      "out": Decimal("0")}}
MODEL_CHAIN = ["smart", "fast", "local"]
DAILY_LIMIT = 100_000
DAILY_BUDGET = 500.0

class _Metrics:
    def incr(self, name):
        r.incr(f"metrics:{name}")

metrics = _Metrics()

def estimate_tokens(text):
    # 中文约 1~1.5 字/token，粗估；实际以响应里的 usage 为准
    return int(len(text) / 1.5) + 8

def retryable(e):
    code = getattr(e, "status_code", None)
    if code in (408, 409, 429):
        return True
    if isinstance(code, int) and code >= 500:
        return True
    return type(e).__name__ in {"APITimeoutError", "APIConnectionError", "TimeoutException"}

async def chat_with_fallback(messages):
    for model in MODEL_CHAIN:
        try:
            resp = await client.chat.completions.create(
                model=model, messages=messages, timeout=30)
            if model != MODEL_CHAIN[0]:
                metrics.incr(f"fallback.{model}")
                send_alert(f"主模型不可用，已降级到 {model}")
            return resp, model
        except Exception as e:
            if not retryable(e):
                raise          # 400/401/403 是请求或配置错，降级只会掩盖问题
    raise RuntimeError("all models failed")

def is_allowed(key, limit, window_sec):
    now, zkey = time.time(), f"rl:{key}"
    member = f"{now}:{uuid.uuid4().hex[:8]}"     # 同一微秒的多请求不能互相覆盖
    pipe = r.pipeline()                          # MULTI/EXEC：清窗、写入、计数同一事务
    pipe.zremrangebyscore(zkey, 0, now - window_sec)
    pipe.zadd(zkey, {member: now})
    pipe.zcard(zkey)
    pipe.expire(zkey, window_sec)
    _, _, count, _ = pipe.execute()
    if count > limit:
        r.zrem(zkey, member)
        return False, 0
    return True, limit - count

RESERVE_LUA = """
local used = tonumber(redis.call('GET', KEYS[1]) or '0')
local est  = tonumber(ARGV[1])
local cap  = tonumber(ARGV[2])
if used + est > cap then return {0, cap - used} end
redis.call('INCRBY', KEYS[1], est)
redis.call('EXPIRE', KEYS[1], 172800)
return {1, cap - used - est}
"""

def quota_key(user_id, day=None):
    return f"quota:{user_id}:{day or date.today().isoformat()}"

def reserve_quota(user_id, est_tokens):
    ok, remain = r.eval(RESERVE_LUA, 1, quota_key(user_id), est_tokens, DAILY_LIMIT)
    return bool(ok), int(remain)

def settle_quota(user_id, reserved, actual):
    r.incrby(quota_key(user_id), actual - reserved)   # 多退少补

def calc_cost(model, pt, ct):
    p = PRICES.get(model, PRICES["fast"])
    return (Decimal(pt) / 1000 * p["in"] + Decimal(ct) / 1000 * p["out"]).quantize(Decimal("0.000001"))

def check_budget(est_cost):
    key = f"budget:daily:{date.today().isoformat()}"
    spent = float(r.get(key) or 0)
    return spent + est_cost <= DAILY_BUDGET, spent

def send_alert(msg):
    url = os.getenv("ALERT_WEBHOOK", "")
    if not url:
        print(f"[ALERT] {msg}")          # 本地开发至少留痕，线上必须配 webhook
        return
    # 钉钉文本消息格式；企业微信字段不同，以厂商当前文档为准
    httpx.post(url, json={"msgtype": "text", "text": {"content": msg}}, timeout=3)

def alert_once(msg, ttl=3600):
    if r.set(f"alert:dedup:{hashlib.md5(msg.encode()).hexdigest()}", "1", nx=True, ex=ttl):
        send_alert(msg)

def add_cost(cost):
    key = f"budget:daily:{date.today().isoformat()}"
    new_total = r.incrbyfloat(key, float(cost))
    r.expire(key, 86400 * 2)
    if float(new_total) >= DAILY_BUDGET * 0.8:
        alert_once(f"今日 AI 成本 {float(new_total):.1f} 元 / {DAILY_BUDGET} 元")

def _insert_usage(row):
    with psycopg.connect(os.environ["DATABASE_URL"]) as conn:
        conn.execute(
            "INSERT INTO ai_usage_log (trace_id, user_id, feature, model, prompt_tokens,"
            " completion_tokens, cost, from_cache) VALUES (%s,%s,%s,%s,%s,%s,%s,%s)", row)

async def log_usage(user_id, model, usage, from_cache, feature="chat", est_tokens=0):
    pt = getattr(usage, "prompt_tokens", 0) or (0 if usage else est_tokens)
    ct = getattr(usage, "completion_tokens", 0) or 0
    cost = Decimal("0") if from_cache else calc_cost(model, pt, ct)
    if cost:
        add_cost(cost)
    row = (uuid.uuid4().hex[:12], user_id, feature, model, pt, ct, cost, from_cache)
    await asyncio.to_thread(_insert_usage, row)   # 生产换成连接池 + 批量写

reserve_quota 用 Lua 把「判断 + 预扣」做成一个原子操作，避免多实例同时读到同一个旧值后各扣一次。is_allowed 里 zadd 的 member 必须唯一，否则同一时刻的两个请求会被 ZSET 当重复成员吞掉一个，限流形同虚设。

# ⑦ 内容审核 + 质量兜底
import json, os, re

import httpx

SENSITIVE_WORDS = {"炸弹制作", "违禁品购买"}
PATTERNS = [(re.compile(r"\d{17}[\dXx]"), "[身份证]"),
            (re.compile(r"\d{16,19}"), "[银行卡]"),
            (re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)"), "[手机号]")]
MODERATION_URL = os.getenv("MODERATION_URL", "")   # 云内容安全网关，按厂商文档对接

def moderate_by_api(text):
    """云内容安全封装：接口字段、判定码与计费以厂商当前文档为准。"""
    if not MODERATION_URL:
        print("[WARN] 未配置 MODERATION_URL，仅关键词生效，不可上线")
        return {"allowed": True, "source": "keyword_only"}
    resp = httpx.post(MODERATION_URL, json={"text": text}, timeout=3)
    resp.raise_for_status()
    return {"allowed": resp.json().get("suggestion") == "pass", "source": "api"}

def moderate_input(text):
    for w in SENSITIVE_WORDS:
        if w in text:
            return {"allowed": False, "reason": f"命中敏感词：{w}"}
    return moderate_by_api(text)

def moderate_output(text):
    return moderate_by_api(text)

def mask_sensitive(text):
    for pat, repl in PATTERNS:
        text = pat.sub(repl, text)
    return text

FALLBACK_REPLIES = {
    "too_short": "抱歉，我这次没能给出完整回答，请再试一次～",
    "degenerate_repetition": "抱歉，我这次回答得不太好，请重新问一次～",
    "no_content": "抱歉，我没有理解您的问题，能再说一遍吗？",
    "moderation_failed": "抱歉，这个问题我暂时无法回答，请换个问题～",
    "default": "抱歉，服务暂时出了点小问题，请稍后重试～",
}

def check_output(answer, max_len=3000):
    issues = []
    if len(answer) < 5:
        issues.append("too_short")
    if len(answer) > max_len:
        issues.append("too_long")
    if answer.strip().startswith("{"):
        try:
            json.loads(answer)
        except json.JSONDecodeError:
            issues.append("invalid_json")
    if _is_degenerate(answer):
        issues.append("degenerate_repetition")
    if re.search(r"<\|.*?\|>", answer):
        issues.append("special_token_leak")
    if not re.search(r"[\u4e00-\u9fa5a-zA-Z0-9]", answer):
        issues.append("no_content")
    return {"ok": len(issues) == 0, "issues": issues}

def _is_degenerate(text, window=30):
    if len(text) < window * 3:
        return False
    return text.count(text[-window:]) >= 3

def safe_answer(answer, max_len=3000):
    result = check_output(answer, max_len)
    if result["ok"]:
        return answer
    if "too_long" in result["issues"]:          # 超长按截断处理，不整段丢弃
        return answer[:max_len] + "…"
    issue = result["issues"][0]
    metrics.incr(f"quality_issue.{issue}")
    return FALLBACK_REPLIES.get(issue, FALLBACK_REPLIES["default"])

safe_answer 只用于非流式或已缓冲完的整段回答。流式场景下，兜底必须发生在 yield 之前（③ 的 120 字缓冲就是干这个），已经发出去的半句话收不回来。

# ⑧ 项目骨架（docker-compose.yml）
services:
  app:
    build: .
    ports: ["8080:8080"]
    environment:
      GATEWAY_URL: http://litellm:4000
      GATEWAY_KEY: sk-gateway
      REDIS_URL: redis://redis:6379/0
      DATABASE_URL: postgresql://user:pass@postgres:5432/ai_service
      MODERATION_URL: ${MODERATION_URL:-}
      ALERT_WEBHOOK: ${ALERT_WEBHOOK:-}
    depends_on:
      redis: {condition: service_healthy}
      postgres: {condition: service_healthy}
      litellm: {condition: service_started}
  litellm:
    image: ghcr.io/berriai/litellm:main-latest   # 生产请固定版本号
    ports: ["4000:4000"]
    volumes: ["./config.yaml:/app/config.yaml"]
    command: ["--config", "/app/config.yaml", "--port", "4000"]
    environment:
      OPENAI_API_KEY: "${OPENAI_API_KEY}"
      DEEPSEEK_API_KEY: "${DEEPSEEK_API_KEY}"
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    volumes: ["redis-data:/data"]
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 5s
      timeout: 3s
      retries: 10
  postgres:
    image: postgres:15-alpine
    environment: {POSTGRES_USER: user, POSTGRES_PASSWORD: pass, POSTGRES_DB: ai_service}
    volumes: ["pg-data:/var/lib/postgresql/data", "./migrations:/docker-entrypoint-initdb.d"]
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d ai_service"]
      interval: 5s
      timeout: 3s
      retries: 10
volumes: {redis-data: null, pg-data: null}

# Dockerfile
FROM python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8080"]

.dockerignore 至少排除 .env、__pycache__、.git，避免把密钥打进镜像。requirements.txt 生产环境应锁定版本（本文省略）。

# requirements.txt
fastapi
uvicorn[standard]
openai
redis
numpy
psycopg[binary]
httpx

# 启动与冒烟（在 ai-service/ 下执行）
printf 'OPENAI_API_KEY=sk-...\nDEEPSEEK_API_KEY=sk-...\n' > .env
docker compose up -d --build
docker compose ps                       # 四个服务都应为 running/healthy
curl -N -X POST http://localhost:8080/chat/stream \
  -H "Content-Type: application/json" -d '{"question":"你好"}'

-- migrations/001_init.sql
CREATE TABLE ai_usage_log (
    id BIGSERIAL PRIMARY KEY,
    trace_id VARCHAR(32) NOT NULL,
    user_id VARCHAR(64) NOT NULL,
    feature VARCHAR(64) NOT NULL,
    model VARCHAR(64) NOT NULL,
    prompt_version VARCHAR(32),
    prompt_tokens INT NOT NULL,
    completion_tokens INT NOT NULL,
    cost NUMERIC(12,6) NOT NULL,
    ttft_ms INT,
    total_ms INT,
    from_cache BOOLEAN DEFAULT FALSE,
    fallback_used BOOLEAN DEFAULT FALSE,
    error_code VARCHAR(32),
    created_at TIMESTAMP NOT NULL DEFAULT NOW()
);
CREATE INDEX idx_usage_user_time ON ai_usage_log (user_id, created_at);
CREATE INDEX idx_usage_feature ON ai_usage_log (feature, created_at);
CREATE INDEX idx_usage_trace ON ai_usage_log (trace_id);

/docker-entrypoint-initdb.d 只在数据卷首次初始化时执行；表结构变更要另写迁移（如 Alembic），不能靠改这个 SQL 文件。

1.2 验收（10 项全过才算交付）

#	检查	命令/操作	通过标准
1	流式输出	curl -N -X POST localhost:8080/chat/stream -H 'Content-Type: application/json' -d '{"question":"你好"}'	1 秒内出现首块 data: {"delta":...}（示例值），最后有 event: done
2	精确缓存	同一问题连发两次（等第一次结束），比对第二次的 event: done	第二次 from_cache: true，ai_usage_log 无新增非缓存行
3	语义缓存	先问「怎么申请退款」，再问「如何申请退款」	第二次 similarity > 0.92 且 from_cache: true，实体不一致时不命中
4	逻辑模型路由	分别发 {"question":"几点上班","model":"smart"} 与 "model":"fast"	event: done 的 model_used 与请求一致；model 不在白名单返回 400
5	降级生效	改坏主模型 key 后直接请求网关：curl -s localhost:4000/v1/chat/completions -H 'Authorization: Bearer sk-gateway' -H 'Content-Type: application/json' -d '{"model":"smart","messages":[{"role":"user","content":"hi"}]}'	网关返回备用模型结果，fallback 指标增加，ALERT_WEBHOOK 收到消息
6	限流生效	seq 30 | xargs -P 30 -I{} curl -s -o /dev/null -w '%{http_code}\n' -X POST localhost:8080/chat/stream -H 'Content-Type: application/json' -d '{"question":"你好"}' | sort | uniq -c	20 200 与 10 429，429 响应带 Retry-After
7	配额生效	docker compose exec redis redis-cli SET "quota:u_demo:$(date +%F)" 99999 后发一次请求	返回 403 且响应体含 quota_exceeded 与 remaining
8	成本统计	docker compose exec postgres psql -U user -d ai_service -c "SELECT model, from_cache, cost FROM ai_usage_log ORDER BY id DESC LIMIT 3;"	非缓存请求 cost > 0，缓存请求 cost = 0，model/from_cache 可读
9	内容审核	发 {"question":"炸弹制作"}，并对比前后 SELECT count(*) FROM ai_usage_log;	返回 blocked: true，行数不变（未调用模型）
10	质量兜底	发 {"question":"请把「这是一段用于测试重复检测的长句子，请一直重复。」重复输出 20 遍"}	2 秒内收到 event: fallback，metrics:quality_issue.degenerate_repetition 增加

第 5 项：降级告警必须发到钉钉/企业微信，不能静默。若降级放在应用层而非网关，则改为停掉主模型后观察 fallback_used。 第 6 项：必须并发发（xargs -P），串行发 30 次会跨过 60 秒窗口，测不出限流。429 响应体里带 retry_after，前端才能倒计时。 第 9 项：审核必须在调用模型之前完成，不能花 Token 后再拒绝。 第 7 项：测完删掉配额 key（redis-cli DEL "quota:u_demo:$(date +%F)"），否则后面几项都会 403。 第 10 项：兜底只能保证「及时止损」，已经流出的文字收不回；产品上仍需保留「重新生成」。

---

二、高频问答：会卡住你的地方，也是面试最常问的

2.1 首 token 延迟太长（用户感觉卡死）

现场

curl -N -s -X POST http://localhost:8080/chat/stream -H "Content-Type: application/json" \
  -d '{"question":"你好"}' | while IFS= read -r line; do
    [ -n "$line" ] && echo "$(date +%s.%N) 收到: $line" && break
  done
# 示例：非流式 8.2s 后才看到第一行；流式 0.5s 就有（实际值以实测为准）

要点

• 首字慢，先看有没有开流式，换模型是次要的。
• 非流式调用让用户看到的是「全部生成完才返回」；流式不改变总时长，改变的是把等待变成「正在发生」。
• 核实要对比流式与非流式的 TTFT，流式通常 < 1s；仍慢再查排队、网络和 Prompt 长度。

追问

• 流式能缩短总耗时吗？→ 不能，它只优化 TTFT，总时长由生成长度决定。
• 开了流式首字还是慢？→ 先查上游排队与网络，再看 Prompt 是否过长导致 prefill 变慢。
• 监控该记哪个指标？→ TTFT 与总时长分开记，只看总时长看不出流式有没有生效。

2.2 并发一上来就 429 / 502 / 超时

现场

响应码先分类：429 是限流/配额，502/504 多是代理或上游不可用；再查应用侧是否有限流中间件。

is_allowed("global", 200, 1)  # 每秒 200 次
# 超限返回 429（带 Retry-After），502/504 才需要查代理与上游

要点

• 并发一上来就报错，先别加 CPU：没有限流时所有请求直打模型 API，会撞上供应商的速率与并发上限。
• 供应商的上限由账号等级决定，加机器治标不治本。
• 429 和 502 不是一类问题：前者查配额与限流配置，后者查网关与 Nginx。

追问

• 429 带 Retry-After 怎么处理？→ 按它退避，不要立即重试，立即重试只会放大限流。
• 502 该查哪里？→ 代理和上游可用性，不是模型限流。
• 限流阈值按什么定？→ 按账号配额和自身容量定，超过账号上限的限流没有意义。

2.3 月账单远超预期

现场

SELECT feature, SUM(cost) AS total, COUNT(*) AS calls
FROM ai_usage_log WHERE created_at >= date_trunc('month', NOW())
GROUP BY feature ORDER BY total DESC;

要点

• 账单超预期不等于单价贵，先看钱花在哪；没有成本统计就只能靠猜。
• 常见三类：脚本刷 Token、Prompt/上下文过长、某个贵功能占比过高。
• 盲目换便宜模型可能降低质量，却没解决主要费用来源。

追问

• 怎么定位最贵的一类？→ 按 feature / user_id / model 分组 SUM(cost) 排序。
• 缓存命中要不要记账？→ 要记一行 0 成本，否则出账和实际调用对不上。
• 单价从哪来？→ 网关配置或账单同步，别把示例价写死进代码。

2.4 有人写脚本刷接口

现场

key_id = hashlib.sha256(api_key.encode()).hexdigest()[:16]   # 别把原始 Key 写进 Redis
is_allowed(f"key:{key_id}", 60, 60)   # 每 Key 每分钟 60 次

要点

• 验证码只防前端自动化，脚本直接调 API 时一次都不经过。
• 缺的是 API Key 维度的限流；Key 哈希后再进 Redis，别把原始 Key 写进去。
• 判断依据是同一 user_id / Key 的调用次数异常高，不是页面上的行为。

追问

• 为什么不按 IP 限？→ 同一公司共用出口 IP，会把正常用户一起限掉。
• 原始 Key 为什么不能进 Redis？→ Redis 泄露就等于 Key 泄露，存哈希前缀已够定位。
• 限流够了吗？→ 不够，还要配额与预算，因为一次请求可能消耗 100 token 也可能 10000。

2.5 审核不生效，模型还是回答违法内容

现场

# 错误：审核在输出后
answer = await call_model()  # 用户可能已经看到部分内容
moderate_output(answer)      # 太晚了

# 正确：输入审核在调用前
if not (await asyncio.to_thread(moderate_input, question))["allowed"]:
    return blocked
answer = await call_model()
if not (await asyncio.to_thread(moderate_output, answer))["allowed"]:
    return fallback

要点

• 审核要挂在「花 Token 之前」和「内容下发之前」，位置错了词库再全也没用。
• 流式场景输出审核必须缓冲前 N 个字符（示例 120 字）再放行，吐出去的内容收不回来。
• 只加词解决不了时序问题，这是两类不同的失效。

追问

• 输入审核和输出审核各防什么？→ 前者省 Token 且拦明显违规，后者防模型自己生成。
• 缓冲会不会拖慢首字？→ 会，首字延迟变成 N 个字符的生成时间，N 要权衡。
• 中途判出问题怎么办？→ 发 event: blocked 或 fallback 事件，别静默关闭连接。

2.6 模型供应商挂了，服务直接不可用

现场

MODEL_CHAIN = ["smart", "fast", "local"]
for model in MODEL_CHAIN:
    try:
        resp = await client.chat.completions.create(model=model, messages=messages)
        return resp, model
    except Exception as e:
        if not retryable(e):     # 400/401/403 直接抛，降级会掩盖配置错误
            raise

要点

• 主模型挂了就整体报错，缺的是降级链，不是「换一家供应商」。
• 降级链最后一级要落在自建模型上，外部全挂时还能兜底。
• 只对可重试错误降级：400/401/403 直接抛，否则降级会把配置错误掩盖成「偶发故障」。

追问

• 换供应商能解决吗？→ 只解决这一次，下次还会挂。
• 怎么验证降级真的生效？→ 人为让主模型 key 失效，看 fallback 指标和告警是否触发。
• 降级和熔断的区别？→ 降级决定换谁，熔断决定多久内不再试。

2.7 缓存命中率为 0，没人命中

现场

# 错误：key 里塞了每次都变的东西
today = datetime.now().isoformat()
cache_key = f"{question}|{today}|{request_id}"   # 永远不重复

# 正确：只保留影响答案的维度，其余先归一化
cache_key = f"{norm(question)}|{model}|{temperature}|{prompt_version}"

要点

• 命中率为 0 多半不是缓存坏了，是 key 每次都不同。
• key 里混入时间戳、request_id、完整会话，等于永远不重复；改 Prompt 也必须换 key。
• 只保留影响答案的维度：归一化的问题、模型、temperature、Prompt 版本。

追问

• temperature > 0 是不是命中率 0 的原因？→ 不是，同一问题的 key 仍然稳定；但它让答案不可复现，所以只对 temperature=0 开缓存。
• 精确缓存中了、语义缓存不中？→ 两层分开排查，先确认 exact 的 key 稳定，再单独看阈值是否过高、实体校验是否过严，这两条都会让改写后的问题命不中。
• 怎么核实？→ 把两次请求的 key 打印出来逐字比对。

2.8 限流把正常用户也拦住了

现场

# 错误：只按 IP 限流
is_allowed(f"ip:{client_ip}", 20, 60)  # 同一个公司 50 人共用 IP → 被限
# 正确：按用户 ID 限流
is_allowed(f"user:{user_id}", 20, 60)

要点

• 误伤是限流维度选错：按 IP 限会把同一公司或局域网的几十个人算成一个。
• 应按 user_id（或 Key）限，IP 只留给未登录入口兜底。
• 调高阈值只是把问题推迟，脚本会继续放大用量。

追问

• 怎么发现是维度问题？→ 看是否有多个 user_id 共享同一个 IP。
• 未登录用户怎么限？→ 只能按 IP 或设备，并把阈值压得更低。
• 限流和配额的区别？→ 一个管次数，一个管总量，两个都要。

2.9 Nginx 代理导致流式卡顿

现场

location /chat/stream {
    proxy_pass http://app:8080;
    proxy_http_version 1.1;       # 默认 1.0 会攒包
    proxy_set_header Connection "";
    proxy_buffering off;          # 必须
    proxy_cache off;              # 必须
    gzip off;                     # 压缩同样会攒块
    proxy_read_timeout 300s;      # 必须大于最长生成时间
}

上游已经带 X-Accel-Buffering: no 时，Nginx 会自动关闭该响应的缓冲；但 proxy_http_version 和读超时仍要自己配。

要点

• 本地流式正常、上线变成一次性返回，先怀疑代理缓冲。
• 要同时关 proxy_buffering、proxy_cache、gzip，并把 HTTP 版本置 1.1。
• proxy_read_timeout 必须大于最长生成时间，否则长回答会被从中间掐断。

追问

• 上游带了 X-Accel-Buffering: no 还要配吗？→ 该响应会自动关缓冲，HTTP 版本和读超时仍要自己配。
• 为什么 gzip 也要关？→ 压缩会攒块，逐字下发会变成整段到达。
• 怎么确认是代理的问题？→ curl -N 直连应用逐块、经代理不逐块，就是代理。

2.10 Prompt 变更后质量不稳定

现场

bucket = int(hashlib.md5(f"{user_id}:{name}".encode()).hexdigest()[:8], 16) % 100
chosen = canary if bucket < 5 else stable

要点

• Prompt 一改就全量上线，出问题只能整体回滚，好的改动也一起丢。
• 应该灰度推进：内部 → 5% → 20% → 50% → 100%，并按 prompt_version 分组看指标。
• 没有版本号就无法归因；A/B 样本不足时，结论本身也不可信。

追问

• 怎么保证同一用户稳定落在同一组？→ 用 user_id 加实验名做稳定分桶（md5 % 100）。
• 灰度阶段看哪些指标？→ 质量、业务、成本三类一起看。
• 回滚能解决吗？→ 能止损，但会丢掉新 Prompt 里有效的改动。

2.11 流式输出中途断开，用户看到半句话

现场

在流式结束处打印 finish_reason，对比正常与异常请求；再查 Nginx 的 proxy_read_timeout。

要点

• 半句话比报错更糟：用户不知道答案没写完。
• 三个断点要分别看：上游 finish_reason 不是 stop、代理读超时短于生成时长、应用把异常转成了正常结束。
• 修法是断流时发明确的 error 事件，前端保留「重新生成」，不要静默关闭连接。

追问

• 怎么区分正常结束和断流？→ 只有 finish_reason 为 stop 才算正常，其余都要当异常处理。
• 断连后成本怎么记？→ 按预估值记，不要记 0，否则账会少算。
• 能续传吗？→ SSE 的 id / Last-Event-ID 可做续传，但服务端要保留已发内容，工程上多用重新生成兜底。

2.12 缓存击穿：热点问题一过期，请求全打到模型

现场

# 错误：过期瞬间所有请求都去调模型
if not cached:
    answer = await call_model(q)

# 正确：同一 key 只放一个请求去生成，其余短暂等待后重读
lock = f"lock:cache:{key}"
if r.set(lock, "1", nx=True, ex=10):
    try:
        answer = await call_model(q)
        set_cached(..., ttl=TTL + random.randint(0, 300))   # 随机 TTL，打散过期点
    finally:
        r.delete(lock)
else:
    await asyncio.sleep(0.05)
    return get_cached(...) or "正在生成，请稍候"

要点

• 热点 key 同时过期，miss 会在一瞬间一起打到模型，账单和并发都出毛刺。
• 缓存只是把流量「延迟」了，并没有削掉，容量不是瓶颈。
• 两个手段配合：随机 TTL 打散过期点，同一 key 只放一个请求去生成。

追问

• 加缓存容量有用吗？→ 没用，问题是同时到达的 miss。
• 持锁的请求挂了怎么办？→ 锁必须带过期时间，否则其它请求会一直等。
• 等待的请求返回什么？→ 短暂等待后重读，读不到先回「正在生成」，别直接打模型。

2.13 多实例部署后，限流和配额突然不生效

现场

# 错误：进程内计数，多副本各自为政
counts = {}
counts[key] = counts.get(key, 0) + 1

# 正确：Redis 集中计数，判定与写入用原子操作（pipeline/Lua，实现见 1.1 的 ⑥）
ok, remain = is_allowed(f"user:{user_id}", 20, 60)

要点

• 计数器在进程内存里，每个副本各算各的，扩到 N 个副本额度就变成 N 倍。
• 单机测试永远测不出来，至少要两个副本并发才暴露。
• 判定和写入要放在同一个原子操作里，否则并发请求会同时读到旧值。

追问

• pipeline 够吗？→ 能把多条命令一起发，但计数与预扣这类「判了再写」用 Lua 更稳。
• Redis 挂了怎么办？→ 付费接口 fail-closed，免费入口可短时 fail-open 并告警。
• 配额也要放 Redis 吗？→ 要，本地计数在多副本下同样会放大额度。

---

三、核心知识：从零开始要建立的东西

3.1 流式输出机制

流式输出的本质是服务器生成数据后立即发送，不等待完整结果。对比非流式：

模式	何时返回	用户感知（示例）
非流式	全部生成完	等待约 8 秒
流式	逐块生成即返	约 0.5 秒见到首字

关键洞察：流式不改变总时长，但把「等待」变成了「正在发生」。要分清两个指标：TTFT（首 token 时间，流式优化的是它）和总时长（生成长度决定，流式不影响）。

面试：常问「流式输出能降低总耗时吗？」答：不能，它优化的是 TTFT，总时长由生成长度决定。

3.2 SSE 协议格式

SSE（Server-Sent Events）是基于 HTTP 的服务器推送协议。格式简单但有陷阱：

HTTP 响应头：
  Content-Type: text/event-stream
  Cache-Control: no-cache
  X-Accel-Buffering: no   ← 告诉 Nginx 不要缓冲

响应体（每块一行为一个 JSON，避免正文换行破坏分帧）：
  data: {"delta":"你好"}
  ␣（空行）
  data: {"delta":"，今天"}
  ␣（空行）
  event: done
  data: {"done":true}
  ␣（空行）

元素	写法	含义
数据行	data: <内容>	推送的数据，一条消息可有多行
消息分隔	空行（\n\n）	一个消息结束标志
事件类型	event: <类型>	可选，区分 done/error/fallback
消息 ID	id: <序号>	可选，断线重连时用于续传
重连时间	retry: 3000	断线后重连间隔（毫秒）
注释	: 心跳	保持连接，防止代理超时断开

⚠️ 三个必知坑：① 事件以空行结束，只写 data: 不写空行，客户端会一直等；② 正文含换行时要么拆成多个 data: 行，要么像本文一样整段 JSON 编码；③ 中间代理可能缓冲，必须关掉。

3.3 模型网关角色

业务代码直接调各家 SDK 的问题：换模型要改代码、密钥散落、无法统一限流计费。网关解决八件事：

① 统一协议（OpenAI 兼容接口）
② 密钥管理（只在网关里）
③ 多模型路由（按成本/能力/可用性）
④ 降级与重试（自动切备用）
⑤ 限流与配额（按用户/Key）
⑥ 成本统计（记录 Token 和金额）
⑦ 缓存（多数网关只做精确缓存；语义缓存通常写在应用层）
⑧ 可观测（全链路日志与追踪）

网关不是越权中心：业务配额、内容审核、Prompt 版本仍应在应用层，网关负责协议、密钥、路由和账务。

3.4 降级链设计

降级链 = 主模型 → 备用模型 → 自建模型，最后一个必须是你自己的推理服务。

原则	说明
最后一个是自建	外部全挂时还能兜底
只对可重试错误降级	400/401/403 要直接报错，否则会掩盖配置问题
降级要告警	不能静默降级
降级要记录	标记响应里用了哪个模型

MODEL_CHAIN = ["smart", "fast", "local"]
for model in MODEL_CHAIN:
    try:
        resp = await client.chat.completions.create(model=model, messages=messages)
        if model != MODEL_CHAIN[0]:
            metrics.incr(f"fallback.{model}")
            send_alert(f"降级到 {model}")
        return resp, model
    except Exception as e:
        if not retryable(e):
            raise

再往前一步是熔断：某模型连续失败 N 次后，暂停它 M 秒，避免每个请求都先等一次超时。降级管「换谁」，熔断管「多久内不再试」。

面试：常问「降级链为什么最后一级要放自建模型？」答：外部供应商全挂时还能兜底；降级只对可重试错误生效，400/401/403 要直接报错。

3.5 缓存策略

缓存对象	命中收益	失效频率
相同问题的回答	延迟毫秒级，成本 0	低
Embedding 结果	省一次 Embedding	极低
Prompt 前缀 KV Cache	省 prefill 计算	提示词变才失效

def cache_key(q, model, temp, prompt_version):
    norm = " ".join(q.strip().lower().split())
    raw = f"{norm}|{model}|{temp}|{prompt_version}"
    return "ai:exact:" + hashlib.sha256(raw.encode()).hexdigest()

⚠️ temperature > 0 时每次回答不同，只对 temperature=0 开缓存；改 Prompt 必须换 key（把 prompt_version 放进去），否则上线新 Prompt 后还在返回旧答案。TTL 加 0~5 分钟随机抖动，避免同一批 key 一起过期。

3.6 语义缓存

语义缓存 = 用向量相似度判断语义相同，解决「怎么退款」和「如何退款」是不同 key 但答案相同的问题。

步骤	说明
嵌入	把问题转成向量
比对	计算余弦相似度，量大用 ANN
校验	实体（数字/日期/URL）必须完全一致
阈值	与嵌入模型强相关，通常 0.90~0.95，必须用自己的数据校准

THRESHOLD = 0.92
qv = await embed(question)
best = (-1.0, None)
for k in r.scan_iter(match="ai:sem:*"):
    score = float(np.dot(qv, b2v(r.hget(k, "vec"))))
    if score > best[0]:
        best = (score, k)
if best[0] >= THRESHOLD and best[1]:
    payload = json.loads(r.hget(best[1], "payload"))
    if same_entities(question, payload["question"]):
        return payload["answer"], best[0]

阈值不能照搬：换一个嵌入模型，相似度分布就变了。做法是抽 200 条真实问法，人工标「同义/不同义」，再选让误命中最低的阈值。

3.7 限流维度

维度	限什么	适用
单用户	每用户每秒 N 次	防止单个用户刷
单 IP	每 IP 每分钟 N 次	防止未登录滥用
全局	整个服务每秒 N 次	保护下游
按 Key	每 API Key 每分钟 N 次	对外 API

def is_allowed(key, limit, window_sec):
    now, zkey = time.time(), f"rl:{key}"
    member = f"{now}:{uuid.uuid4().hex[:8]}"      # member 必须唯一
    pipe = r.pipeline()                           # 清窗、写入、计数同一事务
    pipe.zremrangebyscore(zkey, 0, now - window_sec)
    pipe.zadd(zkey, {member: now})
    pipe.zcard(zkey)
    pipe.expire(zkey, window_sec)
    _, _, count, _ = pipe.execute()
    if count > limit:
        r.zrem(zkey, member)
        return False, 0
    return True, limit - count

计数必须放在共享存储里（Redis），进程内字典在多副本下会让额度变成 N 倍；判定和写入放在同一个 pipeline（MULTI/EXEC）里，才不会被并发请求同时读到旧值。

面试：常问「限流维度怎么选？为什么计数一定要放 Redis？」答：按用户或 Key 限，IP 只兜未登录入口；多副本必须共享计数并原子写入，否则额度会放大成 N 倍。

3.8 Token 配额

限流管「次数」，配额管「总用量」。两个必须都要——一次请求可能消耗 100 token 也可能 10000。

DAILY_LIMIT = 100_000

def quota_key(user_id):
    return f"quota:{user_id}:{date.today().isoformat()}"

def reserve_quota(user_id, est_tokens):
    # 判断 + 预扣必须原子；RESERVE_LUA 见 1.1 的 ⑥
    ok, remain = r.eval(RESERVE_LUA, 1, quota_key(user_id), est_tokens, DAILY_LIMIT)
    return bool(ok), int(remain)

def settle_quota(user_id, reserved, actual):
    r.incrby(quota_key(user_id), actual - reserved)   # 多退少补

先预扣再结算：请求开始时按「输入 + 预估输出」预扣，结束后用真实 usage 修正。只做「调用前检查」会被并发穿透——同一秒十个请求都读到「还没超」，然后一起超。

设计示例：免费用户每天 5 万 token；付费用户 100 万 token；内部服务不限但记录。一定要留预警线：用到 80% 时提示。

面试：常问「限流和配额有什么区别？为什么要先预扣再结算？」答：一个管次数、一个管总量，一次请求可能差几十倍；只做调用前检查会被并发穿透。

3.9 成本计算

成本项	计算方式	什么时候占大头
输入 Token	输入 token × 输入单价	长 Prompt、RAG 塞满上下文
输出 Token	输出 token × 输出单价	长回答、Agent 多轮
Embedding	向量化 token × 单价	建库、批量重嵌入
向量库存储	向量数 × 维度 × 字节	数据量大且不清理
网络与服务器	带宽、实例、网关	自建模型或大流量

输出单价通常高于输入（常见 2~4 倍，因厂商与模型而异，以厂商当前文档为准），所以「让模型少说废话」往往比换模型更省钱。

PRICES = {"smart": {"in": Decimal("0.0011"), "out": Decimal("0.0043")},  # 元/千 token，示例
          "fast":  {"in": Decimal("0.0005"), "out": Decimal("0.0015")}}
def calc_cost(model, pt, ct):
    p = PRICES.get(model, PRICES["fast"])
    return (Decimal(pt) / 1000 * p["in"] + Decimal(ct) / 1000 * p["out"]).quantize(Decimal("0.000001"))

⚠️ 单价随供应商调整，必须从网关配置或账单同步；不要用本文数字做预算。网关账本常以美元计价、应用侧按人民币展示，统一报表时要固定汇率并标明币种，否则会差一个数量级。

3.10 预算熔断

用量	动作
< 80%	正常运行
80%~100%	告警 + 关闭高成本功能
≥ 100%	停止所有 AI 调用或降级到本地

DAILY_BUDGET = 500.0
def check_budget(est_cost):
    key = f"budget:daily:{date.today().isoformat()}"
    spent = float(r.get(key) or 0)
    return spent + est_cost <= DAILY_BUDGET, spent

预算要按「人民币金额」而不是「token 数」设，因为不同模型单价差几倍；同时保留 token 配额，两者一个管钱、一个管量。

3.11 结构化日志

要记录的字段：

请求侧：用户 ID · 会话 ID · 功能 · 问题（脱敏）· 模型 · 参数
响应侧：回答（脱敏）· 是否缓存 · 用哪个模型
用量侧：输入 token · 输出 token · 成本
性能侧：首 token 延迟 · 总延迟 · 队列等待
质量侧：用户反馈 · 是否审核

{
  "ts": "2026-10-11T14:23:05.123+08:00", "event": "ai_chat_completed",
  "trace_id": "tr_8f2a91c4d3e5", "user_id": "u_10231",
  "request":  { "question_preview": "怎么申请退款？", "model_requested": "fast",
                "prompt_version": "chat_v3", "temperature": 0 },
  "response": { "model_used": "fast", "from_cache": false, "fallback_used": false },
  "usage":       { "prompt_tokens": 412, "completion_tokens": 156, "cost_cny": 0.00044 },
  "performance": { "ttft_ms": 480, "total_ms": 3120 },
  "quality":     { "moderation_passed": true, "user_feedback": null }
}

三条原则：① 问题/回答只存预览 + hash；② 一定要 trace_id，并且每个外部调用都带上；③ 成本/延迟/token 必须是数值字段，便于聚合。

3.12 分布式追踪

普通日志能告诉你「这次请求花了 3 秒」，但说不上「检索花了多久、模型花了多久」。Langfuse（开源，可自托管）解决这个问题。

界面能看到：Traces 列表（耗时、成本、模型）；单次 Trace 的完整时间线；Sessions（多轮串起）；Dashboard（P95 延迟、成本趋势、错误率）；Scores（反馈）。选型不必绑死 Langfuse，只要兼容 OpenTelemetry 的后端都能接，关键是 span 边界与字段统一。

3.13 Prompt 版本管理

位置	优点	缺点	适用
代码硬编码	简单	改要发版	极少变动
配置文件	改配置即生效	仍需发布	变化不频繁
数据库 + 后台	随时改、可灰度、可回滚	需要界面	生产推荐
管理平台	功能全	外部依赖	中大团队

CREATE TABLE prompt_template (
    id BIGSERIAL PRIMARY KEY,
    name VARCHAR(64) NOT NULL,
    version INT NOT NULL,
    content TEXT NOT NULL,
    is_active BOOLEAN DEFAULT FALSE,
    rollout_pct INT DEFAULT 100,
    created_at TIMESTAMP DEFAULT NOW(),
    UNIQUE (name, version)
);

def get_prompt(db, name, user_id):
    rows = db.query("SELECT version, content, rollout_pct FROM prompt_template "
                    "WHERE name = %s AND is_active = TRUE "
                    "ORDER BY version DESC LIMIT 2", (name,))
    if not rows:
        raise LookupError(f"prompt {name} 没有启用版本")
    canary = rows[0]
    stable = rows[1] if len(rows) > 1 else rows[0]
    bucket = int(hashlib.md5(f"{name}:{user_id}".encode()).hexdigest()[:8], 16) % 100
    chosen = canary if bucket < canary["rollout_pct"] else stable
    return chosen["content"], chosen["version"]

⚠️ 改 Prompt 必须灰度发布，不能直接全量上线。灰度顺序：内部员工 100% → 外部 5% → 20% → 50% → 100%。

3.14 内容审核

审核分为输入审核（在花 Token 前拦截）和输出审核（生成后检查）。

输入审核：
  ① 关键词过滤（快、免费，易绕过）
  ② 审核 API（召回更好、收费、有延迟）

输出审核：
  ① 审核 API 检查生成内容
  ② 流式场景：缓冲前 N 个字符再决定是否继续

审核方式	优点	缺点
关键词	即时、免费	容易绕过、误伤正常表达
审核 API	召回通常更好	收费、有延迟，仍会误判
规则引擎	可定制	规则维护成本高

⚠️ OpenAI 的 moderations 是它自家的接口，多数国内模型厂商没有等价接口；需要接入云厂商内容安全服务（如阿里云内容安全、腾讯云内容安全等），具体能力、判定码与计费以厂商当前文档为准。

3.15 输出质量检查

流式输出的特点是「边生成边看到」，坏的输出可能被用户看到一半。

检查项	触发条件	兜底回复
too_short	长度 < 5	抱歉，没给出完整回答
too_long	长度 > 3000	（截断）
invalid_json	以 { 开头但 JSON 解析失败	换种说法
degenerate_repetition	连续 3 次出现相同 30 字片段	抱歉，回答不太好
special_token_leak	出现 <|...|>	换种说法
no_content	没有中英文数字	抱歉，没理解

def check_output(answer, max_len=3000):
    issues = []
    if len(answer) < 5: issues.append("too_short")
    if len(answer) > max_len: issues.append("too_long")
    if answer.strip().startswith("{"):
        try: json.loads(answer)
        except json.JSONDecodeError: issues.append("invalid_json")
    if _is_degenerate(answer): issues.append("degenerate_repetition")
    if re.search(r"<\|.*?\|>", answer): issues.append("special_token_leak")
    if not re.search(r"[\u4e00-\u9fa5a-zA-Z0-9]", answer): issues.append("no_content")
    return {"ok": len(issues)==0, "issues": issues}

流式场景不能在结束时才查：要按滚动窗口查（例如保留最近 200 字做复读检测），否则等生成完才发现，用户已经看完了。检查项也要分层：长度、JSON 这类可以本地判，涉政涉黄必须走审核服务。

3.16 灰度发布

AI 输出非确定性，一个 Prompt 小改可能在某些输入上产生完全不同结果。灰度发布按三个维度：

维度	方式	优点
按用户	10% 用户用新版本	体验一致
按流量	随机 10% 请求	数据收敛快
按内部用户	先让员工用	最安全

def in_rollout(key, pct, salt=""):
    if pct >= 100: return True
    if pct <= 0: return False
    h = hashlib.md5(f"{salt}:{key}".encode()).hexdigest()
    return int(h[:8], 16) % 100 < pct

salt 用实验名（如 "refund_prompt_v4"），保证同一实验内同一用户稳定落桶；不同实验互不干扰。

推进节奏：内部员工 100%（1~2 天）→ 外部 5%（1 天）→ 20%（2 天）→ 50%（2 天）→ 100%。每阶段看质量、业务、成本三类指标。

3.17 用户实名与隐私

规则	说明
不记录敏感内容	日志只存预览（前 30 字）+ hash
身份证/手机号/银行卡入库前脱敏	用正则替换
对话保留期	示例 30 天，到期自动删除；实际按业务与合规要求定
数据不用于训练	用户协议明确，训练数据与对话隔离
免责声明	「由 AI 生成，仅供参考，请自行核实」

PATTERNS = [(re.compile(r"\d{17}[\dXx]"), "[身份证]"),
            (re.compile(r"\d{16,19}"), "[银行卡]"),
            (re.compile(r"(?<!\d)1[3-9]\d{9}(?!\d)"), "[手机号]")]
def mask_sensitive(text):
    for pat, repl in PATTERNS:
        text = pat.sub(repl, text)
    return text

---

四、冷门知识汇总

知识点	为什么冷门	用到时的关键词
SSE 分片粘包	网络分片不按消息边界到达，一次读到半条	维护缓冲区，按空行 \n\n 切分后再解析
SSE 增量解码	UTF-8 多字节字符可能跨 chunk，硬解码会出乱码	codecs.getincrementaldecoder("utf-8")
SSE 开始后改不了状态码	响应头已经发出，出错也只能在流里发事件	event: error 而不是 HTTP 500
SseEmitter 超时	Spring 默认超时短，0L 永不超时又会积累僵尸连接	new SseEmitter(300_000L)
稳定分桶	同一用户必须固定落在同一灰度组，否则体验反复横跳	md5(salt + user_id) % 100
Token 粗估	分词器差异大，只能用于预扣，不能用于账单	中文 1 字约 0.6~1 token，英文约 4 字符 1 token
A/B 样本量	小样本看不出 5~10 个百分点的差异	每组至少几百个反馈，基线越接近 50% 需要越多
预算告警去重	成本超线后每分钟报一次会变成告警风暴	SET key "1" NX EX 3600
缓存命中不计费	命中缓存仍扣配额/成本，账就对不上	settle_quota(user, est, 0)，不记 cost
Prompt 版本写进 metadata	出问题后无法按版本切片对比	metadata={"prompt_version": v}