26：Prompt版本控制与A/B测试体系搭建

class PromptVersion:
    def __init__(self, prompt_id, version, content, metadata=None):
        """初始化Prompt版本"""
        self.prompt_id = prompt_id
        self.version = version
        self.content = content
        self.metadata = metadata or {}
        self.created_at = time.time()
        self.created_by = "system"
    
    def to_dict(self):
        """转换为字典"""
        return {
            "prompt_id": self.prompt_id,
            "version": self.version,
            "content": self.content,
            "metadata": self.metadata,
            "created_at": self.created_at,
            "created_by": self.created_by
        }
    
    @classmethod
    def from_dict(cls, data):
        """从字典创建对象"""
        version = cls(
            data["prompt_id"],
            data["version"],
            data["content"],
            data.get("metadata", {})
        )
        version.created_at = data.get("created_at", time.time())
        version.created_by = data.get("created_by", "system")
        return version

# 使用示例
prompt_version = PromptVersion(
    "prompt_001",
    "v1.0.0",
    "请分析以下文本的情感倾向：{text}",
    {"description": "情感分析Prompt", "tags": ["情感分析", "NLP"]}
)
print(prompt_version.to_dict())

class PromptVersionRepository:
    def __init__(self, storage_path):
        """初始化版本库"""
        self.storage_path = storage_path
        os.makedirs(self.storage_path, exist_ok=True)
        self._load_repository()
    
    def _load_repository(self):
        """加载版本库"""
        self.prompts = {}
        for prompt_id in os.listdir(self.storage_path):
            prompt_dir = os.path.join(self.storage_path, prompt_id)
            if os.path.isdir(prompt_dir):
                self.prompts[prompt_id] = {}
                for version_file in os.listdir(prompt_dir):
                    if version_file.endswith(".json"):
                        version_path = os.path.join(prompt_dir, version_file)
                        with open(version_path, 'r', encoding='utf-8') as f:
                            data = json.load(f)
                            version = PromptVersion.from_dict(data)
                            self.prompts[prompt_id][version.version] = version
    
    def save_version(self, prompt_version):
        """保存版本"""
        prompt_dir = os.path.join(self.storage_path, prompt_version.prompt_id)
        os.makedirs(prompt_dir, exist_ok=True)
        version_file = os.path.join(prompt_dir, f"{prompt_version.version}.json")
        with open(version_file, 'w', encoding='utf-8') as f:
            json.dump(prompt_version.to_dict(), f, indent=2, ensure_ascii=False)
        
        # 更新内存中的版本信息
        if prompt_version.prompt_id not in self.prompts:
            self.prompts[prompt_version.prompt_id] = {}
        self.prompts[prompt_version.prompt_id][prompt_version.version] = prompt_version
    
    def get_version(self, prompt_id, version):
        """获取指定版本"""
        if prompt_id in self.prompts and version in self.prompts[prompt_id]:
            return self.prompts[prompt_id][version]
        return None
    
    def get_latest_version(self, prompt_id):
        """获取最新版本"""
        if prompt_id not in self.prompts or not self.prompts[prompt_id]:
            return None
        # 按版本号排序，返回最新版本
        versions = list(self.prompts[prompt_id].keys())
        # 简单的版本号排序，实际应用中可以使用更复杂的版本号比较
        versions.sort(reverse=True)
        return self.prompts[prompt_id][versions[0]]
    
    def get_all_versions(self, prompt_id):
        """获取所有版本"""
        if prompt_id not in self.prompts:
            return []
        return list(self.prompts[prompt_id].values())
    
    def delete_version(self, prompt_id, version):
        """删除版本"""
        if prompt_id in self.prompts and version in self.prompts[prompt_id]:
            # 删除文件
            version_file = os.path.join(self.storage_path, prompt_id, f"{version}.json")
            if os.path.exists(version_file):
                os.remove(version_file)
            # 删除内存中的版本信息
            del self.prompts[prompt_id][version]
            # 如果没有版本了，删除prompt目录
            if not self.prompts[prompt_id]:
                del self.prompts[prompt_id]
                prompt_dir = os.path.join(self.storage_path, prompt_id)
                if os.path.exists(prompt_dir):
                    os.rmdir(prompt_dir)

# 使用示例
repository = PromptVersionRepository("./prompt_versions")
repository.save_version(prompt_version)
latest_version = repository.get_latest_version("prompt_001")
print(latest_version.to_dict())

