import json import csv import pandas as pd import matplotlib.pyplot as plt import seaborn as sns from datetime import datetime import requests import time from typing import Dict, List, Any, Optional import numpy as np from openai import OpenAI import os import re class CarPurchaseEvaluator: def __init__(self, api_config: Dict[str, Any]): """ 初始化购车决策评测器 Args: api_config: API配置信息，包含base_url, api_key等 """ self.api_config = api_config self.results = [] self.conversation_history = [] # 初始化OpenAI客户端 self.client = OpenAI( api_key=api_config.get("api_key", ""), base_url=api_config.get("api_url", "https://dashscope.aliyuncs.com/compatible-mode/v1"), ) # 定义购车决策评测指标和标准 self.evaluation_metrics = { "requirement_understanding": { "name": "需求理解能力", "description": "评估模型理解用户购车需求的能力", "weight": 1.2 }, "recommendation_relevance": { "name": "推荐相关性", "description": "评估模型推荐车型与用户需求的匹配程度", "weight": 1.3 }, "parameter_analysis": { "name": "参数分析能力", "description": "评估模型分析和对比车型参数的能力", "weight": 1.1 }, "context_consistency": { "name": "上下文一致性", "description": "评估模型在多轮对话中保持推荐一致性的能力", "weight": 1.0 }, "decision_support": { "name": "决策支持能力", "description": "评估模型提供最终决策建议的合理性", "weight": 1.2 }, "knowledge_accuracy": { "name": "知识准确性", "description": "评估模型提供车型信息的准确程度", "weight": 1.1 }, "structured_output": { "name": "结构化输出", "description": "评估模型生成表格和结构化信息的能力", "weight": 0.9 }, "communication_naturalness": { "name": "沟通自然度", "description": "评估模型对话的自然流畅程度", "weight": 0.8 } } # 购车决策测试场景 self.test_scenarios = self._load_test_scenarios() def _load_test_scenarios(self) -> List[Dict]: """加载购车决策测试场景""" return [ { "scenario_id": "urban_commuter", "name": "城市通勤场景", "initial_intent": "我最近打算买车，预算大约20万元左右，主要用于城市通勤，希望空间舒适、安全性高。", "follow_up_questions": [ "请考虑动力性能，我希望加速响应快一些", "油耗或电耗怎么样？我每天通勤距离约40公里", "安全配置方面有哪些亮点？", "保值率如何？我可能3-5年后换车", "品牌口碑和售后服务怎么样？" ], "table_request": "请将这几款候选车型的主要性能参数做成对比表格，包括动力、续航/油耗、空间、安全配置等关键指标", "final_decision_request": "综合考虑所有因素，请给出一款你的最终推荐车型，并详细说明理由" } ] def call_model_api(self, messages: List[Dict], max_retries: int = 3, retry_delay: int = 2) -> str: """ 调用Qwen3-Max API进行多轮对话，增加更健壮的错误处理 Args: messages: 对话消息列表 max_retries: 最大重试次数 retry_delay: 重试延迟（秒） Returns: model_response: 模型响应文本 """ for attempt in range(max_retries): try: # 构建extra_body参数 extra_body = {} # Qwen3特定参数 if "enable_thinking" in self.api_config: extra_body["enable_thinking"] = self.api_config["enable_thinking"] if "top_k" in self.api_config: extra_body["top_k"] = self.api_config["top_k"] if "thinking_budget" in self.api_config: extra_body["thinking_budget"] = self.api_config["thinking_budget"] # 构建请求参数 request_params = { "model": self.api_config.get("model_name", "qwen3-max"), "messages": messages, "max_tokens": self.api_config.get("max_tokens", 2000), "temperature": self.api_config.get("temperature", 0.7), "top_p": self.api_config.get("top_p", 0.8), "stream": False, } # 添加extra_body参数（如果有） if extra_body: request_params["extra_body"] = extra_body print(f" 发送API请求 (尝试 {attempt + 1}/{max_retries})...") # 使用OpenAI兼容的API调用方式 completion = self.client.chat.completions.create(**request_params) # 提取响应文本 if completion.choices and completion.choices[0].message: response = completion.choices[0].message.content.strip() print(f" 收到响应，长度: {len(response)} 字符") return response else: print(" API响应格式错误") return "API响应格式错误" except Exception as e: error_msg = str(e) print(f" API调用异常 (尝试 {attempt + 1}/{max_retries}): {error_msg}") # 如果是速率限制错误，等待更长时间 if "rate limit" in error_msg.lower() or "too many requests" in error_msg.lower(): wait_time = retry_delay * (attempt + 1) * 2 # 指数退避 print(f" 检测到速率限制，等待 {wait_time} 秒后重试...") time.sleep(wait_time) elif attempt < max_retries - 1: print(f" 等待 {retry_delay} 秒后重试...") time.sleep(retry_delay) else: return f"API调用失败: {error_msg}" return "API调用失败" def safe_conversation_flow(self, scenario: Dict, max_retries_per_step: int = 3) -> Dict: """ 安全的多轮对话流程，确保对话连贯性 Args: scenario: 测试场景 max_retries_per_step: 每个步骤的最大重试次数 Returns: conversation_result: 对话结果 """ conversation_result = { "stages": [], "candidate_models": [], "final_recommended_model": "", "success": True, "error_message": "" } try: # 初始化对话历史 conversation_history = [ {"role": "system", "content": "你是一个专业的汽车选购顾问，帮助用户根据需求选择合适的车型。请严格遵守以下格式要求：\n" "1. 所有车型名称请统一使用'品牌-车型'格式，例如：丰田-凯美瑞、本田-CRV、比亚迪-汉EV、特斯拉-Model3\n" "2. 品牌使用中文名称，车型使用标准型号（可使用英文缩写如CRV、Model3等）\n" "3. 除车型名称外，回答的其他部分请勿使用横杠符号，例如100-150km续航应改为100到150km续航\n" "4. 除车型可以使用英文缩写外，回答的其他部分请勿使用英文缩写符号\n" "5. 对于用户的需求请至少给出5款车型推荐，均使用上述统一格式\n" "6. 在最终决策阶段，必须只推荐一款最适合的车型，并使用统一格式明确标出\n" "7. 在对比表格中，车型名称列必须使用统一的'品牌-车型'格式"} ] candidate_models = [] # 阶段1: 提出购车意图 print(" 阶段1: 提出购车意图") conversation_history.append({"role": "user", "content": scenario["initial_intent"]}) response1 = self.call_model_api(conversation_history, max_retries=max_retries_per_step) if "API调用失败" in response1: conversation_result["success"] = False conversation_result["error_message"] = f"初始推荐阶段失败: {response1}" return conversation_result conversation_history.append({"role": "assistant", "content": response1}) # 提取候选车型 candidate_models = self.extract_candidate_models(response1) print(f" 提取到候选车型: {candidate_models}") # 评估阶段1 scores1 = self.evaluate_conversation_stage("initial_recommendation", scenario["initial_intent"], response1, scenario) conversation_result["stages"].append({ "stage": "initial_recommendation", "user_input": scenario["initial_intent"], "model_response": response1, "scores": scores1 }) # 等待确保API稳定 time.sleep(2) # 阶段2: 跟进提问分析 follow_up_scores = [] for i, question in enumerate(scenario["follow_up_questions"]): print(f" 阶段2.{i+1}: 跟进提问 - '{question[:30]}...'") conversation_history.append({"role": "user", "content": question}) response = self.call_model_api(conversation_history, max_retries=max_retries_per_step) if "API调用失败" in response: print(f" 跟进提问 {i+1} 失败，跳过此问题") continue conversation_history.append({"role": "assistant", "content": response}) # 评估跟进提问 scores = self.evaluate_conversation_stage("follow_up_analysis", question, response, scenario, candidate_models) conversation_result["stages"].append({ "stage": f"follow_up_analysis_{i+1}", "user_input": question, "model_response": response, "scores": scores }) follow_up_scores.append(scores) # 在提问之间增加等待时间，避免API限制 wait_time = 3 if i < len(scenario["follow_up_questions"]) - 1 else 2 print(f" 等待 {wait_time} 秒后进行下一轮对话...") time.sleep(wait_time) # 检查是否有足够的成功对话 if len([s for s in follow_up_scores if s]) < len(scenario["follow_up_questions"]) * 0.6: print(" 警告: 超过40%的跟进提问失败，对话质量可能受影响") # 阶段3: 参数表格对比 print(" 阶段3: 参数表格对比") conversation_history.append({"role": "user", "content": scenario["table_request"]}) response3 = self.call_model_api(conversation_history, max_retries=max_retries_per_step) if "API调用失败" in response3: print(" 参数表格对比阶段失败，使用空响应继续") response3 = "无法生成参数对比表格" else: conversation_history.append({"role": "assistant", "content": response3}) # 评估表格对比 scores3 = self.evaluate_conversation_stage("parameter_comparison", scenario["table_request"], response3, scenario, candidate_models) conversation_result["stages"].append({ "stage": "parameter_comparison", "user_input": scenario["table_request"], "model_response": response3, "scores": scores3 }) # 等待确保API稳定 time.sleep(5) # 阶段4: 最终决策 print(" 阶段4: 最终决策") conversation_history.append({"role": "user", "content": scenario["final_decision_request"]}) response4 = self.call_model_api(conversation_history, max_retries=max_retries_per_step) if "API调用失败" in response4: print(" 最终决策阶段失败，使用空响应继续") response4 = "无法提供最终推荐" else: conversation_history.append({"role": "assistant", "content": response4}) # 提取最终推荐车型 final_recommended_model = self.extract_final_recommendation(response4, candidate_models) conversation_result["final_recommended_model"] = final_recommended_model # 评估最终决策 scores4 = self.evaluate_conversation_stage("final_decision", scenario["final_decision_request"], response4, scenario, candidate_models) conversation_result["stages"].append({ "stage": "final_decision", "user_input": scenario["final_decision_request"], "model_response": response4, "scores": scores4 }) conversation_result["candidate_models"] = candidate_models print(f" 场景完成，候选车型: {candidate_models}") print(f" 最终推荐车型: {final_recommended_model}") except Exception as e: conversation_result["success"] = False conversation_result["error_message"] = f"对话流程异常: {str(e)}" print(f" 对话流程异常: {e}") return conversation_result def extract_candidate_models(self, response: str) -> List[str]: """从模型响应中提取候选车型""" # 改进的车型提取逻辑 - 专注于'品牌-车型'格式 patterns = [ # 主要模式：品牌-车型 (丰田-凯美瑞, 本田-CRV, 比亚迪-汉EV) r'([\u4e00-\u9fff]{2,6}-[A-Za-z0-9\u4e00-\u9fff]+(?:[A-Za-z0-9]*(?:EV|PHEV|DM|Pro|Max|Plus)?)?)', # 备用模式：中文品牌 车型 (如果模型偶尔不遵循格式) r'([\u4e00-\u9fff]{2,6}\s+[A-Za-z0-9\u4e00-\u9fff]+(?:[A-Za-z0-9]*(?:EV|PHEV|DM)?)?)', # 国际品牌-车型 (Tesla-Model3, BMW-X5, Mercedes-Benz-GLC) r'([A-Z][a-zA-Z]*(?:-[A-Z][a-zA-Z]*)?-[A-Z][a-zA-Z0-9]*)', ] candidates = [] for pattern in patterns: matches = re.findall(pattern, response) candidates.extend(matches) # 去重并清洗 cleaned_candidates = [] seen = set() for candidate in candidates: candidate_clean = candidate.strip() # 验证格式并标准化 standardized_model = self._standardize_model_name(candidate_clean) if (standardized_model and len(standardized_model) >= 3 and len(standardized_model) <= 30 and not self._is_invalid_model_name(standardized_model) and standardized_model not in seen): cleaned_candidates.append(standardized_model) seen.add(standardized_model) # 返回前5个有效候选 return cleaned_candidates[:5] def _standardize_model_name(self, model_name: str) -> str: """标准化车型名称格式""" # 如果已经是品牌-车型格式，直接返回 if '-' in model_name and len(model_name.split('-')) == 2: return model_name # 尝试将空格分隔转换为横杠格式 if ' ' in model_name: parts = model_name.split() if len(parts) >= 2: # 取前两部分作为品牌和车型 return f"{parts[0]}-{''.join(parts[1:])}" return model_name def _is_invalid_model_name(self, name: str) -> bool: """判断是否为无效的车型名称""" invalid_keywords = [ '推荐', '建议', '考虑', '适合', '预算', '价格', '万元', '左右', '大约', '可以', '选择', '购买', '车型', '汽车', '车辆', '品牌', '配置', '参数', '动力', '安全', '舒适', '空间', '油耗', '电耗', '续航', '加速', '性能', '方面', '比较', '对比', '优势', '缺点', '特点', '功能' ] # 检查是否包含无效关键词 if any(keyword in name for keyword in invalid_keywords): return True # 检查是否为纯数字 if name.replace('-', '').isdigit(): return True # 检查是否包含明显的非车型词汇 non_car_words = ['很好', '不错', '优秀', '合适', '满足', '需要', '要求', '用户'] if any(word in name for word in non_car_words): return True # 检查格式是否正确（至少包含一个横杠） if '-' not in name: return True return False def extract_final_recommendation(self, response: str, candidate_models: List[str]) -> str: """ 从最终决策响应中提取最推荐的唯一车型 Args: response: 最终决策阶段的模型响应 candidate_models: 之前阶段提取的候选车型列表 Returns: final_model: 最终推荐的车型名称 """ # 方法1: 寻找明确的推荐关键词 recommendation_keywords = [ '最终推荐', '最推荐', '首选', '最佳选择', '最适合', '最优选', '我的推荐是', '建议选择', '我推荐', '强烈推荐', '最终选择' ] # 按段落分割，寻找包含推荐关键词的段落 paragraphs = response.split('\n') recommendation_paragraphs = [] for para in paragraphs: if any(keyword in para for keyword in recommendation_keywords): recommendation_paragraphs.append(para) # 如果在推荐段落中找到车型，优先使用 for para in recommendation_paragraphs: for model in candidate_models: if model in para: return model # 方法2: 在响应开头部分寻找（通常最终推荐会在开头明确） first_paragraph = paragraphs[0] if paragraphs else response[:200] for model in candidate_models: if model in first_paragraph: return model # 方法3: 统计所有候选车型的出现频率，选择最频繁出现的 model_frequency = {} for model in candidate_models: frequency = response.count(model) if frequency > 0: model_frequency[model] = frequency if model_frequency: return max(model_frequency.items(), key=lambda x: x[1])[0] # 方法4: 重新从响应中提取车型，选择第一个 final_candidates = self.extract_candidate_models(response) if final_candidates: return final_candidates[0] # 方法5: 如果以上都失败，返回候选车型中的第一个 return candidate_models[0] if candidate_models else "无法确定推荐车型" def extract_table_data(self, response: str) -> Dict: """从模型响应中提取表格数据""" table_data = { "has_table": False, "rows": [], "headers": [], "format_quality": 0 } # 检查是否有表格标记 if "|" in response or "---" in response or "表" in response: table_data["has_table"] = True # 简单的表格格式检测 lines = response.split('\n') table_lines = [] in_table = False for line in lines: if "|" in line: in_table = True table_lines.append(line.strip()) elif in_table and ("---" in line or "===" in line or len(line.strip()) == 0): continue elif in_table and "|" not in line: break if table_lines: # 解析表头 if len(table_lines) > 0: headers = [h.strip() for h in table_lines[0].split('|') if h.strip()] table_data["headers"] = headers # 解析数据行 for line in table_lines[1:]: if "|" in line and "---" not in line: row = [cell.strip() for cell in line.split('|') if cell.strip()] if len(row) == len(table_data["headers"]): table_data["rows"].append(row) # 评估表格质量 if len(table_data["rows"]) >= 2 and len(table_data["headers"]) >= 3: table_data["format_quality"] = min(len(table_data["rows"]) * 2, 10) return table_data def evaluate_conversation_stage(self, stage: str, user_input: str, model_response: str, scenario: Dict, candidate_models: List[str] = None) -> Dict[str, float]: """ 评估对话阶段的表现 Args: stage: 对话阶段标识 user_input: 用户输入 model_response: 模型响应 scenario: 测试场景 candidate_models: 候选车型列表 Returns: scores: 各指标得分字典 """ scores = {} if stage == "initial_recommendation": scores = self._evaluate_initial_recommendation(user_input, model_response, scenario) elif stage.startswith("follow_up_analysis"): scores = self._evaluate_follow_up_analysis(user_input, model_response, scenario, candidate_models) elif stage == "parameter_comparison": scores = self._evaluate_parameter_comparison(user_input, model_response, scenario, candidate_models) elif stage == "final_decision": scores = self._evaluate_final_decision(user_input, model_response, scenario, candidate_models) return scores def _evaluate_initial_recommendation(self, user_input: str, response: str, scenario: Dict) -> Dict[str, float]: """评估初始推荐阶段""" scores = {} # 需求理解能力 budget_keywords = ["预算", "价格", "万元", "花费"] usage_keywords = ["通勤", "家用", "出行", "代步"] requirement_keywords = ["空间", "安全", "舒适", "动力"] budget_match = any(keyword in user_input for keyword in budget_keywords) and any(keyword in response for keyword in budget_keywords) usage_match = any(keyword in user_input for keyword in usage_keywords) and any(keyword in response for keyword in usage_keywords) requirement_match = any(keyword in user_input for keyword in requirement_keywords) and any(keyword in response for keyword in requirement_keywords) understanding_score = 5.0 if budget_match: understanding_score += 1.5 if usage_match: understanding_score += 1.5 if requirement_match: understanding_score += 2.0 scores["requirement_understanding"] = min(understanding_score, 10.0) # 推荐相关性 candidates = self.extract_candidate_models(response) relevance_score = 6.0 if len(candidates) >= 3: relevance_score += 2.0 if len(candidates) >= 5: relevance_score += 2.0 scores["recommendation_relevance"] = min(relevance_score, 10.0) # 沟通自然度 naturalness_score = 6.0 if len(response) > 100 and len(response) < 800: naturalness_score += 2.0 if "推荐" in response or "建议" in response: naturalness_score += 1.0 if "考虑" in response or "适合" in response: naturalness_score += 1.0 scores["communication_naturalness"] = min(naturalness_score, 10.0) return scores def _evaluate_follow_up_analysis(self, user_input: str, response: str, scenario: Dict, candidate_models: List[str]) -> Dict[str, float]: """评估跟进分析阶段""" scores = {} # 参数分析能力 analysis_score = 5.0 if any(keyword in response for keyword in ["马力", "扭矩", "加速", "功率"]): analysis_score += 1.0 if any(keyword in response for keyword in ["油耗", "电耗", "续航", "能耗"]): analysis_score += 1.0 if any(keyword in response for keyword in ["安全", "气囊", "辅助", "预警"]): analysis_score += 1.0 if any(keyword in response for keyword in ["保值", "残值", "二手"]): analysis_score += 1.0 if any(keyword in response for keyword in ["品牌", "口碑", "服务", "售后"]): analysis_score += 1.0 scores["parameter_analysis"] = min(analysis_score, 10.0) # 知识准确性 accuracy_score = 6.0 if any(keyword in response for keyword in ["具体", "详细", "参数", "配置"]): accuracy_score += 2.0 if any(keyword in response for keyword in ["大约", "大概", "左右", "约"]): accuracy_score -= 1.0 # 模糊表述扣分 scores["knowledge_accuracy"] = min(max(accuracy_score, 1.0), 10.0) # 上下文一致性 consistency_score = 7.0 if candidate_models: mentioned_models = sum(1 for model in candidate_models if model in response) if mentioned_models >= len(candidate_models) * 0.5: consistency_score += 2.0 elif mentioned_models == 0: consistency_score -= 3.0 scores["context_consistency"] = min(max(consistency_score, 1.0), 10.0) return scores def _evaluate_parameter_comparison(self, user_input: str, response: str, scenario: Dict, candidate_models: List[str]) -> Dict[str, float]: """评估参数对比阶段""" scores = {} # 结构化输出能力 table_data = self.extract_table_data(response) structured_score = 3.0 if table_data["has_table"]: structured_score += 4.0 structured_score += table_data["format_quality"] * 0.3 scores["structured_output"] = min(structured_score, 10.0) # 参数分析能力 comparison_score = 6.0 comparison_indicators = ["对比", "比较", "差异", "优劣", "优势", "劣势"] found_indicators = sum(1 for indicator in comparison_indicators if indicator in response) comparison_score += min(found_indicators * 0.5, 2.0) if table_data["has_table"] and len(table_data["rows"]) >= 2: comparison_score += 2.0 scores["parameter_analysis"] = min(comparison_score, 10.0) # 知识准确性 accuracy_score = 7.0 if table_data["has_table"]: accuracy_score += 1.0 if any(keyword in response for keyword in ["数据", "参数", "配置"]): accuracy_score += 1.0 if "估计" in response or "推测" in response: accuracy_score -= 1.0 scores["knowledge_accuracy"] = min(accuracy_score, 10.0) return scores def _evaluate_final_decision(self, user_input: str, response: str, scenario: Dict, candidate_models: List[str]) -> Dict[str, float]: """评估最终决策阶段""" scores = {} # 决策支持能力 decision_score = 6.0 if "推荐" in response and "理由" in response: decision_score += 2.0 if "综合考虑" in response or "权衡" in response: decision_score += 1.0 if "最适合" in response or "最优选择" in response: decision_score += 1.0 # 检查是否有明确的最终推荐（使用新的提取方法） final_recommendation = self.extract_final_recommendation(response, candidate_models) if final_recommendation and final_recommendation != "无法确定推荐车型": decision_score += 2.0 else: decision_score -= 2.0 scores["decision_support"] = min(decision_score, 10.0) # 上下文一致性 consistency_score = 7.0 if candidate_models and final_recommendation: if final_recommendation in candidate_models: consistency_score += 2.0 else: consistency_score -= 2.0 scores["context_consistency"] = min(max(consistency_score, 1.0), 10.0) # 沟通自然度 naturalness_score = 7.0 reasoning_indicators = ["因为", "由于", "考虑到", "基于"] found_indicators = sum(1 for indicator in reasoning_indicators if indicator in response) naturalness_score += min(found_indicators * 0.5, 2.0) if "希望" in response or "祝" in response: naturalness_score += 1.0 scores["communication_naturalness"] = min(naturalness_score, 10.0) return scores def run_car_purchase_evaluation(self, model_name: str, num_scenarios: int = 1) -> Dict[str, Any]: """ 运行购车决策评测 Args: model_name: 模型名称 num_scenarios: 测试的场景数量 Returns: evaluation_results: 评测结果 """ print(f"开始购车决策评测模型: {model_name}") evaluation_results = { "model_name": model_name, "evaluation_date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "api_config": {k: v for k, v in self.api_config.items() if k != 'api_key'}, "scenario_results": [], "metrics_scores": {}, "overall_score": 0 } tested_scenarios = self.test_scenarios[:num_scenarios] all_scores = {metric: [] for metric in self.evaluation_metrics.keys()} for scenario in tested_scenarios: print(f"\n测试场景: {scenario['name']}") # 使用安全对话流程 conversation_result = self.safe_conversation_flow(scenario, max_retries_per_step=3) if not conversation_result["success"]: print(f" 场景测试失败: {conversation_result['error_message']}") continue scenario_result = { "scenario_id": scenario["scenario_id"], "scenario_name": scenario["name"], "conversation_stages": conversation_result["stages"], "candidate_models": conversation_result["candidate_models"], "final_recommended_model": conversation_result["final_recommended_model"], "stage_scores": {} } # 计算场景各阶段平均分 all_stage_scores = [] for stage in scenario_result["conversation_stages"]: all_stage_scores.append(stage["scores"]) scenario_avg_scores = {} for metric in self.evaluation_metrics.keys(): metric_scores = [s.get(metric, 0) for s in all_stage_scores if metric in s and s[metric] > 0] if metric_scores: scenario_avg_scores[metric] = np.mean(metric_scores) all_scores[metric].append(scenario_avg_scores[metric]) scenario_result["stage_scores"] = scenario_avg_scores evaluation_results["scenario_results"].append(scenario_result) print(f" 最终推荐结果: {conversation_result['final_recommended_model']}") # 计算总体指标得分 for metric, scores_list in all_scores.items(): if scores_list: avg_score = np.mean(scores_list) evaluation_results["metrics_scores"][metric] = { "name": self.evaluation_metrics[metric]["name"], "average_score": round(avg_score, 2), "weight": self.evaluation_metrics[metric]["weight"], "weighted_score": round(avg_score * self.evaluation_metrics[metric]["weight"], 2) } # 计算总体得分 weighted_scores = [score_info["weighted_score"] for score_info in evaluation_results["metrics_scores"].values()] total_weights = sum(self.evaluation_metrics[metric]["weight"] for metric in self.evaluation_metrics) if weighted_scores: overall_score = sum(weighted_scores) / total_weights evaluation_results["overall_score"] = round(overall_score, 2) else: evaluation_results["overall_score"] = 0 self.results.append(evaluation_results) return evaluation_results def generate_report(self, output_format: str = "csv") -> str: """ 生成评测报告 Args: output_format: 输出格式 ('csv', 'excel', 'json') Returns: file_path: 生成的文件路径 """ if not self.results: print("没有评测结果可生成报告") return "" timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") if output_format == "csv": return self._generate_csv_report(timestamp) elif output_format == "excel": return self._generate_excel_report(timestamp) elif output_format == "json": return self._generate_json_report(timestamp) else: print(f"不支持的格式: {output_format}") return "" def _generate_csv_report(self, timestamp: str) -> str: """生成CSV报告""" filename = f"car_purchase_evaluation_report_{timestamp}.csv" with open(filename, 'w', newline='', encoding='utf-8') as csvfile: writer = csv.writer(csvfile) # 写入总体结果 writer.writerow(["购车决策大模型评测报告", f"生成时间: {timestamp}"]) writer.writerow([]) for result in self.results: writer.writerow([f"模型名称: {result['model_name']}"]) writer.writerow([f"评测时间: {result['evaluation_date']}"]) writer.writerow([f"总体得分: {result['overall_score']}/10"]) writer.writerow([]) # 写入指标得分 writer.writerow(["评测指标", "平均得分", "权重", "加权得分"]) for metric, score_info in result["metrics_scores"].items(): writer.writerow([ score_info["name"], score_info["average_score"], score_info["weight"], score_info["weighted_score"] ]) writer.writerow([]) # 写入场景详细结果 for scenario_result in result["scenario_results"]: writer.writerow([f"测试场景: {scenario_result['scenario_name']}"]) writer.writerow([f"候选车型: {', '.join(scenario_result['candidate_models'])}"]) writer.writerow([f"最终推荐: {scenario_result['final_recommended_model']}"]) writer.writerow([]) writer.writerow(["对话阶段", "用户输入", "模型响应摘要", "各指标得分"]) for stage in scenario_result["conversation_stages"]: user_preview = stage["user_input"][:50] + "..." if len(stage["user_input"]) > 50 else stage["user_input"] response_preview = stage["model_response"][:80] + "..." if len(stage["model_response"]) > 80 else stage["model_response"] scores_str = "; ".join([f"{k}: {v:.1f}" for k, v in stage["scores"].items()]) writer.writerow([ stage["stage"], user_preview, response_preview, scores_str ]) writer.writerow([]) writer.writerow([]) print(f"CSV报告已生成: {filename}") return filename def _generate_excel_report(self, timestamp: str) -> str: """生成Excel报告""" try: filename = f"car_purchase_evaluation_report_{timestamp}.xlsx" with pd.ExcelWriter(filename, engine='openpyxl') as writer: for i, result in enumerate(self.results): # 创建总体得分表 summary_data = [] for metric, score_info in result["metrics_scores"].items(): summary_data.append({ '评测指标': score_info["name"], '平均得分': score_info["average_score"], '权重': score_info["weight"], '加权得分': score_info["weighted_score"] }) summary_df = pd.DataFrame(summary_data) summary_df.to_excel(writer, sheet_name=f"{result['model_name']}_总体", index=False) # 创建详细对话记录表 detail_data = [] for scenario_result in result["scenario_results"]: for stage in scenario_result["conversation_stages"]: stage_data = { '测试场景': scenario_result['scenario_name'], '候选车型': ', '.join(scenario_result['candidate_models']), '最终推荐': scenario_result['final_recommended_model'], '对话阶段': stage['stage'], '用户输入': stage['user_input'], '模型响应': stage['model_response'] } # 添加各指标得分 for metric, score in stage['scores'].items(): metric_name = self.evaluation_metrics[metric]["name"] stage_data[metric_name] = score detail_data.append(stage_data) detail_df = pd.DataFrame(detail_data) detail_df.to_excel(writer, sheet_name=f"{result['model_name']}_详细", index=False) print(f"Excel报告已生成: {filename}") return filename except ImportError: print("请安装 openpyxl: pip install openpyxl") return self._generate_csv_report(timestamp) def _generate_json_report(self, timestamp: str) -> str: """生成JSON报告""" filename = f"car_purchase_evaluation_report_{timestamp}.json" with open(filename, 'w', encoding='utf-8') as f: json.dump(self.results, f, ensure_ascii=False, indent=2) print(f"JSON报告已生成: {filename}") return filename def visualize_results(self): """可视化评测结果""" if not self.results: print("没有评测结果可可视化") return # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei'] plt.rcParams['axes.unicode_minus'] = False for result in self.results: metrics = [] scores = [] weighted_scores = [] for metric, score_info in result["metrics_scores"].items(): metrics.append(score_info["name"]) scores.append(score_info["average_score"]) weighted_scores.append(score_info["weighted_score"]) # 创建雷达图 self._create_radar_chart(metrics, scores, result["model_name"]) # 创建柱状图 self._create_bar_chart(metrics, scores, weighted_scores, result["model_name"]) # 创建场景对比图 self._create_scenario_comparison_chart(result) def _create_radar_chart(self, metrics: List[str], scores: List[float], model_name: str): """创建雷达图""" fig = plt.figure(figsize=(10, 8)) ax = fig.add_subplot(111, polar=True) # 完成循环 angles = np.linspace(0, 2 * np.pi, len(metrics), endpoint=False).tolist() angles += angles[:1] scores += scores[:1] ax.plot(angles, scores, 'o-', linewidth=2, label=model_name, color='#1f77b4') ax.fill(angles, scores, alpha=0.25, color='#1f77b4') ax.set_thetagrids(np.degrees(angles[:-1]), metrics) ax.set_ylim(0, 10) ax.set_title(f'{model_name} - 购车决策能力雷达图', size=16, y=1.1) ax.grid(True) plt.savefig(f'{model_name}_car_purchase_radar.png', bbox_inches='tight', dpi=300) plt.close() print(f"雷达图已保存: {model_name}_car_purchase_radar.png") def _create_bar_chart(self, metrics: List[str], scores: List[float], weighted_scores: List[float], model_name: str): """创建柱状图""" try: # 确保所有列表长度一致 min_len = min(len(metrics), len(scores), len(weighted_scores)) if min_len == 0: print("错误: 没有有效的数据用于创建柱状图") return metrics = metrics[:min_len] scores = scores[:min_len] weighted_scores = weighted_scores[:min_len] x = np.arange(len(metrics)) width = 0.35 fig, ax = plt.subplots(figsize=(12, 8)) bars1 = ax.bar(x - width/2, scores, width, label='平均得分', alpha=0.8, color='#2ecc71') bars2 = ax.bar(x + width/2, weighted_scores, width, label='加权得分', alpha=0.8, color='#3498db') ax.set_xlabel('评测指标') ax.set_ylabel('得分') ax.set_title(f'{model_name} - 评测结果对比') ax.set_xticks(x) ax.set_xticklabels(metrics, rotation=45, ha='right') ax.legend() ax.set_ylim(0, 12) ax.grid(axis='y', alpha=0.3) # 在柱子上标注数值 for bar in bars1: height = bar.get_height() ax.annotate(f'{height:.1f}', xy=(bar.get_x() + bar.get_width() / 2, height), xytext=(0, 3), textcoords="offset points", ha='center', va='bottom', fontsize=9) for bar in bars2: height = bar.get_height() ax.annotate(f'{height:.1f}', xy=(bar.get_x() + bar.get_width() / 2, height), xytext=(0, 3), textcoords="offset points", ha='center', va='bottom', fontsize=9) plt.tight_layout() plt.savefig(f'{model_name}_bar_chart.png', bbox_inches='tight', dpi=300) plt.close() print(f"柱状图已保存: {model_name}_bar_chart.png") except Exception as e: print(f"创建柱状图时出错: {e}") import traceback traceback.print_exc() # 打印完整的错误堆栈 def _create_scenario_comparison_chart(self, result: Dict): """创建场景对比图""" if not result.get("scenario_results"): return scenarios = [sr["scenario_name"] for sr in result["scenario_results"]] metrics = list(self.evaluation_metrics.keys()) metric_names = [self.evaluation_metrics[metric]["name"] for metric in metrics] # 创建热力图数据 heatmap_data = [] for scenario_result in result["scenario_results"]: scenario_scores = [] for metric in metrics: score = scenario_result["stage_scores"].get(metric, 0) scenario_scores.append(score) heatmap_data.append(scenario_scores) heatmap_data = np.array(heatmap_data) fig, ax = plt.subplots(figsize=(12, 8)) im = ax.imshow(heatmap_data, cmap='YlOrRd', aspect='auto') # 设置坐标轴 ax.set_xticks(np.arange(len(metric_names))) ax.set_yticks(np.arange(len(scenarios))) ax.set_xticklabels(metric_names, rotation=45, ha='right') ax.set_yticklabels(scenarios) # 在每个格子中显示数值 for i in range(len(scenarios)): for j in range(len(metric_names)): text = ax.text(j, i, f'{heatmap_data[i, j]:.1f}', ha="center", va="center", color="black", fontsize=10) ax.set_title(f"{result['model_name']} - 各场景能力表现热力图") fig.colorbar(im, ax=ax) plt.tight_layout() plt.savefig(f"{result['model_name']}_scenario_heatmap.png", bbox_inches='tight', dpi=300) plt.close() print(f"场景热力图已保存: {result['model_name']}_scenario_heatmap.png") def main(): # API配置 api_config = { "api_url": "https://dashscope.aliyuncs.com/compatible-mode/v1", "api_key": "", "model_name": "qwen3-max", "max_tokens": 1000, "temperature": 0.7, "top_p": 0.8, "enable_thinking": False, } try: # 创建购车决策评测器 evaluator = CarPurchaseEvaluator(api_config) # 运行购车决策评测 results = evaluator.run_car_purchase_evaluation("Qwen3-Max", num_scenarios=1) # 生成报告 csv_report = evaluator.generate_report("csv") json_report = evaluator.generate_report("json") # 生成可视化图表 evaluator.visualize_results() print(f"\n购车决策评测完成！") print(f"总体得分: {results['overall_score']}/10") print(f"报告文件: {csv_report}, {json_report}") # 打印各指标得分 print("\n各指标得分:") for metric, score_info in results["metrics_scores"].items(): print(f" {score_info['name']}: {score_info['average_score']}/10") # 打印最终推荐结果 for scenario_result in results["scenario_results"]: print(f"\n测试场景: {scenario_result['scenario_name']}") print(f"候选车型: {', '.join(scenario_result['candidate_models'])}") print(f"最终推荐: {scenario_result['final_recommended_model']}") except Exception as e: print(f"评测except: {e}") print("API连接已断开") if __name__ == "__main__": main()