MLOps

LLM版本管理的挑战 传统的软件版本管理（Git）无法处理大模型文件（数十GB）。LLM的版本管理需要同时追踪代码、配置、数据、模型权重和评估结果。 版本管理工具链 MLflow模型注册 import mlflow # 记录模型版本 with mlflow.start_run(run_name="qwen3-32b-v2"): mlflow.log_params({ "base_model": "Qwen-3-32B", "fine_tune_method": "LoRA", "learning_rate": 2e-4, "epochs": 3, "dataset": "instruction-v3", }) mlflow.log_metrics({ "eval_loss": 0.45, "eval_accuracy": 0.89, "human_eval_score": 4.2, }) # 注册模型 mlflow.register_model( "runs:/abc123/model", "qwen3-32b-instruct", tags={ "version": "v2.1", "stage": "staging", "creator": "team-agi", } ) DVC管理大文件 # 初始化DVC dvc init # 添加模型文件到DVC dvc add models/qwen3-32b-v2.1/ # 推送到远程存储 dvc remote add -d storage s3://my-bucket/models dvc push # Git只追踪.dvc文件（指针），不追踪实际大文件 git add models/qwen3-32b-v2.1/.dvc git commit -m "Add qwen3-32b v2.1" 版本发布流程 灰度发布 class CanaryDeployment: def __init__(self, stable_version, canary_version, canary_ratio=0.1): self.stable = stable_version self.canary = canary_version self.ratio = canary_ratio self.metrics = {"stable": [], "canary": []} def route(self, request): """灰度路由""" import random if random.random() < self.ratio: self.metrics["canary"].append({"time": time.time()}) return self.canary else: self.metrics["stable"].append({"time": time.time()}) return self.stable def evaluate(self): """评估灰度结果""" canary_latency = self.get_avg_latency("canary") stable_latency = self.get_avg_latency("stable") canary_error = self.get_error_rate("canary") stable_error = self.get_error_rate("stable") # 灰度通过条件 if canary_latency > stable_latency * 1.2: return "rollback", "Canary latency too high" if canary_error > stable_error * 2: return "rollback", "Canary error rate too high" return "promote", "Canary performing well" A/B测试 class ABTest: def __init__(self, models, weights=None): self.models = models self.weights = weights or [1/len(models)] * len(models) self.results = {m: {"satisfied": 0, "total": 0} for m in models} def route(self, user_id): # 基于用户ID的确定性路由 hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) cumulative = 0 for model, weight in zip(self.models, self.weights): cumulative += weight if (hash_val % 1000) / 1000 < cumulative: return model def record_feedback(self, model, satisfied): self.results[model]["total"] += 1 if satisfied: self.results[model]["satisfied"] += 1 def get_winner(self): rates = {m: r["satisfied"]/r["total"] for m, r in self.results.items() if r["total"] > 0} return max(rates, key=rates.get) if rates else None CI/CD管线 # .github/workflows/model-deploy.yml name: Model Deploy Pipeline on: push: tags: ['v*'] jobs: evaluate: runs-on: gpu-runner steps: - uses: actions/checkout@v4 - name: Run evaluation run: | python eval.py --model checkpoints/latest --benchmark mmlu,gsm8k,humaneval - name: Check quality gates run: | python check_gates.py --min_accuracy 0.85 --max_regression 0.02 deploy_staging: needs: evaluate runs-on: deploy-runner steps: - name: Deploy to staging run: | ./deploy.sh --env staging --version ${{ github.ref_name }} - name: Run smoke tests run: | python smoke_test.py --env staging canary: needs: deploy_staging runs-on: deploy-runner steps: - name: Canary deployment (10%) run: | ./deploy.sh --env production --version ${{ github.ref_name }} --canary 0.1 - name: Monitor for 1 hour run: | python monitor.py --duration 3600 --check latency,error_rate full_deploy: needs: canary runs-on: deploy-runner steps: - name: Full deployment run: | ./deploy.sh --env production --version ${{ github.ref_name }} --promote 模型回滚 class ModelRollback: def __init__(self, deployment_manager): self.deployment = deployment_manager self.version_history = [] async def rollback(self, target_version=None): """回滚到指定版本或上一个稳定版本""" if target_version is None: target_version = self.get_previous_stable() logger.info(f"Rolling back to {target_version}") # 快速切换流量 await self.deployment.switch_traffic( from_version="current", to_version=target_version, ratio=1.0 # 100%切换 ) # 验证回滚 health = await self.deployment.health_check(target_version) if not health: logger.error("Rollback target also unhealthy!") return False return True 结语 LLM的版本管理需要结合MLflow（实验追踪）、DVC（大文件管理）和CI/CD（自动化部署）。灰度发布和快速回滚是降低部署风险的关键能力。建立完善的MLOps流程，可以让模型迭代从"手动谨慎"变为"自动自信"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 AI Agent的部署架构从简单的单机脚本演进为复杂的多层云原生系统。2026年，Agent部署不再只是"跑通模型"，而是需要考虑高可用、弹性伸缩、成本优化、安全合规等多维度的系统工程。 部署架构演进 阶段一：单机部署 适用场景： 开发测试、小规模演示、个人项目 ┌─────────────────────────────┐ │ 单机服务器 │ │ ┌───────────────────────┐ │ │ │ Agent 应用进程 │ │ │ │ + 推理引擎 │ │ │ │ + 向量数据库 │ │ │ └───────────────────────┘ │ │ 操作系统 │ └─────────────────────────────┘ 技术栈： 推理：vLLM / Ollama / llama.cpp 向量库：Chroma / FAISS（内存） 缓存：Redis（单机版） 优缺点： ✅ 简单快速，部署成本低 ✅ 调试方便 ❌ 无高可用，无弹性 ❌ 单点故障 阶段二：容器化部署 适用场景： 生产环境、中小规模应用 ┌───────────────────────────────────────┐ │ Kubernetes Cluster │ │ ┌─────────┐ ┌─────────┐ ┌───────┐ │ │ │ Agent-1 │ │ Agent-2 │ │Agent-N│ │ │ │ Pod │ │ Pod │ │ Pod │ │ │ └─────────┘ └─────────┘ └───────┘ │ │ ┌─────────────────────────────────┐ │ │ │ 服务网格 (Istio/Linkerd) │ │ │ └─────────────────────────────────┘ │ │ ┌─────────┐ ┌─────────┐ │ │ │ Redis │ │ Milvus │ │ │ │ Cluster│ │ Cluster │ │ │ └─────────┘ └─────────┘ │ └───────────────────────────────────────┘ 技术栈： ...

