CI

做过提示词工程的人都懂这种痛：上周还好好的回复，今天改了一段系统提示，某个边角场景突然开始胡来，而你根本不知道。解法不是更小心，而是把评测变成自动化测试。本文用 DeepEval 把 LLM 评测写进 pytest，接进 CI。 一条 DeepEval 用例长什么样？ DeepEval 把评测指标封装成断言：用一个样例，包含输入与期望输出，配上一个或多个指标，例如答案相关性、事实一致性、严格匹配。写起来和普通 pytest 几乎一样：定义样例列表，循环调用被测智能体，断言指标不低于阈值。逻辑上就是：对黄金集里的每条用例，跑一遍被测流程，再断言相关分数高于你设的红线。 黄金集该怎么一步步攒起来？ 别一开始就建几百条。先从线上真实工单里挑三五十条「最容易出错」的典型问题，人工标注期望行为。每次线上翻车，就把这个案例补进黄金集——它就会变成一条永远不许再犯的回归用例。黄金集是长出来的，不是设计出来的。 把评测接进 CI 有多大成本？ 评测要调模型，比普通单测慢且花钱。做法是：提交请求级别只跑子集做冒烟，合并到主干才跑全量。再给评测成本设个上限，避免半夜 CI 跑飞烧掉一大笔 API 费。 大模型评测分数该怎么正确看待？ DeepEval 这类「用大模型评大模型」的指标不是绝对真理，它更像一个报警器：分数掉了说明有变化，值不值得修要人判断。别把它当成精确的质量分，要把它当成「改动有没有让事情变坏」的护栏。 指标一开始该选几个才合适？ 一开始别把七八个指标全挂上，每加一个指标，评测成本和噪音都涨。先选两个最贴你业务的：答案相关性和事实一致性。跑稳了，再按需加。指标是手段不是目的，你要的是「改动没让事情变坏」这一条报警线，不是一张看着漂亮却没人看的仪表盘。 为什么提示词也需要回归测试？ 提示词和智能体代码和软件一样需要回归测试。把几十条黄金用例用 DeepEval 写进 pytest，你就拥有了一张安全网——下次再改提示词，心里有底，而不是靠玄学。 去论坛讨论 这篇文章里的观点，欢迎到硅基AGI论坛讨论。按标题搜索：DeepEval回归

提示词工程有个隐性痛点：它一直被当作文档，而不是代码。改一句系统提示、换一个 few-shot 示例，往往是「我觉得更顺了」就上线，没人知道这次改动是不是把某个边缘问题改坏了。promptfoo 要补的就是这个洞——它把提示词连同一组测试用例，变成一个能在命令行重复跑、能接进 CI 的评测脚本。 它怎么工作 写一个配置文件，列三件事：要测哪些提示词变体、一组输入样例、以及判定标准（正则包含、JSON 字段校验、或让另一个模型打分）。然后跑一行命令：npx promptfoo eval，它把每个提示词在每个样例上都跑一遍，输出一张对照表。 再用 npx promptfoo view 打开本地界面，就能并排看：旧提示词和新提示词，在这 50 个用例上分别答成什么样。 接进 CI 的价值 把这一步挂进 GitHub Actions：每次有人改提示词文件，自动跑评测，失败率上升超过阈值就拦住合并。提示词从此有了「测试套」——改坏了能立刻发现，而不是靠用户反馈。 判定标准别贪多。一开始只测硬指标：输出能不能被你的下游 JSON schema 解析、关键事实有没有出现。软的「答得好不好」交给人工抽检，否则你会陷入给模型打分的模型打分。 判断 2026 年做 AI 应用，提示词已经是生产代码的一部分。既然是代码，就该享受代码的待遇：版本控制、评审、回归测试。promptfoo 这类工具把「凭感觉改提示词」变成「带着测试改提示词」，这是从玩具走向产品的必经一步。 补充一个实践点：测试用例要从真实失败里积累。每次用户反馈答得不好，就把那个问题和期望答案补进 promptfoo 的用例库。半年下来，你的测试套就是一份「我们踩过的坑清单」，提示词改动再也不会重蹈覆辙，新成员接手也能一眼看懂这条提示词经历过什么。 提示词是生产代码，就该享受版本控制和回归测试的待遇。 去论坛讨论 这篇文章里的观点，欢迎到硅基AGI论坛讨论。按标题搜索：提示词测试

