LLMOps平台搭建：从实验追踪到模型服务的完整流水线

LLMOps平台搭建从实验追踪到模型服务的完整流水线一、LLMOps的独特挑战LLMOps不是MLOps的简单平移。两者的核心差异在于规模和成本。LLM参数规模从7B到数百B。推理成本是传统ML模型的100-1000倍。提示词工程是全新的变更管理维度。评估标准从精确率变为语义相关性。安全和护栏问题是生产化的第一优先级。传统MLOps管的是模型版本和特征工程。LLMOps还需要管提示词模板、RAG知识库、Agent工具链。这些新维度的组合爆炸性让实验追踪变得极其复杂。graph TD subgraph MLOps传统流程 A1[数据] -- B1[特征工程] B1 -- C1[训练] C1 -- D1[模型注册] D1 -- E1[部署] end subgraph LLMOps新增维度 A2[提示词模板] -- B2[Prompt Registry] C2[RAG文档] -- D2[向量库管理] E2[Agent工具] -- F2[工具注册中心] G2[安全护栏] -- H2[Guardrail配置] end subgraph LLMOps统一平台 I[实验追踪] -- J[评估矩阵] J -- K[部署流水线] K -- L[在线监控] end二、实验追踪从模型到提示词的端到端2.1 统一实验Schema实验追踪是LLMOps的基石。一次LLM实验涉及模型、提示词、温度、检索策略等多个变量。需要统一的Schema记录所有配置。from dataclasses import dataclass, field from typing import Optional, Dict, List, Any from datetime import datetime import uuid import json dataclass class LLMExperiment: LLM实验的统一记录结构 experiment_id: str field(default_factorylambda: str(uuid.uuid4())[:8]) name: str created_at: str field(default_factorylambda: datetime.now().isoformat()) # 模型配置 model_name: str model_provider: str # openai / anthropic / local model_version: str quantization: Optional[str] None # int4, int8, fp16 # 推理超参 temperature: float 0.7 top_p: float 0.95 max_tokens: int 2048 frequency_penalty: float 0.0 presence_penalty: float 0.0 # 提示词配置 system_prompt: str prompt_template: str prompt_variables: Dict[str, str] field(default_factorydict) few_shot_examples: List[Dict] field(default_factorylist) # RAG配置 retrieval_enabled: bool False retriever_type: str top_k: int 3 embedding_model: str chunk_size: int 512 chunk_overlap: int 50 # 评估指标 latency_ms: float 0.0 tokens_generated: int 0 prompt_tokens: int 0 total_tokens: int 0 cost_usd: float 0.0 # 质量指标 accuracy: Optional[float] None rouge_l: Optional[float] None bert_score: Optional[float] None human_rating: Optional[float] None def to_dict(self) - dict: return { k: v for k, v in self.__dict__.items() if not k.startswith(_) } def log(self, backend): 记录实验到后端 backend.log_experiment(self.to_dict())2.2 评估矩阵构建from typing import Callable import numpy as np from rouge_score import rouge_scorer from bert_score import score as bert_score_fn class LLMEvaluator: LLM多维度评估器 def __init__(self): self.rouge_scorer rouge_scorer.RougeScorer( [rouge1, rouge2, rougeL], use_stemmerTrue ) self.metrics {} def evaluate(self, predictions: List[str], references: List[str], prompts: Optional[List[str]] None): 执行完整评估 results {} # ROUGE评估 rouge_scores self._compute_rouge(predictions, references) results[rouge] rouge_scores # BERTScore评估 precision, recall, f1 bert_score_fn( predictions, references, langzh ) results[bert_score] { precision: precision.mean().item(), recall: recall.mean().item(), f1: f1.mean().item(), } # 成本分析 results[cost] self._compute_cost_metrics(predictions) # 延迟统计 results[latency] self._latency_stats() return results def _compute_rouge(self, preds, refs): scores {rouge1: [], rouge2: [], rougeL: []} for p, r in zip(preds, refs): s self.rouge_scorer.score(r, p) for key in scores: scores[key].append(s[key].fmeasure) return {k: np.mean(v) for k, v in scores.items()} def _compute_cost_metrics(self, predictions): return { total_tokens: sum( p.get(total_tokens, 0) for p in predictions ), avg_tokens_per_request: np.mean([ p.get(total_tokens, 0) for p in predictions ]), } def _latency_stats(self): return {p50: 0, p95: 0, p99: 0}三、提示词注册与版本管理3.1 Prompt Registry提示词的版本管理参考Git的语义。支持fork、diff、rollback操作。import hashlib from datetime import datetime from typing import Optional class PromptRegistry: 提示词注册中心 def __init__(self, storage_backend): self.storage storage_backend self.cache {} def register(self, name: str, template: str, version: str None, metadata: dict None) - str: 注册新提示词版本 content_hash hashlib.sha256( template.encode() ).hexdigest()[:12] version version or fv{datetime.now().strftime(%Y%m%d%H%M%S)} prompt_entry { name: name, version: version, template: template, content_hash: content_hash, variables: self._extract_variables(template), metadata: metadata or {}, created_at: datetime.now().isoformat(), status: draft, } self.storage.save(fprompts/{name}/{version}, prompt_entry) self.cache[f{name}:{version}] prompt_entry return version def get(self, name: str, version: str latest) - dict: 获取指定版本的提示词 cache_key f{name}:{version} if cache_key in self.cache: return self.cache[cache_key] if version latest: versions self.storage.list(fprompts/{name}) version sorted(versions)[-1] return self.storage.load(fprompts/{name}/{version}) def diff(self, name: str, v1: str, v2: str) - str: 比较两个版本的差异 p1 self.get(name, v1)[template] p2 self.get(name, v2)[template] return self._generate_diff(p1, p2) def promote(self, name: str, version: str, environment: str): 将提示词版本推广到指定环境 prompt self.get(name, version) prompt[status] fdeployed:{environment} self.storage.save( fprompts/{name}/{version}, prompt ) def _extract_variables(self, template: str) - list: 提取模板中的变量 import re return re.findall(r\{(\w)\}, template) def _generate_diff(self, s1, s2): import difflib return \n.join(difflib.unified_diff( s1.splitlines(), s2.splitlines(), lineterm ))四、模型服务与网关4.1 统一推理网关graph TD A[客户端请求] -- B[API Gateway] B -- C{Prompt装配} C -- D{路由决策} D --|简单任务| E[小模型 7B] D --|复杂推理| F[大模型 70B] D --|特定领域| G[微调模型] E -- H[限流检查] F -- H G -- H H -- I[安全护栏] I -- J{输出检查} J --|通过| K[返回结果] J --|违规| L[拒绝/改写] K -- M[日志记录] L -- M M -- N[指标采集]import asyncio from typing import AsyncIterator class LLMGateway: 统一LLM推理网关 def __init__(self, config): self.providers {} self.rate_limiter TokenBucketRateLimiter( capacityconfig[rate_limit_per_minute], fill_rateconfig[rate_limit_per_minute] / 60, ) self.guardrails self._init_guardrails(config[guardrails]) async def chat(self, messages: list, model: str default, stream: bool False) - dict: 统一推理入口 # 限流检查 if not await self.rate_limiter.acquire(): raise RateLimitExceeded(请求频率超限) # 安全护栏 for guard in self.guardrails: violation await guard.check_input(messages) if violation: return self._reject_response(violation) # 路由到对应provider provider self._route_provider(model) # 执行推理 response await provider.generate( messagesmessages, streamstream ) # 输出安全检查 content response[choices][0][message][content] for guard in self.guardrails: violation await guard.check_output(content) if violation: return self._rewrite_response(content, violation) return response def _route_provider(self, model): 模型路由 - 支持A/B测试 if model in self.providers: return self.providers[model] # 根据任务复杂度自动路由 return self.providers.get(default) def _reject_response(self, violation): return { choices: [{ message: { role: assistant, content: 抱歉您的请求包含违规内容。, }, finish_reason: content_filter, }], guardrail: violation[rule], } def _rewrite_response(self, content, violation): 根据护栏规则改写输出 return { choices: [{ message: { role: assistant, content: f[已重写] {content}, }, finish_reason: stop, }], guardrail: violation[rule], } def _init_guardrails(self, config): guards [] if config.get(pii_detection): guards.append(PIIGuard()) if config.get(toxic_content): guards.append(ToxicContentGuard()) if config.get(prompt_injection): guards.append(InjectionDetector()) return guards五、监控与持续优化5.1 在线监控矩阵生产环境需要四维度监控。请求量监控出现异常的流量尖峰。延迟监控P50/P95/P99的劣化趋势。质量监控输出分布的偏移检测。成本监控token消耗的预算预警。class LLMMonitor: LLM在线监控器 def __init__(self): self.metrics {} def record_inference(self, experiment_id, metrics): 记录单次推理指标 pass def check_distribution_shift(self, window_hours24): 检测输出分布偏移 pass def alert_on_anomaly(self, metric_name, threshold): 异常告警 pass监控维度关键指标告警阈值流量QPS/每分钟请求数突增200%延迟P95延迟 目标SLA × 2质量输出分布KL散度 0.3成本日消耗token数 预算120%总结构建LLMOps平台的五大核心模块。LLMExperiment统一Schema记录模型、提示词、超参数、RAG配置和评估指标。LLMEvaluator实现ROUGE/BERTScore多维度自动评估。PromptRegistry提供提示词版本管理、diff比较和环境推广。LLMGateway统一推理网关集成限流、路由、输入/输出安全护栏。LLMMonitor在线监控四维度流量/延迟/质量/成本并设定告警阈值。强调LLMOps与MLOps的核心差异在于提示词、RAG、Agent工具和安全护栏的维度扩展。