从"散装Prompt"到"Prompt工程体系" 2026年，大型企业平均拥有超过5000个生产环境Prompt。这些Prompt分散在不同团队、不同项目中，由不同开发者编写，使用不同模型，服务于不同场景。如果没有系统化的管理方案，Prompt的维护成本将急剧攀升。 典型问题： 同一业务的Prompt在10个项目中各自维护，修改需要同步10处 离职员工的Prompt无人理解，不敢修改 模型升级后30%的Prompt性能下降，但无人知晓 没有统一的Prompt质量标准，质量参差不齐 本文分享我们在过去两年中构建企业级Prompt管理系统的实践经验。 Prompt模板架构 模板结构设计 from dataclasses import dataclass, field from typing import Optional from enum import Enum class PromptCategory(Enum): SYSTEM = "system" # 系统级Prompt TASK = "task" # 任务级Prompt GUARDRAIL = "guardrail" # 安全护栏 UTILITY = "utility" # 工具函数 EVALUATION = "evaluation" # 评估用 class PromptStatus(Enum): DRAFT = "draft" REVIEW = "review" TESTING = "testing" STAGING = "staging" PRODUCTION = "production" DEPRECATED = "deprecated" @dataclass class PromptTemplate: """Prompt模板定义""" id: str # 唯一标识 name: str # 模板名称 category: PromptCategory # 分类 status: PromptStatus # 状态 # 模板内容 system_prompt: str # 系统提示 user_prompt_template: str # 用户提示模板（含变量） variables: list[dict] # 变量定义 # 元数据 description: str # 描述 author: str # 作者 version: str # 版本号 tags: list[str] # 标签 # 配置 model_config: dict # 模型配置 expected_output: Optional[dict] # 期望输出格式 # 质量指标 quality_score: Optional[float] # 质量评分 latency_p95: Optional[float] # P95延迟 success_rate: Optional[float] # 成功率 # 关联 dependencies: list[str] = field(default_factory=list) # 依赖的其他模板 parent_id: Optional[str] = None # 父模板（继承关系） 模板语法 class PromptTemplateEngine: """ Prompt模板引擎 支持变量插值、条件逻辑、循环和继承 """ # 模板语法示例 TEMPLATE_EXAMPLE = """ {{#system}} 你是{{role}}，专注于{{domain}}领域。 核心规则： {{#rules}} - {{.}} {{/rules}} {{#if strict_mode}} ⚠️ 严格遵守以上规则，不允许偏离。 {{/if}} {{/system}} {{#user}} {{user_input}} {{#if context}} 相关上下文： {{#context}} --- {{.}} --- {{/context}} {{/if}} {{#if examples}} 参考示例： {{#examples}} 输入：{{input}} 输出：{{output}} {{/examples}} {{/if}} {{/user}} """ def __init__(self): self.templates: dict[str, PromptTemplate] = {} self.cache = {} def render(self, template_id: str, variables: dict) -> dict: """渲染模板""" template = self.templates.get(template_id) if not template: raise ValueError(f"模板 {template_id} 不存在") # 合并默认变量 merged_vars = self._merge_defaults(template, variables) # 验证必填变量 self._validate_variables(template, merged_vars) # 渲染 system = self._render_text(template.system_prompt, merged_vars) user = self._render_text(template.user_prompt_template, merged_vars) return { "system": system, "user": user, "model_config": template.model_config, "template_id": template_id, "version": template.version } def _render_text(self, template_text: str, variables: dict) -> str: """渲染模板文本""" # 使用Jinja2或自定义模板引擎 from jinja2 import Template tpl = Template(template_text) return tpl.render(**variables) Prompt注册中心 集中化存储 class PromptRegistry: """ Prompt注册中心 所有Prompt模板的单一可信来源（Single Source of Truth） """ def __init__(self, storage_backend="postgresql"): self.storage = self._init_storage(storage_backend) async def register(self, template: PromptTemplate) -> str: """注册新模板""" # 验证 self._validate_template(template) # 检查命名冲突 if await self._exists(template.name, template.version): raise ValueError(f"模板 {template.name} v{template.version} 已存在") # 存储 template_id = await self.storage.save(template) # 建立索引 await self._update_index(template) return template_id async def get(self, template_id: str) -> PromptTemplate: """获取模板""" return await self.storage.get(template_id) async def search(self, query: dict) -> list[PromptTemplate]: """搜索模板""" # 支持按名称、标签、分类、状态搜索 return await self.storage.search(query) async def update(self, template_id: str, updates: dict) -> PromptTemplate: """更新模板（创建新版本）""" current = await self.get(template_id) # 创建新版本 new_version = self._increment_version(current.version) updated = PromptTemplate( **{**current.__dict__, **updates, "version": new_version, "parent_id": template_id} ) # 注册新版本 new_id = await self.register(updated) # 标记旧版本 await self.storage.update( template_id, {"status": PromptStatus.DEPRECATED} ) return updated 权限管理 class PromptAccessControl: """ Prompt权限管理 """ PERMISSIONS = { "read": "查看模板", "write": "创建/修改模板", "deploy": "部署到生产", "delete": "删除模板", "export": "导出模板", } ROLES = { "viewer": ["read"], "developer": ["read", "write"], "reviewer": ["read", "write"], "admin": ["read", "write", "deploy", "delete", "export"], } def check_permission(self, user_id: str, template_id: str, permission: str) -> bool: """检查用户权限""" user_role = self._get_user_role(user_id) allowed = self.ROLES.get(user_role, []) if permission not in allowed: return False # 项目级权限检查 template = self.registry.get(template_id) if not self._has_project_access(user_id, template.project): return False return True Prompt流水线 CI/CD for Prompts class PromptPipeline: """ Prompt CI/CD 流水线 从开发到部署的完整流程 """ async def run_pipeline(self, template: PromptTemplate): """执行完整流水线""" results = {} # 阶段1: 静态检查 results["lint"] = await self._lint(template) if not results["lint"]["passed"]: return results # 阶段2: 单元测试 results["unit_test"] = await self._unit_test(template) if not results["unit_test"]["passed"]: return results # 阶段3: 安全检查 results["security"] = await self._security_scan(template) if not results["security"]["passed"]: return results # 阶段4: 性能测试 results["performance"] = await self._performance_test(template) # 阶段5: A/B测试准备 results["ab_setup"] = await self._setup_ab_test(template) # 阶段6: 部署 if all(r.get("passed", True) for r in results.values()): results["deploy"] = await self._deploy(template) return results async def _lint(self, template: PromptTemplate) -> dict: """静态检查""" issues = [] # 检查变量完整性 used_vars = self._extract_variables(template.user_prompt_template) defined_vars = [v["name"] for v in template.variables] for var in used_vars: if var not in defined_vars: issues.append(f"未定义的变量: {var}") # 检查长度 if len(template.system_prompt) > 2000: issues.append("System Prompt过长（>2000字符），可能影响性能") # 检查安全 dangerous_patterns = ["ignore previous", "you are now", "system prompt"] for pattern in dangerous_patterns: if pattern in template.user_prompt_template.lower(): issues.append(f"潜在安全风险: 包含 '{pattern}'") return { "passed": len(issues) == 0, "issues": issues } async def _unit_test(self, template: PromptTemplate) -> dict: """单元测试""" test_cases = template.variables.get("test_cases", []) results = [] for case in test_cases: rendered = self.engine.render(template.id, case["input"]) response = await self.llm.generate(rendered) passed = self._evaluate_response( response, case["expected"] ) results.append({ "case_name": case.get("name", "unnamed"), "passed": passed, "response": response[:200] }) pass_rate = sum(r["passed"] for r in results) / len(results) return { "passed": pass_rate >= 0.9, "pass_rate": pass_rate, "results": results } async def _performance_test(self, template: PromptTemplate) -> dict: """性能测试""" import time latencies = [] for _ in range(50): start = time.time() rendered = self.engine.render(template.id, {}) response = await self.llm.generate(rendered) latencies.append(time.time() - start) import numpy as np return { "passed": np.percentile(latencies, 95) < 5.0, # P95 < 5秒 "p50": np.median(latencies), "p95": np.percentile(latencies, 95), "p99": np.percentile(latencies, 99), } Prompt监控系统 实时监控 class PromptMonitor: """ Prompt生产环境监控 """ def __init__(self): self.metrics_store = MetricsStore() self.alerting = AlertingSystem() async def record_invocation(self, template_id: str, version: str, invocation_data: dict): """记录每次Prompt调用""" await self.metrics_store.record({ "template_id": template_id, "version": version, "timestamp": datetime.now(), "input": invocation_data["input"], "output": invocation_data["output"], "latency_ms": invocation_data["latency_ms"], "tokens_used": invocation_data["tokens_used"], "cost": invocation_data["cost"], "success": invocation_data["success"], "user_feedback": invocation_data.get("user_feedback"), }) # 实时检查 await self._check_anomalies(template_id, invocation_data) async def _check_anomalies(self, template_id: str, data: dict): """异常检测""" # 延迟异常 baseline_latency = await self.metrics_store.get_baseline_latency(template_id) if data["latency_ms"] > baseline_latency * 3: await self.alerting.send_alert( level="warning", template_id=template_id, message=f"延迟异常: {data['latency_ms']}ms (基线: {baseline_latency}ms)" ) # 成功率下降 recent_success_rate = await self.metrics_store.get_recent_success_rate( template_id, window_minutes=30 ) if recent_success_rate < 0.85: await self.alerting.send_alert( level="critical", template_id=template_id, message=f"成功率下降: {recent_success_rate:.1%}" ) # 成本异常 daily_cost = await self.metrics_store.get_daily_cost(template_id) if daily_cost > 100: # 日成本超过100元 await self.alerting.send_alert( level="warning", template_id=template_id, message=f"日成本异常: ¥{daily_cost}" ) 仪表盘 class PromptDashboard: """Prompt管理仪表盘数据生成""" def generate_report(self, date_range: tuple) -> dict: return { "overview": { "total_templates": self._count_templates(), "active_templates": self._count_active_templates(), "total_invocations": self._count_invocations(date_range), "total_cost": self._sum_cost(date_range), "avg_success_rate": self._avg_success_rate(date_range), "avg_latency_p95": self._avg_latency(date_range), }, "top_templates": self._top_templates(date_range, n=10), "quality_issues": self._identify_quality_issues(date_range), "cost_breakdown": self._cost_breakdown(date_range), "performance_trends": self._performance_trends(date_range), "recommendations": self._generate_recommendations(date_range), } 模板继承与组合 class PromptInheritance: """ Prompt模板继承系统 支持模板之间的继承和组合 """ def resolve(self, template_id: str) -> PromptTemplate: """ 解析模板继承链，生成最终Prompt """ template = self.registry.get(template_id) if template.parent_id: # 递归解析父模板 parent = self.resolve(template.parent_id) # 合并：子模板覆盖父模板 return self._merge(parent, template) return template def _merge(self, parent: PromptTemplate, child: PromptTemplate) -> PromptTemplate: """合并父子模板""" return PromptTemplate( id=child.id, name=child.name, system_prompt=child.system_prompt or parent.system_prompt, user_prompt_template=child.user_prompt_template or parent.user_prompt_template, variables=self._merge_variables(parent.variables, child.variables), # ... 其他字段 ) 最佳实践总结 Prompt模板管理清单 维度 实践 优先级 存储 集中化注册中心 P0 版本 语义化版本控制 P0 权限 基于角色的访问控制 P1 测试 自动化单元测试 P0 安全 注入扫描+内容审查 P0 监控 延迟/成功率/成本 P0 文档 每个模板附带说明 P1 复用 模板继承与组合 P1 优化 A/B测试框架 P2 治理 定期审查与清理 P1 结语 Prompt模板管理是AI工程化的基础设施。2026年的经验表明：将Prompt视为代码（Prompt as Code）是正确的方向。 版本控制、CI/CD、测试、监控——这些软件工程的成熟实践同样适用于Prompt管理。 ...

