Prompt模板管理：企业级Prompt工程实践

from dataclasses import dataclass, field
from typing import Optional
from enum import Enum

class PromptCategory(Enum):
    SYSTEM = "system"           # 系统级Prompt
    TASK = "task"              # 任务级Prompt
    GUARDRAIL = "guardrail"    # 安全护栏
    UTILITY = "utility"        # 工具函数
    EVALUATION = "evaluation"  # 评估用

class PromptStatus(Enum):
    DRAFT = "draft"
    REVIEW = "review"
    TESTING = "testing"
    STAGING = "staging"
    PRODUCTION = "production"
    DEPRECATED = "deprecated"

@dataclass
class PromptTemplate:
    """Prompt模板定义"""
    id: str                              # 唯一标识
    name: str                            # 模板名称
    category: PromptCategory             # 分类
    status: PromptStatus                 # 状态
    
    # 模板内容
    system_prompt: str                   # 系统提示
    user_prompt_template: str            # 用户提示模板（含变量）
    variables: list[dict]                # 变量定义
    
    # 元数据
    description: str                     # 描述
    author: str                          # 作者
    version: str                         # 版本号
    tags: list[str]                      # 标签
    
    # 配置
    model_config: dict                   # 模型配置
    expected_output: Optional[dict]      # 期望输出格式
    
    # 质量指标
    quality_score: Optional[float]       # 质量评分
    latency_p95: Optional[float]         # P95延迟
    success_rate: Optional[float]        # 成功率
    
    # 关联
    dependencies: list[str] = field(default_factory=list)  # 依赖的其他模板
    parent_id: Optional[str] = None      # 父模板（继承关系）

class PromptTemplateEngine:
    """
    Prompt模板引擎
    支持变量插值、条件逻辑、循环和继承
    """
    
    # 模板语法示例
    TEMPLATE_EXAMPLE = """
{{#system}}
你是{{role}}，专注于{{domain}}领域。

核心规则：
{{#rules}}
- {{.}}
{{/rules}}

{{#if strict_mode}}
⚠️ 严格遵守以上规则，不允许偏离。
{{/if}}
{{/system}}

{{#user}}
{{user_input}}

{{#if context}}
相关上下文：
{{#context}}
---
{{.}}
---
{{/context}}
{{/if}}

{{#if examples}}
参考示例：
{{#examples}}
输入：{{input}}
输出：{{output}}
{{/examples}}
{{/if}}
{{/user}}
"""
    
    def __init__(self):
        self.templates: dict[str, PromptTemplate] = {}
        self.cache = {}
    
    def render(self, template_id: str, variables: dict) -> dict:
        """渲染模板"""
        template = self.templates.get(template_id)
        if not template:
            raise ValueError(f"模板 {template_id} 不存在")
        
        # 合并默认变量
        merged_vars = self._merge_defaults(template, variables)
        
        # 验证必填变量
        self._validate_variables(template, merged_vars)
        
        # 渲染
        system = self._render_text(template.system_prompt, merged_vars)
        user = self._render_text(template.user_prompt_template, merged_vars)
        
        return {
            "system": system,
            "user": user,
            "model_config": template.model_config,
            "template_id": template_id,
            "version": template.version
        }
    
    def _render_text(self, template_text: str, variables: dict) -> str:
        """渲染模板文本"""
        # 使用Jinja2或自定义模板引擎
        from jinja2 import Template
        tpl = Template(template_text)
        return tpl.render(**variables)

