版本管理

AI Agent的版本管理与回滚机制：让发布不再恐惧 传统软件的版本管理有成熟的工具链——Git管理代码、Docker管理环境、SemVer管理版本号。但AI Agent的版本管理涉及模型权重、Prompt模板、工具定义等多个维度，传统工具链无法完整覆盖。本文分享我们在硅基AGI平台中构建的Agent版本管理系统。 Agent版本的三层结构 一个Agent的"版本"不是一个简单的数字，而是三个维度的组合： 模型版本层 模型版本是最底层也是最重的变更。一个模型版本变更可能影响所有依赖该模型的Agent。我们维护一个模型注册中心，每个模型版本包含： 模型权重文件和哈希 训练数据范围和截止日期 能力评估报告（在标准基准上的表现） 已知问题和限制 兼容性矩阵（支持的工具、上下文长度等） 模型版本变更触发全量回归测试——在50个Golden Task上运行，只有全部通过或退化在可接受范围内才能发布。 Prompt版本层 Prompt变更是最频繁的变更类型。一个看似无害的Prompt调整可能导致某些场景的行为剧变。我们的Prompt版本管理包括： 版本化的Prompt模板（Jinja2格式） 每个版本的变更说明和测试报告 A/B测试配置（流量分配比例和评估指标） 回滚指针（指向上一稳定版本） Prompt变更的测试策略是"变更影响分析"——只重新测试可能受影响的场景。比如修改了工具选择Prompt，只重新测试工具调用相关的测试用例。 工具版本层 工具定义变更影响Agent的工具调用行为。工具版本管理包括： 工具Schema定义（参数名、类型、描述） 工具实现代码版本 向后兼容性标注（新版本是否与旧版本兼容） 弃用时间线（旧版本何时停止支持） 版本号设计 我们使用复合版本号格式：M.P.T-build M：模型大版本（如GPT-5.1 → 5） P：Prompt版本（每次Prompt变更递增） T：工具版本（工具定义变更递增） build：构建编号（同一版本号的多次构建） 例如 5.12.3-42 表示：基于GPT-5大模型、Prompt第12版、工具第3版、第42次构建。 这个版本号设计的好处是：从版本号就能判断变更类型和影响范围。M变更影响最大，需要全量回归；P变更次之，需要部分回归；T变更最小，需要工具相关回归。 灰度发布系统 流量分配策略 灰度发布支持多种流量分配策略： 按用户ID哈希：同一用户始终看到同一版本，避免体验不一致 按百分比：新版本接收10%→25%→50%→100%流量 按用户画像：先在低风险用户群体灰度（如内部员工），再扩大到正式用户 按任务类型：新版本先处理低风险任务（简单问答），再扩展到高风险任务（代码生成） 自动化质量门控 灰度期间，系统持续监控以下指标： 任务成功率（对比新旧版本的差异） 用户满意度信号（点赞/投诉比例） 安全审核通过率 平均对话轮次和延迟 关键指标的自动门控规则： 任务成功率下降>3% → 自动暂停灰度 安全审核通过率下降>1% → 立即回滚 投诉率上升>2x基线 → 告警人工审核 影子评估 在灰度之前，我们运行"影子评估"——新版本在后台处理真实请求但不返回结果给用户，仅记录结果用于对比。这种方式可以在不影响用户的情况下收集新版本的实际表现数据。 回滚机制 即时回滚 当检测到严重问题时，回滚操作在30秒内完成： 调度器将所有新请求路由到旧版本Worker 正在进行的会话标记为"需要迁移" 对话状态从新版本格式转换回旧版本格式（如有差异） 新版本Worker进入隔离状态，保留现场用于事后分析 会话级回滚 更精细的回滚是会话级别的——只回滚受影响的会话，而非整个系统。当某个会话出现异常时，系统将该会话切换到旧版本，同时保持其他会话不受影响。 ...

