import requests
import json
import math
import os
class QwenPerplexityCalculator:
    def __init__(self, api_key):
        self.api_key = api_key
        self.api_url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
    
    def calculate_perplexity_process(self, text, probabilities):
        """
        使用Qwen API模拟困惑度计算过程
        
        参数:
            text: 要计算困惑度的文本
            probabilities: 各位置的条件概率值列表
        """
        # 构建详细的prompt，要求模型按照指定流程计算困惑度
        prompt = f"""请按照以下步骤计算文本的困惑度(PPL)，使用我提供的概率值：
文本: "{text}"
条件概率值:
P("今天" | <开始>) = {probabilities[0]}
P("天气" | "今天") = {probabilities[1]}
P("很好" | "今天 天气") = {probabilities[2]}
请逐步计算并解释：
1. 计算对数概率和：log({probabilities[0]}) + log({probabilities[1]}) + log({probabilities[2]})
2. 计算平均负对数概率：- (对数概率和) / N (N=3)
3. 计算困惑度：exp(平均负对数概率)
请详细展示每一步的计算过程和结果，最后给出完整的困惑度值。"""
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
        
        payload = {
            "model": "qwen-max",
            "input": {
                "messages": [{"role": "user", "content": prompt}]
            },
            "parameters": {
                "result_format": "text",
                "max_tokens": 800,
                "temperature": 0.1
            }
        }
        
        try:
            response = requests.post(self.api_url, headers=headers, data=json.dumps(payload))
            result = response.json()
            
            if "output" in result and "text" in result["output"]:
                return result["output"]["text"].strip()
            return None
        except Exception as e:
            print(f"API调用失败: {e}")
            return None
    
    def manual_calculation(self, text, probabilities):
        """
        手动计算困惑度过程，用于验证
        """
        print(f"文本: '{text}'")
        print("\n1. 计算每个位置的条件概率:")
        for i, prob in enumerate(probabilities):
            if i == 0:
                print(f"   P('{text.split()[i]}' | <开始>) = {prob}")
            else:
                context = " ".join(text.split()[:i])
                print(f"   P('{text.split()[i]}' | '{context}') = {prob}")
        
        print("\n2. 计算对数概率和:")
        log_probs = [math.log(p) for p in probabilities]
        log_sum = sum(log_probs)
        log_str = " + ".join([f"log({p})" for p in probabilities])
        print(f"   {log_str} = {log_str.replace('log', str(round(math.log(probabilities[0]), 2))).replace('log', str(round(math.log(probabilities[1]), 2))).replace('log', str(round(math.log(probabilities[2]), 2)))} = {round(log_sum, 2)}")
        
        print("\n3. 计算平均负对数概率:")
        n = len(probabilities)
        avg_neg_log_prob = -log_sum / n
        print(f"   -({round(log_sum, 2)}) / {n} = {round(avg_neg_log_prob, 2)}")
        
        print("\n4. 计算困惑度:")
        perplexity = math.exp(avg_neg_log_prob)
        print(f"   exp({round(avg_neg_log_prob, 2)}) = {round(perplexity, 2)}")
        
        return perplexity
# 使用示例
if __name__ == "__main__":
    # 初始化计算器
    calculator = QwenPerplexityCalculator(os.environ.get("DASHSCOPE_API_KEY"))
    
    # 定义文本和概率值
    text = "今天 天气 很好"
    probabilities = [0.4, 0.6, 0.5]
    
    print("使用Qwen API模拟困惑度计算过程:")
    print("=" * 50)
    
    # 使用Qwen API获取计算过程
    process = calculator.calculate_perplexity_process(text, probabilities)
    if process:
        print(process)
    else:
        print("API调用失败，使用手动计算")
    
    print("\n" + "=" * 50)
    print("手动计算验证:")
    print("=" * 50)
    
    # 手动计算验证
    final_ppl = calculator.manual_calculation(text, probabilities)
    print(f"\n最终困惑度: {final_ppl:.2f}")

