构建生产级 LLM 应用:把前 12 节拧成一首歌

构建生产级 LLM 应用:把前 12 节拧成一首歌

本节摘要:你已经分别搭过提示、嵌入、RAG、函数调用、缓存层、护栏——像练吉他和弦音阶却从不弹一首歌。本节就是那首歌:把第 0112 节的每个组件,拧成一个生产级服务。不是玩具,不是 demo,是一个能扛真实流量、优雅失败、流式输出 token、追踪成本、活过首批 1 万用户的系统。本节带你吃透生产架构七件套(API 网关→提示路由→语义缓存→LLM 调用→护栏→评估日志→成本追踪)的接线;流式输出为何把感知延迟降 90%(SSE 首 token 200500ms);三层错误处理(API 失败、模型失败、应用失败)各自的恢复策略与降级模型链;可观测三支柱(结构化日志、OpenTelemetry 追踪、五项指标仪表盘);生产环境 A/B 测试(影子模式 + 按百分比灰度);以及一份 15 项部署前清单——任何一项未打勾都不许上线。

对应原课程:Phase 11 · Lesson 13 · production-app(原英文 phases/11-llm-engineering/13-production-app/docs/en.md)。本节为 Phase 11 的 capstone(收官节)。

学习目标

阅读完本节,你应当能够:

1. 把第 11 章所有组件(提示、RAG、函数调用、缓存、护栏)拧成一个生产级服务。
2. 实现流式 token 投递、优雅错误处理、请求超时管理。
3. 给应用装上可观测性:请求日志、成本追踪、延迟分位、错误率仪表盘。
4. 用健康检查、限流、厂商宕机降级策略部署应用。

一、问题与直觉

做一个 LLM 功能要一个下午,交付一个 LLM 产品要几个月。差距不在智能,在基础设施。你的原型调 OpenAI、拿到响应、打印出来,笔记本上跑得好好的。然后现实来了:

• 用户发来 5 万 token 的文档,你的上下文窗口溢出。
• 两个用户 4 秒间隔问同一问题,你为两者都付钱。
• API 在凌晨 2 点返回 500,你的服务崩了。
• 用户让模型生成 SQL,模型输出 DROP TABLE users。
• 月账单 12000 美元,你不知道哪个功能烧的。
• 平均响应 8 秒,用户 3 秒就走了。

今天每个生产 LLM 应用——Perplexity、Cursor、ChatGPT、Notion AI——都解决过这些问题。不是靠提示更聪明,而是靠工程更严谨。

本节是收官:你将构建一个完整的生产 LLM 服务,整合提示管理(L0102)、嵌入与向量检索(L0407)、函数调用(L09)、评估(L10)、缓存(L11)、护栏(L12)、流式、错误处理、可观测、成本追踪。一个服务,每个组件都接上。

生产架构

每个严肃的 LLM 应用都遵循同一流程,细节有别,结构不变。

请求经 API 网关进入(处理鉴权与限流);输入护栏在提示路由选模板前检查注入与违禁内容;语义缓存查最近是否答过类似问题;缓存未中则带流式调 LLM;输出护栏校验响应;评估日志记录质量指标;成本追踪给每个 token 计费;响应流式回客户端。七个组件,每个都是你已完成的一节,工程在接线。

技术栈

流式:为什么重要

一个 500 输出 token 的 GPT-5 响应要 38 秒生成完。不流式,用户盯着转圈整段时间;流式,首 token 200500ms 就到。总时间一样,感知延迟降 90%。

三种流式协议:

SSE 是默认选择。OpenAI、Anthropic、Google 都通过 SSE 流式。你的服务器从 LLM API 收 chunk、作为 SSE 事件转发给客户端,客户端用 EventSource(浏览器)或 httpx(Python)消费流。

错误处理:三层

生产 LLM 应用以三种截然不同的方式失败,各需不同恢复策略。

第 1 层:API 失败。 LLM 厂商返回 429(限流)、500(服务器错误)或超时。解法:带抖动的指数退避。从 1 秒起、每次翻倍、加随机抖动防惊群,最多 3 次重试。

