速通 AI（十）：生产环境工程化

上一篇：速通 AI（九）：RAG 与 Agent 评估体系 — RAGAS 框架、Agent 评估五维度、幻觉检测。

目标：掌握 LLM 应用从 Demo 到生产的关键工程化技术——可观测性、安全防护、错误处理、成本优化。

前置要求：第七篇（RAG 流程）、第八篇（Agent 架构、安全设计概念）

本阶段知识依赖图

flowchart TD
    A["RAG + Agent 系统"] --> B["可观测性<br/>Langfuse / Tracing"]
    A --> C["安全防护<br/>Prompt Injection 防御"]
    A --> D["错误处理<br/>重试/降级/熔断"]
    A --> E["成本优化<br/>缓存/模型路由/SSE"]
    A --> F["工具系统设计<br/>数量管理/安全/版本"]
    B --> G["生产级 LLM 应用"]
    C --> G
    D --> G
    E --> G
    F --> G

1. 从 Demo 到生产的鸿沟

很多人的 RAG/Agent 项目停留在 Jupyter Notebook 阶段。但面试官想听到的是你如何把它做成一个可靠的线上服务。

面试官判断"有没有上过线"的关键信号：能不能说清系统的评估指标、能不能描述出错时的处理方案、能不能给出成本优化的具体数字。

2. LLM 应用的可观测性

2.1 为什么需要 Tracing

一个 RAG + Agent 请求可能涉及：查询改写（LLM 调用 1）→向量检索（数据库查询）→Rerank（模型调用）→Agent 规划（LLM 调用 2）→工具调用（API 请求）→最终生成（LLM 调用 3）。

如果最终结果不好，你怎么知道是哪一步出了问题？你需要 Tracing：记录每一步的输入、输出、耗时、Token 数。

2.2 Langfuse 实战

Langfuse 是开源的 LLM 可观测性平台，支持自部署。

from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse()

@observe()  # 自动追踪这个函数的执行
def rag_pipeline(query: str):

    @observe()
    def retrieve(query):
        docs = vector_store.search(query, top_k=5)
        return docs

    @observe()
    def rerank(query, docs):
        scored = reranker.rank(query, docs)
        return scored[:3]

    @observe()
    def generate(query, context):
        response = llm.invoke(
            f"基于以下上下文回答问题：\n{context}\n\n问题：{query}"
        )
        return response

    docs = retrieve(query)
    ranked_docs = rerank(query, docs)
    context = "\n".join([d.content for d in ranked_docs])
    answer = generate(query, context)

    return answer

在 Langfuse 的 Web UI 中你可以看到：每次请求的完整调用链、每步的输入/输出/耗时、Token 消耗和成本统计、用户反馈（点赞/点踩）。

Langfuse vs LangSmith：

3. 安全防护工程实现

第八篇介绍了 Agent 安全的威胁类型（Prompt Injection、工具滥用、无限循环、信息泄露），本节聚焦工程实现。

3.1 Prompt Injection 四层防御

Prompt Injection 是 LLM 应用最大的安全威胁。攻击者通过精心构造的输入，让模型忽略系统指令，执行攻击者的意图。

第 1 层：输入过滤

• 关键词检测：检测"忽略指令"、“ignore previous"等模式
• LLM Guard：专门的安全检测模型，判断输入是否包含攻击意图
• 输入长度限制：防止超长注入攻击

第 2 层：Prompt 加固

• 分隔符隔离：用 XML 标签或特殊分隔符区分系统指令和用户输入
• 指令重复：在 prompt 末尾重复核心约束（“你是一个客服助手，只回答产品相关问题”）
• 角色锚定：强化系统角色设定

第 3 层：输出过滤

• 检测模型输出是否包含敏感信息（PII、密码等）
• 检测是否偏离了预期的回答范围
• NeMo Guardrails：NVIDIA 的输出约束框架

第 4 层：监控告警

• 记录所有异常请求
• 定期审查高频攻击模式
• 告警通知运维团队

3.2 NeMo Guardrails 示例

NeMo Guardrails 用 Colang 语言定义对话规则（以下为 Colang 1.0 语法示例，NeMo Guardrails v0.5.0+ 已默认使用 Colang 2.0，语法有差异）：

define user ask about hacking
  "如何入侵系统"
  "怎么获取别人密码"
  "教我黑客技术"

