← 返回 FDE 培养总览

模块 15 · 大模型应用工程

本章目标：学会把"调通一次 API"升级成"能跑在生产环境、敢给客户用"的大模型应用。核心不是 Prompt 技巧，而是工程化：Prompt 版本管理、流式输出、错误重试、Token 限流、成本熔断、评测回归。

---

15.1 从 Demo 到生产：差在哪

你在笔记本上调通了一个 200 行的 Python 脚本，让模型总结文档。客户说"好，上线吧"。这中间差的不是模型能力，是工程化：

维度	Demo 阶段	生产阶段
Prompt	硬编码在 .py 里	版本化、可灰度、A/B 测试
错误处理	报错就挂	重试、降级、兜底回复
延迟	等就等了	流式输出 + 超时控制
成本	不关心	Token 计量、月度预算熔断
监控	看 print	trace 全链路、幻觉告警
测试	手动试几句	回归评测集，每次改 Prompt 跑一遍

---

15.2 Prompt 工程化：别把提示词写死在代码里

新手常犯的错：Prompt 字符串直接写在业务代码里，改一句话要发版。生产做法是把 Prompt 当配置/数据：

# prompts/summarize_v2.yaml
system: |
  你是一个专业的文档摘要助手。请：
  1. 用 3-5 句话概括核心内容
  2. 保留关键数据和时间点
  3. 不要添加原文没有的信息
  4. 输出 JSON: {"summary": "...", "key_points": [...]}
user_template: "请总结以下文档：\n\n{document}"
model: "gpt-4o-mini"
temperature: 0.1
max_tokens: 500

import yaml

class PromptRegistry:
    def __init__(self, path="prompts/"):
        self.templates = {}
        # 启动时加载所有版本
        for f in Path(path).glob("*.yaml"):
            self.templates[f.stem] = yaml.safe_load(f.read_text())

    def render(self, name, **vars):
        tpl = self.templates[name]
        return {
            "system": tpl["system"],
            "user": tpl["user_template"].format(**vars),
            "model": tpl["model"],
            "temperature": tpl["temperature"],
            "max_tokens": tpl["max_tokens"],
        }

好处：

• 改 Prompt 不用发版，改 YAML 重启即可。
• 可以同时存 v1/v2，按用户灰度切流量。
• 每次模型调用记录用了哪个 Prompt 版本，出问题能复现。

---

15.3 流式输出：别让用户干等 10 秒

长文本生成时，用户盯着空白转圈是体验灾难。必须用 SSE（Server-Sent Events）流式吐字：

# FastAPI 后端：流式转发模型输出
from fastapi.responses import StreamingResponse
import openai

@app.get("/chat/stream")
async def chat_stream(q: str):
    async def gen():
        stream = await openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": q}],
            stream=True,
        )
        async for chunk in stream:
            delta = chunk.choices[0].delta.content
            if delta:
                # SSE 格式：data: xxx\n\n
                yield f"data: {delta}\n\n"
        yield "data: [DONE]\n\n"
    return StreamingResponse(gen(), media_type="text/event-stream")

// 前端接收流式输出
const evtSource = new EventSource(`/chat/stream?q=${encodeURIComponent(query)}`);
evtSource.onmessage = (e) => {
  if (e.data === "[DONE]") { evtSource.close(); return; }
  answerBox.textContent += e.data;  // 逐字追加
};

体验差距：非流式等 8 秒出整段 vs 流式 0.5 秒开始吐字，感知延迟差一个量级。

---

15.4 错误重试与降级：模型 API 不是 100% 可靠

生产环境大模型 API 会遇到：限流（429）、超时、500、网络抖动。不能裸调：

import tenacity
import openai

@tenacity.retry(
    stop=tenacity.stop_after_attempt(3),
    wait=tenacity.wait_exponential(multiplier=1, min=2, max=10),
    retry=tenacity.retry_if_exception_type((
        openai.RateLimitError,
        openai.APIConnectionError,
        openai.InternalServerError,
    )),
)
def call_llm(messages, **kwargs):
    return openai.chat.completions.create(
        model=kwargs.get("model", "gpt-4o-mini"),
        messages=messages,
        timeout=30,
    )

def safe_chat(messages):
    try:
        resp = call_llm(messages)
        return resp.choices[0].message.content
    except Exception as e:
        # 降级：重试三次还失败，返回兜底文案
        log.error(f"LLM failed: {e}")
        return "抱歉，AI 服务暂时繁忙，请稍后重试。您也可以拨打人工客服电话。"

关键策略：

• 指数退避：等 2s→4s→8s，别立刻重试打爆。
• 只重试可恢复错误（限流、网络、500），参数错（400）重试也没用。
• 降级兜底：最终失败了要给用户一个能用的回复，不能白屏。

---

15.5 Token 计量与成本熔断

大模型按 Token 计费，客户最怕账单爆炸。必须做计量和预算控制：