import requests
import json
import re
import os
class QwenIntrinsicEvaluator:
    def __init__(self, api_key):
        self.api_key = api_key
        self.api_url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
    
    def estimate_perplexity_with_details(self, text):
        """
        估算文本困惑度并返回详细计算过程
        
        参数:
            text: 要计算困惑度的文本
        
        返回:
            包含困惑度值和计算细节的字典
        """
        # 构建详细的prompt，要求模型解释计算过程
        prompt = f"""作为语言模型专家，请分析以下文本的语言建模难度，并估算其困惑度(PPL)值。
请详细解释你的计算过程，包括以下内容：
1. 文本的语言复杂性分析
2. 词汇的常见程度评估
3. 句法结构的复杂性
4. 最终估算的困惑度值及其解释
文本: "{text}"
请按以下格式回复：
【文本分析】
[对文本复杂性的分析]
【词汇评估】
[对词汇常见程度的评估]
【句法分析】
[对句法复杂性的分析]
【困惑度估算】
估算值: [数值]
解释: [对估算值的解释]
"""
        
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
        
        payload = {
            "model": "qwen-max",
            "input": {
                "messages": [{"role": "user", "content": prompt}]
            },
            "parameters": {
                "result_format": "text",
                "max_tokens": 800,
                "temperature": 0.1
            }
        }
        
        try:
            response = requests.post(self.api_url, headers=headers, data=json.dumps(payload))
            result = response.json()
            
            if "output" in result and "text" in result["output"]:
                response_text = result["output"]["text"].strip()
                
                # 解析响应内容
                details = self._parse_response_details(response_text)
                
                # 提取困惑度值
                match = re.search(r"估算值:\s*([\d\.]+)", response_text)
                if match:
                    details['perplexity'] = float(match.group(1))
                else:
                    # 备用方法：尝试从文本中提取任何数字
                    num_match = re.search(r"(\d+\.\d+|\d+)", response_text)
                    if num_match:
                        details['perplexity'] = float(num_match.group())
                    else:
                        details['perplexity'] = None
                
                details['full_response'] = response_text
                return details
            return None
        except Exception as e:
            print(f"API调用失败: {e}")
            return None
    
    def _parse_response_details(self, response_text):
        """解析模型响应中的详细信息"""
        details = {
            'text_analysis': '',
            'vocabulary_assessment': '',
            'syntax_analysis': '',
            'explanation': ''
        }
        
        # 使用正则表达式提取各个部分
        text_analysis_match = re.search(r"【文本分析】\s*(.*?)(?=【|$)", response_text, re.DOTALL)
        vocab_match = re.search(r"【词汇评估】\s*(.*?)(?=【|$)", response_text, re.DOTALL)
        syntax_match = re.search(r"【句法分析】\s*(.*?)(?=【|$)", response_text, re.DOTALL)
        explanation_match = re.search(r"解释:\s*(.*?)(?=【|$)", response_text, re.DOTALL)
        
        if text_analysis_match:
            details['text_analysis'] = text_analysis_match.group(1).strip()
        if vocab_match:
            details['vocabulary_assessment'] = vocab_match.group(1).strip()
        if syntax_match:
            details['syntax_analysis'] = syntax_match.group(1).strip()
        if explanation_match:
            details['explanation'] = explanation_match.group(1).strip()
        
        return details
# 使用示例
if __name__ == "__main__":
    evaluator = QwenIntrinsicEvaluator(os.environ.get("DASHSCOPE_API_KEY"))
    text = "语言模型的评测是一个重要的研究领域"
    
    # 获取详细的困惑度分析
    result = evaluator.estimate_perplexity_with_details(text)
    
    if result and 'perplexity' in result:
        print(f"文本: '{text}'")
        print(f"估算困惑度值: {result['perplexity']}")
        print("\n=== 详细分析 ===")
        print(f"文本分析: {result.get('text_analysis', '无')}")
        print(f"词汇评估: {result.get('vocabulary_assessment', '无')}")
        print(f"句法分析: {result.get('syntax_analysis', '无')}")
        print(f"解释: {result.get('explanation', '无')}")
        
        print("\n=== 完整响应 ===")
        print(result.get('full_response', '无'))
    else:
        print("无法计算困惑度")