AI 写代码越来越快，瓶颈反而卡在「谁来审」上。9 月底 OpenAI DevDay 之后，带云电脑的编码智能体已经能自己提 PR，但合并前那道人工评审，仍是团队质量的最后一道闸。把这道闸的初筛交给 AI，是性价比极高的一步——不需要买重型平台，一个 GitHub Actions 工作流就够。 工作流长什么样 核心是三步：触发 PR、拉取 diff、调模型评审、把结果写回评论。在仓库 .github/workflows/ai-review.yml 里： on: [pull_request]，限定 opened / synchronize； 用 actions/checkout@v4 拉代码，git diff origin/main...HEAD 拿到本次改动； 把 diff 作为 user message 发给任意兼容 OpenAI 协议的接口（DeepSeek、智谱、或自建网关都行），API key 存在仓库 Secrets 的 LLM_API_KEY； 用 actions/github-script 把模型返回贴成 PR 评论。 提示词只抓四件事：明显 bug、空指针或越界、安全性（注入、硬编码密钥）、以及「这次改动是否超过了它该做的范围」。 两个容易踩的坑 第一，别让它写风格意见。一旦允许模型评论「变量名不够语义化」，PR 评论区会被无效意见淹没，开发者很快把机器人静音。把输出限定为「必须修改」与「建议确认」两档。 第二，控制 diff 体积。一次改两千行的 PR 直接喂模型，既慢又贵还漏看。超过 800 行就只评审变更文件的关键函数，或拆成多轮。 落地建议 先在内部仓库灰度两周，让团队投票保留哪些意见类型。AI 评审不会替代人，但它把新人最容易犯的那类低级错误挡在人工评审之前，Reviewer 的注意力才能留给真正难的设计权衡。这才是它该待的位置。 补充一个实操细节：首次接入时，建议先把评审设为「仅建议、不拦截」，运行一周收集机器人给出的意见，让团队投票保留哪几类。那些真正挡住 bug 的意见留下，只会制造噪音的删掉。此外，对大改动 PR 可以只做增量评审——用 git diff 限定到本次变更的函数，避免模型把精力花在历史代码上，也省 token。 ...

当AI遇见CI/CD 传统CI/CD流水线是确定性的规则系统：触发条件→构建→测试→部署。这套体系在过去十年极大提升了软件交付效率，但也暴露了明显的僵化——流水线无法理解代码变更的语义，对所有提交执行相同的检查流程。 Codex智能体的引入，使得流水线具备了语义感知能力：根据变更内容动态调整检查策略、自动生成测试用例、预测部署风险。 集成架构总览 开发者提交PR │ ▼ ┌──────────────┐ ┌──────────────┐ │ GitHub/GitLab│────▶│ 触发器 │ │ Webhook │ │ (事件路由) │ └──────────────┘ └──────┬───────┘ │ ┌────────▼────────┐ │ Codex智能体 │ │ (语义分析层) │ └────────┬────────┘ │ ┌─────────────┼─────────────┐ ▼ ▼ ▼ ┌─────────┐ ┌──────────┐ ┌──────────┐ │ 测试生成 │ │ 风险评估 │ │ 审查路由 │ └────┬────┘ └─────┬────┘ └─────┬────┘ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌──────────┐ ┌──────────┐ │ 运行测试 │ │ 决策门禁 │ │ 人工/自动 │ └────┬────┘ └─────┬────┘ └──────────┘ │ │ └──────┬──────┘ ▼ ┌───────────┐ │ 部署决策 │ └───────────┘ 关键集成点 1. 语义变更分析 传统流水线根据文件路径和diff大小决定检查策略。Codex则能理解变更的语义意图： ...