define flow
  user ask about hacking
  bot refuse to help with hacking
  bot inform cannot assist with harmful activities

4. 生产错误处理模式

Demo 阶段出错了就重启。生产阶段需要优雅地处理各种失败——这是面试官判断你有没有上过线的关键信号。

4.1 重试与退避（Retry with Backoff）

为什么需要指数退避？LLM API 限流（Rate Limit）时，立即重试会被继续拒绝。第 1 次等 1 秒，第 2 次等 2 秒，第 3 次等 4 秒——给服务端恢复的时间。加随机抖动（jitter）防止所有客户端同时重试（惊群效应）。

import tenacity

@tenacity.retry(
    stop=tenacity.stop_after_attempt(4),              # 最多尝试 4 次（含首次，最多重试 3 次）
    wait=tenacity.wait_exponential(min=1, max=30),     # 指数退避：1s, 2s, 4s...
    retry=tenacity.retry_if_exception_type(            # 只对特定异常重试
        (TimeoutError, RateLimitError, ServiceUnavailableError)
    ),
    before_sleep=lambda retry_state: logger.warning(   # 重试前记日志
        f"LLM 调用失败，第 {retry_state.attempt_number} 次重试"
    )
)
def call_llm(prompt: str, model: str = "deepseek-v4") -> str:
    return llm.invoke(prompt, model=model)

4.2 模型降级链（Fallback Chain）

async def call_with_fallback(prompt: str) -> str:
    models = [
        ("deepseek-v4", "主力模型"),
        ("deepseek-v3", "备用模型"),
        ("qwen-2.5-72b", "兜底模型"),
    ]

    for model_name, desc in models:
        try:
            return await asyncio.to_thread(call_llm, prompt, model=model_name)
        except Exception as e:
            logger.warning(f"{desc}({model_name})失败: {e}，尝试下一个")

    return "抱歉，服务暂时不可用，请稍后重试。"  # 所有模型都失败的兜底

面试话术：“我们的模型调用有三层降级：首选 DeepSeek-V4，超时或限流自动切 V3，如果 V3 也不行，降级到本地部署的 Qwen。这样保证了 99.9% 的可用性。”

4.3 熔断器（Circuit Breaker）

如果下游 LLM 服务持续故障，每个请求都要等超时后才失败→大量请求堆积→线程池耗尽→你的服务也跟着挂。熔断器解决这个问题。

stateDiagram-v2
    [*] --> CLOSED
    CLOSED --> OPEN : 失败率超过阈值
    OPEN --> HALF_OPEN : 冷却时间到
    HALF_OPEN --> CLOSED : 探测成功
    HALF_OPEN --> OPEN : 探测失败

• CLOSED（正常）：正常转发请求，统计失败率
• OPEN（直接拒绝）：直接返回降级结果，不调用下游（保护系统）
• HALF-OPEN（试探性放行）：定时放行少量请求，测试下游是否恢复

实际场景：如果某个 LLM API 供应商宕机了，熔断器自动切断请求，走降级链到备用模型，同时后台定时探测是否恢复。恢复后自动切回主力模型。

4.4 超时与取消

每一步都要独立设超时，而非只在最外层设一个总超时——这样你能知道"到底是哪一步超时了”。

import asyncio

async def rag_with_timeout(query: str, timeout: float = 30.0) -> str:
    try:
        # rag_pipeline 是同步函数，需要用 asyncio.to_thread 包装
        result = await asyncio.wait_for(
            asyncio.to_thread(rag_pipeline, query),
            timeout=timeout
        )
        return result
    except asyncio.TimeoutError:
        logger.error(f"RAG 管道超时 ({timeout}s): {query}")
        return await asyncio.to_thread(simple_llm_answer, query)  # 降级为不检索直接回答

5. 成本优化

可观测性让你"看到"问题，安全防护让你"阻止"攻击，错误处理让你"容忍"故障。但还有一个生产级问题不可忽视——成本。LLM 应用的成本大头是 Token 消耗。以下是四种核心优化策略。

5.1 Prompt 缓存

相同或相似的请求直接返回缓存结果。工具：GPTCache、Redis + 向量相似度。根据请求重复率，可节省 30-50% 的 API 调用（具体取决于请求模式）。

5.2 模型路由（Model Router）

简单问题用小模型（如 8B），复杂问题用大模型（如 70B）。先用小模型判断问题难度，再路由到合适的模型。当简单问题占多数时，可显著降低成本（节省幅度取决于简单/复杂问题的比例）。

