Prompt 版本控制：比代码版本管理更复杂的那点事-PySuper

在 LLM 应用开发中，Prompt 是连接用户需求与模型能力的核心桥梁。与传统的代码不同，Prompt 的管理和版本控制面临着独特的挑战：非确定性输出、上下文依赖、评估困难等问题。

笔者在构建企业级 LLM 应用的过程中，积累了一些 Prompt 版本管理的经验。本文将深入探讨 Prompt 管理的难点、解决方案，以及如何构建一个完整的 Prompt 管理系统。

一、Prompt 版本管理的难点

1.1 为什么 Prompt 管理如此特殊

┌─────────────────────────────────────────────────────────────────┐
│                  Prompt 管理的独特挑战                           │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   ┌─────────────────┐      ┌─────────────────┐                 │
│   │    代码管理    │      │   Prompt 管理   │                 │
│   └────────┬────────┘      └────────┬────────┘                 │
│            │                        │                          │
│            ▼                        ▼                          │
│   ┌─────────────────┐      ┌─────────────────┐                 │
│   │ 确定性输出      │      │ 非确定性输出    │                 │
│   │ 相同输入→      │      │ 相同 Prompt →   │                 │
│   │ 相同输出        │      │ 不同输出         │                 │
│   └─────────────────┘      └─────────────────┘                 │
│            │                        │                          │
│            ▼                        ▼                          │
│   ┌─────────────────┐      ┌─────────────────┐                 │
│   │ 单元测试验证    │      │ 评估困难        │                 │
│   │ 断言输出        │      │ 如何定义"正确" │                 │
│   └─────────────────┘      └─────────────────┘                 │
│            │                        │                          │
│            ▼                        ▼                          │
│   ┌─────────────────┐      ┌─────────────────┐                 │
│   │ 自动化回归     │      │ 上下文依赖     │                 │
│   │ 简单可靠       │      │ Prompt 效果     │                 │
│   └─────────────────┘      │ 依赖对话历史   │                 │
│                            └─────────────────┘                 │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

1.2 Prompt 的不确定性来源

# nondeterminism_analysis.py
"""
Prompt 不确定性分析
分析导致 Prompt 输出不一致的各种因素
"""

from dataclasses import dataclass
from typing import List, Dict, Any, Optional
from enum import Enum


class UncertaintySource(Enum):
    """不确定性来源"""
    # 模型层面
    MODEL_TEMPERATURE = "temperature"           # 温度参数
    MODEL_TOP_P = "top_p"                        # Top-p 采样
    MODEL_TOP_K = "top_k"                        # Top-k 采样
    
    # 提示层面
    PROMPT_VARIATION = "prompt_variation"        # Prompt 措辞变化
    CONTEXT_OVERFLOW = "context_overflow"        # 上下文溢出
    ORDER_EFFECT = "order_effect"               # 顺序效应
    
    # 模型层面
    MODEL_VERSION = "model_version"              # 模型版本变化
    BATCH_EFFECT = "batch_effect"               # GPU 批次效应
    
    # 系统层面
    SYSTEM_LOAD = "system_load"                  # 系统负载
    NETWORK_LATENCY = "network_latency"          # 网络延迟


@dataclass
class UncertaintyFactor:
    """不确定性因素"""
    source: UncertaintySource
    description: str
    severity: float  # 0-1，影响严重程度
    controllable: bool  # 是否可控
    mitigation: str  # 缓解方法


UNCERTAINTY_FACTORS = [
    UncertaintyFactor(
        source=UncertaintySource.MODEL_TEMPERATURE,
        description="温度参数控制输出的随机性，temperature=0 时最确定",
        severity=0.9,
        controllable=True,
        mitigation="生产环境使用 temperature=0 或极低值"
    ),
    UncertaintyFactor(
        source=UncertaintySource.MODEL_VERSION,
        description="不同模型版本对同一 Prompt 可能有不同理解",
        severity=0.8,
        controllable=True,
        mitigation="固定模型版本，Promp t与模型版本绑定"
    ),
    UncertaintyFactor(
        source=UncertaintySource.CONTEXT_OVERFLOW,
        description="长对话可能导致早期上下文被"遗忘"",
        severity=0.7,
        controllable=True,
        mitigation="使用滑动窗口或摘要机制"
    ),
    UncertaintyFactor(
        source=UncertaintySource.ORDER_EFFECT,
        description="列表项的顺序可能影响模型注意力分配",
        severity=0.5,
        controllable=True,
        mitigation="打乱顺序多次测试取平均"
    ),
    UncertaintyFactor(
        source=UncertaintySource.BATCH_EFFECT,
        description="GPU 并行推理可能引入微小数值差异",
        severity=0.3,
        controllable=False,
        mitigation="多次采样取最常见结果"
    ),
]


def print_uncertainty_analysis():
    """打印不确定性分析报告"""
    print("=" * 70)
    print("Prompt 不确定性分析报告")
    print("=" * 70)
    
    print("\n📊 不确定性因素汇总:\n")
    print(f"{'来源':<25} {'严重程度':<12} {'可控':<8} {'描述'}")
    print("-" * 70)
    
    for factor in UNCERTAINTY_FACTORS:
        severity_bar = "█" * int(factor.severity * 10) + "░" * (10 - int(factor.severity * 10))
        controllable = "✅" if factor.controllable else "❌"
        
        print(f"{factor.source.value:<25} [{severity_bar}]  {controllable:<6}  {factor.description}")
    
    print("\n" + "=" * 70)
    print("🎯 关键结论:")
    print("=" * 70)
    controllable_factors = [f for f in UNCERTAINTY_FACTORS if f.controllable]
    uncontrollable_factors = [f for f in UNCERTAINTY_FACTORS if not f.controllable]
    
    print(f"\n✅ 可控因素 ({len(controllable_factors)} 个):")
    for f in controllable_factors:
        print(f"   • {f.source.value}: {f.mitigation}")
    
    print(f"\n⚠️ 不可控因素 ({len(uncontrollable_factors)} 个):")
    for f in uncontrollable_factors:
        print(f"   • {f.source.value}: {f.description}")


if __name__ == "__main__":
    print_uncertainty_analysis()

1.3 上下文依赖问题

┌─────────────────────────────────────────────────────────────────┐
│                    Prompt 上下文依赖问题                         │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   简单场景（独立 Prompt）:                                      │
│   ┌─────────────────────────────────────────────────────────┐   │
│   │  Prompt: "将以下文本翻译成英文: {text}"                  │   │
│   │                                                         │   │
│   │  输入: "你好世界"                                       │   │
│   │  输出: "Hello World" (✓ 确定)                          │   │
│   └─────────────────────────────────────────────────────────┘   │
│                                                                 │
│   复杂场景（上下文依赖）:                                        │
│   ┌─────────────────────────────────────────────────────────┐   │
│   │  System: "你是一个客服助手，擅长解答{product}相关问题"  │   │
│   │  History: [对话历史...]                                  │   │
│   │  Prompt: "{user_message}"                               │   │
│   │                                                         │   │
│   │  问题:                                                   │   │
│   │  • Product 变化时效果如何？                             │   │
│   │  • 对话历史有多长的影响？                               │   │
│   │  • 不同用户的个性化如何处理？                          │   │
│   └─────────────────────────────────────────────────────────┘   │
│                                                                 │
│   Prompt 效果 = f(Prompt内容, 对话历史, 用户画像, 模型状态)    │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

二、Prompt 模板化设计

2.1 模板引擎实现

# prompt_template_engine.py
"""
Prompt 模板引擎
支持变量插值、条件逻辑、循环、继承等高级特性
"""

import re
import json
import hashlib
from typing import Dict, Any, List, Optional, Callable, Union
from dataclasses import dataclass, field
from enum import Enum
from abc import ABC, abstractmethod
from datetime import datetime
from string import Template


class TemplateType(Enum):
    """模板类型"""
    SIMPLE = "simple"           # 简单模板
    CHAT = "chat"              # 对话模板
    FEW_SHOT = "few_shot"      # Few-shot 模板
    CHAIN_OF_THOUGHT = "cot"   # 思维链模板


@dataclass
class PromptVariable:
    """Prompt 变量定义"""
    name: str
    type_hint: str = "str"     # str, int, float, bool, list, dict
    description: str = ""
    required: bool = True
    default: Any = None
    validator: Optional[Callable] = None
    
    def validate(self, value: Any) -> tuple[bool, str]:
        """验证变量值"""
        if value is None:
            if self.required:
                return False, f"变量 {self.name} 是必填的"
            return True, ""
        
        # 类型检查
        type_map = {
            "str": str,
            "int": int,
            "float": (int, float),
            "bool": bool,
            "list": list,
            "dict": dict,
        }
        
        expected_type = type_map.get(self.type_hint)
        if expected_type and not isinstance(value, expected_type):
            return False, f"变量 {self.name} 类型错误，期望 {self.type_hint}"
        
        # 自定义验证器
        if self.validator:
            return self.validator(value)
        
        return True, ""