为什么需要自动化评测管线 LLM上线后并非一劳永逸——模型供应商的静默更新、prompt的迭代、工具接口的变化，都可能导致线上效果漂移。人工评测周期长、成本高、不可重复。自动化评测管线是保障线上质量的唯一可行路径。 一条完整的评测管线应该做到：定时执行、自动评分、异常告警、趋势可视化。 整体架构 ┌─────────────┐ ┌──────────────┐ ┌─────────────┐ │ 测试集仓库 │───→│ 评测调度器 │───→│ 评测执行器 │ │ (Git) │ │ (Cron/CI) │ │ (Parallel) │ └─────────────┘ └──────────────┘ └──────┬──────┘ │ ┌───────────────────────┘ ↓ ┌────────────────┐ ┌──────────────┐ │ 评分引擎 │───→│ 结果存储 │ │ (Multi-Judge) │ │ (Timeseries)│ └────────────────┘ └──────┬───────┘ │ ┌───────────┼───────────┐ ↓ ↓ ↓ ┌────────┐ ┌────────┐ ┌────────┐ │ 告警 │ │ 仪表盘 │ │ 报告 │ │(Alert) │ │(Grafana)│ │(Report)│ └────────┘ └────────┘ └────────┘ 测试集管理 测试集结构 from pydantic import BaseModel, Field from typing import Any from enum import Enum import json class Difficulty(Enum): EASY = "easy" MEDIUM = "medium" HARD = "hard" class TestCategory(Enum): REASONING = "reasoning" CODING = "coding" SAFETY = "safety" INSTRUCTION_FOLLOWING = "instruction_following" class TestCase(BaseModel): id: str category: TestCategory difficulty: Difficulty prompt: str expected_output: str | None = None # 精确匹配 expected_pattern: str | None = None # 正则匹配 evaluator_config: dict = Field(default_factory=dict) # 评测器特定配置 tags: list[str] = [] metadata: dict = {} class TestSuite(BaseModel): name: str version: str description: str cases: list[TestCase] def to_file(self, path: str): with open(path, "w", encoding="utf-8") as f: json.dump(self.model_dump(), f, ensure_ascii=False, indent=2) # 构建测试集 suite = TestSuite( name="core-abilities-v2", version="2.1.0", description="核心能力评测集", cases=[ TestCase( id="reason_001", category=TestCategory.REASONING, difficulty=Difficulty.MEDIUM, prompt="一个房间有3个开关控制隔壁房间的3盏灯。你只能去隔壁房间一次。如何确定每个开关对应哪盏灯？", expected_output="打开开关1等5分钟，关掉开关1打开开关2，去隔壁房间。亮的灯对应开关2，摸起来热的对应开关1，凉的对应开关3。", evaluator_config={"method": "llm_judge", "criteria": "逻辑正确性"}, tags=["logic", "puzzle"] ), TestCase( id="code_001", category=TestCategory.CODING, difficulty=Difficulty.EASY, prompt="写一个Python函数，输入列表返回第二大的元素", expected_pattern=r"def\s+\w+\(.*\).*:.*\n.*return.*\n", evaluator_config={"method": "unit_test", "test_function": "find_second_largest"}, tags=["python", "algorithm"] ) ] ) 评测执行器 并行执行框架 import asyncio from dataclasses import dataclass from datetime import datetime @dataclass class EvaluationContext: model_name: str model_version: str temperature: float test_suite_name: str test_suite_version: str run_id: str timestamp: str class ParallelEvaluator: def __init__(self, llm_client, max_concurrent: int = 10): self.llm = llm_client self.semaphore = asyncio.Semaphore(max_concurrent) async def run_suite(self, suite: TestSuite, ctx: EvaluationContext) -> list[dict]: tasks = [ self._run_single(case, ctx) for case in suite.cases ] results = await asyncio.gather(*tasks, return_exceptions=True) return [r for r in results if not isinstance(r, Exception)] async def _run_single(self, case: TestCase, ctx: EvaluationContext) -> dict: async with self.semaphore: start = datetime.now() # 调用LLM response = await self.llm.chat( model=ctx.model_name, messages=[{"role": "user", "content": case.prompt}], temperature=ctx.temperature ) elapsed = (datetime.now() - start).total_seconds() # 评测 score, detail = await self._evaluate(case, response) return { "case_id": case.id, "category": case.category.value, "difficulty": case.difficulty.value, "response": response, "score": score, "detail": detail, "latency_s": elapsed, "input_tokens": response.usage.prompt_tokens, "output_tokens": response.usage.completion_tokens, "run_id": ctx.run_id } async def _evaluate(self, case: TestCase, response) -> tuple[float, dict]: method = case.evaluator_config.get("method", "exact") if method == "exact": return self._exact_match(response.content, case.expected_output), {"method": "exact"} elif method == "regex": return self._regex_match(response.content, case.expected_pattern), {"method": "regex"} elif method == "llm_judge": return await self._llm_judge(response.content, case.evaluator_config) elif method == "unit_test": return await self._unit_test(response.content, case.evaluator_config) return 0.0, {"error": "unknown method"} async def _llm_judge(self, response: str, config: dict) -> tuple[float, dict]: criteria = config.get("criteria", "正确性") judge_prompt = f"""请评估以下回答的质量，维度: {criteria} 回答: {response} 评分标准: - 0.0-0.3: 完全错误 - 0.4-0.6: 部分正确，有缺陷 - 0.7-0.8: 基本正确，小瑕疵 - 0.9-1.0: 完全正确且优秀 只输出一个0到1的浮点数。""" judge_resp = await self.llm.chat( model="gpt-4o", messages=[{"role": "user", "content": judge_prompt}], temperature=0.0 ) try: score = float(judge_resp.content.strip()) except ValueError: score = 0.0 return score, {"method": "llm_judge", "raw": judge_resp.content} 结果存储与趋势分析 时序数据存储 from dataclasses import asdict import sqlite3 import json from datetime import datetime class EvalResultStore: def __init__(self, db_path: str = "eval_results.db"): self.conn = sqlite3.connect(db_path) self._init_db() def _init_db(self): self.conn.execute(""" CREATE TABLE IF NOT EXISTS eval_results ( run_id TEXT, case_id TEXT, category TEXT, difficulty TEXT, score REAL, latency_s REAL, input_tokens INTEGER, output_tokens INTEGER, model_name TEXT, model_version TEXT, test_suite TEXT, timestamp TEXT, detail TEXT ) """) self.conn.execute(""" CREATE INDEX IF NOT EXISTS idx_run ON eval_results(run_id) """) self.conn.commit() def save(self, results: list[dict]): for r in results: self.conn.execute( "INSERT INTO eval_results VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (r["run_id"], r["case_id"], r["category"], r["difficulty"], r["score"], r["latency_s"], r["input_tokens"], r["output_tokens"], r.get("model_name",""), r.get("model_version",""), r.get("test_suite",""), datetime.now().isoformat(), json.dumps(r["detail"])) ) self.conn.commit() def get_trend(self, category: str, days: int = 30) -> list[dict]: rows = self.conn.execute(""" SELECT date(timestamp) as date, avg(score) as avg_score, count(*) as n, avg(latency_s) as avg_latency FROM eval_results WHERE category = ? AND timestamp >= date('now', ?) GROUP BY date(timestamp) ORDER BY date """, (category, f"-{days} days")).fetchall() return [{"date": r[0], "score": r[1], "count": r[2], "latency": r[3]} for r in rows] 告警系统 漂移检测与告警 from collections import deque import statistics class DriftAlerter: def __init__(self, config: dict): self.window = deque(maxlen=config.get("window_size", 50)) self.threshold = config.get("threshold", 0.05) self.alert_channels = config.get("channels", []) self.category_baselines: dict[str, float] = {} def set_baseline(self, category: str, score: float): self.category_baselines[category] = score async def check(self, results: list[dict]): category_scores = {} for r in results: cat = r["category"] category_scores.setdefault(cat, []).append(r["score"]) alerts = [] for cat, scores in category_scores.items(): avg = statistics.mean(scores) baseline = self.category_baselines.get(cat) if baseline and (baseline - avg) > self.threshold: alerts.append({ "severity": "warning", "category": cat, "baseline": baseline, "current": avg, "drop": baseline - avg, "message": f"类别 {cat} 得分从 {baseline:.3f} 下降到 {avg:.3f}" }) # 绝对值告警 if avg < 0.5: alerts.append({ "severity": "critical", "category": cat, "current": avg, "message": f"类别 {cat} 得分低于0.5: {avg:.3f}" }) for alert in alerts: await self._send_alert(alert) return alerts async def _send_alert(self, alert: dict): for channel in self.alert_channels: if channel["type"] == "webhook": await self._send_webhook(channel["url"], alert) elif channel["type"] == "email": await self._send_email(channel["address"], alert) 调度器：CI/CD集成 Cron + GitHub Actions混合调度 # .github/workflows/llm-eval.yml name: LLM Evaluation Pipeline on: schedule: - cron: "0 2,14 * * *" # 每天2:00和14:00执行 workflow_dispatch: {} # 手动触发 jobs: evaluate: runs-on: ubuntu-latest steps: - uses: actions/checkout@v4 - name: Setup Python uses: actions/setup-python@v5 with: python-version: "3.12" - name: Install Dependencies run: pip install -r eval/requirements.txt - name: Run Evaluation env: OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }} ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }} run: | python -m eval.pipeline \ --suite test-suites/core-abilities-v2.json \ --models config/models.yaml \ --output results/$(date +%Y%m%d_%H%M).json - name: Check Alerts run: python -m eval.alerter --results results/*.json - name: Update Dashboard if: always() run: python -m eval.dashboard --update --results results/*.json - name: Upload Results if: always() uses: actions/upload-artifact@v4 with: name: eval-results path: results/ 仪表盘指标 指标 类型 说明 各类别平均分 时序线图 趋势监控 评分分布 直方图 离散度分析 延迟P50/P95 时序线图 性能监控 Token消耗 柱状图 成本监控 告警次数 计数器 质量监控 通过率 百分比 合规监控 总结 自动化评测管线是LLM系统的"免疫系统"。核心设计要点： ...

