LLMOps 实践：生产环境大模型服务的全生命周期管理

把大模型跑起来很容易，但在生产环境稳定运行是另一回事。LLMOps 是 2025 年增长最快的运维子领域。

LLMOps vs MLOps vs AIOps

维度MLOpsLLMOps
核心产物训练好的模型权重Prompt + 模型配置
版本管理模型版本Prompt 版本 + 模型版本
评估指标准确率/AUC相关性/幻觉率/延迟
主要风险数据漂移幻觉/Prompt 注入/成本失控
部署单元模型服务LLM 应用（含 RAG/Agent）

可观测性：追踪每一次 LLM 调用

生产 LLM 服务必须追踪的指标：

# 使用 Langfuse 追踪 LLM 调用
from langfuse import Langfuse
from langfuse.decorators import observe, langfuse_context
from openai import OpenAI

langfuse = Langfuse(
    public_key="pk-...",
    secret_key="sk-...",
    host="https://your-langfuse.example.com"  # 可自部署
)

client = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")

@observe()  # 自动追踪输入、输出、延迟、token 用量
def analyze_alert(alert_text: str) -> str:
    langfuse_context.update_current_observation(
        metadata={"alert_type": "pod_crash", "severity": "critical"}
    )

    response = client.chat.completions.create(
        model="qwen2.5:14b",
        messages=[
            {"role": "system", "content": "你是运维专家，分析告警并给出处理建议。"},
            {"role": "user", "content": alert_text}
        ]
    )

    result = response.choices[0].message.content

    # 记录自定义评分
    langfuse_context.score_current_observation(
        name="response_quality",
        value=1,  # 人工评分后更新
        comment="待人工审核"
    )

    return result

关键指标看板：

  • TTFT（Time to First Token）：首 token 延迟，影响用户体验
  • TPS（Tokens per Second）：吞吐量
  • 幻觉率：通过人工抽样或自动评估检测
  • 成本/请求：token 用量 × 单价

Prompt 版本管理

Prompt 是 LLM 应用的核心资产，必须像代码一样管理：

# prompts/alert_analysis/v2.yaml
version: "2.1"
description: "告警分析 Prompt，增加了置信度输出"
model: "qwen2.5:14b"
temperature: 0.2
system: |
  你是一个资深 SRE 工程师，专注于 Kubernetes 和 Linux 系统运维。
  分析告警时请：
  1. 判断故障原因（给出置信度 0-100%）
  2. 评估影响范围
  3. 给出立即处理步骤（可执行命令）
  4. 给出预防措施

  输出格式为 JSON：
  {
    "cause": "故障原因",
    "confidence": 85,
    "impact": "影响范围",
    "actions": ["步骤1", "步骤2"],
    "prevention": "预防措施"
  }
import yaml
from pathlib import Path

class PromptRegistry:
    def __init__(self, prompts_dir: str = "prompts"):
        self.dir = Path(prompts_dir)

    def get(self, name: str, version: str = "latest") -> dict:
        if version == "latest":
            versions = sorted(self.dir.glob(f"{name}/v*.yaml"))
            path = versions[-1]
        else:
            path = self.dir / name / f"{version}.yaml"
        return yaml.safe_load(path.read_text())

registry = PromptRegistry()
prompt = registry.get("alert_analysis")  # 自动获取最新版本

金丝雀发布 Prompt

import random

class PromptABTest:
    def __init__(self, control_version: str, experiment_version: str,
                 experiment_ratio: float = 0.1):
        self.control = registry.get("alert_analysis", control_version)
        self.experiment = registry.get("alert_analysis", experiment_version)
        self.ratio = experiment_ratio

    def get_prompt(self) -> tuple[dict, str]:
        if random.random() < self.ratio:
            return self.experiment, "experiment"
        return self.control, "control"

ab_test = PromptABTest("v2.0", "v2.1", experiment_ratio=0.1)

def analyze_with_ab(alert: str) -> str:
    prompt_config, variant = ab_test.get_prompt()

    # 记录使用了哪个变体
    langfuse_context.update_current_observation(
        metadata={"prompt_variant": variant}
    )

    response = client.chat.completions.create(
        model=prompt_config["model"],
        temperature=prompt_config["temperature"],
        messages=[
            {"role": "system", "content": prompt_config["system"]},
            {"role": "user", "content": alert}
        ]
    )
    return response.choices[0].message.content

安全护栏

import re

class LLMGuardrails:
    # 禁止输入的敏感模式
    INPUT_BLOCKLIST = [
        r"ignore previous instructions",
        r"你现在是",
        r"system prompt",
        r"<\|.*\|>",  # 特殊 token 注入
    ]

    # 禁止输出的内容模式
    OUTPUT_BLOCKLIST = [
        r"\b(rm -rf|dd if=|mkfs)\b",  # 危险命令
        r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b",  # IP 地址泄露
    ]

    def check_input(self, text: str) -> bool:
        for pattern in self.INPUT_BLOCKLIST:
            if re.search(pattern, text, re.IGNORECASE):
                return False
        return True

    def check_output(self, text: str) -> str:
        for pattern in self.OUTPUT_BLOCKLIST:
            text = re.sub(pattern, "[已过滤]", text)
        return text

guardrails = LLMGuardrails()

def safe_analyze(alert: str) -> str:
    if not guardrails.check_input(alert):
        return "输入包含不允许的内容，请重新描述问题。"

    result = analyze_alert(alert)
    return guardrails.check_output(result)

成本控制

# 按 token 用量设置预算告警
class CostTracker:
    def __init__(self, daily_budget_yuan: float = 100.0):
        self.budget = daily_budget_yuan
        self.spent = 0.0
        # Qwen2.5-14B 自部署成本约 ¥0.001/1K tokens
        self.cost_per_1k_tokens = 0.001

    def track(self, tokens: int) -> bool:
        cost = tokens / 1000 * self.cost_per_1k_tokens
        self.spent += cost
        if self.spent > self.budget * 0.8:
            send_alert(f"LLM 成本已达预算 80%：¥{self.spent:.2f}")
        return self.spent < self.budget

LLMOps 的核心是把 AI 服务当成普通服务来运维：有监控、有版本管理、有灰度发布、有安全边界。工具链还在快速演进，但这些基本原则是稳定的。

← 返回文章列表