@dataclass
class TemplateMetadata:
    """模板元数据"""
    name: str
    version: str
    description: str = ""
    author: str = ""
    created_at: str = ""
    tags: List[str] = field(default_factory=list)
    variables: List[PromptVariable] = field(default_factory=list)
    examples: List[Dict[str, Any]] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)


class TemplateRenderer:
    """
    Prompt 模板渲染器
    支持简单变量、条件、循环等
    """
    
    def __init__(self):
        self._filters: Dict[str, Callable] = {}
        self._register_builtin_filters()
    
    def _register_builtin_filters(self):
        """注册内置过滤器"""
        self._filters["upper"] = lambda x: str(x).upper()
        self._filters["lower"] = lambda x: str(x).lower()
        self._filters["capitalize"] = lambda x: str(x).capitalize()
        self._filters["first"] = lambda x: x[0] if x else ""
        self._filters["last"] = lambda x: x[-1] if x else ""
        self._filters["join"] = lambda x, sep="": sep.join(str(i) for i in x)
        self._filters["length"] = lambda x: len(x)
        self._filters["to_json"] = lambda x: json.dumps(x, ensure_ascii=False)
        self._filters["default"] = lambda x, d="": x if x is not None else d
        self._filters["indent"] = lambda x, n=2: "\n".join(" " * n + line for line in str(x).split("\n"))
    
    def register_filter(self, name: str, func: Callable):
        """注册自定义过滤器"""
        self._filters[name] = func
    
    def render(
        self,
        template: str,
        context: Dict[str, Any],
        strict: bool = True
    ) -> str:
        """
        渲染模板
        
        Args:
            template: 模板字符串
            context: 变量上下文
            strict: 是否严格模式（未定义变量报错）
        
        Returns:
            渲染后的字符串
        """
        # 提取所有变量引用
        pattern = r'\$\{([^}]+)\}'
        
        def replace_var(match):
            var_expr = match.group(1)
            
            # 处理过滤器
            if '|' in var_expr:
                var_name, *filters = var_expr.split('|')
                var_name = var_name.strip()
                value = self._get_nested_value(context, var_name)
                
                for f in filters:
                    f = f.strip()
                    if '(' in f:
                        # 带参数的过滤器
                        f_name, args_str = f.split('(', 1)
                        args_str = args_str.rstrip(')')
                        args = [a.strip() for a in args_str.split(',')]
                        # 将上下文中的值替换参数
                        resolved_args = []
                        for a in args:
                            if a in context:
                                resolved_args.append(context[a])
                            else:
                                resolved_args.append(a)
                        value = self._filters[f_name](value, *resolved_args)
                    else:
                        # 无参数过滤器
                        if f in self._filters:
                            value = self._filters[f](value)
                        else:
                            raise ValueError(f"未知的过滤器: {f}")
                
                return str(value) if value is not None else ""
            else:
                # 简单变量
                value = self._get_nested_value(context, var_expr)
                if value is None:
                    if strict:
                        raise ValueError(f"未定义的变量: {var_expr}")
                    return ""
                return str(value)
        
        result = re.sub(pattern, replace_var, template)
        
        # 处理条件块 {% if %} ... {% endif %}
        result = self._render_conditionals(result, context)
        
        # 处理循环块 {% for %} ... {% endfor %}
        result = self._render_loops(result, context)
        
        return result
    
    def _get_nested_value(self, data: Dict, path: str) -> Any:
        """获取嵌套字典的值"""
        keys = path.split('.')
        value = data
        for key in keys:
            if isinstance(value, dict):
                value = value.get(key)
            else:
                return None
            if value is None:
                return None
        return value
    
    def _render_conditionals(self, template: str, context: Dict) -> str:
        """渲染条件块"""
        pattern = r'\{%\s*if\s+([^%]+)\s*%\}(.*?)\{%\s*endif\s*%\}'
        
        def replace_if(match):
            condition = match.group(1).strip()
            content = match.group(2)
            
            # 简单条件判断
            # 支持: var, not var, var == value, var > value 等
            if self._evaluate_condition(condition, context):
                return content
            return ""
        
        return re.sub(pattern, replace_if, template, flags=re.DOTALL)
    
    def _render_loops(self, template: str, context: Dict) -> str:
        """渲染循环块"""
        pattern = r'\{%\s*for\s+(\w+)\s+in\s+([^%]+)\s*%\}(.*?)\{%\s*endfor\s*%\}'
        
        def replace_for(match):
            loop_var = match.group(1)
            iterable_path = match.group(2).strip()
            content = match.group(3)
            
            iterable = self._get_nested_value(context, iterable_path)
            if not iterable:
                return ""
            
            results = []
            for item in iterable:
                loop_context = context.copy()
                loop_context[loop_var] = item
                loop_context["loop"] = {
                    "index": len(results),
                    "index1": len(results) + 1,
                    "first": len(results) == 0,
                    "last": len(results) == len(iterable) - 1,
                }
                results.append(self.render(content, loop_context, strict=False))
            
            return "".join(results)
        
        return re.sub(pattern, replace_for, template, flags=re.DOTALL)
    
    def _evaluate_condition(self, condition: str, context: Dict) -> bool:
        """评估条件表达式"""
        condition = condition.strip()
        
        # 处理 not
        if condition.startswith("not "):
            var = condition[4:].strip()
            value = self._get_nested_value(context, var)
            return not bool(value)
        
        # 处理 == 
        if " == " in condition:
            left, right = condition.split(" == ", 1)
            left_value = self._get_nested_value(context, left.strip())
            # 尝试解析右侧为字符串或数字
            right = right.strip().strip('"\'')
            if right.isdigit():
                right = int(right)
            elif right.replace('.', '', 1).isdigit():
                right = float(right)
            return str(left_value) == str(right)
        
        # 简单变量检查
        value = self._get_nested_value(context, condition)
        return bool(value)


class PromptTemplate:
    """
    Prompt 模板类
    封装模板定义、验证、渲染功能
    """
    
    def __init__(self, metadata: TemplateMetadata, content: str):
        self.metadata = metadata
        self.content = content
        self.renderer = TemplateRenderer()
    
    def validate(self, context: Dict[str, Any]) -> tuple[bool, List[str]]:
        """
        验证上下文是否满足模板要求
        
        Returns:
            (是否有效, 错误消息列表)
        """
        errors = []
        
        for var in self.metadata.variables:
            value = context.get(var.name)
            valid, msg = var.validate(value)
            if not valid:
                errors.append(msg)
        
        return len(errors) == 0, errors
    
    def render(self, context: Dict[str, Any], strict: bool = True) -> str:
        """
        渲染模板
        
        Args:
            context: 变量上下文
            strict: 是否严格模式
        
        Returns:
            渲染后的 Prompt
        """
        valid, errors = self.validate(context)
        if not valid and strict:
            raise ValueError(f"上下文验证失败: {errors}")
        
        return self.renderer.render(self.content, context, strict=strict)
    
    def render_messages(self, context: Dict[str, Any]) -> List[Dict[str, str]]:
        """
        渲染为对话消息格式
        
        Returns:
            消息列表 [{"role": "system", "content": "..."}]
        """
        rendered = self.render(context)
        
        # 解析对话格式
        messages = []
        lines = rendered.strip().split("\n")
        current_role = "user"
        current_content = []
        
        for line in lines:
            line = line.strip()
            if not line:
                continue
            
            # 检测角色标记
            if line.startswith("system:") or line.startswith("[SYSTEM]"):
                if current_content:
                    messages.append({"role": current_role, "content": "\n".join(current_content)})
                current_role = "system"
                current_content = [line.split(":", 1)[1].strip()]
            elif line.startswith("assistant:") or line.startswith("[ASSISTANT]"):
                if current_content:
                    messages.append({"role": current_role, "content": "\n".join(current_content)})
                current_role = "assistant"
                current_content = [line.split(":", 1)[1].strip()]
            elif line.startswith("user:") or line.startswith("[USER]"):
                if current_content:
                    messages.append({"role": current_role, "content": "\n".join(current_content)})
                current_role = "user"
                current_content = [line.split(":", 1)[1].strip()]
            else:
                current_content.append(line)
        
        if current_content:
            messages.append({"role": current_role, "content": "\n".join(current_content)})
        
        return messages
    
    def get_hash(self) -> str:
        """计算模板的 hash 值"""
        content_to_hash = f"{self.metadata.name}:{self.metadata.version}:{self.content}"
        return hashlib.sha256(content_to_hash.encode()).hexdigest()[:16]
    
    def to_dict(self) -> Dict:
        """导出为字典"""
        return {
            "metadata": {
                "name": self.metadata.name,
                "version": self.metadata.version,
                "description": self.metadata.description,
                "author": self.metadata.author,
                "created_at": self.metadata.created_at,
                "tags": self.metadata.tags,
                "variables": [
                    {
                        "name": v.name,
                        "type_hint": v.type_hint,
                        "description": v.description,
                        "required": v.required,
                        "default": v.default,
                    }
                    for v in self.metadata.variables
                ],
                "examples": self.metadata.examples,
            },
            "content": self.content,
            "hash": self.get_hash(),
        }