Codex CLI 2026 现状 2026年，OpenAI Codex已不再是2021年那个代码生成模型，而是一套完整的AI编程智能体： CLI版本：v0.130.0（2026-05-08） GitHub Star：83,200+ 核心能力：读代码、改代码、跑命令、做审查 支持模型：gpt-5-codex、gpt-5.6 Sol 与GitHub Copilot相比，Codex CLI的优势在于直接操作本地文件的能力——不只是给你代码片段让你粘贴，而是直接改你的项目。 环境搭建 系统要求 要求 最低版本 推荐版本 操作系统 macOS 12+ / Ubuntu 20.04+ / Win11(WSL2) macOS 14+ / Ubuntu 22.04+ Node.js 18.0+ 22.0+ npm 10.0+ 最新版 Git 2.23+ 2.40+ 内存 4GB 8GB+ ⚠️ Windows注意：Codex CLI暂不支持原生Windows，需要通过WSL2或使用Codex App。 安装 # 方式一：npm安装（推荐） npm install -g @openai/codex # 国内加速 npm install -g @openai/codex --registry=https://registry.npmmirror.com # 方式二：Codex App（Windows用户） # 从OpenAI官网下载Codex App 认证配置 # 启动并登录 codex # 系统自动唤起浏览器进行ChatGPT账号授权 # Token自动保存，无需每次输入API Key # 配置默认模型 # ~/.codex/config.toml [model] default = "gpt-5-codex" reasoning = "high" # low/medium/high 三种使用模式 1. CLI交互模式 $ codex > 帮我审查src/auth目录下的代码，重点关注安全问题 # Codex会： # 1. 读取src/auth/下所有文件 # 2. 分析代码逻辑和安全风险 # 3. 输出审查报告和修改建议 # 4. 可选择直接应用修改 2. 非交互模式 # 直接执行任务 codex "创建一个Hello World的Python脚本" # 指定目录 codex --dir /path/to/project "修复所有TypeScript错误" # 管道输入 cat error.log | codex "分析这个错误日志，找出根本原因" 3. 云端网页版 访问 chatgpt.com/codex： ...

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自动化测试已经形成了一套完整的方法论和工具链。本文将系统介绍如何构建LLM自动化测试体系。 LLM测试的挑战 挑战一：输入空间无限 传统软件的输入是结构化的（如表单字段），而LLM的输入是自由文本，可能的组合无限多。 挑战二：输出不确定 同一个输入，LLM可能给出不同的输出（即使temperature=0，不同版本也可能不同）。 挑战三：质量主观 “好的回答"的定义因任务、用户、上下文而异。 挑战四：成本高 每次测试都需要调用LLM API，大规模测试成本可观。 测试分层架构 ┌─────────────────────────────────┐ │ 端到端测试 (E2E) │ ← 模拟真实用户场景 ├─────────────────────────────────┤ │ 集成测试 │ ← 测试模块间协作 ├─────────────────────────────────┤ │ 提示测试 │ ← 测试提示效果 ├─────────────────────────────────┤ │ 单元测试 │ ← 测试单个功能 └─────────────────────────────────┘ 第一层：单元测试 测试LLM的基本功能： class TestLLMUnit: def test_json_output(self): """测试JSON输出格式""" response = llm.generate("输出JSON格式：{\"name\": \"test\"}") assert is_valid_json(response) def test_language_detection(self): """测试语言识别""" response = llm.generate("用中文回复：Hello") assert is_chinese(response) def test_length_constraint(self): """测试长度约束""" response = llm.generate("用50字以内回答：什么是AI？") assert len(response) <= 100 # 中文字符数 第二层：提示测试 测试提示在不同输入上的表现： class TestPrompt: def test_sentiment_analysis(self): """测试情感分析提示""" test_cases = [ {"input": "太棒了！", "expected": "positive"}, {"input": "太差了！", "expected": "negative"}, {"input": "还行吧。", "expected": "neutral"}, ] for case in test_cases: response = llm.generate(prompt.format(input=case["input"])) assert case["expected"] in response.lower() def test_robustness(self): """测试提示鲁棒性""" # 测试不同表述方式 inputs = [ "这部电影怎么样？", "你觉得这部电影如何？", "评价一下这部电影", ] responses = [llm.generate(prompt.format(input=i)) for i in inputs] # 检查输出是否一致（语义相似度） for i in range(len(responses)-1): similarity = compute_similarity(responses[i], responses[i+1]) assert similarity > 0.7 第三层：集成测试 测试多个模块协作： ...

引言 Agent系统的CI/CD比传统应用复杂——除了代码变更外，Prompt模板变更、工具定义变更、模型版本切换都可能影响系统行为。一个完整的Agent CI/CD流水线需要覆盖代码、配置、模型三个维度的变更管理，并建立严格的质量门禁确保每次发布都不会降低系统质量。 CI/CD流水线全景 ┌─────────────────────────────────────────────────────────────┐ │ Agent CI/CD Pipeline │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Code │ │ Build │ │ Test │ │ Deploy │ │ │ │ Commit │──▶│ & Push │──▶│ & QA │──▶│ to Prod │ │ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ │ │ │ ▼ ▼ ▼ ▼ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Lint & │ │ Image │ │ Unit │ │ Staging │ │ │ │ Format │ │ Build │ │ Tests │ │ Deploy │ │ │ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ ┌──────────┐ ┌──────────┐│ │ │ Integration│ │ Canary ││ │ │ Tests │ │ Deploy ││ │ └──────────┘ └──────────┘│ │ │ │ │ ┌──────────┐ │ │ │ GA Deploy│ │ │ └──────────┘ │ └─────────────────────────────────────────────────────────────┘ 代码提交与构建 # .github/workflows/ci.yml name: Agent CI Pipeline on: push: branches: [main, develop] pull_request: branches: [main] jobs: lint: runs-on: ubuntu-latest steps: - uses: actions/checkout@v3 - name: Set up Python uses: actions/setup-python@v4 with: python-version: '3.11' - name: Install dependencies run: pip install flake8 black mypy - name: Lint with flake8 run: flake8 agent/ --max-line-length=120 --ignore=E203,W503 - name: Format check with black run: black --check agent/ - name: Type check with mypy run: mypy agent/ --ignore-missing-imports build: needs: lint runs-on: self-hosted # 需要Docker支持 steps: - uses: actions/checkout@v3 - name: Set up Docker Buildx uses: docker/setup-buildx-action@v2 - name: Log in to Container Registry uses: docker/login-action@v2 with: registry: ${{ secrets.REGISTRY_URL }} username: ${{ secrets.REGISTRY_USER }} password: ${{ secrets.REGISTRY_PASSWORD }} - name: Build and Push Docker Image uses: docker/build-push-action@v4 with: context: . push: true tags: | ${{ secrets.REGISTRY_URL }}/agent-service:${{ github.sha }} ${{ secrets.REGISTRY_URL }}/agent-service:latest cache-from: type=gha cache-to: type=gha,mode=max - name: Scan Image for Vulnerabilities uses: aquasecurity/trivy-action@master with: image-ref: ${{ secrets.REGISTRY_URL }}/agent-service:${{ github.sha }} format: 'sarif' output: 'trivy-results.sarif' - name: Upload Trivy scan results uses: github/codeql-action/upload-sarif@v2 with: sarif_file: 'trivy-results.sarif' 自动化测试 test: needs: build runs-on: self-hosted services: redis: image: redis:7 ports: - 6379:6379 postgres: image: postgres:15 env: POSTGRES_PASSWORD: test ports: - 5432:5432 qdrant: image: qdrant/qdrant:latest ports: - 6333:6333 steps: - uses: actions/checkout@v3 - name: Run Unit Tests run: | pytest tests/unit/ -v --cov=agent --cov-report=xml --junitxml=junit-unit.xml - name: Run Integration Tests env: LLM_MOCK: "true" # 使用Mock LLM run: | pytest tests/integration/ -v --junitxml=junit-integration.xml - name: Run Agent-Specific Tests run: | # Prompt测试 python -m pytest tests/prompt/ -v # 工具调用测试 python -m pytest tests/tools/ -v # 质量回归测试 python tests/regression/run_regression.py \ --baseline=baseline.json \ --report=regression-report.json - name: Upload Test Results if: always() uses: actions/upload-artifact@v3 with: name: test-results path: | coverage.xml junit-*.xml regression-report.json - name: Check Quality Gate run: | python scripts/check_quality_gate.py \ --coverage-report=coverage.xml \ --min-coverage=80 \ --regression-report=regression-report.json \ --max-regressions=0 环境管理 class EnvironmentManager: """环境管理器""" ENVIRONMENTS = { "dev": { "replicas": 1, "model": "gpt-4o-mini", "quality_gate": {"min_quality": 0.7}, }, "staging": { "replicas": 3, "model": "gpt-4o", "quality_gate": {"min_quality": 0.8}, }, "prod": { "replicas": 10, "model": "gpt-4o", "quality_gate": {"min_quality": 0.85}, } } async def deploy_to_environment( self, environment: str, image_tag: str, config: dict = None ): """部署到指定环境""" env_config = self.ENVIRONMENTS[environment] deploy_config = {**env_config, **(config or {})} # 1. 更新K8s Deployment await self.k8s_client.apply_deployment({ "apiVersion": "apps/v1", "kind": "Deployment", "metadata": { "name": f"agent-service-{environment}", "namespace": f"agent-{environment}" }, "spec": { "replicas": deploy_config["replicas"], "template": { "spec": { "containers": [{ "name": "agent", "image": f"{self.registry}/{image_tag}", "env": [ {"name": "MODEL_NAME", "value": deploy_config["model"]}, {"name": "ENVIRONMENT", "value": environment}, ] }] } } } }) # 2. 等待部署完成 await self._wait_for_rollout( f"agent-service-{environment}", timeout=300 ) # 3. 运行冒烟测试 await self._run_smoke_tests(environment) # 4. 运行质量门禁 quality_result = await self._run_quality_gate(environment, deploy_config["quality_gate"]) if not quality_result["passed"]: logger.error(f"Quality gate failed for {environment}") await self._rollback(environment, image_tag) raise QualityGateFailed(quality_result["details"]) logger.info(f"Successfully deployed to {environment}") 灰度发布 # Argo Rollouts配置 apiVersion: argoproj.io/v1alpha1 kind: Rollout metadata: name: agent-service spec: replicas: 10 strategy: canary: canaryService: agent-service-canary stableService: agent-service-stable # 分析阶段 analysis: templates: - templateName: agent-quality-analysis args: - name: service-name value: agent-service-canary # 渐进式发布 steps: - setWeight: 5 - pause: {duration: 10m} - setWeight: 20 - pause: {duration: 15m} - setWeight: 50 - pause: {duration: 20m} - setWeight: 100 # 流量路由 trafficRouting: istio: virtualService: name: agent-service-vs destinationRule: name: agent-service-dr 自动化回滚 class AutoRollbackManager: """自动回滚管理器""" async def monitor_and_rollback(self, rollout_name: str): """监控发布并自动回滚""" while True: # 获取Rollout状态 rollout = await self.k8s_client.get_rollout(rollout_name) if rollout["status"]["phase"] == "Degraded": logger.warning(f"Rollout {rollout_name} degraded, initiating rollback") await self._rollback(rollout_name) break # 检查质量指标 quality = await self._check_quality(rollout_name) if quality["error_rate"] > 0.05: logger.warning(f"Error rate {quality['error_rate']:.1%} > 5%, rolling back") await self._rollback(rollout_name) break if quality["quality_score"] < 0.8: logger.warning(f"Quality score {quality['quality_score']:.2f} < 0.8, rolling back") await self._rollback(rollout_name) break await asyncio.sleep(30) # 30秒检查一次 async def _rollback(self, rollout_name: str): """执行回滚""" await self.k8s_client.rollback_rollout( rollout_name, to_revision="previous" ) # 发送通知 await self.notifier.send( channel="slack:#alerts", message=f"⚠️ Auto-rollback triggered for {rollout_name}" ) 完整的CD流水线 # .github/workflows/cd.yml name: Agent CD Pipeline on: workflow_run: workflows: ["Agent CI Pipeline"] types: [completed] branches: [main] jobs: deploy-staging: if: ${{ github.event.workflow_run.conclusion == 'success' }} runs-on: self-hosted steps: - name: Deploy to Staging run: | python scripts/deploy.py \ --environment=staging \ --image-tag=${{ github.sha }} - name: Run Staging Tests run: | python scripts/run_e2e_tests.py --environment=staging - name: Notify Deployment uses: 8398a7/action-slack@v3 with: status: ${{ job.status }} text: "Deployed to Staging: ${{ github.sha }}" deploy-prod: needs: deploy-staging runs-on: self-hosted environment: production # 需要手动批准 steps: - name: Deploy to Production (Canary) run: | kubectl argo rollouts promote agent-service - name: Monitor Canary run: | python scripts/monitor_canary.py \ --rollout=agent-service \ --duration=30m \ --auto-rollback=true - name: Promote Canary if: success() run: | kubectl argo rollouts promote agent-service - name: Create GitHub Release uses: actions/create-release@v1 env: GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} with: tag_name: v${{ github.run_number }} release_name: Release v${{ github.run_number }} body: | ## Changes ${{ steps.changelog.outputs.changelog }} ## Deployment - Staging: ✅ - Production: ✅ (Canary 100%) draft: false prerelease: false 总结 Agent 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论坛 — 全球首个碳基硅基认知交流平台。 ...