Prompt 也是代码，也需要版本管理 2026 年，头部 AI 团队的 Prompt 库已经增长到数千条，涉及数百个应用场景。没有版本管理，Prompt 的变更是灾难性的——“谁改了什么？为什么改？改了之后效果变好了还是变差了？“这些问题无法回答。Prompt 版本管理平台已成为 AI 工程化的基础设施。 一、Prompt 版本管理的核心需求 1.1 与 Git 的异同 维度 代码 Git Prompt 版本管理 版本控制 ✅ 文件差异 ✅ Prompt 差异 分支管理 ✅ 功能分支 ✅ 实验分支 代码审查 ✅ PR ✅ Prompt 评审 CI/CD ✅ 自动测试 ✅ 效果评估 回滚 ✅ 任意版本 ✅ 任意版本 性能指标 ❌ 不内置 ✅ 必须内置 多环境 dev/staging/prod draft/staging/prod A/B测试 ❌ 不内置 ✅ 核心功能 1.2 平台架构 ┌────────────────────────────────────────────┐ │ Web UI / CLI │ ├────────────────────────────────────────────┤ │ 版本管理 │ A/B测试 │ 灰度发布 │ 监控面板 │ ├────────────────────────────────────────────┤ │ Prompt 存储引擎 │ │ ┌─────────┐ ┌──────────┐ ┌────────────┐ │ │ │版本树 │ │元数据 │ │评估结果 │ │ │ └─────────┘ └──────────┘ └────────────┘ │ ├────────────────────────────────────────────┤ │ 集成层 │ │ ┌─────────┐ ┌──────────┐ ┌────────────┐ │ │ │LLM API │ │CI/CD │ │监控系统 │ │ │ └─────────┘ └──────────┘ └────────────┘ │ └────────────────────────────────────────────┘ 二、数据模型设计 from dataclasses import dataclass, field from datetime import datetime from typing import List, Optional, Dict from enum import Enum class PromptStatus(Enum): DRAFT = "draft" IN_REVIEW = "in_review" STAGING = "staging" PRODUCTION = "production" DEPRECATED = "deprecated" ARCHIVED = "archived" class ChangeType(Enum): CREATED = "created" MODIFIED = "modified" PROMOTED = "promoted" ROLLED_BACK = "rolled_back" DEPRECATED = "deprecated" @dataclass class PromptVersion: """Prompt 版本模型""" id: str prompt_id: str # Prompt 唯一标识 version: str # 语义化版本号 e.g. "2.3.1" parent_version: Optional[str] # 父版本 # Prompt 内容 system_prompt: str user_template: str variables_schema: Dict # 变量定义 # 元数据 author: str created_at: datetime status: PromptStatus # 变更说明 change_type: ChangeType change_description: str # 评估结果 evaluation: Optional[Dict] = None # {'accuracy': 0.92, 'safety': 0.99, 'latency_ms': 1200, ...} # 部署信息 deployed_at: Optional[datetime] = None deployed_by: Optional[str] = None traffic_percentage: int = 0 # 灰度比例 @dataclass class PromptBranch: """Prompt 分支""" name: str base_version: str head_version: str purpose: str # 实验目的 created_at: datetime experiments: List[str] = field(default_factory=list) @dataclass class ABTest: """A/B 测试配置""" id: str prompt_id: str variants: Dict[str, str] # {'A': 'v2.3.0', 'B': 'v2.3.1'} traffic_split: Dict[str, int] # {'A': 50, 'B': 50} start_time: datetime end_time: Optional[datetime] = None success_metrics: List[str] # ['accuracy', 'user_satisfaction'] results: Optional[Dict] = None 三、版本控制引擎 class PromptVersionControl: """Prompt 版本控制引擎""" def __init__(self, storage_backend='postgresql'): self.storage = self._init_storage(storage_backend) def create_prompt(self, prompt_id: str, system_prompt: str, user_template: str, author: str, variables_schema: dict = None) -> PromptVersion: """创建新 Prompt""" version = PromptVersion( id=self._generate_id(), prompt_id=prompt_id, version="1.0.0", parent_version=None, system_prompt=system_prompt, user_template=user_template, variables_schema=variables_schema or {}, author=author, created_at=datetime.now(), status=PromptStatus.DRAFT, change_type=ChangeType.CREATED, change_description="初始版本" ) self.storage.save(version) return version def commit(self, prompt_id: str, system_prompt: str = None, user_template: str = None, author: str = "", change_description: str = "") -> PromptVersion: """提交新版本（类似 git commit）""" latest = self.storage.get_latest(prompt_id) new_version = self._increment_version(latest.version, change_description) version = PromptVersion( id=self._generate_id(), prompt_id=prompt_id, version=new_version, parent_version=latest.version, system_prompt=system_prompt or latest.system_prompt, user_template=user_template or latest.user_template, variables_schema=latest.variables_schema, author=author, created_at=datetime.now(), status=PromptStatus.DRAFT, change_type=ChangeType.MODIFIED, change_description=change_description ) self.storage.save(version) return version def diff(self, version_a: str, version_b: str) -> dict: """比较两个版本的差异""" va = self.storage.get(version_a) vb = self.storage.get(version_b) return { 'system_prompt_diff': self._text_diff( va.system_prompt, vb.system_prompt), 'user_template_diff': self._text_diff( va.user_template, vb.user_template), 'version_a': version_a, 'version_b': version_b, 'metadata_changes': { 'author': f"{va.author} → {vb.author}", 'change_type': vb.change_type.value, } } def promote(self, version: str, target_env: str) -> PromptVersion: """版本晋升（draft → staging → production）""" pv = self.storage.get(version) if target_env == "staging": pv.status = PromptStatus.STAGING elif target_env == "production": # 检查前置条件 if pv.evaluation is None: raise ValueError("版本未评估，不能上线") if pv.evaluation.get('safety', 0) < 0.95: raise ValueError("安全评估未达标") # 将之前的 production 版本标记为 deprecated old_prod = self.storage.get_production_version(pv.prompt_id) if old_prod: old_prod.status = PromptStatus.DEPRECATED self.storage.save(old_prod) pv.status = PromptStatus.PRODUCTION pv.deployed_at = datetime.now() pv.traffic_percentage = 100 self.storage.save(pv) return pv def rollback(self, prompt_id: str, target_version: str = None) -> PromptVersion: """回滚到指定版本""" if target_version is None: # 回滚到上一个 production 版本 versions = self.storage.get_version_history(prompt_id) prod_versions = [v for v in versions if v.status in [PromptStatus.DEPRECATED]] if not prod_versions: raise ValueError("没有可回滚的版本") target_version = prod_versions[0].version target = self.storage.get(target_version) current_prod = self.storage.get_production_version(prompt_id) if current_prod: current_prod.status = PromptStatus.DEPRECATED target.status = PromptStatus.PRODUCTION target.change_type = ChangeType.ROLLED_BACK target.deployed_at = datetime.now() self.storage.save(current_prod) self.storage.save(target) return target def _increment_version(self, current: str, change_desc: str) -> str: """语义化版本号递增""" major, minor, patch = map(int, current.split('.')) if change_desc.startswith('BREAKING') or '重大修改' in change_desc: major += 1 minor = 0 patch = 0 elif '新增' in change_desc or '优化' in change_desc: minor += 1 patch = 0 else: patch += 1 return f"{major}.{minor}.{patch}" def _text_diff(self, text_a: str, text_b: str) -> str: """生成文本差异""" import difflib diff = difflib.unified_diff( text_a.splitlines(keepends=True), text_b.splitlines(keepends=True), fromfile='old', tofile='new' ) return ''.join(diff) 四、A/B 测试引擎 class PromptABTestEngine: """Prompt A/B 测试引擎""" def __init__(self, version_control: PromptVersionControl, llm_client, evaluator): self.vc = version_control self.llm = llm_client self.evaluator = evaluator self.active_tests: Dict[str, ABTest] = {} def create_test(self, prompt_id: str, variant_a: str, variant_b: str, traffic_split: dict = None, duration_days: int = 7) -> ABTest: """创建 A/B 测试""" test = ABTest( id=self._generate_id(), prompt_id=prompt_id, variants={'A': variant_a, 'B': variant_b}, traffic_split=traffic_split or {'A': 50, 'B': 50}, start_time=datetime.now(), end_time=datetime.now().replace( hour=datetime.now().hour + duration_days * 24), success_metrics=['accuracy', 'safety', 'user_satisfaction'], ) self.active_tests[test.id] = test return test def route_request(self, prompt_id: str, user_id: str) -> PromptVersion: """路由用户请求到对应的 Prompt 版本""" import hashlib # 查找活跃测试 test = self._find_active_test(prompt_id) if not test: # 没有测试，返回 production 版本 return self.vc.storage.get_production_version(prompt_id) # 确定性路由（同一用户总是看到同一版本） hash_value = int(hashlib.md5(user_id.encode()).hexdigest(), 16) bucket = hash_value % 100 cumulative = 0 for variant, percentage in test.traffic_split.items(): cumulative += percentage if bucket < cumulative: version = test.variants[variant] return self.vc.storage.get(version) return self.vc.storage.get_production_version(prompt_id) def evaluate_test(self, test_id: str) -> dict: """评估 A/B 测试结果""" test = self.active_tests[test_id] results = {} for variant, version in test.variants.items(): pv = self.vc.storage.get(version) results[variant] = { 'version': version, 'metrics': pv.evaluation or {}, 'sample_size': self._get_sample_size(version), } # 统计显著性检验 significance = self._statistical_test( results['A']['metrics'], results['B']['metrics'] ) test.results = { 'variants': results, 'significance': significance, 'winner': self._determine_winner(results, significance), 'recommendation': self._recommend(test, results, significance) } return test.results def _statistical_test(self, metrics_a: dict, metrics_b: dict) -> dict: """统计显著性检验""" from scipy import stats results = {} for metric in ['accuracy', 'safety', 'user_satisfaction']: if metric in metrics_a and metric in metrics_b: # 简化：假设已有足够样本 z_stat, p_value = stats.ttest_ind( [metrics_a[metric]], [metrics_b[metric]] ) results[metric] = { 'p_value': p_value, 'significant': p_value < 0.05 } return results 五、CI/CD 集成 class PromptCIPipeline: """Prompt CI/CD 管道""" def __init__(self, version_control, evaluator, safety_checker): self.vc = version_control self.evaluator = evaluator self.safety = safety_checker def run_pipeline(self, prompt_version: PromptVersion) -> dict: """运行完整 CI 管道""" results = { 'version': prompt_version.version, 'stages': [], 'passed': True, 'blocking_issues': [] } # Stage 1: 格式检查 stage = self._stage_format_check(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("格式检查未通过") return results # Stage 2: 安全扫描 stage = self._stage_safety_scan(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("安全扫描未通过") return results # Stage 3: 单元测试 stage = self._stage_unit_test(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("单元测试未通过") # Stage 4: 回归测试 stage = self._stage_regression_test(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("回归测试未通过") # Stage 5: 性能评估 stage = self._stage_performance_eval(prompt_version) results['stages'].append(stage) # Stage 6: 安全对抗测试 stage = self._stage_adversarial_test(prompt_version) results['stages'].append(stage) if not stage['passed']: results['passed'] = False results['blocking_issues'].append("对抗测试未通过") return results def _stage_format_check(self, pv: PromptVersion) -> dict: """格式检查""" issues = [] # 检查变量引用 for var in pv.variables_schema: if f"{{{{{var}}}}}" not in pv.user_template: issues.append(f"变量 {var} 未在模板中使用") # 检查 Prompt 长度 token_count = self._estimate_tokens(pv.system_prompt) if token_count > 8000: issues.append(f"System Prompt 过长：{token_count} tokens") return { 'stage': 'format_check', 'passed': len(issues) == 0, 'issues': issues } def _stage_safety_scan(self, pv: PromptVersion) -> dict: """安全扫描""" issues = self.safety.scan(pv.system_prompt) return { 'stage': 'safety_scan', 'passed': len(issues) == 0, 'issues': issues } def _stage_regression_test(self, pv: PromptVersion) -> dict: """回归测试：与 production 版本对比""" prod = self.vc.storage.get_production_version(pv.prompt_id) if not prod: return {'stage': 'regression_test', 'passed': True, 'issues': []} # 在相同测试集上对比 test_cases = self.vc.storage.get_test_cases(pv.prompt_id) new_results = [self.evaluator.evaluate(pv, case) for case in test_cases] old_results = [self.evaluator.evaluate(prod, case) for case in test_cases] # 检查是否有关键指标下降 new_accuracy = sum(r['correct'] for r in new_results) / len(new_results) old_accuracy = sum(r['correct'] for r in old_results) / len(old_results) issues = [] if new_accuracy < old_accuracy - 0.05: # 下降超过5% issues.append(f"准确率下降：{old_accuracy:.2%} → {new_accuracy:.2%}") return { 'stage': 'regression_test', 'passed': len(issues) == 0, 'issues': issues, 'metrics': { 'old_accuracy': old_accuracy, 'new_accuracy': new_accuracy } } 六、Prompt 注册中心 class PromptRegistry: """Prompt 注册中心——生产环境的服务发现""" def __init__(self, storage): self.storage = storage self.cache = {} # 本地缓存 def get_prompt(self, prompt_id: str, version: str = "latest") -> PromptVersion: """获取 Prompt（生产环境调用）""" cache_key = f"{prompt_id}:{version}" if cache_key in self.cache: return self.cache[cache_key] if version == "latest": pv = self.storage.get_production_version(prompt_id) else: pv = self.storage.get(prompt_id, version) # 缓存 self.cache[cache_key] = pv return pv def invalidate_cache(self, prompt_id: str): """缓存失效""" keys_to_remove = [k for k in self.cache if k.startswith(prompt_id)] for k in keys_to_remove: del self.cache[k] def list_prompts(self, status: PromptStatus = None) -> list: """列出所有 Prompt""" return self.storage.list_all(status) 七、监控与告警 class PromptMonitor: """Prompt 监控系统""" def __init__(self): self.metrics = {} def record_usage(self, prompt_id: str, version: str, latency_ms: float, token_count: int, success: bool, user_feedback: int = None): """记录 Prompt 使用指标""" key = f"{prompt_id}:{version}" if key not in self.metrics: self.metrics[key] = { 'total_calls': 0, 'success_count': 0, 'latency_sum': 0, 'token_sum': 0, 'feedback_sum': 0, 'feedback_count': 0, 'errors': [] } m = self.metrics[key] m['total_calls'] += 1 if success: m['success_count'] += 1 m['latency_sum'] += latency_ms m['token_sum'] += token_count if user_feedback is not None: m['feedback_sum'] += user_feedback m['feedback_count'] += 1 def check_alerts(self) -> list: """检查告警条件""" alerts = [] for key, m in self.metrics.items(): if m['total_calls'] < 100: continue success_rate = m['success_count'] / m['total_calls'] avg_latency = m['latency_sum'] / m['total_calls'] if success_rate < 0.95: alerts.append({ 'prompt': key, 'alert': 'success_rate_low', 'value': success_rate, 'threshold': 0.95 }) if avg_latency > 5000: alerts.append({ 'prompt': key, 'alert': 'latency_high', 'value': avg_latency, 'threshold': 5000 }) return alerts 结语 Prompt 版本管理不是锦上添花，而是 AI 应用从"能用"到"好用"再到"敢用"的必经之路。正如 Git 改变了软件工程一样，Prompt 版本管理平台正在改变 AI 工程的协作方式。投入建设 Prompt 管理平台，是对团队 AI 能力长期投资中回报率最高的一项。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