# 示例：创建企业级 Prompt 模板
def create_sentiment_analysis_template() -> PromptTemplate:
    """创建情感分析 Prompt 模板"""
    
    content = """
{% if language == 'zh' %}
【系统提示】
你是一个专业的情感分析助手。你的任务是对给定的文本进行情感分类。

【分析规则】
1. 只输出以下三种结果之一：positive、negative、neutral
2. 不要添加任何解释或额外文字
3. 严格按照以下标准判断：
   - positive: 正面情感，如喜欢、满意、开心、赞扬等
   - negative: 负面情感，如讨厌、不满、悲伤、批评等
   - neutral: 中性表达，无明显情感倾向

【待分析文本】
${text}

【输出格式】
${output_format}
{% else %}
[SYSTEM]
You are a professional sentiment analysis assistant. Analyze the following text and classify it as positive, negative, or neutral.

Text: ${text}

Output format: ${output_format}
{% endif %}
"""
    
    metadata = TemplateMetadata(
        name="sentiment-analysis",
        version="v2.1.0",
        description="情感分析 Prompt，支持多语言",
        author="prompt-team",
        created_at=datetime.now().isoformat(),
        tags=["classification", "sentiment", "multilingual"],
        variables=[
            PromptVariable(
                name="text",
                type_hint="str",
                description="待分析的文本",
                required=True,
            ),
            PromptVariable(
                name="language",
                type_hint="str",
                description="文本语言",
                required=True,
                default="zh",
                validator=lambda x: x in ["zh", "en", "ja", "ko"] or True,
            ),
            PromptVariable(
                name="output_format",
                type_hint="str",
                description="输出格式",
                required=False,
                default="JSON",
            ),
        ],
        examples=[
            {
                "input": {
                    "text": "这个产品太棒了，我非常满意！",
                    "language": "zh",
                },
                "expected": "positive",
            },
            {
                "input": {
                    "text": "The service was terrible and slow.",
                    "language": "en",
                },
                "expected": "negative",
            },
        ],
    )
    
    return PromptTemplate(metadata, content)


def create_customer_service_template() -> PromptTemplate:
    """创建客服对话 Prompt 模板"""
    
    content = """
[SYSTEM]
你是一个专业的{product}客服助手，名字叫{assistant_name}。

【你的特点】
1. 专业、耐心、友好
2. 使用{language}回复
3. 回答简洁明了，不超过{response_length}字
{% if customer_tier == 'vip' %}
4. VIP客户享受优先服务
{% endif %}

【服务条款】
- 尊重客户隐私
- 不承诺超出服务范围的事项
- 遇到无法解决的问题，及时转人工

[CONTEXT]
客户: ${customer_message}
{% if previous_messages %}
对话历史:
{% for msg in previous_messages %}
${msg.role}: ${msg.content}
{% endfor %}
{% endif %}

[KNOWLEDGE]
${knowledge_base}

[INSTRUCTIONS]
请根据以上信息，回答客户的问题。如果需要更多信息，请礼貌地询问。
"""
    
    metadata = TemplateMetadata(
        name="customer-service",
        version="v1.5.0",
        description="通用客服对话 Prompt",
        author="prompt-team",
        created_at=datetime.now().isoformat(),
        tags=["conversation", "customer-service", "multi-turn"],
        variables=[
            PromptVariable(
                name="product",
                type_hint="str",
                description="产品名称",
                required=True,
            ),
            PromptVariable(
                name="assistant_name",
                type_hint="str",
                description="助手名称",
                required=False,
                default="小助手",
            ),
            PromptVariable(
                name="language",
                type_hint="str",
                description="回复语言",
                required=True,
                default="zh",
            ),
            PromptVariable(
                name="response_length",
                type_hint="int",
                description="回复最大长度",
                required=False,
                default=200,
            ),
            PromptVariable(
                name="customer_tier",
                type_hint="str",
                description="客户等级",
                required=False,
                default="normal",
            ),
            PromptVariable(
                name="customer_message",
                type_hint="str",
                description="客户消息",
                required=True,
            ),
            PromptVariable(
                name="previous_messages",
                type_hint="list",
                description="对话历史",
                required=False,
                default=[],
            ),
            PromptVariable(
                name="knowledge_base",
                type_hint="str",
                description="知识库内容",
                required=True,
            ),
        ],
    )
    
    return PromptTemplate(metadata, content)


# 使用示例
def main():
    """使用示例"""
    
    # 创建情感分析模板
    template = create_sentiment_analysis_template()
    print("模板信息:")
    print(f"  名称: {template.metadata.name}")
    print(f"  版本: {template.metadata.version}")
    print(f"  Hash: {template.get_hash()}")
    print()
    
    # 渲染模板
    context = {
        "text": "这个产品太棒了，我非常满意！",
        "language": "zh",
        "output_format": "JSON",
    }
    
    result = template.render(context)
    print("渲染结果:")
    print(result)
    print()
    
    # 渲染为消息格式
    messages = template.render_messages(context)
    print("对话消息格式:")
    for msg in messages:
        print(f"  {msg['role']}: {msg['content'][:50]}...")


if __name__ == "__main__":
    main()

三、版本对比与 A/B 测试

3.1 Prompt 版本对比工具

# prompt_versioning.py
"""
Prompt 版本管理工具
支持版本对比、回滚、A/B 测试等
"""

import os
import re
import hashlib
import difflib
import json
from typing import Dict, List, Optional, Any, Tuple
from dataclasses import dataclass, field, asdict
from datetime import datetime
from enum import Enum
from pathlib import Path


class ChangeType(Enum):
    """变更类型"""
    ADDED = "added"           # 新增
    REMOVED = "removed"       # 删除
    MODIFIED = "modified"     # 修改
    UNCHANGED = "unchanged"  # 未变


@dataclass
class VersionDiff:
    """版本差异"""
    change_type: ChangeType
    line_number: int
    old_content: str
    new_content: str
    context: str  # 周围内容用于理解


@dataclass
class VersionComparison:
    """版本对比结果"""
    old_version: str
    new_version: str
    diffs: List[VersionDiff]
    summary: Dict[str, int]  # 变更统计
    
    def get_html_diff(self) -> str:
        """生成 HTML 格式的差异"""
        html_parts = ['<div class="prompt-diff">']
        
        for diff in self.diffs:
            if diff.change_type == ChangeType.ADDED:
                html_parts.append(f'<div class="added">+ {self._escape_html(diff.new_content)}</div>')
            elif diff.change_type == ChangeType.REMOVED:
                html_parts.append(f'<div class="removed">- {self._escape_html(diff.old_content)}</div>')
            else:
                html_parts.append(f'<div class="unchanged">  {self._escape_html(diff.new_content)}</div>')
        
        html_parts.append('</div>')
        return '\n'.join(html_parts)
    
    @staticmethod
    def _escape_html(text: str) -> str:
        """转义 HTML 特殊字符"""
        return (text
                .replace("&", "&amp;")
                .replace("<", "&lt;")
                .replace(">", "&gt;")
                .replace('"', "&quot;"))