尝试 1: 立即
尝试 2: 1s + random(0, 0.5s)
尝试 3: 2s + random(0, 1.0s)
尝试 4: 4s + random(0, 2.0s)
放弃: 返回降级响应
​

第 2 层:模型失败。 模型返回畸形 JSON、幻觉函数名、产出校验失败的输出。解法:带纠错提示的重试,把错误纳入重试消息让模型自纠正。

第 3 层:应用失败。 下游服务不可达、向量库慢、护栏抛异常。解法:优雅降级。RAG 上下文不可用就跳过,缓存挂了就绕过。永远不让次要系统搞垮主流程。

降级模型链。主模型不可用时,沿链下探:

claude-sonnet-5 → gpt-4o → gpt-4o-mini → 缓存响应 → "服务暂时不可用"
​

每一步以质量换可用性,用户总能拿到东西。

可观测性:测什么

你改善不了你看不见的。每个生产 LLM 应用需要可观测三支柱。

结构化日志。 每个请求产出一条 JSON 日志:请求 ID、用户 ID、提示模板名、所用模型、输入 token、输出 token、延迟(ms)、缓存命中/未中、护栏通过/失败、成本(美元)、任何错误。

追踪(Tracing)。 单个用户请求触及 5~8 个组件。OpenTelemetry 追踪让你看完整旅程:嵌入花了多久?缓存命中没?LLM 调用多长?护栏加了多少延迟?没追踪,调试生产问题是猜。

指标仪表盘。 每个 LLM 团队盯的五个数:

生产环境 A/B 测试提示

提示不是「能跑」就算完,是「有数据证明它胜过替代方案」才算完。

影子模式(Shadow Mode)。 在 100% 流量上跑新提示,但只记录结果、不展示给用户。与当前提示比质量指标,零用户风险、满数据。

百分比灰度。 把 10% 流量导到新提示,盯指标;质量稳就升到 25%、50%、100%;质量掉就即时回滚。

用用户 ID 的确定性哈希,而非随机选择——这保证每个用户在同一实验里跨请求体验一致。

真实架构示例

Perplexity。 用户查询进入,搜索引擎检索 10~20 个网页,网页分块、嵌入、重排,top 5 块成 RAG 上下文。LLM 带引用生成答案,实时流式回。两个模型:快的做查询改写,强的做答案合成。估计日查询 5000 万+。

Cursor。 打开的文件、周边文件、最近编辑、终端输出组成上下文。提示路由决定:小模型做自动补全(Cursor-small,~20ms),大模型做聊天(Claude Sonnet 4.6/GPT-5,~3s)。上下文激进压缩——只相关代码段而非整文件。代码库嵌入提供长程上下文。推测式编辑流式 diff 而非整文件。MCP 集成让第三方工具无需逐工具改代码即可插入。

ChatGPT。 插件、函数调用、MCP 服务器让模型访问网页、跑代码、生图、查数据库。路由层决定调用哪些能力。Memory 跨会话持久化用户偏好。系统提示是 1500+ token 的行为规则,靠提示缓存。多模型服务不同功能:GPT-5 聊天、GPT-Image 出图、Whisper 语音、o4-mini 深度推理。

扩展

关键扩展模式:

• 到处异步。 永不阻塞 Web 服务器线程在 LLM 调用上,用 asyncio 与 httpx.AsyncClient。
• 基于队列的处理。 非实时任务(摘要、分析)推进队列(Redis、SQS)用 worker 处理,返回任务 ID 让客户端轮询。
• 连接池。 复用与 LLM 厂商的 HTTP 连接,每次新建 TLS 连接加 100~200ms。
• 水平扩展。 LLM 应用是 I/O 密集而非 CPU 密集,单个异步服务器扛 100+ 并发请求,扩展服务器而非核心。

成本预估

上线前估月成本,这张表决定你的商业模式成不成立。

月 LLM 成本:

• 输入:32500 × 1500 × 30 / 100万 × 2.5 = 3656 美元
• 输出:32500 × 400 × 30 / 100万 × 10 = 3900 美元
• 合计:7556 美元/月(缓存省约 4070 美元/月)

