企业官网建设流程全解析

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_factory=lambda: str(uuid.uuid4())[:8]) name: str = "" created_at: str = field(default_factory=lambda: 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_factory=dict) few_shot_examples: List[Dict] = field(default_factory=list) # 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_stemmer=True ) 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, lang='zh' ) 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 f"v{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(f'prompts/{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(f'prompts/{name}') version = sorted(versions)[-1] return self.storage.load(f'prompts/{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'] = f'deployed:{environment}' self.storage.save( f'prompts/{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( capacity=config['rate_limit_per_minute'], fill_rate=config['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( messages=messages, stream=stream ) # 输出安全检查 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_hours=24): """检测输出分布偏移""" pass def alert_on_anomaly(self, metric_name, threshold): """异常告警""" pass

总结：构建LLMOps平台的五大核心模块。LLMExperiment统一Schema记录模型、提示词、超参数、RAG配置和评估指标。LLMEvaluator实现ROUGE/BERTScore多维度自动评估。PromptRegistry提供提示词版本管理、diff比较和环境推广。LLMGateway统一推理网关集成限流、路由、输入/输出安全护栏。LLMMonitor在线监控四维度（流量/延迟/质量/成本）并设定告警阈值。强调LLMOps与MLOps的核心差异在于提示词、RAG、Agent工具和安全护栏的维度扩展。