@dataclass
class PromptVersion:
    """Prompt 版本"""
    version_id: str
    template_name: str
    version: str
    content: str
    variables: Dict[str, Any]
    metadata: Dict[str, Any]
    created_by: str
    created_at: str
    parent_version: Optional[str] = None
    change_log: str = ""
    
    def get_hash(self) -> str:
        """计算内容 hash"""
        content_to_hash = f"{self.template_name}:{self.version}:{self.content}"
        return hashlib.sha256(content_to_hash.encode()).hexdigest()[:16]
    
    def compare_with(self, other: "PromptVersion") -> VersionComparison:
        """
        与另一个版本对比
        
        Args:
            other: 要对比的版本
        
        Returns:
            版本对比结果
        """
        old_lines = self.content.splitlines(keepends=True)
        new_lines = other.content.splitlines(keepends=True)
        
        diffs = []
        
        # 使用 difflib 计算差异
        matcher = difflib.SequenceMatcher(None, old_lines, new_lines)
        
        line_num_old = 0
        line_num_new = 0
        
        for tag, i1, i2, j1, j2 in matcher.get_opcodes():
            if tag == 'replace':
                for i in range(i1, i2):
                    diffs.append(VersionDiff(
                        change_type=ChangeType.REMOVED,
                        line_number=line_num_old + i + 1,
                        old_content=old_lines[i].rstrip(),
                        new_content="",
                        context=self._get_context(old_lines, i),
                    ))
                for j in range(j1, j2):
                    diffs.append(VersionDiff(
                        change_type=ChangeType.ADDED,
                        line_number=line_num_new + j + 1,
                        old_content="",
                        new_content=new_lines[j].rstrip(),
                        context=self._get_context(new_lines, j),
                    ))
                line_num_old += i2 - i1
                line_num_new += j2 - j1
            elif tag == 'delete':
                for i in range(i1, i2):
                    diffs.append(VersionDiff(
                        change_type=ChangeType.REMOVED,
                        line_number=line_num_old + i + 1,
                        old_content=old_lines[i].rstrip(),
                        new_content="",
                        context=self._get_context(old_lines, i),
                    ))
                line_num_old += i2 - i1
            elif tag == 'insert':
                for j in range(j1, j2):
                    diffs.append(VersionDiff(
                        change_type=ChangeType.ADDED,
                        line_number=line_num_new + j + 1,
                        old_content="",
                        new_content=new_lines[j].rstrip(),
                        context=self._get_context(new_lines, j),
                    ))
                line_num_new += j2 - j1
            elif tag == 'equal':
                line_num_old += i2 - i1
                line_num_new += j2 - j1
        
        # 统计
        summary = {
            "added": sum(1 for d in diffs if d.change_type == ChangeType.ADDED),
            "removed": sum(1 for d in diffs if d.change_type == ChangeType.REMOVED),
            "total": len(diffs),
        }
        
        return VersionComparison(
            old_version=self.version,
            new_version=other.version,
            diffs=diffs,
            summary=summary,
        )
    
    @staticmethod
    def _get_context(lines: List[str], index: int, window: int = 2) -> str:
        """获取行周围的上下文"""
        start = max(0, index - window)
        end = min(len(lines), index + window + 1)
        return "".join(f"{i}: {lines[i].rstrip()}\n" for i in range(start, end))
    
    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            "version_id": self.version_id,
            "template_name": self.template_name,
            "version": self.version,
            "content": self.content,
            "variables": self.variables,
            "metadata": self.metadata,
            "created_by": self.created_by,
            "created_at": self.created_at,
            "parent_version": self.parent_version,
            "change_log": self.change_log,
            "hash": self.get_hash(),
        }


@dataclass 
class ABTestConfig:
    """A/B 测试配置"""
    test_id: str
    template_name: str
    versions: List[str]  # 参与测试的版本列表
    traffic_split: Dict[str, float]  # 流量分配比例
    metrics: List[str]  # 评估指标
    min_sample_size: int = 1000  # 最小样本量
    duration_hours: int = 24  # 测试持续时间
    start_time: str = ""
    end_time: Optional[str] = None
    status: str = "pending"  # pending, running, completed, cancelled
    
    def validate_split(self) -> Tuple[bool, str]:
        """验证流量分配是否合理"""
        total = sum(self.traffic_split.values())
        if abs(total - 1.0) > 0.001:
            return False, f"流量分配总和必须为1，当前为 {total}"
        
        if set(self.traffic_split.keys()) != set(self.versions):
            return False, "流量分配键必须与版本列表一致"
        
        return True, ""


@dataclass
class ABTestResult:
    """A/B 测试结果"""
    test_id: str
    version_results: Dict[str, Dict[str, float]]  # 版本 -> 指标结果
    winner: Optional[str]
    confidence: float  # 置信度
    recommendation: str
    raw_data: Dict[str, Any]
    
    def is_significant(self, threshold: float = 0.95) -> bool:
        """判断结果是否统计显著"""
        return self.confidence >= threshold


class PromptVersionManager:
    """
    Prompt 版本管理器
    管理 Prompt 的版本、对比、A/B 测试等
    """
    
    def __init__(self, storage_path: str = "./prompt_registry"):
        self.storage_path = Path(storage_path)
        self.storage_path.mkdir(parents=True, exist_ok=True)
        
        self._versions: Dict[str, List[PromptVersion]] = {}  # template_name -> versions
        self._active_versions: Dict[str, str] = {}  # template_name -> active version
        self._ab_tests: Dict[str, ABTestConfig] = {}
        self._ab_results: Dict[str, ABTestResult] = {}
        
        self._load_registry()
    
    def _get_registry_file(self, template_name: str) -> Path:
        """获取模板的注册文件"""
        safe_name = re.sub(r'[^\w\-]', '_', template_name)
        return self.storage_path / f"{safe_name}_registry.json"
    
    def _load_registry(self):
        """加载注册表"""
        for file in self.storage_path.glob("*_registry.json"):
            with open(file, 'r', encoding='utf-8') as f:
                data = json.load(f)
                template_name = data.get("template_name")
                self._versions[template_name] = [
                    PromptVersion(**v) for v in data.get("versions", [])
                ]
                self._active_versions[template_name] = data.get("active_version", "")
    
    def _save_registry(self, template_name: str):
        """保存注册表"""
        versions = self._versions.get(template_name, [])
        data = {
            "template_name": template_name,
            "versions": [v.to_dict() for v in versions],
            "active_version": self._active_versions.get(template_name, ""),
        }
        
        file = self._get_registry_file(template_name)
        with open(file, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)
    
    def register_version(
        self,
        template_name: str,
        content: str,
        version: str,
        variables: Dict[str, Any],
        created_by: str,
        metadata: Optional[Dict[str, Any]] = None,
        change_log: str = ""
    ) -> PromptVersion:
        """
        注册新版本
        
        Args:
            template_name: 模板名称
            content: Prompt 内容
            version: 版本号
            variables: 变量定义
            created_by: 创建者
            metadata: 元数据
            change_log: 变更日志
        
        Returns:
            PromptVersion 对象
        """
        if template_name not in self._versions:
            self._versions[template_name] = []
        
        # 获取父版本
        parent = self._versions[template_name][-1] if self._versions[template_name] else None
        
        pv = PromptVersion(
            version_id=f"{template_name}:{version}",
            template_name=template_name,
            version=version,
            content=content,
            variables=variables,
            metadata=metadata or {},
            created_by=created_by,
            created_at=datetime.now().isoformat(),
            parent_version=parent.version if parent else None,
            change_log=change_log,
        )
        
        self._versions[template_name].append(pv)
        self._save_registry(template_name)
        
        print(f"✅ 已注册版本: {template_name}@{version}")
        return pv
    
    def get_version(self, template_name: str, version: str) -> Optional[PromptVersion]:
        """获取指定版本"""
        versions = self._versions.get(template_name, [])
        for v in versions:
            if v.version == version:
                return v
        return None
    
    def get_latest_version(self, template_name: str) -> Optional[PromptVersion]:
        """获取最新版本"""
        versions = self._versions.get(template_name, [])
        return versions[-1] if versions else None
    
    def get_active_version(self, template_name: str) -> Optional[PromptVersion]:
        """获取当前活跃版本"""
        version = self._active_versions.get(template_name)
        if version:
            return self.get_version(template_name, version)
        return self.get_latest_version(template_name)
    
    def set_active_version(self, template_name: str, version: str):
        """设置活跃版本"""
        pv = self.get_version(template_name, version)
        if not pv:
            raise ValueError(f"版本不存在: {template_name}@{version}")
        
        self._active_versions[template_name] = version
        self._save_registry(template_name)
        print(f"✅ 已设置活跃版本: {template_name}@{version}")
    
    def compare_versions(
        self,
        template_name: str,
        version_a: str,
        version_b: str
    ) -> VersionComparison:
        """对比两个版本"""
        v_a = self.get_version(template_name, version_a)
        v_b = self.get_version(template_name, version_b)
        
        if not v_a or not v_b:
            raise ValueError("指定的版本不存在")
        
        return v_a.compare_with(v_b)
    
    def create_ab_test(
        self,
        template_name: str,
        versions: List[str],
        traffic_split: Dict[str, float],
        metrics: List[str],
        **kwargs
    ) -> ABTestConfig:
        """
        创建 A/B 测试
        
        Args:
            template_name: 模板名称
            versions: 参与测试的版本
            traffic_split: 流量分配
            metrics: 评估指标
            **kwargs: 其他参数
        
        Returns:
            A/B 测试配置
        """
        # 验证版本存在
        for v in versions:
            if not self.get_version(template_name, v):
                raise ValueError(f"版本不存在: {template_name}@{v}")
        
        test_id = f"{template_name}_ab_{datetime.now().strftime('%Y%m%d%H%M%S')}"
        
        config = ABTestConfig(
            test_id=test_id,
            template_name=template_name,
            versions=versions,
            traffic_split=traffic_split,
            metrics=metrics,
            start_time=datetime.now().isoformat(),
            **kwargs
        )
        
        valid, msg = config.validate_split()
        if not valid:
            raise ValueError(f"流量分配验证失败: {msg}")
        
        self._ab_tests[test_id] = config
        print(f"✅ 已创建 A/B 测试: {test_id}")
        return config
    
    def record_ab_result(
        self,
        test_id: str,
        version: str,
        metric: str,
        value: float
    ):
        """记录 A/B 测试结果"""
        if test_id not in self._ab_results:
            self._ab_results[test_id] = ABTestResult(
                test_id=test_id,
                version_results={},
                winner=None,
                confidence=0.0,
                recommendation="",
                raw_data={},
            )
        
        if version not in self._ab_results[test_id].version_results:
            self._ab_results[test_id].version_results[version] = {}
        
        self._ab_results[test_id].version_results[version][metric] = value
    
    def get_version_history(self, template_name: str) -> List[PromptVersion]:
        """获取版本历史"""
        return self._versions.get(template_name, [])
    
    def rollback(self, template_name: str, target_version: str):
        """回滚到指定版本"""
        current = self.get_active_version(template_name)
        target = self.get_version(template_name, target_version)
        
        if not target:
            raise ValueError(f"目标版本不存在: {target_version}")
        
        self.set_active_version(template_name, target_version)
        print(f"🔄 已回滚: {template_name}@{current.version} -> {template_name}@{target_version}")
    
    def export_template(self, template_name: str) -> Dict:
        """导出版本（用于备份/迁移）"""
        return {
            "template_name": template_name,
            "versions": [v.to_dict() for v in self._versions.get(template_name, [])],
            "active_version": self._active_versions.get(template_name),
            "exported_at": datetime.now().isoformat(),
        }
    
    def import_template(self, data: Dict):
        """导入版本"""
        template_name = data["template_name"]
        versions = [PromptVersion(**v) for v in data["versions"]]
        
        self._versions[template_name] = versions
        self._active_versions[template_name] = data.get("active_version", "")
        self._save_registry(template_name)
        
        print(f"✅ 已导入模板: {template_name}，共 {len(versions)} 个版本")


