Prompt 优化与评估：生产级提示词的量化测试与持续改进

import json
from typing import List, Dict, Any
from dataclasses import dataclass

@dataclass
class TestCase:
    """单个测试用例"""
    id: str
    input: str
    expected_output: str
    category: str
    difficulty: str  # easy/medium/hard
    notes: str = ""

def load_test_suite(filepath: str) -> List[TestCase]:
    """从 JSON 文件加载测试数据集"""
    with open(filepath, 'r', encoding='utf-8') as f:
        data = json.load(f)
    return [TestCase(**item) for item in data]

# 测试数据集示例
test_cases = [
    TestCase(
        id="extract-001",
        input="John Smith, born March 15, 1990, works at Google as a Senior Engineer. Contact: john@google.com",
        expected_output='{"name":"John Smith","birth_date":"1990-03-15","company":"Google","role":"Senior Engineer","email":"john@google.com"}',
        category="信息提取",
        difficulty="easy",
    ),
    TestCase(
        id="extract-002",
        input="Dr. Sarah Johnson-Williams（出生于1985年7月22日）目前在 Meta Platforms, Inc. 担任 AI Research Scientist。",
        expected_output='{"name":"Sarah Johnson-Williams","birth_date":"1985-07-22","company":"Meta Platforms, Inc.","role":"AI Research Scientist","email":"sarah.jw@meta.com"}',
        category="信息提取",
        difficulty="medium",
        notes="包含连字符姓名、中文日期格式",
    ),
    TestCase(
        id="extract-003",
        input="张三 / 李四 / 王五 三人于 2023-01-15 共同创立了 XYZ 科技有限公司，注册资本 1000 万元。",
        expected_output='{"company":"XYZ 科技有限公司","founding_date":"2023-01-15","capital":"1000 万元"}',
        category="信息提取",
        difficulty="hard",
        notes="多条实体、结构化数据提取",
    ),
]

import re
import json
from typing import Dict, List, Tuple

class PromptEvaluator:
    """Prompt 评估器"""

    def __init__(self, api_client, prompt_template: str):
        self.api = api_client
        self.prompt_template = prompt_template
        self.results: List[Dict] = []

    def run_test(self, test_case: TestCase, temperature: float = 0.0) -> Dict[str, Any]:
        """运行单个测试用例"""
        prompt = self.prompt_template.format(input=test_case.input)
        response = self.api.generate(
            prompt=prompt,
            temperature=temperature,
            max_tokens=1024
        )
        score = self._score_output(response.text, test_case.expected_output)
        result = {
            "test_id": test_case.id,
            "category": test_case.category,
            "difficulty": test_case.difficulty,
            "prompt_tokens": response.usage.prompt_tokens,
            "completion_tokens": response.usage.completion_tokens,
            "total_tokens": response.usage.total_tokens,
            "latency_ms": response.latency_ms,
            "score": score,
            "output": response.text,
        }
        self.results.append(result)
        return result

    def _score_output(self, actual: str, expected: str) -> Dict[str, float]:
        """计算输出评分"""
        try:
            actual_data = json.loads(actual.strip())
            expected_data = json.loads(expected.strip())
        except json.JSONDecodeError:
            return {"accuracy": 0.0, "parse_success": False}

        scores = {}
        for key in expected_data:
            if key in actual_data:
                exact_match = 1.0 if str(actual_data[key]) == str(expected_data[key]) else 0.0
                fuzzy_match = self._fuzzy_match(str(actual_data[key]), str(expected_data[key]))
                scores[key] = max(exact_match, fuzzy_match)
            else:
                scores[key] = 0.0

        extra_fields = set(actual_data.keys()) - set(expected_data.keys())
        penalty = len(extra_fields) * 0.05
        overall = (sum(scores.values()) / len(scores)) - penalty
        return {
            "accuracy": max(0.0, min(1.0, overall)),
            "parse_success": True,
            "field_scores": scores,
            "extra_fields": list(extra_fields),
        }

    def _fuzzy_match(self, a: str, b: str) -> float:
        """模糊匹配：处理格式差异"""
        a_norm = re.sub(r'[^\w]', '', a.lower())
        b_norm = re.sub(r'[^\w]', '', b.lower())
        if a_norm == b_norm: return 1.0
        if a_norm in b_norm or b_norm in a_norm: return 0.8
        return 0.0

    def run_stability_test(self, test_case: TestCase, n_runs: int = 10) -> Dict:
        """稳定性测试：同一输入多次运行"""
        scores = []
        for _ in range(n_runs):
            result = self.run_test(test_case, temperature=0.7)
            scores.append(result["score"]["accuracy"])
        import statistics
        return {
            "test_id": test_case.id,
            "mean_score": statistics.mean(scores),
            "std_dev": statistics.stdev(scores) if len(scores) > 1 else 0,
            "min_score": min(scores),
            "max_score": max(scores),
        }

def optimize_few_shot_examples(
    examples: List[Dict],
    target_count: int = 3,
    api_client=None
) -> List[Dict]:
    """智能选择最优的 Few-Shot 示例"""
    if api_client:
        # 使用模型自己评估每个示例的价值
        scores = []
        for ex in examples:
            eval_prompt = f"评估以下示例对信息提取任务的帮助程度（1-10分）：\n输入: {ex['input']}\n输出: {ex['output']}"
            response = api_client.generate(eval_prompt)
            scores.append(int(response.text.strip()))
        ranked = sorted(zip(examples, scores), key=lambda x: x[1], reverse=True)
        return [ex for ex, _ in ranked[:target_count]]
    
    # 无模型时，按输入长度和复杂度排序
    examples_by_complexity = sorted(
        examples,
        key=lambda x: len(x['input']) + len(x['output']),
        reverse=True
    )
    return examples_by_complexity[:target_count]