import requests
import json
import numpy as np
import pandas as pd
from sentence_transformers import SentenceTransformer
import faiss
import re
from typing import List, Dict, Any
import time
import os
class QwenAccuracyEvaluator:
    def __init__(self, api_key: str, knowledge_base: List[str] = None):
        """
        基于Qwen API的外在评测准确率计算系统
        
        参数:
            api_key: Qwen API密钥
            knowledge_base: 知识库文本列表（可选，用于检索增强）
        """
        self.api_key = api_key
        self.api_url = "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation"
        
        # 如果有知识库，构建Faiss索引
        if knowledge_base:
            self.knowledge_base = knowledge_base
            self.embedder = SentenceTransformer('D:/modelscope/hub/models/sentence-transformers/all-MiniLM-L6-v2')
            self._build_faiss_index()
            self.use_rag = True
        else:
            self.use_rag = False
        
        # 测试数据集 - 问答对
        self.qa_test_set = [
            {
                "id": 1,
                "question": "Python中如何创建一个空列表？",
                "expected_answer": "可以使用方括号 [] 或 list() 函数创建空列表",
                "acceptable_variations": [
                    "使用 [] 创建空列表",
                    "用 list() 创建空列表",
                    "空列表可以通过 [] 或 list() 创建"
                ]
            },
            {
                "id": 2,
                "question": "什么是机器学习？",
                "expected_answer": "机器学习是人工智能的一个子领域，使计算机能够从数据中学习而不需要明确编程",
                "acceptable_variations": [
                    "机器学习是AI的一个分支，让计算机从数据中学习",
                    "ML是计算机通过数据自动学习和改进的技术"
                ]
            },
            {
                "id": 3,
                "question": "如何在Python中打印'Hello, World!'？",
                "expected_answer": "使用 print('Hello, World!')",
                "acceptable_variations": [
                    "print('Hello, World!')",
                    "使用print函数打印Hello, World!"
                ]
            }
        ]
    
    def _build_faiss_index(self):
        """构建Faiss向量索引（如果使用知识库）"""
        if not hasattr(self, 'knowledge_base'):
            return
            
        # 生成知识库文本的向量
        knowledge_vectors = self.embedder.encode(self.knowledge_base)
        self.dimension = knowledge_vectors.shape[1]
        
        # 创建Faiss索引
        self.index = faiss.IndexFlatL2(self.dimension)
        self.index.add(knowledge_vectors.astype('float32'))
        
        print(f"知识库索引构建完成，包含 {len(self.knowledge_base)} 条知识")
    
    def _retrieve_relevant_texts(self, query: str, top_k: int = 3) -> List[str]:
        """检索相关文本（如果使用知识库）"""
        if not hasattr(self, 'index'):
            return []
            
        # 生成查询向量
        query_vector = self.embedder.encode([query])
        
        # 检索相似文本
        D, I = self.index.search(query_vector.astype('float32'), top_k)
        
        retrieved_texts = []
        for i in range(top_k):
            if I[0][i] < len(self.knowledge_base):
                retrieved_texts.append(self.knowledge_base[I[0][i]])
        
        return retrieved_texts
    
    def call_qwen_api(self, prompt: str, max_tokens: int = 500) -> str:
        """调用Qwen API生成文本"""
        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}"
        }
        
        payload = {
            "model": "qwen-max",  # 可以根据需要选择不同模型
            "input": {
                "messages": [
                    {
                        "role": "user",
                        "content": prompt
                    }
                ]
            },
            "parameters": {
                "result_format": "text",
                "max_tokens": max_tokens,
                "temperature": 0.3  # 降低随机性，提高确定性
            }
        }
        
        try:
            response = requests.post(self.api_url, headers=headers, data=json.dumps(payload))
            response.raise_for_status()
            result = response.json()
            
            if "output" in result and "text" in result["output"]:
                return result["output"]["text"].strip()
            return None
        except Exception as e:
            print(f"API调用失败: {e}")
            return None
    
    def generate_answer(self, question: str) -> str:
        """生成答案"""
        if self.use_rag:
            # 使用检索增强生成(RAG)
            relevant_texts = self._retrieve_relevant_texts(question)
            context = "\n".join(relevant_texts)
            
            prompt = f"""基于以下上下文信息回答问题。如果信息不足，请明确说明。
上下文：
{context}
问题：
{question}
请提供准确、简洁的答案："""
        else:
            # 直接回答问题
            prompt = f"""请回答以下问题。提供准确、完整的回答。
问题：{question}
回答："""
        
        return self.call_qwen_api(prompt)
    
    def evaluate_accuracy(self, generated_answer: str, expected_answer: str, 
                         acceptable_variations: List[str] = None) -> Dict[str, Any]:
        """
        评估生成答案的准确率
        
        参数:
            generated_answer: 模型生成的答案
            expected_answer: 期望的标准答案
            acceptable_variations: 可接受的答案变体列表
        
        返回:
            包含准确率评估结果的字典
        """
        if not generated_answer:
            return {
                "is_correct": False,
                "confidence": 0.0,
                "match_type": "no_answer",
                "explanation": "模型未生成答案"
            }
        
        # 标准化答案（小写、去除标点）
        def normalize_text(text):
            text = text.lower()
            text = re.sub(r'[^\w\s]', '', text)  # 移除标点符号
            return text.strip()
        
        gen_norm = normalize_text(generated_answer)
        exp_norm = normalize_text(expected_answer)
        
        # 1. 精确匹配检查
        if gen_norm == exp_norm:
            return {
                "is_correct": True,
                "confidence": 1.0,
                "match_type": "exact",
                "explanation": "生成答案与标准答案完全匹配"
            }
        
        # 2. 检查可接受的变体
        if acceptable_variations:
            for variation in acceptable_variations:
                if gen_norm == normalize_text(variation):
                    return {
                        "is_correct": True,
                        "confidence": 0.9,
                        "match_type": "acceptable_variation",
                        "explanation": "生成答案与可接受变体匹配"
                    }
        
        # 3. 关键词匹配检查
        expected_keywords = set(exp_norm.split())
        generated_keywords = set(gen_norm.split())
        common_keywords = expected_keywords & generated_keywords
        
        if len(common_keywords) >= len(expected_keywords) * 0.7:  # 70%关键词匹配
            keyword_coverage = len(common_keywords) / len(expected_keywords)
            return {
                "is_correct": True,
                "confidence": keyword_coverage * 0.8,  # 最高0.8置信度
                "match_type": "keyword_based",
                "explanation": f"关键词匹配度: {keyword_coverage:.2f}"
            }
        
        # 4. 使用Qwen API进行语义评估
        evaluation_prompt = f"""请判断以下两个答案是否在语义上等价：
答案1: {expected_answer}
答案2: {generated_answer}
请只回复"是"或"否"，不要包含其他内容。"""
        
        semantic_eval = self.call_qwen_api(evaluation_prompt, max_tokens=10)
        
        if semantic_eval and "是" in semantic_eval:
            return {
                "is_correct": True,
                "confidence": 0.7,
                "match_type": "semantic",
                "explanation": "语义评估认为答案等价"
            }
        
        # 5. 所有检查都失败，答案不正确
        return {
            "is_correct": False,
            "confidence": 0.0,
            "match_type": "incorrect",
            "explanation": "答案不正确"
        }
    
    def run_accuracy_evaluation(self) -> List[Dict[str, Any]]:
        """运行准确率评测"""
        results = []
        correct_count = 0
        
        for test_case in self.qa_test_set:
            print(f"\n处理测试用例 {test_case['id']}: {test_case['question']}")
            
            # 生成答案
            generated_answer = self.generate_answer(test_case['question'])
            if not generated_answer:
                print("生成答案失败")
                results.append({
                    "test_case_id": test_case["id"],
                    "question": test_case["question"],
                    "generated_answer": None,
                    "expected_answer": test_case["expected_answer"],
                    "is_correct": False,
                    "confidence": 0.0,
                    "match_type": "no_answer"
                })
                continue
            
            print(f"生成答案: {generated_answer}")
            
            # 评估准确率
            accuracy_result = self.evaluate_accuracy(
                generated_answer, 
                test_case["expected_answer"],
                test_case.get("acceptable_variations", [])
            )
            
            # 更新正确计数
            if accuracy_result["is_correct"]:
                correct_count += 1
            
            # 保存结果
            result = {
                "test_case_id": test_case["id"],
                "question": test_case["question"],
                "generated_answer": generated_answer,
                "expected_answer": test_case["expected_answer"],
                "is_correct": accuracy_result["is_correct"],
                "confidence": accuracy_result["confidence"],
                "match_type": accuracy_result["match_type"],
                "explanation": accuracy_result["explanation"]
            }
            
            results.append(result)
            
            # 打印当前结果
            status = "正确" if accuracy_result["is_correct"] else "错误"
            print(f"评估结果: {status} (置信度: {accuracy_result['confidence']:.2f})")
            print(f"匹配类型: {accuracy_result['match_type']}")
            
            # 添加延迟避免API限制
            time.sleep(1)
        
        # 计算总体准确率
        total_cases = len(results)
        accuracy = correct_count / total_cases if total_cases > 0 else 0
        
        return results, accuracy
    
    def generate_accuracy_report(self, results: List[Dict[str, Any]], accuracy: float) -> str:
        """生成准确率评测报告"""
        if not results:
            return "无评测结果"
        
        # 创建报告
        report_lines = [
            "Qwen API外在评测准确率报告",
            "=" * 60,
            f"评测时间: {time.strftime('%Y-%m-%d %H:%M:%S')}",
            f"测试用例数量: {len(results)}",
            f"正确回答数量: {sum(1 for r in results if r['is_correct'])}",
            f"总体准确率: {accuracy:.2%}",
            "\n详细结果:",
            "=" * 60
        ]
        
        for result in results:
            status = "✓" if result["is_correct"] else "✗"
            report_lines.extend([
                f"\n{status} 用例 {result['test_case_id']}:",
                f"问题: {result['question']}",
                f"生成答案: {result['generated_answer']}",
                f"期望答案: {result['expected_answer']}",
                f"匹配类型: {result['match_type']}",
                f"置信度: {result['confidence']:.2f}",
                f"解释: {result['explanation']}"
            ])
        
        # 添加匹配类型统计
        match_types = {}
        for result in results:
            match_type = result["match_type"]
            match_types[match_type] = match_types.get(match_type, 0) + 1
        
        report_lines.extend([
            "\n匹配类型统计:",
            "=" * 60
        ])
        
        for match_type, count in match_types.items():
            report_lines.append(f"{match_type}: {count}次")
        
        return "\n".join(report_lines)