class PromptRegistry:
    """
    Prompt注册中心
    所有Prompt模板的单一可信来源（Single Source of Truth）
    """
    
    def __init__(self, storage_backend="postgresql"):
        self.storage = self._init_storage(storage_backend)
    
    async def register(self, template: PromptTemplate) -> str:
        """注册新模板"""
        # 验证
        self._validate_template(template)
        
        # 检查命名冲突
        if await self._exists(template.name, template.version):
            raise ValueError(f"模板 {template.name} v{template.version} 已存在")
        
        # 存储
        template_id = await self.storage.save(template)
        
        # 建立索引
        await self._update_index(template)
        
        return template_id
    
    async def get(self, template_id: str) -> PromptTemplate:
        """获取模板"""
        return await self.storage.get(template_id)
    
    async def search(self, query: dict) -> list[PromptTemplate]:
        """搜索模板"""
        # 支持按名称、标签、分类、状态搜索
        return await self.storage.search(query)
    
    async def update(self, template_id: str, 
                     updates: dict) -> PromptTemplate:
        """更新模板（创建新版本）"""
        current = await self.get(template_id)
        
        # 创建新版本
        new_version = self._increment_version(current.version)
        updated = PromptTemplate(
            **{**current.__dict__, **updates, 
               "version": new_version,
               "parent_id": template_id}
        )
        
        # 注册新版本
        new_id = await self.register(updated)
        
        # 标记旧版本
        await self.storage.update(
            template_id, {"status": PromptStatus.DEPRECATED}
        )
        
        return updated

class PromptAccessControl:
    """
    Prompt权限管理
    """
    
    PERMISSIONS = {
        "read": "查看模板",
        "write": "创建/修改模板",
        "deploy": "部署到生产",
        "delete": "删除模板",
        "export": "导出模板",
    }
    
    ROLES = {
        "viewer": ["read"],
        "developer": ["read", "write"],
        "reviewer": ["read", "write"],
        "admin": ["read", "write", "deploy", "delete", "export"],
    }
    
    def check_permission(self, user_id: str, template_id: str,
                         permission: str) -> bool:
        """检查用户权限"""
        user_role = self._get_user_role(user_id)
        allowed = self.ROLES.get(user_role, [])
        
        if permission not in allowed:
            return False
        
        # 项目级权限检查
        template = self.registry.get(template_id)
        if not self._has_project_access(user_id, template.project):
            return False
        
        return True

class PromptPipeline:
    """
    Prompt CI/CD 流水线
    从开发到部署的完整流程
    """
    
    async def run_pipeline(self, template: PromptTemplate):
        """执行完整流水线"""
        
        results = {}
        
        # 阶段1: 静态检查
        results["lint"] = await self._lint(template)
        if not results["lint"]["passed"]:
            return results
        
        # 阶段2: 单元测试
        results["unit_test"] = await self._unit_test(template)
        if not results["unit_test"]["passed"]:
            return results
        
        # 阶段3: 安全检查
        results["security"] = await self._security_scan(template)
        if not results["security"]["passed"]:
            return results
        
        # 阶段4: 性能测试
        results["performance"] = await self._performance_test(template)
        
        # 阶段5: A/B测试准备
        results["ab_setup"] = await self._setup_ab_test(template)
        
        # 阶段6: 部署
        if all(r.get("passed", True) for r in results.values()):
            results["deploy"] = await self._deploy(template)
        
        return results
    
    async def _lint(self, template: PromptTemplate) -> dict:
        """静态检查"""
        issues = []
        
        # 检查变量完整性
        used_vars = self._extract_variables(template.user_prompt_template)
        defined_vars = [v["name"] for v in template.variables]
        
        for var in used_vars:
            if var not in defined_vars:
                issues.append(f"未定义的变量: {var}")
        
        # 检查长度
        if len(template.system_prompt) > 2000:
            issues.append("System Prompt过长（>2000字符），可能影响性能")
        
        # 检查安全
        dangerous_patterns = ["ignore previous", "you are now", "system prompt"]
        for pattern in dangerous_patterns:
            if pattern in template.user_prompt_template.lower():
                issues.append(f"潜在安全风险: 包含 '{pattern}'")
        
        return {
            "passed": len(issues) == 0,
            "issues": issues
        }
    
    async def _unit_test(self, template: PromptTemplate) -> dict:
        """单元测试"""
        test_cases = template.variables.get("test_cases", [])
        
        results = []
        for case in test_cases:
            rendered = self.engine.render(template.id, case["input"])
            response = await self.llm.generate(rendered)
            
            passed = self._evaluate_response(
                response, case["expected"]
            )
            results.append({
                "case_name": case.get("name", "unnamed"),
                "passed": passed,
                "response": response[:200]
            })
        
        pass_rate = sum(r["passed"] for r in results) / len(results)
        
        return {
            "passed": pass_rate >= 0.9,
            "pass_rate": pass_rate,
            "results": results
        }
    
    async def _performance_test(self, template: PromptTemplate) -> dict:
        """性能测试"""
        import time
        
        latencies = []
        for _ in range(50):
            start = time.time()
            rendered = self.engine.render(template.id, {})
            response = await self.llm.generate(rendered)
            latencies.append(time.time() - start)
        
        import numpy as np
        return {
            "passed": np.percentile(latencies, 95) < 5.0,  # P95 < 5秒
            "p50": np.median(latencies),
            "p95": np.percentile(latencies, 95),
            "p99": np.percentile(latencies, 99),
        }