class BranchManager:
    def __init__(self, repository):
        """初始化分支管理器"""
        self.repository = repository
        self.branches = {}
        self._load_branches()
    
    def _load_branches(self):
        """加载分支信息"""
        # 从存储中加载分支信息
        # 这里简化处理，实际应用中可以存储在数据库或文件中
        pass
    
    def create_branch(self, branch_name, prompt_id, base_version):
        """创建分支"""
        self.branches[branch_name] = {
            "prompt_id": prompt_id,
            "base_version": base_version,
            "created_at": time.time()
        }
        # 保存分支信息
        # 这里简化处理，实际应用中可以存储在数据库或文件中
    
    def switch_branch(self, branch_name):
        """切换分支"""
        if branch_name not in self.branches:
            raise ValueError(f"Branch {branch_name} does not exist")
        branch_info = self.branches[branch_name]
        prompt_id = branch_info["prompt_id"]
        base_version = branch_info["base_version"]
        return self.repository.get_version(prompt_id, base_version)
    
    def merge_branch(self, source_branch, target_branch):
        """合并分支"""
        if source_branch not in self.branches:
            raise ValueError(f"Source branch {source_branch} does not exist")
        if target_branch not in self.branches:
            raise ValueError(f"Target branch {target_branch} does not exist")
        
        # 获取源分支和目标分支的信息
        source_info = self.branches[source_branch]
        target_info = self.branches[target_branch]
        
        # 检查是否可以合并
        if source_info["prompt_id"] != target_info["prompt_id"]:
            raise ValueError("Cannot merge branches from different prompts")
        
        # 这里简化处理，实际应用中需要更复杂的合并逻辑
        # 例如，处理冲突、生成新版本等
        
        return True
    
    def list_branches(self):
        """列出所有分支"""
        return list(self.branches.keys())

# 使用示例
branch_manager = BranchManager(repository)
branch_manager.create_branch("feature-1", "prompt_001", "v1.0.0")
branches = branch_manager.list_branches()
print("Branches:", branches)

class ABTestConfig:
    def __init__(self, test_id, prompt_versions, test_cases, metrics=None, sample_size=100):
        """初始化A/B测试配置"""
        self.test_id = test_id
        self.prompt_versions = prompt_versions  # 要测试的Prompt版本列表
        self.test_cases = test_cases  # 测试用例列表
        self.metrics = metrics or ["accuracy", "response_time", "satisfaction"]  # 评估指标
        self.sample_size = sample_size  # 样本大小
        self.created_at = time.time()
    
    def to_dict(self):
        """转换为字典"""
        return {
            "test_id": self.test_id,
            "prompt_versions": self.prompt_versions,
            "test_cases": self.test_cases,
            "metrics": self.metrics,
            "sample_size": self.sample_size,
            "created_at": self.created_at
        }
    
    @classmethod
    def from_dict(cls, data):
        """从字典创建对象"""
        return cls(
            data["test_id"],
            data["prompt_versions"],
            data["test_cases"],
            data.get("metrics", ["accuracy", "response_time", "satisfaction"]),
            data.get("sample_size", 100)
        )

# 使用示例
test_cases = [
    {"input": "这部电影非常精彩", "expected": "正面"},
    {"input": "这个餐厅的服务态度很差", "expected": "负面"},
    {"input": "今天天气一般", "expected": "中性"}
]

prompt_versions = ["v1.0.0", "v1.1.0"]

test_config = ABTestConfig(
    "test_001",
    prompt_versions,
    test_cases,
    sample_size=50
)
print(test_config.to_dict())

class ABTestExecutor:
    def __init__(self, repository, model):
        """初始化测试执行器"""
        self.repository = repository
        self.model = model
    
    def execute_test(self, test_config):
        """执行A/B测试"""
        results = {}
        
        for version in test_config.prompt_versions:
            prompt_version = self.repository.get_version("prompt_001", version)
            if not prompt_version:
                continue
            
            version_results = []
            for test_case in test_config.test_cases:
                # 执行测试
                result = self._execute_test_case(prompt_version, test_case)
                version_results.append(result)
            
            results[version] = version_results
        
        return results
    
    def _execute_test_case(self, prompt_version, test_case):
        """执行单个测试用例"""
        # 构建测试Prompt
        test_prompt = prompt_version.content.replace("{text}", test_case["input"])
        
        # 记录开始时间
        start_time = time.time()
        
        # 执行模型调用
        response = self.model(test_prompt)
        
        # 记录结束时间
        end_time = time.time()
        response_time = end_time - start_time
        
        # 评估结果
        correct = self._evaluate_result(response, test_case["expected"])
        
        return {
            "input": test_case["input"],
            "expected": test_case["expected"],
            "actual": response,
            "correct": correct,
            "response_time": response_time
        }
    
    def _evaluate_result(self, actual, expected):
        """评估测试结果"""
        # 简单的评估方法，实际应用中可以使用更复杂的评估方法
        return expected.lower() in actual.lower()

# 使用示例
from langchain.chat_models import ChatOpenAI