# 使用示例
if __name__ == "__main__":
    # 配置参数
    API_KEY = os.environ.get("DASHSCOPE_API_KEY")
    
    # 可选的知识库数据
    knowledge_base = [
        "Python中可以使用方括号[]或list()函数创建空列表",
        "机器学习是人工智能的一个子领域，使计算机能够从数据中学习",
        "在Python中使用print()函数可以输出文本到控制台"
    ]
    
    # 初始化评测器（可以选择是否使用知识库）
    evaluator = QwenAccuracyEvaluator(API_KEY, knowledge_base)
    
    # 运行准确率评测
    print("开始准确率评测...")
    results, accuracy = evaluator.run_accuracy_evaluation()
    
    # 生成报告
    if results:
        report = evaluator.generate_accuracy_report(results, accuracy)
        print(report)
        
        # 保存报告到文件
        with open('qwen_accuracy_report.txt', 'w', encoding='utf-8') as f:
            f.write(report)
        print("\n准确率报告已保存至 qwen_accuracy_report.txt")
    else:
        print("评测失败，无结果生成")

if gen_norm == exp_norm:
    return {
        "is_correct": True,
        "confidence": 1.0,
        "match_type": "exact",
        "explanation": "生成答案与标准答案完全匹配"
    }

for variation in acceptable_variations:
    if gen_norm == normalize_text(variation):
        return {
            "is_correct": True,
            "confidence": 0.9,
            "match_type": "acceptable_variation",
            "explanation": "生成答案与可接受变体匹配"
        }

if len(common_keywords) >= len(expected_keywords) * 0.7:
    keyword_coverage = len(common_keywords) / len(expected_keywords)
    return {
        "is_correct": True,
        "confidence": keyword_coverage * 0.8,
        "match_type": "keyword_based",
        "explanation": f"关键词匹配度: {keyword_coverage:.2f}"
    }

evaluation_prompt = f"""请判断以下两个答案是否在语义上等价：
答案1: {expected_answer}
答案2: {generated_answer}
请只回复"是"或"否"，不要包含其他内容。"""
semantic_eval = self.call_qwen_api(evaluation_prompt, max_tokens=10)