不缓存同样流量要 11625 美元/月。35% 命中率省 35% LLM 成本——这就是第 11 节存在的理由。

部署清单

15 项,每项打勾才许上线。

二、从零实现

这是 capstone,一个文件,每个组件接上。代码构建一个完整生产 LLM 服务:FastAPI 服务器(健康检查 + CORS)、带版本与 A/B 的提示模板管理、基于嵌入余弦的语义缓存、输入输出护栏、带流式(SSE)的模拟 LLM 调用、带抖动指数退避与降级模型链、按请求与汇总的成本追踪、带请求 ID 的结构化日志、评估日志。完整代码见原课程 code/production_app.py,这里给出关键骨架。

步骤 1:核心基础设施

配置、日志、每个组件都依赖的数据结构。

import asyncio, hashlib, json, math, os, random, re, time, uuid
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum

class ModelName(Enum):
    CLAUDE_SONNET = "claude-sonnet-5"
    GPT_4O = "gpt-4o"
    GPT_4O_MINI = "gpt-4o-mini"

MODEL_PRICING = {
    ModelName.CLAUDE_SONNET: {"input": 3.00, "output": 15.00},
    ModelName.GPT_4O:        {"input": 2.50, "output": 10.00},
    ModelName.GPT_4O_MINI:   {"input": 0.15, "output": 0.60},
}
PRIMARY = ModelName.CLAUDE_SONNET
FALLBACK_CHAIN = [PRIMARY] + [m for m in ModelName if m is not PRIMARY]

@dataclass
class RequestLog:
    request_id: str; user_id: str; timestamp: str
    prompt_template: str; prompt_version: str; model: str
    input_tokens: int; output_tokens: int; latency_ms: float
    cache_hit: bool; guardrail_input_pass: bool; guardrail_output_pass: bool
    cost_usd: float; error: str | None = None

@dataclass
class CostTracker:
    total_cost_usd: float = 0.0
    total_requests: int = 0
    total_cache_hits: int = 0
    cost_by_user: dict = field(default_factory=lambda: defaultdict(float))
    def record(self, user_id, cost):
        self.total_cost_usd += cost
        self.total_requests += 1
        self.cost_by_user[user_id] += cost
    def summary(self):
        return {"total_requests": self.total_requests,
                "total_cost_usd": round(self.total_cost_usd, 6),
                "cache_hit_rate_pct": round(self.total_cache_hits/max(self.total_requests,1)*100, 2)}
​

步骤 2:提示管理

带 A/B 测试的版本化提示模板,路由按请求上下文与实验分桶选模板。

@dataclass
class PromptTemplate:
    name: str; version: str; template: str
    model: ModelName = ModelName.GPT_4O
    max_output_tokens: int = 1024

PROMPT_TEMPLATES = {
    "general_chat": {
        "v1": PromptTemplate("general_chat", "v1",
            "你是一个有用的 AI 助手。清晰简洁地回答用户问题。\n\n用户问题:{query}"),
        "v2": PromptTemplate("general_chat", "v2",
            "你是一个给出精确、可操作答案的 AI 助手。不确定就说不确定,绝不编造。\n\n问题:{query}\n\n答案:"),
    },
    "rag_answer": {
        "v1": PromptTemplate("rag_answer", "v1",
            "只根据提供的上下文回答问题。上下文不含答案就说『我没有足够的信息』。\n\n上下文:\n{context}\n\n问题:{query}\n\n答案:",
            max_output_tokens=512),
    },
}

AB_EXPERIMENTS = {
    "general_chat_v2_test": {"template": "general_chat", "control": "v1",
                             "variant": "v2", "traffic_pct": 10},
}

def select_prompt(template_name, user_id, variables):
    versions = PROMPT_TEMPLATES[template_name]
    version = "v1"
    for exp_name, exp in AB_EXPERIMENTS.items():
        if exp["template"] == template_name:
            # 确定性哈希分桶:同一用户在同一实验里始终拿同一版本
            bucket = int(hashlib.md5(f"{user_id}:{exp_name}".encode()).hexdigest(), 16) % 100
            version = exp["variant"] if bucket < exp["traffic_pct"] else exp["control"]
            break
    tpl = versions[version]
    return tpl, tpl.template.format(**variables)
