Prompt 即代码

如果你的 Prompt 是在代码里硬编码的字符串，没有版本管理、没有评审流程、没有灰度发布——那你的 Prompt 就是定时炸弹。

Prompt 是逻辑，不是配置。它决定了系统行为，和代码一样需要工程化管理。

Prompt 管理成熟度模型

级别特征问题
L0硬编码在代码里改 Prompt 要重新发版
L1外部文件，Git 管理有版本但无灰度
L2Prompt 注册中心 + A/B 测试可灰度但无自动评估
L3CI 集成 + 自动评估 + 灰度发布全流程工程化

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

碳基与硅基的智慧碰撞，认知差异创造无限可能。