# 使用示例
def main():
    """使用示例"""
    
    # 初始化管理器
    manager = PromptVersionManager("./prompt_registry")
    
    # 注册版本 v1.0.0
    v1_content = """
[SYSTEM]
你是一个客服助手。
用户: ${user_message}
"""
    manager.register_version(
        template_name="customer-service",
        content=v1_content,
        version="v1.0.0",
        variables={"user_message": {"type": "str", "required": True}},
        created_by="zhangsan",
        change_log="初始版本"
    )
    
    # 注册版本 v1.1.0
    v2_content = """
[SYSTEM]
你是一个专业的客服助手，态度友好。
用户: ${user_message}
{% if user_tier == 'vip' %}
注意：这是VIP客户，请提供优质服务。
{% endif %}
"""
    manager.register_version(
        template_name="customer-service",
        content=v2_content,
        version="v1.1.0",
        variables={
            "user_message": {"type": "str", "required": True},
            "user_tier": {"type": "str", "required": False, "default": "normal"},
        },
        created_by="lisi",
        change_log="添加VIP用户支持"
    )
    
    # 对比版本
    print("\n" + "=" * 60)
    print("版本对比: v1.0.0 vs v1.1.0")
    print("=" * 60)
    
    comparison = manager.compare_versions("customer-service", "v1.0.0", "v1.1.0")
    print(f"\n变更统计: {comparison.summary}")
    
    for diff in comparison.diffs:
        prefix = {"added": "+", "removed": "-", "modified": "~"}.get(diff.change_type.value, " ")
        print(f"{prefix} {diff.change_type.value.upper()}: {diff.new_content or diff.old_content}")
    
    # 创建 A/B 测试
    print("\n" + "=" * 60)
    print("创建 A/B 测试")
    print("=" * 60)
    
    ab_test = manager.create_ab_test(
        template_name="customer-service",
        versions=["v1.0.0", "v1.1.0"],
        traffic_split={"v1.0.0": 0.5, "v1.1.0": 0.5},
        metrics=["satisfaction", "response_time", "resolution_rate"],
        min_sample_size=500,
        duration_hours=48
    )
    print(f"测试ID: {ab_test.test_id}")
    print(f"流量分配: {ab_test.traffic_split}")


if __name__ == "__main__":
    main()

四、基于 Git 的 Prompt 管理方案

4.1 目录结构设计

prompt-repo/
├── .git/
│
├── templates/                    # Prompt 模板目录
│   ├── customer-service/
│   │   ├── _meta/
│   │   │   ├── config.yaml       # 模板配置
│   │   │   └── variables.yaml    # 变量定义
│   │   ├── v1.0.0/
│   │   │   ├── system.txt        # 系统提示词
│   │   │   ├── user.txt          # 用户提示词模板
│   │   │   └── examples.json     # 示例
│   │   ├── v1.1.0/
│   │   │   └── ...
│   │   └── _current -> v1.1.0/   # 符号链接指向当前版本
│   │
│   ├── sentiment-analysis/
│   │   ├── _meta/
│   │   ├── v1.0.0/
│   │   └── v2.0.0/
│   │
│   └── summarization/
│       ├── _meta/
│       └── ...
│
├── tests/                        # 测试用例
│   ├── customer-service/
│   │   ├── v1.0.0_test.yaml
│   │   ├── v1.1.0_test.yaml
│   │   └── golden/
│   │       └── expected_outputs.jsonl
│   │
│   └── sentiment-analysis/
│       └── ...
│
├── scripts/                      # 管理脚本
│   ├── deploy.py
│   ├── test.py
│   └── export.py
│
├── config.yaml                   # 全局配置
├── .promptrc                     # 本地配置
│
└── README.md

4.2 模板配置文件

# templates/customer-service/_meta/config.yaml
name: customer-service
description: 通用客服对话 Prompt
category: conversation
owner: customer-service-team

variables:
  - name: user_message
    type: str
    required: true
    description: 用户消息
  
  - name: user_tier
    type: str
    required: false
    default: normal
    description: 用户等级
  
  - name: language
    type: str
    required: false
    default: zh
    description: 回复语言

model:
  provider: openai
  model: gpt-4
  parameters:
    temperature: 0.7
    max_tokens: 500

metrics:
  - name: satisfaction
    type: float
    target: > 4.5
  
  - name: response_time_ms
    type: int
    target: < 1000

deployment:
  staging: customer-service-staging
  production: customer-service-prod

versions:
  v1.0.0:
    status: archived
    created_by: zhangsan
    created_at: "2024-01-15"
  
  v1.1.0:
    status: production
    created_by: lisi
    created_at: "2024-02-20"
    changelog: |
      - 添加 VIP 用户支持
      - 优化开场白

4.3 CI/CD 集成

# .github/workflows/prompt-ci.yml
name: Prompt CI/CD

on:
  push:
    paths:
      - 'templates/**'
      - 'tests/**'
  pull_request:
    paths:
      - 'templates/**'
      - 'tests/**'

jobs:
  # 1. 语法检查
  lint:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      
      - name: Install linters
        run: |
          pip install promptlint pyyaml
      
      - name: Lint templates
        run: |
          promptlint templates/
      
      - name: Validate YAML configs
        run: |
          find templates -name "*.yaml" -exec python -c "
            import yaml, sys
            yaml.safe_load(open(sys.argv[1]))
          " {} \;

  # 2. 单元测试
  test:
    runs-on: ubuntu-latest
    needs: lint
    strategy:
      matrix:
        template:
          - customer-service
          - sentiment-analysis
          - summarization
        version:
          - v1.0.0
          - v1.1.0
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      
      - name: Install dependencies
        run: pip install pytest prompttest
      
      - name: Run tests
        run: |
          pytest tests/${{ matrix.template }}/ \
            --template=${{ matrix.template }} \
            --version=${{ matrix.version }} \
            --verbose
      
      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: coverage.xml

  # 3. 集成测试 (与实际 API)
  integration-test:
    runs-on: ubuntu-latest
    needs: test
    if: github.ref == 'refs/heads/main'
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Deploy to staging
        run: |
          python scripts/deploy.py \
            --template customer-service \
            --version ${{ matrix.version }} \
            --env staging
      
      - name: Run E2E tests
        run: |
          pytest tests/e2e/ \
            --env staging \
            --parallel

  # 4. 性能基准测试
  benchmark:
    runs-on: ubuntu-latest
    needs: test
    if: github.event_name == 'pull_request'
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Run benchmarks
        run: |
          python scripts/benchmark.py \
            --template ${{ matrix.template }} \
            --versions v1.0.0 v1.1.0 \
            --samples 100
      
      - name: Comment results
        uses: actions/github-script@v6
        with:
          script: |
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: 'Benchmark results: ...'
            })

  # 5. 部署到生产
  deploy:
    runs-on: ubuntu-latest
    needs: [integration-test, benchmark]
    if: github.event_name == 'release'
    
    environment:
      name: production
      url: https://prompts.example.com
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Deploy to production
        run: |
          python scripts/deploy.py \
            --template ${{ matrix.template }} \
            --version ${{ matrix.version }} \
            --env production \
            --approve
      
      - name: Notify
        run: |
          curl -X POST ${{ secrets.SLACK_WEBHOOK }} \
            -d "Deployment completed: ${{ matrix.template }}@${{ matrix.version }}"

