9.7 AI 基础设施（LLMOps）

LLMOps 是 AI 应用的「运维层」——可观测性、AI 网关、Prompt 缓存、评估系统，让 AI 应用从 Demo 走向生产。

LLMOps 全景图

┌───────────────────────────────────────────────────┐
│                  AI 应用层                         │
│    Chat / RAG / Agent / 多模态                     │
├───────────────────────────────────────────────────┤
│                LLMOps 基础设施                     │
│                                                    │
│  ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│  │ AI 网关  │ │ 可观测性 │ │Prompt   │ │ 评估    │ │
│  │LiteLLM  │ │LangSmith│ │Caching  │ │ RAGAS   │ │
│  │Portkey  │ │Langfuse │ │         │ │DeepEval │ │
│  └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
├───────────────────────────────────────────────────┤
│               模型提供商                           │
│  OpenAI / Anthropic / Google / DeepSeek / 本地模型  │
└───────────────────────────────────────────────────┘

AI 网关（Gateway）

为什么需要 AI 网关？

没有网关（直连模型 API）：

  App → OpenAI API
  问题：单点故障、无法切换、无统一监控

有了网关：

  App → AI 网关 → OpenAI API
                → Anthropic API（Fallback）
                → 本地模型（降级）
  优势：统一接口、自动容错、成本控制、流量管理

LiteLLM（推荐）

python

# pip install litellm

from litellm import completion

# 统一接口调用任意模型
response = completion(
    model="gpt-4o",
    messages=[{"role": "user", "content": "你好"}]
)

# 切换模型只需改 model 名称
response = completion(
    model="claude-sonnet-4-20250514",
    messages=[{"role": "user", "content": "你好"}]
)

response = completion(
    model="deepseek/deepseek-chat",
    messages=[{"role": "user", "content": "你好"}]
)

Fallback 容错链：

python

from litellm import Router

# 配置多模型路由 + Fallback
router = Router(
    model_list=[
        {
            "model_name": "main-model",
            "litellm_params": {
                "model": "gpt-4o",
                "api_key": "sk-xxx"
            }
        },
        {
            "model_name": "main-model",  # 同名 = Fallback
            "litellm_params": {
                "model": "claude-sonnet-4-20250514",
                "api_key": "sk-ant-xxx"
            }
        },
        {
            "model_name": "main-model",  # 第三级 Fallback
            "litellm_params": {
                "model": "deepseek/deepseek-chat",
                "api_key": "sk-ds-xxx"
            }
        }
    ],
    fallbacks=[{"main-model": ["main-model"]}],
    retry_policy={"retry_count": 2}
)

# 使用：自动 Fallback
response = await router.acompletion(
    model="main-model",
    messages=[{"role": "user", "content": "你好"}]
)
# GPT-4o 失败 → 自动切换 Claude → 再失败 → DeepSeek

LiteLLM Proxy（独立部署）：

yaml

# litellm_config.yaml
model_list:
  - model_name: gpt-4o
    litellm_params:
      model: openai/gpt-4o
      api_key: sk-xxx

  - model_name: claude-sonnet
    litellm_params:
      model: anthropic/claude-sonnet-4-20250514
      api_key: sk-ant-xxx

general_settings:
  master_key: sk-my-proxy-key  # Proxy 认证密钥

bash

# 启动 Proxy
litellm --config litellm_config.yaml --port 4000

# 所有应用统一通过 Proxy 访问
curl http://localhost:4000/v1/chat/completions \
  -H "Authorization: Bearer sk-my-proxy-key" \
  -d '{"model": "gpt-4o", "messages": [{"role":"user","content":"hello"}]}'

可观测性（Observability）

为什么 LLM 需要专门的可观测性？

传统 Web：请求 → 处理 → 响应（确定性，毫秒级）
LLM 应用：请求 → Prompt 组装 → 检索 → LLM 调用 → 后处理（非确定性，秒级）

需要追踪：
  - 每次 LLM 调用的输入/输出/Token 用量/延迟/成本
  - RAG 检索的相关度
  - Agent 的工具调用链路
  - Prompt 版本变化对质量的影响

LangSmith

python

# pip install langsmith
# export LANGSMITH_API_KEY=ls-xxx
# export LANGSMITH_PROJECT=my-ai-app

from langsmith import traceable

@traceable(name="chat_with_rag")
async def chat_with_rag(query: str) -> str:
    """自动追踪整个调用链路"""
    
    # Step 1: 检索（自动追踪）
    docs = await retrieve(query)
    
    # Step 2: LLM 调用（自动追踪 Token/延迟/成本）
    response = await llm.chat(
        messages=[
            {"role": "system", "content": build_prompt(docs)},
            {"role": "user", "content": query}
        ]
    )
    
    return response.content

# LangSmith Dashboard 自动显示：
# - 调用链路（检索 → LLM → 后处理）
# - 每步的输入/输出
# - Token 用量和成本
# - 延迟分布
# - 错误追踪

Langfuse（开源替代）

python

# pip install langfuse
from langfuse import Langfuse
from langfuse.decorators import observe

langfuse = Langfuse(
    public_key="pk-xxx",
    secret_key="sk-xxx",
    host="https://cloud.langfuse.com"  # 或自部署地址
)

@observe()
def chat(query: str) -> str:
    # 自动追踪，功能类似 LangSmith
    pass