model = ChatOpenAI(model="gpt-4-turbo")
executor = ABTestExecutor(repository, model)
test_results = executor.execute_test(test_config)
print("Test results:", test_results)

class ABTestAnalyzer:
    def __init__(self):
        """初始化测试分析器"""
        pass
    
    def analyze_results(self, test_results, metrics):
        """分析测试结果"""
        analysis = {}
        
        for version, results in test_results.items():
            version_analysis = {}
            
            # 计算准确率
            if "accuracy" in metrics:
                correct_count = sum(1 for r in results if r["correct"])
                accuracy = correct_count / len(results) * 100
                version_analysis["accuracy"] = accuracy
            
            # 计算响应时间
            if "response_time" in metrics:
                response_times = [r["response_time"] for r in results]
                avg_response_time = sum(response_times) / len(response_times)
                version_analysis["response_time"] = avg_response_time
            
            # 计算满意度（这里简化处理，实际应用中可以通过用户反馈等方式获取）
            if "satisfaction" in metrics:
                # 假设满意度与准确率正相关
                correct_count = sum(1 for r in results if r["correct"])
                satisfaction = correct_count / len(results) * 100
                version_analysis["satisfaction"] = satisfaction
            
            analysis[version] = version_analysis
        
        # 计算版本间的差异
        if len(test_results) > 1:
            analysis["comparison"] = self._compare_versions(analysis)
        
        return analysis
    
    def _compare_versions(self, analysis):
        """比较不同版本的性能"""
        comparison = {}
        versions = list(analysis.keys())
        
        for i in range(len(versions)):
            for j in range(i + 1, len(versions)):
                version1 = versions[i]
                version2 = versions[j]
                comparison[f"{version1}_vs_{version2}"] = {}
                
                metrics = list(analysis[version1].keys())
                for metric in metrics:
                    value1 = analysis[version1][metric]
                    value2 = analysis[version2][metric]
                    difference = value2 - value1
                    percentage_change = (difference / value1) * 100 if value1 != 0 else 0
                    
                    comparison[f"{version1}_vs_{version2}"][metric] = {
                        "value1": value1,
                        "value2": value2,
                        "difference": difference,
                        "percentage_change": percentage_change
                    }
        
        return comparison

# 使用示例
analyzer = ABTestAnalyzer()
analysis = analyzer.analyze_results(test_results, ["accuracy", "response_time", "satisfaction"])
print("Analysis results:", analysis)

def calculate_overall_score(metrics, weights):
    """计算综合评分"""
    total_weight = sum(weights.values())
    weighted_score = 0
    
    for metric, value in metrics.items():
        if metric in weights:
            # 归一化处理
            if metric == "response_time":
                # 响应时间越短越好，需要反向归一化
                normalized_value = 100 - min(value * 10, 100)  # 假设10秒以上为0分
            else:
                normalized_value = value
            
            weighted_score += normalized_value * (weights[metric] / total_weight)
    
    return weighted_score

# 使用示例
metrics = {
    "accuracy": 90,
    "response_time": 0.5,
    "satisfaction": 85,
    "consistency": 95,
    "robustness": 80
}

weights = {
    "accuracy": 40,
    "response_time": 20,
    "satisfaction": 20,
    "consistency": 10,
    "robustness": 10
}

overall_score = calculate_overall_score(metrics, weights)
print(f"Overall score: {overall_score:.2f}")

# 创建不同版本的情感分析Prompt
v1_prompt = PromptVersion(
    "sentiment_analysis",
    "v1.0.0",
    "请分析以下文本的情感倾向，输出正面、负面或中性：\n{text}",
    {"description": "基础情感分析Prompt"}
)

v2_prompt = PromptVersion(
    "sentiment_analysis",
    "v1.1.0",
    "请仔细分析以下文本的情感倾向，考虑上下文和语气，输出正面、负面或中性：\n{text}",
    {"description": "优化版情感分析Prompt"}
)

# 保存版本
repository.save_version(v1_prompt)
repository.save_version(v2_prompt)

# 配置A/B测试
test_cases = [
    {"input": "这部电影非常精彩，演员表演出色", "expected": "正面"},
    {"input": "这个餐厅的服务态度很差，食物也不好吃", "expected": "负面"},
    {"input": "今天天气一般，不冷不热", "expected": "中性"},
    {"input": "这个产品质量很好，性价比高", "expected": "正面"},
    {"input": "服务态度恶劣，下次再也不来了", "expected": "负面"}
]

test_config = ABTestConfig(
    "sentiment_test_001",
    ["v1.0.0", "v1.1.0"],
    test_cases
)

# 执行测试
test_results = executor.execute_test(test_config)