五、自建 Prompt Management System 架构

5.1 系统架构图

┌─────────────────────────────────────────────────────────────────┐
│                   Prompt Management System 架构                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│                         ┌─────────────────┐                      │
│                         │   Web UI        │                      │
│                         │  (React/Vue)    │                      │
│                         └────────┬────────┘                      │
│                                  │                               │
│                         ┌────────▼────────┐                      │
│                         │   API Gateway   │                      │
│                         │   (Kong/Nginx)  │                      │
│                         └────────┬────────┘                      │
│                                  │                               │
│   ┌──────────────────────────────┼──────────────────────────┐   │
│   │                              │                          │   │
│   ▼                              ▼                          ▼   │
│ ┌───────────┐            ┌───────────────┐            ┌─────────┐│
│ │  Template │            │    Test      │            │ Deploy  ││
│ │  Service  │            │   Service    │            │ Service ││
│ └─────┬─────┘            └───────┬───────┘            └────┬────┘│
│       │                          │                         │      │
│       └──────────────────────────┼─────────────────────────┘      │
│                                  │                                 │
│                         ┌────────▼────────┐                        │
│                         │  Result Agg    │                        │
│                         │    Service     │                        │
│                         └────────┬────────┘                        │
│                                  │                                 │
│   ┌──────────────────────────────┼──────────────────────────┐     │
│   │                              │                          │     │
│   ▼                              ▼                          ▼     │
│ ┌───────────┐            ┌───────────────┐            ┌─────────┐ │
│ │ PostgreSQL│            │    Redis      │            │   S3    │ │
│ │(Metadata) │            │   (Cache)     │            │(Templates│ │
│ └───────────┘            └───────────────┘            └─────────┘ │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

5.2 完整 API 实现

# prompt_management_api.py
"""
Prompt 管理系统 API 实现
FastAPI + SQLAlchemy + Redis
"""

import os
import re
import json
import hashlib
import asyncio
from typing import Dict, List, Optional, Any
from datetime import datetime
from enum import Enum
from dataclasses import dataclass, asdict
from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from sqlalchemy import Column, String, Text, DateTime, JSON, Boolean, Integer, ForeignKey, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session, relationship
from sqlalchemy.pool import QueuePool
import redis.asyncio as redis
import aiohttp

# ============== 配置 ==============
DATABASE_URL = os.getenv("DATABASE_URL", "postgresql://prompt:prompt@localhost:5432/prompts")
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
API_SECRET = os.getenv("API_SECRET", "secret")

# ============== 数据库模型 ==============
Base = declarative_base()


class PromptTemplateDB(Base):
    """Prompt 模板表"""
    __tablename__ = "prompt_templates"
    
    id = Column(String(100), primary_key=True)
    name = Column(String(100), nullable=False)
    category = Column(String(50))
    description = Column(Text)
    owner_team = Column(String(100))
    created_by = Column(String(100))
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    versions = relationship("PromptVersionDB", back_populates="template", cascade="all, delete-orphan")


class PromptVersionDB(Base):
    """Prompt 版本表"""
    __tablename__ = "prompt_versions"
    
    id = Column(String(100), primary_key=True)
    template_id = Column(String(100), ForeignKey("prompt_templates.id", ondelete="CASCADE"))
    version = Column(String(50), nullable=False)
    content = Column(Text, nullable=False)
    variables = Column(JSON, default=dict)
    model_config = Column(JSON, default=dict)
    test_cases = Column(JSON, default=list)
    
    content_hash = Column(String(64))  # 内容 hash，用于去重
    parent_version = Column(String(50))
    change_log = Column(Text)
    
    status = Column(String(20), default="draft")  # draft, testing, approved, production, archived
    
    created_by = Column(String(100))
    created_at = Column(DateTime, default=datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
    
    template = relationship("PromptTemplateDB", back_populates="versions")
    test_results = relationship("TestResultDB", back_populates="version", cascade="all, delete-orphan")


class TestResultDB(Base):
    """测试结果表"""
    __tablename__ = "test_results"
    
    id = Column(String(100), primary_key=True, default=lambda: f"tr_{datetime.now().strftime('%Y%m%d%H%M%S')}")
    version_id = Column(String(100), ForeignKey("prompt_versions.id", ondelete="CASCADE"))
    
    test_type = Column(String(50))  # unit, integration, benchmark
    input_data = Column(JSON)
    expected_output = Column(Text)
    actual_output = Column(Text)
    
    metrics = Column(JSON, default=dict)  # 性能指标
    
    passed = Column(Boolean)
    error_message = Column(Text)
    
    created_at = Column(DateTime, default=datetime.utcnow)


class ABTestDB(Base):
    """A/B 测试表"""
    __tablename__ = "ab_tests"
    
    id = Column(String(100), primary_key=True)
    template_id = Column(String(100), ForeignKey("prompt_templates.id", ondelete="CASCADE"))
    
    name = Column(String(100))
    description = Column(Text)
    
    # 测试配置
    versions = Column(JSON)  # 参与的版本
    traffic_split = Column(JSON)  # 流量分配
    metrics = Column(JSON)  # 评估指标
    
    # 时间配置
    start_time = Column(DateTime)
    end_time = Column(DateTime)
    
    # 结果
    results = Column(JSON)
    winner = Column(String(50))
    confidence = Column.Float)
    
    status = Column(String(20), default="pending")  # pending, running, completed, cancelled
    
    created_by = Column(String(100))
    created_at = Column(DateTime, default=datetime.utcnow)


# ============== Pydantic 模型 ==============
class TemplateType(str, Enum):
    CONVERSATION = "conversation"
    CLASSIFICATION = "classification"
    GENERATION = "generation"
    EXTRACTION = "extraction"


class VersionStatus(str, Enum):
    DRAFT = "draft"
    TESTING = "testing"
    APPROVED = "approved"
    PRODUCTION = "production"
    ARCHIVED = "archived"


class TemplateCreate(BaseModel):
    name: str
    category: Optional[str] = None
    description: Optional[str] = None
    owner_team: Optional[str] = None


class TemplateResponse(BaseModel):
    id: str
    name: str
    category: Optional[str]
    description: Optional[str]
    owner_team: Optional[str]
    created_by: str
    created_at: datetime
    latest_version: Optional[str]
    production_version: Optional[str]
    
    class Config:
        from_attributes = True


class VersionCreate(BaseModel):
    version: str
    content: str
    variables: Dict[str, Any] = Field(default_factory=dict)
    model_config: Dict[str, Any] = Field(default_factory=dict)
    test_cases: List[Dict[str, Any]] = Field(default_factory=list)
    parent_version: Optional[str] = None
    change_log: Optional[str] = None


class VersionResponse(BaseModel):
    id: str
    template_id: str
    version: str
    content: str
    variables: Dict[str, Any]
    model_config: Dict[str, Any]
    test_cases: List[Dict[str, Any]]
    content_hash: str
    parent_version: Optional[str]
    change_log: Optional[str]
    status: str
    created_by: str
    created_at: datetime
    
    class Config:
        from_attributes = True


class TestCaseExecute(BaseModel):
    input_data: Dict[str, Any]
    expected_output: Optional[str] = None
    model_provider: str = "openai"
    model_name: str = "gpt-4"
    parameters: Dict[str, Any] = Field(default_factory=dict)


class TestResultResponse(BaseModel):
    id: str
    version_id: str
    test_type: str
    input_data: Dict[str, Any]
    expected_output: Optional[str]
    actual_output: Optional[str]
    metrics: Dict[str, Any]
    passed: bool
    error_message: Optional[str]
    created_at: datetime
    
    class Config:
        from_attributes = True


class ABTestCreate(BaseModel):
    name: str
    description: Optional[str] = None
    versions: List[str]
    traffic_split: Dict[str, float]
    metrics: List[str]
    duration_hours: int = 24


class ABTestResponse(BaseModel):
    id: str
    template_id: str
    name: str
    description: Optional[str]
    versions: List[str]
    traffic_split: Dict[str, float]
    metrics: List[str]
    status: str
    winner: Optional[str]
    confidence: Optional[float]
    created_at: datetime
    
    class Config:
        from_attributes = True


# ============== API 实现 ==============
app = FastAPI(title="Prompt Management System", version="1.0.0")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 数据库
engine = create_engine(DATABASE_URL, poolclass=QueuePool, pool_size=10)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# Redis
redis_client: Optional[redis.Redis] = None


@app.on_event("startup")
async def startup():
    global redis_client
    redis_client = redis.from_url(REDIS_URL)
    Base.metadata.create_all(bind=engine)


@app.on_event("shutdown")
async def shutdown():
    if redis_client:
        await redis_client.close()


# ============== API 端点 ==============

@app.get("/health")
async def health():
    return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}