维度	LangSmith	Langfuse
提供商	LangChain 公司	开源社区
部署	SaaS	SaaS + 自部署
定价	免费 5K traces/月	免费 50K obs/月
LangChain 集成	原生深度	良好
自定义	有限	完全自控
推荐	快速上手	数据自控

Prompt Caching

原理

LLM API 对重复的 Prompt 前缀只计费一次，后续请求仅计算新增部分。

请求 1：[系统提示(2000 token)] + [用户问题 A(100 token)]
  → 费用：2100 token ← 完整计费

请求 2：[系统提示(2000 token)] + [用户问题 B(80 token)]
  → 费用：80 token ← 前缀缓存命中，仅新增部分计费
  → 节省：95%！

支持情况

提供商	功能名称	最小前缀	折扣
Anthropic	Prompt Caching	1024 token	缓存读取 90% 折扣
OpenAI	Automatic Caching	128 token	50% 折扣
Google	Context Caching	32K token	75% 折扣
DeepSeek	自动缓存	64 token	90% 折扣

实战：最大化缓存命中

python

# 关键：保持 System Prompt 不变，让前缀尽可能长且一致

# ❌ 错误：每次请求的 System Prompt 都不同
messages = [
    {"role": "system", "content": f"当前时间：{datetime.now()}。你是客服助手..."},
    # ← 时间变化导致缓存永远不命中
]

# ✅ 正确：固定的 System Prompt + 动态信息放后面
messages = [
    {"role": "system", "content": "你是客服助手，负责回答用户关于产品的问题..."},
    # ← 固定前缀，缓存命中
    {"role": "user", "content": f"当前时间：{datetime.now()}。{user_query}"},
    # ← 动态信息放在用户消息中
]

评估系统

RAGAS（RAG 专用评估）

python

# pip install ragas
from ragas import evaluate
from ragas.metrics import (
    faithfulness,      # 忠实度：答案是否基于检索内容
    answer_relevancy,  # 相关性：答案是否回答了问题
    context_precision, # 精确度：检索内容是否相关
    context_recall     # 召回率：是否检索到关键信息
)

# 准备评估数据
eval_data = {
    "question": ["MCP 是什么？", "如何部署 vLLM？"],
    "answer": ["MCP 是...", "部署步骤..."],
    "contexts": [["MCP 文档片段..."], ["vLLM 文档片段..."]],
    "ground_truth": ["MCP 的标准定义...", "vLLM 部署的正确步骤..."]
}

# 运行评估
result = evaluate(
    dataset=eval_data,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall]
)

print(result)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, 
#  'context_precision': 0.85, 'context_recall': 0.90}

DeepEval（通用 LLM 评估）

python

# pip install deepeval
from deepeval import evaluate
from deepeval.metrics import (
    AnswerRelevancyMetric,
    FaithfulnessMetric,
    HallucinationMetric,
    ToxicityMetric
)
from deepeval.test_case import LLMTestCase

# 定义测试用例
test_case = LLMTestCase(
    input="什么是向量数据库？",
    actual_output="向量数据库是专门存储和检索向量嵌入的数据库...",
    expected_output="向量数据库用于存储高维向量并支持相似度搜索...",
    retrieval_context=["向量数据库文档片段..."]
)

# 评估
metrics = [
    AnswerRelevancyMetric(threshold=0.7),
    FaithfulnessMetric(threshold=0.8),
    HallucinationMetric(threshold=0.5),
    ToxicityMetric(threshold=0.1)
]

evaluate(test_cases=[test_case], metrics=metrics)

LLMOps 工具选型指南

团队规模与选型建议：

  个人 / 小团队（1-5 人）：
    网关：LiteLLM（Python 库直接用）
    追踪：LangSmith 免费版
    评估：手动 + LLM-as-Judge
    缓存：API 侧自动缓存

  中型团队（5-20 人）：
    网关：LiteLLM Proxy（独立部署）
    追踪：Langfuse（自部署，数据自控）
    评估：RAGAS + DeepEval 自动化
    缓存：Redis 精确缓存 + API 侧前缀缓存

  大型团队（20+ 人）：
    网关：Portkey / 自建网关
    追踪：Langfuse 企业版 / 自建
    评估：自定义评估 Pipeline
    缓存：多级缓存架构（见 7.3）

学习资源

9.7 AI 基础设施（LLMOps） ​

LLMOps 全景图 ​

AI 网关（Gateway） ​

为什么需要 AI 网关？ ​

LiteLLM（推荐） ​

可观测性（Observability） ​

为什么 LLM 需要专门的可观测性？ ​

LangSmith ​

Langfuse（开源替代） ​

Prompt Caching ​

原理 ​

支持情况 ​

实战：最大化缓存命中 ​

评估系统 ​

RAGAS（RAG 专用评估） ​

DeepEval（通用 LLM 评估） ​

LLMOps 工具选型指南 ​

9.7 AI 基础设施（LLMOps）

LLMOps 全景图

AI 网关（Gateway）

为什么需要 AI 网关？

LiteLLM（推荐）

可观测性（Observability）

为什么 LLM 需要专门的可观测性？

LangSmith

Langfuse（开源替代）

Prompt Caching

原理

支持情况

实战：最大化缓存命中

评估系统

RAGAS（RAG 专用评估）

DeepEval（通用 LLM 评估）

LLMOps 工具选型指南