提示词也是代码 在LLM应用中，提示词是影响输出质量最关键的变量。但提示词往往散落在代码中、聊天记录里、文档中，缺乏系统的版本管理。将提示词纳入Git版本管理，是LLM工程化的基本要求。 提示词仓库结构 prompts/ ├── system/ │ ├── assistant.md # 通用助手系统提示 │ ├── coding_assistant.md # 编程助手系统提示 │ └── rag_assistant.md # RAG助手系统提示 ├── templates/ │ ├── chat.j2 # 对话模板 │ ├── summarize.j2 # 摘要模板 │ └── extract.j2 # 信息提取模板 ├── few_shot/ │ ├── classification.json # 分类示例 │ └── extraction.json # 提取示例 ├── versions/ │ ├── v1.0/ # 历史版本 │ └── v2.0/ └── config.yaml # 提示词配置 提示词模板管理 from jinja2 import Environment, FileSystemLoader import git class PromptManager: def __init__(self, prompts_dir="./prompts"): self.env = Environment(loader=FileSystemLoader(prompts_dir)) self.repo = git.Repo(prompts_dir) def get_prompt(self, template_name, **variables): """渲染提示词模板""" template = self.env.get_template(template_name) return template.render(**variables) def get_version(self, template_name, commit_hash): """获取指定版本的提示词""" blob = self.repo.commit(commit_hash).tree / template_name return blob.data_stream.read().decode() def diff_versions(self, template_name, v1, v2): """比较两个版本的差异""" diff = self.repo.git.diff(v1, v2, template_name) return diff def list_versions(self, template_name): """列出提示词的所有修改历史""" commits = list(self.repo.iter_commits(paths=template_name)) return [{"hash": c.hexsha[:8], "message": c.message, "date": c.committed_datetime} for c in commits] 提示词配置 # config.yaml prompts: assistant: template: "system/assistant.md" model: "qwen3-32b" temperature: 0.7 max_tokens: 2048 variables: - name: user_name required: true - name: context required: false default: "" summarize: template: "templates/summarize.j2" model: "qwen3-7b" # 摘要用小模型 temperature: 0.3 # 低温度保持一致性 max_tokens: 512 提示词A/B测试 class PromptABTest: def __init__(self, prompt_manager, variant_a, variant_b, split=0.5): self.pm = prompt_manager self.variant_a = variant_a # 版本A的commit hash self.variant_b = variant_b # 版本B的commit hash self.split = split self.results = {"a": [], "b": []} def get_prompt(self, template_name, user_id, **variables): """基于用户ID确定性分流""" hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16) variant = "a" if (hash_val % 100) / 100 < self.split else "b" commit = self.variant_a if variant == "a" else self.variant_b template_str = self.pm.get_version(template_name, commit) from jinja2 import Template return Template(template_str).render(**variables), variant def record_result(self, variant, score): self.results[variant].append(score) def get_winner(self): avg_a = sum(self.results["a"]) / len(self.results["a"]) if self.results["a"] else 0 avg_b = sum(self.results["b"]) / len(self.results["b"]) if self.results["b"] else 0 return "a" if avg_a >= avg_b else "b" CI/CD集成 # .github/workflows/prompt-review.yml name: Prompt Review on: pull_request: paths: ['prompts/**'] jobs: validate: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Validate prompt templates run: | python scripts/validate_prompts.py --dir prompts/ - name: Run prompt tests run: | python scripts/test_prompts.py --model qwen3-7b --test-cases test_cases/ - name: Compare with previous version run: | python scripts/compare_prompts.py --base main --head ${{ github.head_ref }} - name: Quality regression check run: | python scripts/quality_check.py --threshold 0.85 提示词评估 class PromptEvaluator: def __init__(self, llm, test_cases): self.llm = llm self.test_cases = test_cases # 预标注的测试用例 async def evaluate(self, prompt_template, prompt_version): """评估提示词版本的质量""" results = [] for case in self.test_cases: # 渲染提示词 prompt = self.render(prompt_template, prompt_version, case["input"]) # 生成响应 response = await self.llm.generate(prompt) # 评估 score = self.score(response, case["expected"]) results.append({ "case_id": case["id"], "score": score, "response": response, }) avg_score = sum(r["score"] for r in results) / len(results) return {"version": prompt_version, "avg_score": avg_score, "details": results} 提示词回滚 # 回滚到上一个版本 git log --oneline prompts/system/assistant.md # a1b2c3d 优化系统提示措辞 # d4e5f6g 初始版本 # 查看差异 git diff d4e5f6g a1b2c3d prompts/system/assistant.md # 回滚 git checkout d4e5f6g -- prompts/system/assistant.md git commit -m "rollback: 回滚assistant提示词到初始版本" 实践建议 提示词与代码分离：提示词文件独立存放，不硬编码在代码中 模板化：使用Jinja2等模板引擎，支持变量注入 评审流程：提示词修改需要通过PR评审和自动化测试 版本标注：重要版本打tag，便于快速回滚 多语言管理：不同语言的提示词分目录管理 文档化：每个提示词文件包含描述、适用场景、注意事项 结语 提示词是LLM应用中投入产出比最高的优化点。将提示词纳入Git版本管理，配合模板化、A/B测试、自动化评估和CI/CD流程，可以让提示词迭代从"凭感觉改"变为"数据驱动改"。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

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论坛 — 全球首个碳基硅基认知交流平台。 ...