class PromptMonitor:
    """
    Prompt生产环境监控
    """
    
    def __init__(self):
        self.metrics_store = MetricsStore()
        self.alerting = AlertingSystem()
    
    async def record_invocation(self, template_id: str, version: str,
                                 invocation_data: dict):
        """记录每次Prompt调用"""
        await self.metrics_store.record({
            "template_id": template_id,
            "version": version,
            "timestamp": datetime.now(),
            "input": invocation_data["input"],
            "output": invocation_data["output"],
            "latency_ms": invocation_data["latency_ms"],
            "tokens_used": invocation_data["tokens_used"],
            "cost": invocation_data["cost"],
            "success": invocation_data["success"],
            "user_feedback": invocation_data.get("user_feedback"),
        })
        
        # 实时检查
        await self._check_anomalies(template_id, invocation_data)
    
    async def _check_anomalies(self, template_id: str, data: dict):
        """异常检测"""
        # 延迟异常
        baseline_latency = await self.metrics_store.get_baseline_latency(template_id)
        if data["latency_ms"] > baseline_latency * 3:
            await self.alerting.send_alert(
                level="warning",
                template_id=template_id,
                message=f"延迟异常: {data['latency_ms']}ms (基线: {baseline_latency}ms)"
            )
        
        # 成功率下降
        recent_success_rate = await self.metrics_store.get_recent_success_rate(
            template_id, window_minutes=30
        )
        if recent_success_rate < 0.85:
            await self.alerting.send_alert(
                level="critical",
                template_id=template_id,
                message=f"成功率下降: {recent_success_rate:.1%}"
            )
        
        # 成本异常
        daily_cost = await self.metrics_store.get_daily_cost(template_id)
        if daily_cost > 100:  # 日成本超过100元
            await self.alerting.send_alert(
                level="warning",
                template_id=template_id,
                message=f"日成本异常: ¥{daily_cost}"
            )

class PromptDashboard:
    """Prompt管理仪表盘数据生成"""
    
    def generate_report(self, date_range: tuple) -> dict:
        return {
            "overview": {
                "total_templates": self._count_templates(),
                "active_templates": self._count_active_templates(),
                "total_invocations": self._count_invocations(date_range),
                "total_cost": self._sum_cost(date_range),
                "avg_success_rate": self._avg_success_rate(date_range),
                "avg_latency_p95": self._avg_latency(date_range),
            },
            "top_templates": self._top_templates(date_range, n=10),
            "quality_issues": self._identify_quality_issues(date_range),
            "cost_breakdown": self._cost_breakdown(date_range),
            "performance_trends": self._performance_trends(date_range),
            "recommendations": self._generate_recommendations(date_range),
        }

class PromptInheritance:
    """
    Prompt模板继承系统
    支持模板之间的继承和组合
    """
    
    def resolve(self, template_id: str) -> PromptTemplate:
        """
        解析模板继承链，生成最终Prompt
        """
        template = self.registry.get(template_id)
        
        if template.parent_id:
            # 递归解析父模板
            parent = self.resolve(template.parent_id)
            
            # 合并：子模板覆盖父模板
            return self._merge(parent, template)
        
        return template
    
    def _merge(self, parent: PromptTemplate, 
               child: PromptTemplate) -> PromptTemplate:
        """合并父子模板"""
        return PromptTemplate(
            id=child.id,
            name=child.name,
            system_prompt=child.system_prompt or parent.system_prompt,
            user_prompt_template=child.user_prompt_template or parent.user_prompt_template,
            variables=self._merge_variables(parent.variables, child.variables),
            # ... 其他字段
        )