class DynamicPromptRouter:
    """根据输入特征动态选择最优 Prompt 策略"""
    
    def __init__(self):
        self.strategies = {
            "simple": {
                "prompt": "提取以下文本中的关键信息，输出 JSON：{input}",
                "temperature": 0.0,
                "max_tokens": 256,
            },
            "complex": {
                "prompt": "你是一个专业信息提取器。请从以下文本中提取：name, birth_date, company, role, email。严格输出 JSON 格式。缺失字段设为 null。日期统一 YYYY-MM-DD。{text}",
                "temperature": 0.0,
                "max_tokens": 512,
            },
            "ambiguous": {
                "prompt": "请仔细分析以下文本，提取人物信息。注意文本可能包含歧义。输出 JSON，如不确定请在 uncertainty 字段标注原因。{text}",
                "temperature": 0.3,
                "max_tokens": 768,
            },
        }
    
    def classify_input(self, text: str) -> str:
        """分类输入复杂度"""
        word_count = len(text.split())
        has_multiple = text.count(',') > 3 or text.count('、') > 2
        has_dates = bool(re.search(r'\d{4}[-/]\d{1,2}[-/]\d{1,2}', text))
        complexity = min(word_count / 50, 3) + (2 if has_multiple else 0) + (1 if has_dates else 0)
        if complexity <= 2: return "simple"
        if complexity <= 5: return "complex"
        return "ambiguous"
    
    def route(self, text: str) -> Dict:
        """根据输入选择最优策略"""
        strategy_name = self.classify_input(text)
        strategy = self.strategies[strategy_name]
        return {
            "strategy": strategy_name,
            "prompt": strategy["prompt"].format(text=text),
            "temperature": strategy["temperature"],
            "max_tokens": strategy["max_tokens"],
        }

# 动态路由的成本效益
# simple (60% 占比): 180 tokens -> 加权 108
# complex (30% 占比): 350 tokens -> 加权 105
# ambiguous (10% 占比): 520 tokens -> 加权 52
# 总计: 265 tokens (对比固定 complex 策略 350 tokens，节省 24%)

import hashlib
import json
from datetime import datetime
from pathlib import Path

class PromptVersionManager:
    """Prompt 版本管理器"""
    
    def __init__(self, prompts_dir: str = "prompts"):
        self.prompts_dir = Path(prompts_dir)
        self.prompts_dir.mkdir(exist_ok=True)
        self.history = self._load_history()
    
    def register(self, name: str, template: str, metadata: Dict = None) -> str:
        """注册新版本的 Prompt"""
        version_id = hashlib.md5(f"{name}:{template}".encode()).hexdigest()[:8]
        record = {
            "name": name, "version": version_id,
            "template": template, "metadata": metadata or {},
            "created_at": datetime.now().isoformat(),
        }
        filepath = self.prompts_dir / f"{name}_{version_id}.txt"
        filepath.write_text(template, encoding='utf-8')
        if name not in self.history:
            self.history[name] = []
        self.history[name].append(record)
        self._save_history()
        return version_id
    
    def rollback(self, name: str, version_id: str) -> Dict:
        """回滚到指定版本"""
        if name not in self.history:
            raise ValueError(f"Prompt '{name}' not found")
        for record in self.history[name]:
            if record["version"] == version_id:
                return self.register(name, record["template"], {
                    **record["metadata"],
                    "rolled_back_from": self.history[name][-1]["version"],
                })
        raise ValueError(f"Version '{version_id}' not found")

class RegressionDetector:
    """Prompt 回归检测器"""
    
    def __init__(self, evaluator: PromptEvaluator, test_suite: List[TestCase]):
        self.evaluator = evaluator
        self.test_suite = test_suite
        self.baseline_file = Path("baseline_scores.json")
    
    def save_baseline(self, prompt_name: str):
        """保存当前 Prompt 的基线分数"""
        results = self._run_all_tests(prompt_name)
        baseline = {
            "prompt_name": prompt_name,
            "timestamp": datetime.now().isoformat(),
            "per_test_scores": {r["test_id"]: r["score"]["accuracy"] for r in results},
            "overall_accuracy": sum(r["score"]["accuracy"] for r in results) / len(results),
        }
        self.baseline_file.write_text(json.dumps(baseline, indent=2))
        return baseline
    
    def check_regression(self, prompt_name: str) -> Dict:
        """检查是否有回归"""
        if not self.baseline_file.exists():
            return {"status": "no_baseline"}
        baseline = json.loads(self.baseline_file.read_text())
        current_results = self._run_all_tests(prompt_name)
        current_scores = {r["test_id"]: r["score"]["accuracy"] for r in current_results}
        regressions = []
        for test_id, baseline_score in baseline["per_test_scores"].items():
            current_score = current_scores.get(test_id, 0)
            diff = current_score - baseline_score
            if diff < -0.1:  # 下降超过 10%
                regressions.append({"test_id": test_id, "baseline": baseline_score, "current": current_score, "diff": diff})
        return {
            "status": "regression" if regressions else "ok",
            "regressions": regressions,
            "current_accuracy": sum(current_scores.values()) / len(current_scores),
            "baseline_accuracy": baseline["overall_accuracy"],
        }