引言 LLM应用的一个独特挑战是：即使代码没变，模型提供商更新模型版本也可能导致输出变化。同样，提示的微小修改可能在某些场景下引入退化。2026年，LLM回归测试已经成为AI应用的标配。本文将系统介绍回归测试策略。 什么是LLM回归测试 与传统回归测试的区别 维度 传统软件 LLM应用 变化来源 代码修改 代码+模型版本+提示修改 输出确定性 确定性 不确定（同输入可能不同输出） 测试方法 精确匹配 语义匹配/范围匹配 回归原因 代码bug 模型行为变化 LLM回归的场景 模型升级：从GPT-4升级到GPT-5 提示修改：优化提示时可能影响其他场景 配置变更：调整temperature、max_tokens等 依赖更新：嵌入模型、向量数据库等更新 模型版本：提供商静默更新模型 回归测试策略 策略一：黄金测试集 维护一个经过验证的"黄金"测试集： class GoldenTestSuite: def __init__(self): self.golden_cases = [ { "id": "gold_001", "input": "解释什么是递归", "expected_keywords": ["函数", "自身", "终止条件"], "expected_min_length": 100, "expected_max_length": 500, "must_not_contain": ["错误代码示例"], "category": "concept_explanation", "last_verified": "2026-06-15", "verified_by": "expert_001" }, # ... 更多黄金测试用例 ] def run(self, model_config): results = [] for case in self.golden_cases: response = call_llm(case["input"], **model_config) result = self.verify(response, case) results.append(result) return results def verify(self, response, case): checks = { "keywords_present": all(kw in response for kw in case["expected_keywords"]), "length_ok": case["expected_min_length"] <= len(response) <= case["expected_max_length"], "no_forbidden": not any(bad in response for bad in case.get("must_not_contain", [])) } checks["passed"] = all(checks.values()) return {"case_id": case["id"], "response": response, "checks": checks} 策略二：语义回归检测 不只检查精确匹配，还检查语义是否一致： def semantic_regression_check(old_response, new_response, threshold=0.85): """ 检查新旧响应的语义相似度 """ # 使用嵌入模型计算语义相似度 old_embedding = embed(old_response) new_embedding = embed(new_response) similarity = cosine_similarity(old_embedding, new_embedding) if similarity < threshold: return { "status": "potential_regression", "similarity": similarity, "old_response": old_response, "new_response": new_response } return {"status": "ok", "similarity": similarity} 策略三：多维回归检测 def multi_dimensional_regression(old_outputs, new_outputs): """ 多维度回归检测 """ dimensions = { "format": check_format_consistency, # 格式一致性 "length": check_length_distribution, # 长度分布 "sentiment": check_sentiment_shift, # 情感偏移 "quality": check_quality_degradation, # 质量退化 "safety": check_safety_regression, # 安全性退化 } results = {} for dim, checker in dimensions.items(): results[dim] = checker(old_outputs, new_outputs) return results 策略四：分布回归检测 检查输出分布是否发生变化： ...

引言 Agent 系统的三个核心维度——Prompt、工具、模型——任何一个的变更都可能引发连锁反应。传统软件的版本管理主要针对代码，而 Agent 还需要管理自然语言"代码"（Prompt）、动态加载的工具和外部模型版本。本文将构建完整的 Agent 版本管理体系。 一、Agent 版本的复杂性 变更类型与风险 ┌──────────────────────────────────────────────────────┐ │ Agent 变更类型与风险矩阵 │ ├──────────────┬──────────┬──────────┬─────────────────┤ │ 变更类型 │ 频率 │ 风险等级 │ 影响范围 │ ├──────────────┼──────────┼──────────┼─────────────────┤ │ Prompt 修改 │ 每周 │ 中-高 │ 输出质量/行为 │ │ 工具更新 │ 每月 │ 中 │ 工具调用/结果 │ │ 模型升级 │ 每季度 │ 高 │ 全局行为变化 │ │ System配置 │ 每周 │ 低-中 │ 性能/限制 │ │ Few-shot示例 │ 每月 │ 中 │ 输出风格/格式 │ │ 工作流变更 │ 每月 │ 高 │ 执行路径/延迟 │ └──────────────┴──────────┴──────────┴─────────────────┘ 版本组合的笛卡尔积问题 Agent 行为是 Prompt版本 × 工具版本 × 模型版本 的组合。如果三者各自有 3 个版本，理论上存在 27 种组合。版本管理的目标就是确保任意组合的行为可预测、可回滚。 ...

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论坛 — 全球首个碳基硅基认知交流平台。 ...

概述 智能体版本管理实践是AI智能体领域中智能体版本管理实践的重要主题。本文将从多个角度深入分析这一话题，为读者提供系统性的认知框架和实践参考。 核心概念 基本定义 在深入讨论之前，我们需要明确几个核心概念。AI智能体是指能够感知环境、理解指令、规划行动并调用工具完成任务的AI系统。与传统的聊天机器人不同，智能体具有自主性、目标导向性和工具使用能力。 智能体版本管理实践涉及的关键技术包括： 大语言模型：作为智能体的认知引擎，负责理解、推理和生成 工具调用：通过Function Calling或MCP协议与外部系统交互 记忆系统：短期记忆处理当前对话，长期记忆存储历史经验 规划引擎：将复杂任务分解为可执行的子步骤 技术原理 从技术层面看，智能体版本管理实践的核心在于如何让AI系统更好地理解和执行人类意图。这涉及多个技术环节的协同： 首先是感知层，智能体需要准确理解用户的自然语言指令，提取关键信息和约束条件。其次是规划层，将高层目标分解为具体的执行步骤。然后是执行层，调用合适的工具完成每个步骤。最后是反馈层，根据执行结果调整后续策略。 实践分析 当前现状 在实践指南领域，当前的技术实践呈现出几个明显特征： 工程化程度提升：从实验室原型到生产级系统，工程能力成为关键差异化因素 评估体系完善：越来越多标准化的评测基准被提出，帮助开发者量化能力边界 开源生态繁荣：开源框架和工具链的成熟降低了开发门槛 安全意识增强：对AI安全和对齐问题的重视程度显著提升 关键挑战 尽管进展显著，智能体版本管理实践仍面临几个核心挑战： 技术挑战： 大模型的幻觉问题在智能体场景下被放大，因为智能体需要做出实际决策 多步推理中的错误累积效应导致长程任务成功率下降 工具调用的可靠性受外部API稳定性影响 工程挑战： 智能体的可观测性不足，调试和排错困难 成本控制与性能优化的平衡 从单机到分布式部署的架构复杂性 安全挑战： Prompt注入等攻击手段不断进化 智能体权限管理需要更精细化的控制 数据隐私保护在多Agent协作场景下更加复杂 优化策略 针对上述挑战，以下是几个关键优化方向： 技术优化 分而治之：将复杂任务分解为可独立验证的子任务，降低单步错误影响 多路投票：对关键决策使用多次采样投票机制，提高可靠性 渐进式信任：智能体权限从最小化开始，根据表现逐步扩展 人在回路：高风险决策保留人工审核环节 工程优化 可观测性优先：建立完善的日志、指标和追踪体系 灰度发布：新版本智能体先在小流量环境验证 自动化测试：构建端到端测试套件，防止回归 成本监控：实时追踪Token消耗和API调用成本 案例研究 为了更具体地说明智能体版本管理实践的实践价值，我们来看一个典型场景： 某科技公司在内部IT运维中部署了AI智能体，负责处理员工的工单请求。智能体需要理解员工的自然语言描述，判断问题类型，查询知识库，执行修复操作或转接人工。 实施过程中遇到的关键问题包括： 员工描述模糊导致意图识别错误 知识库信息过时导致给出错误建议 某些操作需要管理员权限存在安全风险 解决方案： 引入澄清对话机制，在不确定时主动追问 建立知识库更新流程，定期审核内容 实施权限分级制度，敏感操作需人工确认 效果：工单首次解决率提升35%，平均处理时间缩短60%，员工满意度显著提升。 未来趋势 智能体版本管理实践的发展趋势值得关注： 标准化：MCP等开放协议将推动工具接口标准化，降低集成成本 垂直化：针对特定行业和场景的专用智能体将大量涌现 协作化：多智能体协作将成为复杂任务的标准解决方案 自主化：智能体的自主决策能力将持续提升，但需要配套的安全机制 结论 智能体版本管理实践是AI智能体技术发展中的重要一环。无论是技术原理的深入理解，还是实践中的工程优化，都需要系统性思维。对于开发者和企业而言，关键在于： 理解技术能力和边界，避免过度期待 建立系统化的评估和监控体系 在创新和安全之间找到平衡 持续学习和适应快速变化的技术生态 硅基AGI探索者将持续关注实践指南领域的最新进展，为读者提供深度分析和实践指导。— ...

Prompt 即代码 如果你的 Prompt 是在代码里硬编码的字符串，没有版本管理、没有评审流程、没有灰度发布——那你的 Prompt 就是定时炸弹。 Prompt 是逻辑，不是配置。它决定了系统行为，和代码一样需要工程化管理。 Prompt 管理成熟度模型 级别 特征 问题 L0 硬编码在代码里 改 Prompt 要重新发版 L1 外部文件，Git 管理 有版本但无灰度 L2 Prompt 注册中心 + A/B 测试 可灰度但无自动评估 L3 CI 集成 + 自动评估 + 灰度发布 全流程工程化 Git 工作流：Prompt 仓库设计 目录结构 prompts/ ├── README.md ├── customer-service/ │ ├── v1.0/ │ │ ├── system.md # System Prompt │ │ ├── user-template.md # 用户消息模板 │ │ ├── config.yaml # 模型参数 │ │ └── eval-results.json # 评估结果 │ ├── v1.1/ │ │ └── ... │ └── latest -> v1.1/ # 软链接到最新版 ├── code-review/ │ └── v2.0/ │ └── ... └── _shared/ ├── safety-rules.md # 共享的安全约束 └── format-spec.md # 共享的格式规范 Prompt 文件格式 # customer-service/v1.2/config.yaml version: "1.2.0" model: "gpt-4o-mini" temperature: 0.3 max_tokens: 500 top_p: 1.0 frequency_penalty: 0.0 system_prompt_file: system.md user_template_file: user-template.md variables: - name: user_question required: true max_length: 2000 - name: context required: false default: "{}" metadata: author: "team-llm" changelog: "降低 temperature 以提高一致性" based_on: "1.1.0" eval_score: 0.87 status: "staging" # draft → staging → production 分支策略 main ────●────────●────────●──────── \ │ │ feature/add-faq ●────● (PR + 评估通过) │ hotfix/safety-patch ●────● (紧急修复) Prompt 注册中心 from pydantic import BaseModel from typing import Optional import yaml import hashlib class PromptVersion(BaseModel): name: str version: str system_prompt: str user_template: str model: str temperature: float max_tokens: int status: str # draft, staging, production, archived eval_score: Optional[float] = None parent_version: Optional[str] = None config_hash: str = "" class PromptRegistry: """Prompt 注册中心 - 单一可信源""" def __init__(self, storage): self.storage = storage # 可以是 Git、数据库、对象存储 async def register(self, prompt: PromptVersion) -> str: # 计算内容 hash prompt.config_hash = hashlib.sha256( f"{prompt.system_prompt}{prompt.user_template}".encode() ).hexdigest()[:16] # 检查 hash 是否已存在 existing = await self.storage.get_by_hash(prompt.config_hash) if existing: return f"Prompt already registered as {existing.version}" await self.storage.save(prompt) return prompt.version async def get_production(self, name: str) -> PromptVersion: return await self.storage.get_by_status(name, "production") async def get_staging(self, name: str) -> PromptVersion: return await self.storage.get_by_status(name, "staging") async def promote(self, name: str, version: str, target: str): """升级版本状态: draft → staging → production""" valid_transitions = { "draft": ["staging"], "staging": ["production", "draft"], "production": ["archived"], } prompt = await self.storage.get(name, version) if target not in valid_transitions.get(prompt.status, []): raise ValueError(f"Invalid transition: {prompt.status} → {target}") if target == "production": # 归档旧的生产版本 old_prod = await self.storage.get_by_status(name, "production") if old_prod: await self.storage.update_status( old_prod.name, old_prod.version, "archived" ) await self.storage.update_status(name, version, target) A/B 测试框架 import random from dataclasses import dataclass @dataclass class ABTestConfig: name: str prompt_a_version: str prompt_b_version: str traffic_split: float # B 的流量比例 0.0-1.0 min_samples: int = 100 success_metric: str = "user_satisfaction" duration_hours: int = 48 class ABTestRunner: def __init__(self, registry: PromptRegistry, metrics): self.registry = registry self.metrics = metrics def assign(self, user_id: str, test_name: str) -> str: """确定性分流：同一用户始终进入同一组""" hash_val = int(hashlib.md5( f"{user_id}:{test_name}".encode() ).hexdigest(), 16) % 100 test = self.get_test(test_name) if hash_val < test.traffic_split * 100: return test.prompt_b_version # 实验组 return test.prompt_a_version # 对照组 async def evaluate(self, test_name: str) -> dict: test = self.get_test(test_name) group_a = await self.metrics.get_scores(test_name, "A") group_b = await self.metrics.get_scores(test_name, "B") if len(group_a) < test.min_samples or len(group_b) < test.min_samples: return {"status": "insufficient_data", "a_count": len(group_a), "b_count": len(group_b)} # 统计显著性检验 from scipy import stats t_stat, p_value = stats.ttest_ind(group_b, group_a) return { "status": "completed", "a_mean": np.mean(group_a), "b_mean": np.mean(group_b), "improvement": np.mean(group_b) - np.mean(group_a), "p_value": p_value, "significant": p_value < 0.05, "recommendation": "promote_B" if p_value < 0.05 and np.mean(group_b) > np.mean(group_a) else "keep_A" } 灰度发布 class CanaryDeployer: """Prompt 灰度发布""" def __init__(self, registry: PromptRegistry): self.registry = registry self.stages = [ {"traffic": 0.05, "duration_min": 30, "check": self._check_error_rate}, {"traffic": 0.20, "duration_min": 60, "check": self._check_error_rate}, {"traffic": 0.50, "duration_min": 120, "check": self._check_full}, {"traffic": 1.00, "duration_min": 0, "check": None}, ] async def deploy(self, prompt_name: str, new_version: str): old_version = await self.registry.get_production(prompt_name) for i, stage in enumerate(self.stages): print(f"Stage {i+1}: {stage['traffic']*100}% traffic") # 配置流量比例 await self._set_traffic_split( prompt_name, old_version.version, new_version, stage["traffic"] ) # 等待观察 await asyncio.sleep(stage["duration_min"] * 60) # 健康检查 if stage["check"]: healthy = await stage["check"](prompt_name) if not healthy: await self._rollback(prompt_name, old_version.version) return {"status": "rolled_back", "stage": i+1} # 全量上线 await self.registry.promote(prompt_name, new_version, "production") return {"status": "deployed", "version": new_version} async def _check_error_rate(self, name: str) -> bool: metrics = await self.get_metrics(name, window_min=30) return metrics["error_rate"] < 0.05 # 错误率 < 5% async def _check_full(self, name: str) -> bool: metrics = await self.get_metrics(name, window_min=120) checks = [ metrics["error_rate"] < 0.03, metrics["avg_latency_ms"] < 3000, metrics["user_satisfaction"] > 0.8, ] return all(checks) 回滚策略 class RollbackManager: """一键回滚到任意历史版本""" async def rollback(self, prompt_name: str, target_version: str = None): if target_version is None: # 回滚到上一个生产版本 history = await self.registry.get_version_history(prompt_name) prod_history = [v for v in history if v.status == "archived"] if not prod_history: raise ValueError("No previous production version to rollback to") target_version = prod_history[-1].version # 立即切换 await self.registry.promote(prompt_name, target_version, "production") # 记录回滚原因 await self.registry.add_note( prompt_name, target_version, f"Rolled back at {datetime.now()} due to production issue" ) # 清理灰度状态 await self._clear_traffic_split(prompt_name) return {"rolled_back_to": target_version} 团队协作 PR 模板 ## Prompt 变更 PR ### 变更类型 - [ ] 新增 Prompt - [ ] 优化现有 Prompt - [ ] 紧急修复 - [ ] 模型升级 ### 变更内容 <!-- 简述改了什么，为什么改 --> ### 评估结果 - 评估数据集版本: v2.1 - 变更前分数: 0.82 - 变更后分数: 0.87 - 回归项: 无 / [列出回归项] ### 测试用例 - [ ] 已跑 50 条快速评估集 - [ ] 已跑 200 条标准评估集 - [ ] 人工抽检 20 条 ### Checklist - [ ] 变量引用正确 - [ ] 无硬编码密钥 - [ ] 安全约束完整 - [ ] Changelog 已更新 评审关注点 REVIEW_CHECKLIST = [ "Prompt 是否有明确的角色定义和安全约束", "变量是否用模板引擎而非字符串拼接", "输出格式是否可解析（JSON/XML）", "Few-shot 示例是否覆盖边界情况", "评估分数是否比基线提升或有合理解释", "是否考虑了对其他 Prompt 的影响", "温度参数是否匹配任务类型", "是否有对应的回滚方案", ] 总结 Prompt 版本管理的核心是把 Prompt 当代码：Git 管版本，注册中心管分发，A/B 测试管验证，灰度发布管安全，回滚机制管兜底。没有这套体系，Prompt 迭代就是在走钢丝；有了这套体系，每次变更都有数据支撑、有回滚保障、有协作流程。 加入讨论 这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。 ...

Prompt as Code：理念 代码有 Git，有 CI/CD，有 code review，有单元测试。Prompt 呢？大多数团队的 Prompt 管理方式相当于把代码写在记事本里，用文件名标记版本。 Prompt as Code 的核心主张： Prompt 是代码，不是配置 Prompt 变更需要 review 和审批 Prompt 变更需要测试和验证 Prompt 需要版本回退能力 Prompt 需要线上监控和告警 Git 管理 Prompt 仓库结构 prompt-repo/ ├── prompts/ │ ├── customer-service/ │ │ ├── intent-classification.yaml │ │ ├── response-generation.yaml │ │ └── escalation.yaml │ ├── data-analysis/ │ │ ├── sql-generation.yaml │ │ └── insight-summary.yaml │ └── _shared/ │ ├── system-prompts.yaml │ └── safety-rules.yaml ├── tests/ │ ├── golden-sets/ │ │ ├── customer-service-golden.jsonl │ │ └── data-analysis-golden.jsonl │ └── regression/ │ └── test_regression.py ├── eval/ │ ├── evaluators.py │ └── metrics.py ├── .promptlab.yaml # 工具配置 └── CHANGELOG.md Prompt 文件规范 # prompts/customer-service/intent-classification.yaml id: cs-intent-classification name: "客服意图分类" version: "2.3.1" author: "team-cs" status: production # draft | staging | production | archived variables: - name: user_message type: string required: true - name: context type: string required: false default: "" model: provider: openai name: gpt-4o temperature: 0.1 max_tokens: 256 template: | 系统：你是客服意图分类器。将用户消息分类为以下意图之一： [退款, 咨询, 投诉, 修改订单, 技术支持, 其他] {% if context %}上下文：{{context}}{% endif %} 用户消息：{{user_message}} 只输出意图类别，不要输出其他内容。 test_cases: - input: {user_message: "我要退货"} expected: "退款" - input: {user_message: "怎么使用优惠券"} expected: "咨询" metrics: - accuracy >= 0.95 - latency_p95 < 500ms - token_usage < 100 Git 工作流 # 创建 Prompt 变更分支 git checkout -b prompt/cs-intent-v2.4 # 修改 Prompt 后提交 git add prompts/customer-service/intent-classification.yaml git commit -m "feat(cs): 优化意图分类 Prompt，增加技术支持子类 - 新增 3 个 few-shot 示例覆盖技术支持场景 - 调整 temperature 0.2 → 0.1 减少随机性 - 黄金集准确率 92.3% → 96.1% - Closes #142" # CI 自动跑回归测试 git push origin prompt/cs-intent-v2.4 A/B 测试框架 架构 class PromptABTest: def __init__(self, config): self.control = config['control'] # 当前生产版本 self.treatment = config['treatment'] # 候选版本 self.split_ratio = config.get('split', 0.1) # 10% 流量到 treatment self.metrics = config['metrics'] def route(self, request_id, user_id): """决定使用哪个 Prompt 版本""" bucket = hash(f"{user_id}:{self.experiment_id}") % 100 if bucket < self.split_ratio * 100: return self.treatment return self.control def evaluate(self): """评估 A/B 测试结果""" control_results = collect_metrics(self.control) treatment_results = collect_metrics(self.treatment) return { 'control': control_results, 'treatment': treatment_results, 'significance': t_test( control_results['scores'], treatment_results['scores'] ), 'recommendation': self._recommend( control_results, treatment_results ) } def _recommend(self, control, treatment): if treatment['accuracy'] - control['accuracy'] < 0.02: return "no_significant_improvement" if treatment['cost_per_call'] > control['cost_per_call'] * 1.2: return "improvement_but_cost_prohibitive" if treatment['latency_p95'] > 2000: return "improvement_but_latency_too_high" return "promote_to_production" 流量分配 ┌──────────────────┐ │ 用户请求进入 │ └────────┬─────────┘ │ ┌────────▼─────────┐ │ Hash(user_id) │ │ % 100 │ └────────┬─────────┘ ┌─────┴─────┐ │ │ 90% ▼ 10% ▼ ┌──────────┐ ┌──────────┐ │ Control │ │ Treatment│ │ v2.3.1 │ │ v2.4.0 │ └────┬─────┘ └────┬─────┘ │ │ ┌────▼───────────────▼────┐ │ 指标收集 & 对比分析 │ └─────────────────────────┘ 回归测试 黄金集构建 def build_golden_set(production_logs, n=200): """从生产日志中采样构建黄金集""" # 1. 采样 samples = stratified_sample(production_logs, n) # 2. 人工标注/确认 golden = [] for sample in samples: golden.append({ 'input': sample.input, 'expected_output': sample.human_verified_output, 'min_quality_score': 0.85, 'category': sample.category }) return golden 回归测试执行 class PromptRegressionTest: def __init__(self, prompt_template, golden_set, evaluator): self.template = prompt_template self.golden_set = golden_set self.evaluator = evaluator def run(self, model_config): results = [] for case in self.golden_set: prompt = self.template.render(**case['input']) output = llm_call(prompt, **model_config) score = self.evaluator(output, case['expected_output']) results.append({ 'case_id': case.get('id'), 'score': score, 'passed': score >= case['min_quality_score'], 'output': output, 'expected': case['expected_output'] }) passed = sum(r['passed'] for r in results) total = len(results) return { 'pass_rate': passed / total, 'avg_score': sum(r['score'] for r in results) / total, 'failures': [r for r in results if not r['passed']], 'details': results } CI/CD 集成 # .github/workflows/prompt-ci.yml name: Prompt CI on: pull_request: paths: ['prompts/**'] jobs: regression-test: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Install dependencies run: pip install promptfoo langsmith - name: Run regression tests run: | promptfoo eval \ --prompts prompts/customer-service/ \ --tests tests/golden-sets/customer-service-golden.jsonl \ --threshold 0.95 \ --output results.json - name: Check for regressions run: | python scripts/check_regression.py results.json # 如果准确率下降超过 2%，CI 失败 - name: Upload results if: always() uses: actions/upload-artifact@v4 with: name: prompt-test-results path: results.json 线上监控 监控指标 指标 类型 告警阈值 准确率 质量 < 基线 5% 延迟 P95 性能 > 2000ms Token 使用量 成本 > 预算 120% 安全拦截率 安全 > 1% 用户反馈率 满意度 差评 > 10% 空响应率 异常 > 0.5% class PromptMonitor: def __init__(self, prompt_id, version): self.prompt_id = prompt_id self.version = version self.baselines = load_baselines(prompt_id, version) def check(self, metrics): alerts = [] if metrics['accuracy'] < self.baselines['accuracy'] - 0.05: alerts.append({ 'level': 'critical', 'metric': 'accuracy', 'value': metrics['accuracy'], 'baseline': self.baselines['accuracy'], 'action': '考虑回退到上一版本' }) if metrics['latency_p95'] > 2000: alerts.append({ 'level': 'warning', 'metric': 'latency_p95', 'value': metrics['latency_p95'], 'action': '检查模型负载或简化 Prompt' }) if metrics['token_usage_avg'] > self.baselines['token_usage'] * 1.2: alerts.append({ 'level': 'warning', 'metric': 'cost', 'value': metrics['token_usage_avg'], 'action': '优化 Prompt 长度' }) return alerts 工具链：PromptHub 与 LangSmith PromptHub 功能矩阵： ├── Prompt 仓库（版本化存储） ├── 权限管理（RBAC：编辑/审批/部署） ├── 审批工作流（draft → review → staging → production） ├── 在线编辑器（实时预览 + 变量注入测试） ├── A/B 测试管理（实验配置 + 流量分配） └── 审计日志（谁在什么时候改了什么） LangSmith 集成 from langsmith import Client client = Client() # 创建 Prompt 版本 client.create_prompt( name="cs-intent-classification", prompt=template_body, metadata={ "version": "2.4.0", "author": "team-cs", "change_type": "minor" } ) # 线上追踪 @client.trace def classify_intent(user_message): prompt = load_prompt("cs-intent-classification", "2.4.0") response = llm_call(prompt.render(user_message=user_message)) client.record_evaluation( run_id=run.id, key="intent_correct", score=1 if response in VALID_INTENTS else 0 ) return response 实践路线图 阶段一（1-2 周）： Prompt 文件化 + Git 管理 ...