class LLMGateway:
    def __init__(self):
        self.monthly_budget = 1000.0   # 月度预算 $1000
        self.spent = load_from_db("llm_spending")

    def before_call(self, estimated_tokens):
        cost_est = estimated_tokens * 0.00001   # 粗估
        if self.spent + cost_est > self.monthly_budget:
            raise BudgetExceeded("月度 LLM 预算已用完，请联系管理员充值")

    def after_call(self, usage):
        cost = usage.prompt_tokens * 0.000005 + usage.completion_tokens * 0.000015
        self.spent += cost
        save_to_db("llm_spending", self.spent)
        # 每天打报表

三个必做：

1. 每次调用记录 token 数和成本，按用户/部门/功能维度统计。
2. 设月度/每日预算阈值，超了自动降级到便宜模型或拒绝服务。
3. 异常用量告警：某个用户一小时调了 10 万次，可能是 bug 或被刷。

---

15.6 输出解析与校验：模型输出不一定是合法 JSON

你让模型输出 JSON，它可能输出带 markdown 代码块、前后有废话、字段缺漏。必须做健壮的解析：

import json
import re
from pydantic import BaseModel, ValidationError

class Summary(BaseModel):
    summary: str
    key_points: list[str]

def parse_llm_json(raw: str) -> dict:
    # 1. 去掉 ```json ... ``` 包裹
    m = re.search(r"```(?:json)?\s*(.*?)\s*```", raw, re.DOTALL)
    if m: raw = m.group(1)
    # 2. 找第一个 { 到最后一个 }
    start = raw.find("{")
    end = raw.rfind("}")
    if start == -1 or end == -1:
        raise ValueError("模型输出里没有 JSON")
    return json.loads(raw[start:end])

def safe_summarize(doc: str) -> Summary:
    raw = call_llm([
        {"role": "system", "content": "输出 JSON 格式..."},
        {"role": "user", "content": doc},
    ])
    try:
        data = parse_llm_json(raw)
        return Summary(**data)
    except (json.JSONDecodeError, ValidationError) as e:
        log.warning(f"LLM JSON 解析失败: {e}, raw={raw[:200]}")
        # 降级：纯文本返回，至少别崩
        return Summary(summary=raw[:500], key_points=[])

---

15.7 动手练习（可折叠答案）

练习 1：把一个硬编码 Prompt 的脚本改成配置化 Prompt 注册中心

答案思路：

1. 把 Prompt 文本抽成 YAML 文件，按功能命名（如 summarize_v1.yaml）。
2. 启动时加载所有 YAML 到字典。
3. 业务代码通过名字 + 变量调用 render()。
4. Prompt 里的 model、temperature、max_tokens 也一起配置化。

练习 2：用户反馈"AI 回复要等好久才出来"，你怎么优化？

答案思路：

1. 先上流式输出（SSE），让用户立刻看到字在动。
2. 换更快的模型（如从 gpt-4o 换到 gpt-4o-mini，或本地 vLLM）。
3. 缩短 Prompt：RAG 检索不要塞太多片段，控制在合理 token 数。
4. 设 max_tokens，避免模型写超长废话。
5. 非实时场景（如文档摘要）改异步任务模式，生成完通知用户。

练习 3：客户说"昨天 AI 花了 5000 块"，你怎么排查和预防？

答案思路：

1. 查用量日志：按用户/接口/时间维度拆，看是谁在什么时候调了多少。
2. 常见原因：死循环里反复调 LLM、RAG 每次塞了超长文档、某个 bug 导致无限重试。
3. 预防：加每用户每分钟限流、月度预算熔断、异常调用量告警。
4. 优化：长文档先做摘要再喂给模型、用更便宜的模型做简单任务。

---

15.8 本章面试题

1. "生产级 LLM 应用和 Demo 的区别？"→ 答：Demo 只调通 API；生产要做 Prompt 版本管理、流式输出、错误重试降级、Token 计量和预算熔断、输出校验、监控告警、回归评测。
2. "模型输出不是预期格式怎么办？"→ 答：Prompt 里明确格式要求 + 低 temperature；代码层做健壮解析（去 markdown、抽 JSON）；用 Pydantic 校验字段；解析失败降级到纯文本，不让接口崩。
3. "LLM 成本失控怎么治理？"→ 答：全链路 token 计量 → 按维度统计 → 设预算阈值熔断 → 异常用量告警 → 简单任务用便宜模型 → Prompt 优化减少冗余 token。
4. "为什么要做流式输出？"→ 答：长文本生成非流式时用户要干等数秒到十几秒；流式（SSE）逐字返回，感知延迟大幅降低，体验天差地别。

---

15.9 小结

• LLM 应用工程化 = Prompt 配置化 + 流式输出 + 重试降级 + 成本管控 + 输出校验。
• Prompt 不要写死在代码里，用 YAML 管理版本，支持灰度和 A/B。
• 大模型 API 不可靠，必须加重试（指数退避）和降级兜底。
• Token 就是钱，每次调用都要计量，月度预算熔断是底线。
• 模型输出不可控，代码要做防御性解析和校验。