5.3 Token 优化

• 压缩上下文：摘要替代原文
• 动态截断：只保留最相关的 chunk
• 减少系统 prompt 长度

5.4 流式输出（SSE）

不等全部生成完再返回，用 Server-Sent Events 逐 token 推送。用户体验更好，首 token 延迟更低。

from fastapi import FastAPI
from fastapi.responses import StreamingResponse

app = FastAPI()

@app.post("/chat")
async def chat(query: str):
    async def generate():
        async for chunk in llm.astream(query):
            yield f"data: {chunk.content}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

6. 大模型部署方案选型

vLLM、TGI、llama.cpp、SGLang 是当前主流的四种部署方案：

Prompt Caching 原理：当多个请求共享相同的 System Prompt 前缀时，前缀的 KV Cache 可以复用，不需要重复计算。例如 100 个用户共享同一个 1000 token 的 System Prompt，Prompt Caching 可以将前缀计算从 100 次降到 1 次。

量化方案对比（与第八篇 NF4 量化互补）：

选择建议：微调用 NF4（QLoRA）；部署推理用 AWQ（精度好）或 GPTQ（速度快）；精度敏感场景用 INT8。

7. 知识库工程

知识库动态与持续更新

RAG 知识库不是一次性构建就完事的——文档会更新、新增、删除。知识库的持续更新是生产环境的常见需求。

增量索引策略：

1. 变更检测：监控源文档的修改时间/哈希值，只重新索引变更的文档
2. 增量写入：新文档直接写入向量数据库，不影响已有数据
3. 版本管理：为每个 chunk 附带文档版本号，查询时可按版本过滤
4. 删除处理：文档删除时，同步删除对应的向量（向量数据库需支持按 metadata 过滤）

挑战：Embedding 模型更新后，新旧向量的语义空间不一致——要么全量重新索引（成本高），要么用向量空间对齐技术（如线性变换）将旧向量映射到新空间。

不同来源文档的知识冲突

当多个来源的文档对同一事实有不同描述时，RAG 系统需要决定"信谁"。

冲突解决策略：

实际工程中，通常将冲突解决逻辑放在 Reranker 阶段——在重排时考虑来源可信度和时间戳，让最终进入 LLM 上下文的是最可靠的信息。

8. 工具系统生产设计

前面几节讨论了 LLM 调用层面的工程化。但 Agent 的核心能力来自工具调用——工具系统的设计直接影响 Agent 的准确率和安全性。

8.1 工具数量管理

问题：工具太多时 LLM 选择准确率下降。

以上为典型量级估计（趋势来自 Gorilla、ToolLLM 等研究），具体数值因模型和工具集不同会有差异。

动态工具加载三种策略：

1. 意图分类→按类别加载：用户输入→分类器判断意图类别（天气/数据库/文件/…）→只加载该类别的 3-5 个工具
2. 两阶段选择：第一阶段用 Embedding 相似度粗筛（从 50 个中选 top-10），第二阶段将 10 个工具描述送入 LLM 精选 3 个
3. 工具分组+路由：Agent A（通用助手）→基础工具、Agent B（数据分析）→数据库工具、Agent C（代码助手）→代码执行工具→由路由 Agent 分发

8.2 工具执行安全

三层安全机制：

8.3 工具版本管理与热更新

生产中的工具需要版本管理：version 1.0.0→1.1.0（新增参数）→旧版本的 Agent 还在运行，不能直接替换。

策略：语义化版本（major.minor.patch）→向后兼容（新参数用默认值）→灰度发布（新版本先给 10% 流量测试）→回滚机制（发现问题秒级回滚）。

WebSocket vs SSE 通信对比

LLM 应用中，SSE 是最常用的流式输出方案——服务端逐 token 推送，客户端实时显示。WebSocket 适用于需要双向通信的场景（如语音 Agent，客户端需要实时发送音频流）。

工具调用容错

核心原则：不要让工具调用失败直接中断整个 Agent 流程。将错误信息反馈给 LLM，让它决定是重试、换工具还是跳过。

推荐补充资源

模块小结：生产环境工程化

下一篇预告：本文完成了从 Demo 到生产的工程化全链路。建议结合速通 AI 面试八股文检验掌握程度，面试前重点复习自己答不上来的题目。