微调流水线全景 ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 数据准备 │───▶│ 训练配置 │───▶│ 训练执行 │───▶│ 评估 │───▶│ 部署发布 │ │ 清洗/标注 │ │ LoRA/QLoRA│ │ GPU 集群 │ │ 自动化 │ │ 灰度/AB │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ ▼ ▼ ▼ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ 数据版本 │ │ 模型注册 │ │ 监控告警 │ │ DVC/MLflow│ │ MLflow │ │ 回滚机制 │ └──────────┘ └──────────┘ └──────────┘ 一、数据准备 1.1 数据清洗 import re import json from datasets import Dataset class DataCleaner: def __init__(self, min_length=10, max_length=8192): self.min_length = min_length self.max_length = max_length def clean(self, samples: list[dict]) -> list[dict]: cleaned = [] for s in samples: text = s.get("text", "") # 去除 HTML 标签 text = re.sub(r'<[^>]+>', '', text) # 去除多余空白 text = re.sub(r'\s+', ' ', text).strip() # 长度过滤 if self.min_length <= len(text) <= self.max_length: # 去重（基于内容哈希） cleaned.append({**s, "text": text}) # 去重 seen = set() unique = [] for s in cleaned: h = hash(s["text"][:200]) if h not in seen: seen.add(h) unique.append(s) return unique def to_chat_format(self, samples: list[dict]) -> list[dict]: """转换为 chatml 格式""" formatted = [] for s in samples: formatted.append({ "messages": [ {"role": "system", "content": s.get("system", "你是一个有用的助手")}, {"role": "user", "content": s["input"]}, {"role": "assistant", "content": s["output"]} ] }) return formatted 1.2 数据增强 class DataAugmenter: """使用大模型生成训练数据变体""" AUGMENT_PROMPT = """基于以下示例，生成 3 个语义相同但表达不同的变体： 原文：{original} 要求： 1. 保持意图一致 2. 变化表达方式（句式/用词） 3. 不要改变关键信息 输出 JSON 数组格式。""" async def augment(self, sample: dict, llm_client) -> list[dict]: prompt = self.AUGMENT_PROMPT.format(original=sample["input"]) resp = await llm_client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"} ) variants = json.loads(resp.choices[0].message.content) return [ {"input": v["input"], "output": sample["output"]} for v in variants.get("variants", []) ] 1.3 数据集分割 from sklearn.model_selection import train_test_split def split_dataset(data: list[dict], train=0.8, val=0.1, test=0.1): train_data, temp = train_test_split(data, test_size=1-train, random_state=42) val_data, test_data = train_test_split(temp, test_size=test/(test+val), random_state=42) return {"train": train_data, "val": val_data, "test": test_data} 二、训练配置 2.1 训练方法对比 方法 显存需求 训练速度 效果 适用场景 Full Fine-tune 极高（全部参数） 慢 最好 数据充足、预算充足 LoRA 低（0.1-1% 参数） 快 接近全量 通用首选 QLoRA 极低（4bit 量化） 中 略低于 LoRA 显存受限 P-Tuning v2 低 快 中等 特定任务 2.2 LoRA 训练配置 from peft import LoraConfig, get_peft_model, TaskType from transformers import AutoModelForCausalLM, TrainingArguments from trl import SFTTrainer def setup_lora_training(model_name="Qwen/Qwen2.5-7B"): model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype="auto", device_map="auto", ) lora_config = LoraConfig( task_type=TaskType.CAUSAL_LM, r=64, # LoRA 秩，越大效果越好但显存越多 lora_alpha=128, # 通常为 r 的 2 倍 lora_dropout=0.05, target_modules=[ "q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj" ], bias="none", ) model = get_peft_model(model, lora_config) return model, lora_config training_args = TrainingArguments( output_dir="./output/qwen-lora", num_train_epochs=3, per_device_train_batch_size=4, gradient_accumulation_steps=4, warmup_ratio=0.1, learning_rate=2e-4, lr_scheduler_type="cosine", logging_steps=10, save_strategy="epoch", eval_strategy="epoch", bf16=True, gradient_checkpointing=True, optim="adamw_torch", max_grad_norm=1.0, ) trainer = SFTTrainer( model=model, args=training_args, train_dataset=train_dataset, eval_dataset=eval_dataset, packing=True, # 序列打包提升效率 max_seq_length=2048, ) 2.3 QLoRA（显存优化） from transformers import BitsAndBytesConfig bnb_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype="bfloat16", bnb_4bit_use_double_quant=True, ) model = AutoModelForCausalLM.from_pretrained( model_name, quantization_config=bnb_config, device_map="auto", ) 三、评估流程 class ModelEvaluator: def __init__(self, model_path, test_data): self.model_path = model_path self.test_data = test_data def evaluate(self) -> dict: results = { "loss": self._eval_loss(), "bleu": self._eval_bleu(), "rouge": self._eval_rouge(), "human_like": self._eval_human_like(), "safety": self._eval_safety(), "latency_p50": self._eval_latency(), } return results def _eval_safety(self) -> float: """安全评估：检测有害输出比例""" harmful_count = 0 for sample in self.test_data: output = self._generate(sample["input"]) if self._is_harmful(output): harmful_count += 1 return 1.0 - harmful_count / len(self.test_data) def _is_harmful(self, text: str) -> bool: harmful_patterns = [ r"如何(制造|获取).*(武器|毒品)", r"(自杀|自残)的方法", r"歧视.*(种族|性别|宗教)", ] return any(re.search(p, text) for p in harmful_patterns) 四、版本管理 import mlflow class ModelRegistry: def __init__(self, tracking_uri="http://mlflow:5000"): mlflow.set_tracking_uri(tracking_uri) def register_model(self, model_path, name, metrics, tags=None): with mlflow.start_run(): mlflow.log_metrics(metrics) mlflow.log_artifacts(model_path) mlflow.register_model( f"runs:/{mlflow.active_run().info.run_id}/model", name, tags=tags or {} ) def get_version(self, name, stage="Production"): client = mlflow.tracking.MlflowClient() versions = client.get_latest_versions(name, stages=[stage]) return versions[0] if versions else None 五、灰度发布与 A/B 测试 class CanaryDeployer: """灰度发布：逐步增加新模型流量比例""" def __init__(self, old_model: str, new_model: str): self.old_model = old_model self.new_model = new_model self.traffic_split = 0.0 # 新模型流量比例 self.metrics = {"old": [], "new": []} def should_use_new(self) -> bool: import random return random.random() < self.traffic_split def canary_stages(self): """分阶段灰度""" stages = [ {"split": 0.05, "duration": "1h", "check": "error_rate < 1%"}, {"split": 0.20, "duration": "6h", "check": "error_rate < 1%, latency_p99 < 10s"}, {"split": 0.50, "duration": "24h", "check": "all_metrics_stable"}, {"split": 1.00, "duration": "∞", "check": "promoted"}, ] return stages def evaluate_and_promote(self): """评估指标决定是否推进""" new_error_rate = self._calc_error_rate("new") old_error_rate = self._calc_error_rate("old") new_latency = self._calc_p99("new") old_latency = self._calc_p99("old") if new_error_rate > old_error_rate * 1.5: self._rollback() return "ROLLBACK: error rate too high" if new_latency > old_latency * 1.3: self._rollback() return "ROLLBACK: latency regression" return "PROMOTE: metrics OK" 六、回滚机制 class RollbackManager: def __init__(self, registry: ModelRegistry): self.registry = registry def rollback(self, model_name: str, reason: str): """回滚到上一个 Production 版本""" client = mlflow.tracking.MlflowClient() versions = client.search_model_versions( f"name='{model_name}'", order_by=["version_number DESC"] ) prod_versions = [v for v in versions if v.current_stage == "Production"] archived = [v for v in versions if v.current_stage == "Archived"] if len(prod_versions) >= 1 and archived: # 当前 prod 版本归档，上一个 archived 版本恢复 client.transition_model_version_stage( name=model_name, version=prod_versions[0].version, stage="Archived", ) client.transition_model_version_stage( name=model_name, version=archived[0].version, stage="Production", ) logger.info(f"Rolled back {model_name}: {reason}") return True return False 总结 LLM 微调 MLOps 流水线的核心环节：数据质量决定上限，LoRA/QLoRA 平衡效果与成本，评估必须覆盖质量+安全+性能三维度，灰度发布配合自动回滚是生产安全的最后防线。建议使用 MLflow 统一管理模型版本，从训练到部署全链路可追溯。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