​

步骤 3:语义缓存

基于嵌入的缓存,匹配语义相近的查询——两个措辞不同、含义相同的提问会命中。

def simple_embedding(text, dim=64):
    """演示用哈希嵌入(生产换成真实嵌入 API)。"""
    h = hashlib.sha256(text.lower().strip().encode()).hexdigest()
    raw = [int(h[i:i+2], 16)/255.0 for i in range(0, dim*2, 2)]
    while len(raw) < dim:
        h = hashlib.sha256(f"{text}_{len(raw)}".encode()).hexdigest()
        raw.extend(int(h[i:i+2],16)/255.0 for i in range(0, (dim-len(raw))*2, 2))
    raw = raw[:dim]
    norm = math.sqrt(sum(x*x for x in raw))
    return [x/norm if norm else 0.0 for x in raw]

def cosine_similarity(a, b):
    na, nb = math.sqrt(sum(x*x for x in a)), math.sqrt(sum(x*x for x in b))
    return sum(x*y for x,y in zip(a,b))/(na*nb) if na and nb else 0.0

class SemanticCache:
    def __init__(self, similarity_threshold=0.92, max_entries=10000, ttl_seconds=3600):
        self.threshold, self.max_entries, self.ttl = similarity_threshold, max_entries, ttl_seconds
        self.entries, self.hits, self.misses = [], 0, 0
    def get(self, query):
        q_emb = simple_embedding(query)
        now = time.time()
        best_score, best_entry = 0.0, None
        for e in self.entries:
            if now - e["timestamp"] > self.ttl:
                continue
            s = cosine_similarity(q_emb, e["embedding"])
            if s > best_score:
                best_score, best_entry = s, e
        if best_entry and best_score >= self.threshold:
            self.hits += 1
            return {"response": best_entry["response"], "similarity": round(best_score, 4)}
        self.misses += 1
        return None
    def put(self, query, response):
        if len(self.entries) >= self.max_entries:
            self.entries.sort(key=lambda e: e["timestamp"])
            self.entries = self.entries[len(self.entries)//4:]   # 踢最旧 1/4
        self.entries.append({"query": query, "embedding": simple_embedding(query),
                             "response": response, "timestamp": time.time()})
​

步骤 4:护栏

输入校验在 LLM 看到前拦注入与 PII,输出校验在用户看到前拦不安全内容,两道墙,无一漏检。

INJECTION_PATTERNS = [
    r"ignore\s+(all\s+)?previous\s+instructions",
    r"you\s+are\s+now\s+DAN",
    r"<\s*system\s*>",
    r"jailbreak",
]
PII_PATTERNS = {
    "ssn": r"\b\d{3}-\d{2}-\d{4}\b",
    "card": r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b",
    "email": r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b",
}
BANNED_OUTPUT = [r"(?i)(DROP|DELETE|TRUNCATE)\s+TABLE", r"(?i)rm\s+-rf\s+/", r"(?i)__import__\s*\("]

@dataclass
class GuardrailResult:
    passed: bool
    blocked_reason: str | None = None
    pii_detected: list = field(default_factory=list)
    modified_text: str | None = None

def check_input_guardrails(text):
    for p in INJECTION_PATTERNS:
        if re.search(p, text, re.IGNORECASE):
            return GuardrailResult(passed=False, blocked_reason="疑似提示注入")
    pii_found = [t for t, p in PII_PATTERNS.items() if re.search(p, text)]
    if pii_found:
        redacted = text
        for t, p in PII_PATTERNS.items():
            redacted = re.sub(p, f"[{t.upper()}_REDACTED]", redacted)
        return GuardrailResult(passed=True, pii_detected=pii_found, modified_text=redacted)
    return GuardrailResult(passed=True)

def check_output_guardrails(text):
    for p in BANNED_OUTPUT:
        if re.search(p, text):
            return GuardrailResult(passed=False, blocked_reason="响应含潜在不安全内容")
    return GuardrailResult(passed=True)
​

步骤 5:带重试与流式的 LLM 调用

核心 LLM 接口:失败时带抖动指数退避,沿模型链降级,支持逐 token 流式。

def estimate_tokens(text):
    return max(1, len(text.split()) * 4 // 3)

def calculate_cost(model, in_tok, out_tok):
    p = MODEL_PRICING.get(model, MODEL_PRICING[ModelName.GPT_4O])
    return round(in_tok/1_000_000*p["input"] + out_tok/1_000_000*p["output"], 8)

async def call_llm_with_retry(prompt, model, max_retries=3):
    for attempt in range(max_retries + 1):
        try:
            # 模拟偶发 API 失败(生产里这是真实 HTTP 调用)
            if random.random() < (0.15 if attempt == 0 else 0.05):
                raise ConnectionError(f"{model.value} 返回 500")
            await asyncio.sleep(random.uniform(0.1, 0.3))
            text = f"关于「{prompt[:40]}」,这是模拟响应……"
            return {"text": text, "model": model.value,
                    "input_tokens": estimate_tokens(prompt),
                    "output_tokens": estimate_tokens(text)}
        except (ConnectionError, TimeoutError):
            if attempt < max_retries:
                backoff = min(2**attempt + random.uniform(0, 1), 10)   # 指数 + 抖动
                await asyncio.sleep(backoff)
            else:
                raise
    raise ConnectionError(f"{max_retries} 次重试耗尽")

async def call_with_fallback(prompt, preferred=None):
    """沿降级链尝试,每个模型失败就下一个,用户总能拿到东西。"""
    chain = list(FALLBACK_CHAIN)
    if preferred and preferred in chain:
        chain.remove(preferred); chain.insert(0, preferred)
    last_err = None
    for model in chain:
        try:
            return await call_llm_with_retry(prompt, model)
        except ConnectionError as e:
            last_err = e; continue
    return {"text": "抱歉,暂时无法处理您的请求,请稍后再试。", "model": "fallback",
            "input_tokens": estimate_tokens(prompt), "output_tokens": 20, "error": str(last_err)}

async def stream_response(text):
    """逐 token 流式(模拟 SSE 行为)。"""
    for i, word in enumerate(text.split()):
        yield word if i == 0 else " " + word
        await asyncio.sleep(random.uniform(0.02, 0.08))
​

步骤 6:请求管线

编排器:取一个原始用户请求,过每个组件,返回结构化结果。

class ProductionLLMService:
    def __init__(self):
        self.cache = SemanticCache(similarity_threshold=0.92, ttl_seconds=3600)
        self.cost_tracker = CostTracker()
        self.request_logs, self.eval_results = [], []

    async def handle_request(self, user_id, query, template_name="general_chat", variables=None):
        request_id = str(uuid.uuid4())[:12]
        start = time.time()
        variables = variables or {}; variables["query"] = query
        # 1. 输入护栏
        input_check = check_input_guardrails(query)
        if not input_check.passed:
            return {"request_id": request_id, "blocked": True, "reason": input_check.blocked_reason}
        effective = input_check.modified_text or query
        if input_check.modified_text:
            variables["query"] = effective
        # 2. 语义缓存
        cached = self.cache.get(effective)
        if cached:
            self.cost_tracker.total_cache_hits += 1
            return {"request_id": request_id, "response": cached["response"],
                    "cache_hit": True, "similarity": cached["similarity"],
                    "latency_ms": round((time.time()-start)*1000, 2), "cost_usd": 0.0}
        # 3. 提示路由 + LLM 调用(带降级)
        template, rendered = select_prompt(template_name, user_id, variables)
        result = await call_with_fallback(rendered, template.model)
        # 4. 输出护栏
        output_check = check_output_guardrails(result["text"])
        if not output_check.passed:
            result["text"] = "该响应被安全系统标记,无法提供。"
            result["output_tokens"] = estimate_tokens(result["text"])
        # 5. 成本 + 日志 + 缓存写入
        model = ModelName(result["model"]) if result["model"] != "fallback" else ModelName.GPT_4O_MINI
        cost = calculate_cost(model, result["input_tokens"], result["output_tokens"])
        latency_ms = round((time.time()-start)*1000, 2)
        self.cost_tracker.record(user_id, cost)
        self.cache.put(effective, result["text"])
        self.request_logs.append(RequestLog(
            request_id, user_id, datetime.now(timezone.utc).isoformat(),
            template_name, template.version, result["model"],
            result["input_tokens"], result["output_tokens"], latency_ms,
            False, True, output_check.passed, cost, result.get("error")))
        return {"request_id": request_id, "response": result["text"],
                "model": result["model"], "cache_hit": False,
                "input_tokens": result["input_tokens"], "output_tokens": result["output_tokens"],
                "latency_ms": latency_ms, "cost_usd": cost,
                "pii_detected": input_check.pii_detected}

    def health_check(self):
        return {"status": "healthy", "cache": self.cache.stats(),
                "cost": self.cost_tracker.summary(), "total_requests": len(self.request_logs)}
​

💡 管线顺序是工程声明:输入护栏(防注入/PII)→ 缓存(省钱)→ 提示路由 + LLM(带降级)→ 输出护栏(防有害)→ 成本/日志/缓存写。每一层都在前一层基础上,缺一层就少一道防线。这个顺序是经过 Perplexity/Cursor/ChatGPT 验证的默认形态。

三、框架对比

FastAPI 服务器(生产部署)

demo 跑成脚本;生产里用 FastAPI 包成正经端点。

# from fastapi import FastAPI
# from fastapi.middleware.cors import CORSMiddleware
# from fastapi.responses import StreamingResponse
# from pydantic import BaseModel
# import uvicorn
# app = FastAPI(title="Production LLM Service")
# app.add_middleware(CORSMiddleware, allow_origins=["https://yourdomain.com"], allow_methods=["POST","GET"])
# service = ProductionLLMService()
# class ChatRequest(BaseModel):
#     query: str; user_id: str; template: str = "general_chat"; stream: bool = False
# @app.post("/v1/chat")
# async def chat(req: ChatRequest):
#     if req.stream:
#         result = await service.handle_request(req.user_id, req.query, req.template)
#         async def generate():
#             async for token in stream_response(result["response"]):
#                 yield f"data: {json.dumps({'token': token}, ensure_ascii=False)}\n\n"
#             yield "data: [DONE]\n\n"
#         return StreamingResponse(generate(), media_type="text/event-stream")
#     return await service.handle_request(req.user_id, req.query, req.template)
# @app.get("/health")
# async def health(): return service.health_check()
# @app.get("/v1/costs")
# async def costs(): return service.cost_tracker.summary()
# if __name__ == "__main__":
#     uvicorn.run(app, host="0.0.0.0", port=8000)
​

跑成真实服务器:pip install fastapi uvicorn,访问 http://localhost:8000/docs 看自动生成的 API 文档。

真实 API 集成

把模拟 LLM 调用换成真实厂商 SDK。

# import openai, anthropic
# async def call_openai(prompt, model="gpt-4o"):
#     client = openai.AsyncOpenAI()
#     resp = await client.chat.completions.create(model=model, stream=True,
#         messages=[{"role":"user","content":prompt}])
#     async for chunk in resp:
#         delta = chunk.choices[0].delta.content or ""
#         yield delta
# async def call_anthropic(prompt, model="claude-sonnet-5"):
#     client = anthropic.AsyncAnthropic()
#     async with client.messages.stream(model=model, max_tokens=1024,
#         messages=[{"role":"user","content":prompt}]) as stream:
#         async for text in stream.text_stream:
#             yield text
​

Docker 部署

# FROM python:3.12-slim
# WORKDIR /app
# COPY requirements.txt .
# RUN pip install --no-cache-dir -r requirements.txt
# COPY . .
# EXPOSE 8000
# CMD ["uvicorn", "production_app:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]
​

四个 worker,每个处理异步 I/O。一台 4 worker 的盒子服务 400+ 并发 LLM 请求,因为它们都在等网络 I/O 而非 CPU。LangServe 把 LangChain 的链/agent 直接暴露成符合本节架构的 FastAPI 端点;LangSmith、Helicone、Arize Phoenix 提供托管的可观测 + 评估平台,把本节手写的日志/追踪/指标变成托管服务。

四、可复用产物

本节产出两个可复用文件(位于原课程 outputs/):

• prompt-architecture-reviewer.md:一个元提示,对照生产清单审查任何 LLM 应用的架构。给它你的系统描述,它返回差距分析。
• skill-production-checklist.md:一个决策框架,用于把 LLM 应用交付到生产,覆盖本节每个组件,含具体阈值与通过/失败标准。

Python 代码(code/production_app.py)是一个完整生产服务,把 call_llm_with_retry 里的模拟换成真实 OpenAI/Anthropic SDK、把 FastAPI 注释取消,即可上线;管线、缓存、护栏、成本追踪、日志逻辑均无需修改。

五、练习

1. 加 RAG 集成:建一个带 20 份文档的内存向量库。模板为 rag_answer 时,嵌入查询、找 3 份最相似文档、注入为上下文。测有/无 RAG 上下文时响应质量如何变。把检索延迟与 LLM 延迟分开追踪。

2. 实现真实函数调用:把第 09 节的工具注册表加进服务。用户问需要外部数据的问题(天气、计算、搜索)时,管线应检测、执行工具、把结果纳入提示。给响应加 tools_used 字段。

3. 建成本告警系统:按用户按天追踪成本。用户超 0.50 美元/天就切到 gpt-4o-mini;日总成本超 100 美元就进应急模式:重复查询只返缓存、其余全 gpt-4o-mini、拒超 2000 输入 token 的请求。用模拟流量峰值测。

4. 带回滚的提示版本管理:存所有提示版本(带时间戳)。加一个端点展示每版本质量指标(延迟、用户评分、错误率)。实现自动回滚:若新版本在 100 请求里错误率是上一版本的 2 倍,自动回退。

5. 加 OpenTelemetry 追踪:把每个组件(缓存查找、护栏检查、LLM 调用、成本计算)埋成独立 span,每个 span 记录时长,导出到控制台。展示单个请求的完整 trace,让每个组件对总延迟的贡献可见。

本节要点回顾

1. 做功能一下午,做产品要几个月:差距在基础设施,不在智能——上下文溢出、重复付费、API 500、SQL 注入、账单失控、延迟赶人。
2. 生产架构七件套:API 网关→提示路由→语义缓存→LLM 调用→护栏→评估日志→成本追踪,工程在接线。
3. 流式降感知延迟 90%:SSE 首 token 200~500ms,总时间不变但用户不再盯转圈。
4. 三层错误处理:API 失败带抖动指数退避、模型失败带纠错重试、应用失败优雅降级(永不让次要系统搞垮主流程)。
5. 降级模型链:claude-sonnet→gpt-4o→gpt-4o-mini→缓存→「不可用」,每步以质量换可用。
6. 可观测三支柱:结构化日志(带请求 ID)、OpenTelemetry 追踪(看组件旅程)、五项指标(P50/P99 延迟、缓存命中率、护栏拦截率、每请求成本)。
7. A/B 测试要确定性:影子模式零风险验证、按百分比灰度、用户 ID 哈希分桶保体验一致。
8. 成本靠缓存:35% 命中率省 35% LLM 成本,月账单从 11625 降到 7556 美元。
9. 扩展看 I/O:LLM 应用是 I/O 密集,单异步服务器扛 100+ 并发,扩展服务器而非核心。
10. 15 项部署清单:从 API key 存环境变量到 100 并发负载测试,任何一项未打勾都不许上线。

下一节,我们把视角从「单应用」拉到「跨应用」,讲模型上下文协议 MCP——如何用一套开放标准让工具定义一次、被任何兼容客户端消费,把本节手写的工具注册表升级为可共享的协议层。