Prompt 版本管理平台搭建：Git for Prompts

┌────────────────────────────────────────────┐
│              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)

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

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
            }
        }

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