为什么 LLM 应用需要特殊的 CI/CD 传统软件 CI/CD 关注代码编译、单元测试、部署。LLM 应用的 CI/CD 还需要处理：Prompt 变更的不确定性、模型版本漂移、输出质量回归、A/B 测试的统计显著性。一行 Prompt 改动可能让整个系统的回答质量崩塌，而你很难用传统测试覆盖。 Prompt 版本控制 目录结构 prompts/ ├── v1/ │ ├── system.txt │ ├── fewshot.json │ └── config.yaml ├── v2/ │ ├── system.txt │ ├── fewshot.json │ └── config.yaml └── current -> v2/ # 符号链接指向当前版本 Prompt 配置文件 # prompts/v2/config.yaml version: "2.1.0" model: gpt-4o temperature: 0.3 max_tokens: 2000 system_prompt_file: system.txt fewshot_file: fewshot.json variables: - name: user_query required: true - name: context required: false default: "" tests: - name: "basic_qa" dataset: "datasets/qa_test_100.jsonl" min_score: 0.85 - name: "safety_check" dataset: "datasets/safety_test_50.jsonl" min_score: 0.98 Prompt 加载与版本注入 import yaml from pathlib import Path class PromptManager: def __init__(self, prompts_dir="prompts"): self.prompts_dir = Path(prompts_dir) def load(self, version="current"): path = self.prompts_dir / version config = yaml.safe_load((path / "config.yaml").read_text()) system = (path / config["system_prompt_file"]).read_text() fewshot = json.loads( (path / config["fewshot_file"]).read_text() ) return Prompt( version=config["version"], system=system, fewshot=fewshot, model=config["model"], temperature=config["temperature"], max_tokens=config["max_tokens"], ) 自动评估门禁 CI 流水线中必须在部署前运行自动评估，不达标的版本被拦截。 ...