# --- 模板管理 ---

@app.post("/templates", response_model=TemplateResponse)
async def create_template(
    template: TemplateCreate,
    db: Session = Depends(get_db)
):
    """创建 Prompt 模板"""
    template_id = f"tpl_{template.name}_{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    db_template = PromptTemplateDB(
        id=template_id,
        name=template.name,
        category=template.category,
        description=template.description,
        owner_team=template.owner_team,
        created_by="system",
    )
    db.add(db_template)
    db.commit()
    db.refresh(db_template)
    
    return TemplateResponse(
        id=db_template.id,
        name=db_template.name,
        category=db_template.category,
        description=db_template.description,
        owner_team=db_template.owner_team,
        created_by=db_template.created_by,
        created_at=db_template.created_at,
        latest_version=None,
        production_version=None,
    )


@app.get("/templates", response_model=List[TemplateResponse])
async def list_templates(
    category: Optional[str] = None,
    skip: int = 0,
    limit: int = 100,
    db: Session = Depends(get_db)
):
    """列出所有模板"""
    query = db.query(PromptTemplateDB)
    if category:
        query = query.filter(PromptTemplateDB.category == category)
    
    templates = query.offset(skip).limit(limit).all()
    
    results = []
    for t in templates:
        versions = db.query(PromptVersionDB).filter(
            PromptVersionDB.template_id == t.id
        ).all()
        
        latest = max((v for v in versions), key=lambda x: x.created_at, default=None)
        production = next((v for v in versions if v.status == "production"), None)
        
        results.append(TemplateResponse(
            id=t.id,
            name=t.name,
            category=t.category,
            description=t.description,
            owner_team=t.owner_team,
            created_by=t.created_by,
            created_at=t.created_at,
            latest_version=latest.version if latest else None,
            production_version=production.version if production else None,
        ))
    
    return results


@app.get("/templates/{template_id}", response_model=TemplateResponse)
async def get_template(template_id: str, db: Session = Depends(get_db)):
    """获取模板详情"""
    template = db.query(PromptTemplateDB).filter(
        PromptTemplateDB.id == template_id
    ).first()
    
    if not template:
        raise HTTPException(status_code=404, detail="Template not found")
    
    versions = db.query(PromptVersionDB).filter(
        PromptVersionDB.template_id == template_id
    ).all()
    
    latest = max((v for v in versions), key=lambda x: x.created_at, default=None)
    production = next((v for v in versions if v.status == "production"), None)
    
    return TemplateResponse(
        id=template.id,
        name=template.name,
        category=template.category,
        description=template.description,
        owner_team=template.owner_team,
        created_by=template.created_by,
        created_at=template.created_at,
        latest_version=latest.version if latest else None,
        production_version=production.version if production else None,
    )


# --- 版本管理 ---

@app.post("/templates/{template_id}/versions", response_model=VersionResponse)
async def create_version(
    template_id: str,
    version_data: VersionCreate,
    db: Session = Depends(get_db)
):
    """创建新版本"""
    template = db.query(PromptTemplateDB).filter(
        PromptTemplateDB.id == template_id
    ).first()
    
    if not template:
        raise HTTPException(status_code=404, detail="Template not found")
    
    # 计算内容 hash
    content_hash = hashlib.sha256(version_data.content.encode()).hexdigest()
    
    # 检查是否重复
    existing = db.query(PromptVersionDB).filter(
        PromptVersionDB.template_id == template_id,
        PromptVersionDB.content_hash == content_hash
    ).first()
    
    if existing:
        raise HTTPException(
            status_code=409,
            detail=f"Duplicate content, existing version: {existing.version}"
        )
    
    # 创建版本
    version_id = f"{template_id}:{version_data.version}"
    
    db_version = PromptVersionDB(
        id=version_id,
        template_id=template_id,
        version=version_data.version,
        content=version_data.content,
        variables=version_data.variables,
        model_config=version_data.model_config,
        test_cases=version_data.test_cases,
        content_hash=content_hash,
        parent_version=version_data.parent_version,
        change_log=version_data.change_log,
        status="draft",
        created_by="system",
    )
    db.add(db_version)
    db.commit()
    db.refresh(db_version)
    
    return VersionResponse(
        id=db_version.id,
        template_id=db_version.template_id,
        version=db_version.version,
        content=db_version.content,
        variables=db_version.variables,
        model_config=db_version.model_config,
        test_cases=db_version.test_cases,
        content_hash=db_version.content_hash,
        parent_version=db_version.parent_version,
        change_log=db_version.change_log,
        status=db_version.status,
        created_by=db_version.created_by,
        created_at=db_version.created_at,
    )


@app.get("/templates/{template_id}/versions", response_model=List[VersionResponse])
async def list_versions(
    template_id: str,
    status: Optional[str] = None,
    db: Session = Depends(get_db)
):
    """列出所有版本"""
    query = db.query(PromptVersionDB).filter(
        PromptVersionDB.template_id == template_id
    )
    
    if status:
        query = query.filter(PromptVersionDB.status == status)
    
    versions = query.order_by(PromptVersionDB.created_at.desc()).all()
    
    return [
        VersionResponse(
            id=v.id,
            template_id=v.template_id,
            version=v.version,
            content=v.content,
            variables=v.variables,
            model_config=v.model_config,
            test_cases=v.test_cases,
            content_hash=v.content_hash,
            parent_version=v.parent_version,
            change_log=v.change_log,
            status=v.status,
            created_by=v.created_by,
            created_at=v.created_at,
        )
        for v in versions
    ]


@app.get("/templates/{template_id}/versions/{version}", response_model=VersionResponse)
async def get_version(
    template_id: str,
    version: str,
    db: Session = Depends(get_db)
):
    """获取指定版本"""
    db_version = db.query(PromptVersionDB).filter(
        PromptVersionDB.template_id == template_id,
        PromptVersionDB.version == version
    ).first()
    
    if not db_version:
        raise HTTPException(status_code=404, detail="Version not found")
    
    return VersionResponse(
        id=db_version.id,
        template_id=db_version.template_id,
        version=db_version.version,
        content=db_version.content,
        variables=db_version.variables,
        model_config=db_version.model_config,
        test_cases=db_version.test_cases,
        content_hash=db_version.content_hash,
        parent_version=db_version.parent_version,
        change_log=db_version.change_log,
        status=db_version.status,
        created_by=db_version.created_by,
        created_at=db_version.created_at,
    )


# --- 测试执行 ---

@app.post("/templates/{template_id}/versions/{version}/test")
async def execute_test(
    template_id: str,
    version: str,
    test_case: TestCaseExecute,
    db: Session = Depends(get_db)
):
    """执行测试用例"""
    db_version = db.query(PromptVersionDB).filter(
        PromptVersionDB.template_id == template_id,
        PromptVersionDB.version == version
    ).first()
    
    if not db_version:
        raise HTTPException(status_code=404, detail="Version not found")
    
    # 渲染 Prompt
    try:
        # 简单变量替换
        content = db_version.content
        for key, value in test_case.input_data.items():
            content = content.replace(f"${{{key}}}", str(value))
            content = content.replace(f"${key}", str(value))
        
        # 调用 LLM API
        actual_output = await call_llm_api(
            content=content,
            provider=test_case.model_provider,
            model=test_case.model_name,
            parameters=test_case.parameters,
        )
        
        # 计算匹配度
        passed = True
        if test_case.expected_output:
            # 简单的包含检查，实际应该更复杂
            passed = test_case.expected_output.lower() in actual_output.lower()
        
        # 记录结果
        result_id = f"tr_{datetime.now().strftime('%Y%m%d%H%M%S%f')}"
        test_result = TestResultDB(
            id=result_id,
            version_id=db_version.id,
            test_type="unit",
            input_data=test_case.input_data,
            expected_output=test_case.expected_output,
            actual_output=actual_output,
            passed=passed,
        )
        db.add(test_result)
        db.commit()
        
        return {
            "result_id": result_id,
            "passed": passed,
            "actual_output": actual_output,
        }
        
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


async def call_llm_api(
    content: str,
    provider: str,
    model: str,
    parameters: Dict[str, Any]
) -> str:
    """调用 LLM API"""
    # 这里简化实现，实际应该调用真实的 API
    if provider == "openai":
        # 实际调用 OpenAI API
        api_key = os.getenv("OPENAI_API_KEY")
        # ... 调用逻辑
        pass
    
    # 返回模拟结果
    return "这是模拟的 LLM 输出"