# 分析结果
analysis = analyzer.analyze_results(test_results, ["accuracy", "response_time", "satisfaction"])
print("情感分析Prompt测试结果:", analysis)

# 创建不同版本的实体识别Prompt
v1_prompt = PromptVersion(
    "entity_recognition",
    "v1.0.0",
    "请从以下文本中识别出人名、地名和组织名：\n{text}",
    {"description": "基础实体识别Prompt"}
)

v2_prompt = PromptVersion(
    "entity_recognition",
    "v1.1.0",
    "请从以下文本中准确识别出人名、地名和组织名，格式为：人名：XXX；地名：XXX；组织名：XXX：\n{text}",
    {"description": "优化版实体识别Prompt"}
)

# 保存版本
repository.save_version(v1_prompt)
repository.save_version(v2_prompt)

# 配置A/B测试
test_cases = [
    {"input": "张三在北京大学工作，住在北京", "expected": "人名：张三；地名：北京大学、北京；组织名：无"},
    {"input": "李四是阿里巴巴的员工，来自上海", "expected": "人名：李四；地名：上海；组织名：阿里巴巴"},
    {"input": "王五在腾讯工作，毕业于清华大学", "expected": "人名：王五；地名：清华大学；组织名：腾讯"}
]

test_config = ABTestConfig(
    "entity_test_001",
    ["v1.0.0", "v1.1.0"],
    test_cases
)

# 执行测试
test_results = executor.execute_test(test_config)

# 分析结果
analysis = analyzer.analyze_results(test_results, ["accuracy", "response_time", "satisfaction"])
print("实体识别Prompt测试结果:", analysis)

# 创建不同版本的问答系统Prompt
v1_prompt = PromptVersion(
    "qa_system",
    "v1.0.0",
    "请根据以下上下文回答问题：\n上下文：{context}\n问题：{question}\n答案：",
    {"description": "基础问答系统Prompt"}
)

v2_prompt = PromptVersion(
    "qa_system",
    "v1.1.0",
    "请仔细阅读以下上下文，然后准确回答问题，确保答案直接来自上下文：\n上下文：{context}\n问题：{question}\n答案：",
    {"description": "优化版问答系统Prompt"}
)

# 保存版本
repository.save_version(v1_prompt)
repository.save_version(v2_prompt)

# 配置A/B测试
test_cases = [
    {
        "input": "上下文：北京是中国的首都，也是中国的政治、文化和国际交往中心。\n问题：中国的首都是哪里？",
        "expected": "北京"
    },
    {
        "input": "上下文：苹果是一种常见的水果，富含维生素C和纤维素。\n问题：苹果富含什么？",
        "expected": "维生素C和纤维素"
    },
    {
        "input": "上下文：爱因斯坦是著名的物理学家，提出了相对论。\n问题：爱因斯坦提出了什么理论？",
        "expected": "相对论"
    }
]

test_config = ABTestConfig(
    "qa_test_001",
    ["v1.0.0", "v1.1.0"],
    test_cases
)

# 执行测试
test_results = executor.execute_test(test_config)

# 分析结果
analysis = analyzer.analyze_results(test_results, ["accuracy", "response_time", "satisfaction"])
print("问答系统Prompt测试结果:", analysis)

from prompt_versioning import PromptVersion, PromptVersionRepository

# 初始化版本库
repository = PromptVersionRepository("./prompt_versions")

# 创建并保存版本
prompt_version = PromptVersion(
    "prompt_id",
    "v1.0.0",
    "Prompt content",
    {"description": "Prompt description"}
)
repository.save_version(prompt_version)

# 获取版本
latest_version = repository.get_latest_version("prompt_id")
specific_version = repository.get_version("prompt_id", "v1.0.0")

# 列出所有版本
all_versions = repository.get_all_versions("prompt_id")

from prompt_testing import ABTestConfig, ABTestExecutor, ABTestAnalyzer
from langchain.chat_models import ChatOpenAI

# 初始化模型
model = ChatOpenAI(model="gpt-4-turbo")

# 初始化测试执行器和分析器
executor = ABTestExecutor(repository, model)
analyzer = ABTestAnalyzer()

# 配置测试
test_cases = [
    {"input": "测试输入1", "expected": "预期输出1"},
    {"input": "测试输入2", "expected": "预期输出2"}
]

test_config = ABTestConfig(
    "test_id",
    ["v1.0.0", "v1.1.0"],
    test_cases
)

# 执行测试
test_results = executor.execute_test(test_config)

# 分析结果
analysis = analyzer.analyze_results(test_results, ["accuracy", "response_time", "satisfaction"])
print(analysis)

# 安装依赖
pip install langchain openai

# 配置环境变量
export OPENAI_API_KEY=your-api-key

# 运行示例
python prompt_versioning.py
python prompt_testing.py