@app.get("/templates/{template_id}/versions/{version}/results")
async def get_test_results(
    template_id: str,
    version: str,
    db: Session = Depends(get_db)
):
    """获取测试结果"""
    db_version = db.query(PromptVersionDB).filter(
        PromptVersionDB.template_id == template_id,
        PromptVersionDB.version == version
    ).first()
    
    if not db_version:
        raise HTTPException(status_code=404, detail="Version not found")
    
    results = db.query(TestResultDB).filter(
        TestResultDB.version_id == db_version.id
    ).order_by(TestResultDB.created_at.desc()).limit(100).all()
    
    return [
        TestResultResponse(
            id=r.id,
            version_id=r.version_id,
            test_type=r.test_type,
            input_data=r.input_data,
            expected_output=r.expected_output,
            actual_output=r.actual_output,
            metrics=r.metrics or {},
            passed=r.passed,
            error_message=r.error_message,
            created_at=r.created_at,
        )
        for r in results
    ]


# --- A/B 测试 ---

@app.post("/templates/{template_id}/ab-tests")
async def create_ab_test(
    template_id: str,
    ab_test: ABTestCreate,
    db: Session = Depends(get_db)
):
    """创建 A/B 测试"""
    # 验证版本存在
    for v in ab_test.versions:
        version = db.query(PromptVersionDB).filter(
            PromptVersionDB.template_id == template_id,
            PromptVersionDB.version == v
        ).first()
        if not version:
            raise HTTPException(status_code=404, detail=f"Version {v} not found")
    
    test_id = f"ab_{template_id}_{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    db_ab_test = ABTestDB(
        id=test_id,
        template_id=template_id,
        name=ab_test.name,
        description=ab_test.description,
        versions=ab_test.versions,
        traffic_split=ab_test.traffic_split,
        metrics=ab_test.metrics,
        duration_hours=ab_test.duration_hours,
        start_time=datetime.utcnow(),
        status="pending",
        created_by="system",
    )
    db.add(db_ab_test)
    db.commit()
    
    return ABTestResponse(
        id=db_ab_test.id,
        template_id=db_ab_test.template_id,
        name=db_ab_test.name,
        description=db_ab_test.description,
        versions=db_ab_test.versions,
        traffic_split=db_ab_test.traffic_split,
        metrics=db_ab_test.metrics,
        status=db_ab_test.status,
        winner=None,
        confidence=None,
        created_at=db_ab_test.created_at,
    )


# 启动
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

六、与 CI/CD 集成

6.1 完整的 CI/CD 流程

# .github/workflows/prompt-full-ci.yml
name: Prompt Management CI/CD

on:
  push:
    branches: [main]
    paths:
      - 'prompts/**'
  pull_request:
    paths:
      - 'prompts/**'
  release:
    types: [published]

env:
  PROMPT_REGISTRY_URL: ${{ secrets.PROMPT_REGISTRY_URL }}
  PROMPT_REGISTRY_TOKEN: ${{ secrets.PROMPT_REGISTRY_TOKEN }}

jobs:
  # 阶段 1: 验证
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.10'
      
      - name: Install tools
        run: pip install prompt-linter pyyaml jsonschema
      
      - name: Validate prompt syntax
        run: |
          python -m prompt_tools.lint prompts/
      
      - name: Validate YAML schemas
        run: |
          python -c "
            import yaml, glob
            for f in glob.glob('prompts/**/*.yaml', recursive=True):
                yaml.safe_load(open(f))
                print(f'✓ {f}')
          "
      
      - name: Check for secrets in prompts
        run: |
          grep -r '\${.*API.*KEY}' prompts/ || echo "No API keys found in prompts"

  # 阶段 2: 单元测试
  unit-test:
    runs-on: ubuntu-latest
    needs: validate
    strategy:
      matrix:
        python-version: ['3.9', '3.10', '3.11']
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Set up Python ${{ matrix.python-version }}
        uses: actions/setup-python@v4
        with:
          python-version: ${{ matrix.python-version }}
      
      - name: Install dependencies
        run: |
          pip install -e .[dev]
          pip install pytest pytest-cov
      
      - name: Run unit tests
        run: |
          pytest tests/unit/ \
            --cov=prompt_tools \
            --cov-report=xml
      
      - name: Upload coverage
        uses: codecov/codecov-action@v3
        with:
          files: coverage.xml

  # 阶段 3: 集成测试
  integration-test:
    runs-on: ubuntu-latest
    needs: unit-test
    if: github.event_name == 'pull_request'
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Deploy to staging
        run: |
          curl -X POST "$PROMPT_REGISTRY_URL/api/deploy" \
            -H "Authorization: Bearer $PROMPT_REGISTRY_TOKEN" \
            -d "{\"env\": \"staging\", \"sha\": \"$GITHUB_SHA\"}"
      
      - name: Run integration tests
        run: |
          pytest tests/integration/ \
            --base-url=${{ secrets.STAGING_URL }} \
            --api-key=${{ secrets.TEST_API_KEY }}
      
      - name: Run regression tests
        run: |
          pytest tests/regression/ \
            --baseline=main \
            --current=$GITHUB_SHA

  # 阶段 4: 性能测试
  benchmark:
    runs-on: ubuntu-latest
    needs: unit-test
    if: github.event_name == 'pull_request'
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Run benchmark
        run: |
          python scripts/benchmark.py \
            --template customer-service \
            --versions v1.0.0 v1.1.0 \
            --samples 1000 \
            --output benchmark_results.json
      
      - name: Store benchmark results
        uses: actions/upload-artifact@v3
        with:
          name: benchmark-results
          path: benchmark_results.json
      
      - name: Compare with baseline
        run: |
          python scripts/compare_benchmarks.py \
            --current benchmark_results.json \
            --baseline ${{ secrets.BASELINE_BENCHMARK }}

  # 阶段 5: 部署到生产
  deploy:
    runs-on: ubuntu-latest
    needs: [integration-test, benchmark]
    if: github.event_name == 'release'
    
    environment:
      name: production
      url: https://prompts.example.com
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Create release tag
        run: |
          TAG="v$(date +'%Y%m%d')-${{ github.sha }}"
          git tag $TAG
          git push origin $TAG
      
      - name: Deploy to production
        run: |
          curl -X POST "$PROMPT_REGISTRY_URL/api/deploy" \
            -H "Authorization: Bearer $PROMPT_REGISTRY_TOKEN" \
            -d "{
              \"env\": \"production\",
              \"sha\": \"$GITHUB_SHA\",
              \"release\": \"${{ github.event.release.tag_name }}\"
            }"
      
      - name: Notify deployment
        run: |
          curl -X POST "${{ secrets.SLACK_WEBHOOK }}" \
            -H 'Content-Type: application/json' \
            -d "{
              \"text\": \"🚀 Prompt deployed to production: ${{ github.event.release.tag_name }}\",
              \"attachments\": [{
                \"color\": \"good\",
                \"fields\": [
                  {\"title\": \"Version\", \"value\": \"${{ github.event.release.tag_name }}\", \"short\": true},
                  {\"title\": \"Commit\", \"value\": \"${{ github.sha }}\", \"short\": true}
                ]
              }]
            }"

  # 阶段 6: 监控
  monitor:
    runs-on: ubuntu-latest
    needs: deploy
    if: github.event_name == 'release'
    
    steps:
      - uses: actions/checkout@v4
      
      - name: Setup monitoring
        run: |
          python scripts/setup_monitoring.py \
            --version ${{ github.event.release.tag_name }} \
            --metrics-endpoint ${{ secrets.METRICS_ENDPOINT }}
      
      - name: Wait for baseline
        run: sleep 3600  # 等待 1 小时收集基线数据
      
      - name: Generate report
        run: |
          python scripts/generate_deployment_report.py \
            --version ${{ github.event.release.tag_name }} \
            --output deployment_report.md
      
      - name: Create GitHub release notes
        uses: actions/create-release@v1
        env:
          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
        with:
          tag_name: ${{ github.event.release.tag_name }}
          release_name: ${{ github.event.release.name }}
          body_path: deployment_report.md

总结

本文系统性地介绍了 Prompt 版本管理的全链路解决方案，从面临的独特挑战，到模板化设计、版本对比、A/B 测试，再到完整的工程化实现。

核心要点

理解特殊性：Prompt 管理不同于代码管理，需要考虑非确定性、上下文依赖、评估困难等问题
模板化设计：

变量系统：支持类型检查、默认值、验证器
模板继承：复用通用组件
条件/循环：支持动态内容生成

版本控制：

完整的历史记录和变更日志
语义化版本号
差异对比（文本 diff + 语义 diff）

质量保障：

单元测试：验证变量渲染
集成测试：端到端验证
性能测试：延迟、吞吐量基准

A/B 测试：

流量分配控制
多指标评估
统计显著性检验

工程化集成：

Git 管理代码和 Prompt 统一
CI/CD 自动化测试和部署
监控告警和回滚机制

笔者的实践经验

Prompt 管理的最佳实践是：像管理代码一样管理 Prompt，像管理模型一样管理版本。

不要忽视 Prompt 的版本控制，它可能比模型版本更频繁地变化。

希望本文对你构建 Prompt 管理体系有所帮助！

相关资源：

目录CONTENT

Prompt 版本控制：比代码版本管理更复杂的那点事