948 lines
34 KiB
Python
948 lines
34 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
大模型微调验证系统
|
||
用于评估模型输出与参考答案的相似度和质量
|
||
|
||
LLM评估配置：
|
||
- 默认使用模拟评估（基于传统指标）
|
||
- 如需使用真实LLM，请配置 llm_config.py 文件
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
import multiprocessing as mp
|
||
from collections import Counter
|
||
from typing import Dict, List, Tuple, Any
|
||
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed
|
||
|
||
import numpy as np
|
||
from nltk.translate.bleu_score import sentence_bleu, SmoothingFunction
|
||
from nltk.tokenize import word_tokenize
|
||
import jieba
|
||
import jieba.posseg as pseg
|
||
from difflib import SequenceMatcher
|
||
import pandas as pd
|
||
from tqdm import tqdm
|
||
import requests
|
||
|
||
# 导入LLM配置
|
||
from llm_config import (
|
||
USE_REAL_LLM,
|
||
MAX_CONCURRENT_WORKERS,
|
||
SHOW_DETAILED_PROGRESS,
|
||
MAX_API_RETRIES,
|
||
RETRY_DELAY,
|
||
OPENAI_CONFIG,
|
||
EVALUATION_WEIGHTS,
|
||
PROMPT_TEMPLATE,
|
||
validate_openai_config,
|
||
print_config_info
|
||
)
|
||
|
||
|
||
class ModelEvaluator:
|
||
"""模型评估器"""
|
||
|
||
def __init__(self):
|
||
# 初始化jieba分词
|
||
jieba.initialize()
|
||
|
||
# 停用词列表（中文常见停用词）
|
||
self.stopwords = {
|
||
'的', '了', '是', '在', '我', '有', '和', '就', '不', '人', '都', '一', '一个',
|
||
'上', '也', '很', '到', '说', '要', '去', '你', '会', '着', '没有', '看', '好',
|
||
'自己', '这', '那', '能', '下', '过', '他', '来', '对', '时候', '后', '如果'
|
||
}
|
||
|
||
# 初始化平滑函数
|
||
self.smoothing_function = SmoothingFunction()
|
||
|
||
def get_llm_evaluation_prompt(self, reference: str, candidate: str, question: str = "") -> str:
|
||
"""生成LLM评估提示词"""
|
||
# 使用配置中的提示词模板
|
||
return PROMPT_TEMPLATE.format(
|
||
question=question,
|
||
reference=reference,
|
||
candidate=candidate
|
||
)
|
||
|
||
def call_llm_for_evaluation(self, prompt: str, max_retries: int = 3, retry_delay: float = 1.0) -> Tuple[int, str]:
|
||
"""调用大语言模型进行评估（带自动重试）
|
||
|
||
使用配置：
|
||
- API配置来自 llm_config.py 中的 OPENAI_CONFIG
|
||
- 支持环境变量和直接配置
|
||
|
||
参数：
|
||
- prompt: 评估提示词
|
||
- max_retries: 最大重试次数（默认3次）
|
||
- retry_delay: 重试延迟时间（秒，默认1秒）
|
||
|
||
配置方法：
|
||
1. 设置环境变量：export OPENAI_API_KEY='your-api-key'
|
||
2. 在 llm_config.py 中直接修改 OPENAI_CONFIG
|
||
"""
|
||
# 验证配置
|
||
config = validate_openai_config()
|
||
|
||
# 调用API（使用传统HTTP请求，带重试机制）
|
||
for attempt in range(max_retries + 1):
|
||
try:
|
||
# 构造请求体
|
||
payload = {
|
||
"model": config["model"],
|
||
"messages": [
|
||
{"role": "system", "content": "你是一个专业的文本质量评估专家。"},
|
||
{"role": "user", "content": prompt}
|
||
],
|
||
"temperature": config["temperature"],
|
||
"max_tokens": config["max_tokens"]
|
||
}
|
||
|
||
# 发送HTTP请求
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {config['api_key']}"
|
||
}
|
||
|
||
response = requests.post(
|
||
f"{config['api_base']}/chat/completions",
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=config["timeout"]
|
||
)
|
||
|
||
# 检查响应状态
|
||
response.raise_for_status()
|
||
|
||
# 解析JSON响应
|
||
import json
|
||
response_data = response.json()
|
||
content = response_data["choices"][0]["message"]["content"]
|
||
try:
|
||
result = json.loads(content)
|
||
score = int(result.get("score", 0))
|
||
reason = str(result.get("reason", ""))
|
||
return score, reason
|
||
except (json.JSONDecodeError, KeyError, ValueError):
|
||
# 如果无法解析JSON，尝试从文本中提取
|
||
# 这里可以添加正则表达式来提取评分
|
||
raise ValueError(f"无法解析LLM响应：{content}")
|
||
|
||
except (requests.exceptions.RequestException, requests.exceptions.Timeout,
|
||
requests.exceptions.ConnectionError, RuntimeError) as e:
|
||
# 网络相关错误，可以重试
|
||
if attempt < max_retries:
|
||
wait_time = retry_delay * (2 ** attempt) # 指数退避
|
||
print(f"API调用失败（尝试 {attempt + 1}/{max_retries + 1}），{wait_time:.1f}秒后重试: {str(e)[:100]}")
|
||
time.sleep(wait_time)
|
||
continue
|
||
else:
|
||
print(f"API调用失败，已重试 {max_retries} 次")
|
||
raise RuntimeError(f"API调用失败（已重试{max_retries}次）：{str(e)}")
|
||
|
||
except Exception as e:
|
||
# 其他错误（如JSON解析错误），不重试
|
||
print(f"API调用失败（不可重试的错误）: {str(e)}")
|
||
raise RuntimeError(f"API调用失败：{str(e)}")
|
||
|
||
# 这里不应该到达，但如果到达了，抛出异常
|
||
raise RuntimeError("API调用失败：达到最大重试次数")
|
||
|
||
def get_mock_llm_evaluation(self, reference: str, candidate: str, question: str = "") -> Tuple[int, str]:
|
||
"""获取模拟的LLM评估结果（用于演示）
|
||
|
||
使用配置：
|
||
- 权重来自 llm_config.py 中的 EVALUATION_WEIGHTS
|
||
"""
|
||
# 计算一些基本指标作为参考
|
||
bleu = self.calculate_bleu_score(reference, candidate)
|
||
rouge_l = self.calculate_rouge_l(reference, candidate)
|
||
exact_match = self.calculate_exact_match_rate(reference, candidate)
|
||
keyword_overlap = self.calculate_keyword_overlap(reference, candidate)
|
||
|
||
# 使用配置中的权重进行加权计算
|
||
composite_score = (
|
||
bleu * EVALUATION_WEIGHTS['bleu_score'] +
|
||
rouge_l * EVALUATION_WEIGHTS['rouge_l_score'] +
|
||
exact_match * EVALUATION_WEIGHTS['exact_match_rate'] +
|
||
keyword_overlap * EVALUATION_WEIGHTS['keyword_overlap_rate']
|
||
)
|
||
score = int(composite_score * 10)
|
||
|
||
# 生成评价理由
|
||
reason_parts = []
|
||
|
||
if exact_match > 0.9:
|
||
reason_parts.append("生成答案与参考答案内容完全一致")
|
||
elif bleu > 0.1 or rouge_l > 0.3:
|
||
reason_parts.append("生成答案在内容上与参考答案较为一致")
|
||
else:
|
||
reason_parts.append("生成答案与参考答案存在较大差异")
|
||
|
||
if keyword_overlap > 0.5:
|
||
reason_parts.append("关键词重叠度较高")
|
||
elif keyword_overlap > 0.3:
|
||
reason_parts.append("关键词重叠度中等")
|
||
else:
|
||
reason_parts.append("关键词重叠度较低")
|
||
|
||
reason = "；".join(reason_parts) + f"。综合指标：BLEU={bleu:.3f}, ROUGE-L={rouge_l:.3f}, 完全匹配率={exact_match:.3f}, 关键词重叠率={keyword_overlap:.3f}"
|
||
|
||
return min(10, max(1, score)), reason
|
||
|
||
def tokenize_chinese(self, text: str) -> List[str]:
|
||
"""中文分词"""
|
||
if not text:
|
||
return []
|
||
# 使用jieba进行分词
|
||
words = jieba.cut(text)
|
||
return [word.strip() for word in words if len(word.strip()) > 0]
|
||
|
||
def get_keywords(self, text: str) -> List[str]:
|
||
"""提取关键词（名词、动词、形容词）"""
|
||
if not text:
|
||
return []
|
||
|
||
words = pseg.cut(text)
|
||
keywords = []
|
||
for word, flag in words:
|
||
# 选择名词(n)、动词(v)、形容词(a)、区别词(b)等作为关键词
|
||
if flag.startswith(('n', 'v', 'a', 'b', 'i')) and word not in self.stopwords and len(word) > 1:
|
||
keywords.append(word)
|
||
return keywords
|
||
|
||
def calculate_bleu_score(self, reference: str, candidate: str) -> float:
|
||
"""计算BLEU分数"""
|
||
if not reference or not candidate:
|
||
return 0.0
|
||
|
||
# 分词
|
||
ref_tokens = self.tokenize_chinese(reference)
|
||
cand_tokens = self.tokenize_chinese(candidate)
|
||
|
||
if not cand_tokens:
|
||
return 0.0
|
||
|
||
# 计算BLEU分数（使用1-gram到4-gram）
|
||
try:
|
||
score = sentence_bleu(
|
||
[ref_tokens],
|
||
cand_tokens,
|
||
smoothing_function=self.smoothing_function.method1
|
||
)
|
||
return score
|
||
except:
|
||
return 0.0
|
||
|
||
def calculate_rouge_l(self, reference: str, candidate: str) -> float:
|
||
"""计算ROUGE-L分数"""
|
||
if not reference or not candidate:
|
||
return 0.0
|
||
|
||
ref_tokens = self.tokenize_chinese(reference)
|
||
cand_tokens = self.tokenize_chinese(candidate)
|
||
|
||
if not ref_tokens or not cand_tokens:
|
||
return 0.0
|
||
|
||
# 计算最长公共子序列
|
||
lcs = self._lcs_length(ref_tokens, cand_tokens)
|
||
|
||
# 计算ROUGE-L F1分数
|
||
ref_len = len(ref_tokens)
|
||
cand_len = len(cand_tokens)
|
||
|
||
if ref_len == 0 or cand_len == 0:
|
||
return 0.0
|
||
|
||
precision = lcs / cand_len
|
||
recall = lcs / ref_len
|
||
|
||
if precision + recall == 0:
|
||
return 0.0
|
||
|
||
rouge_l = (2 * precision * recall) / (precision + recall)
|
||
return rouge_l
|
||
|
||
def _lcs_length(self, seq1: List[str], seq2: List[str]) -> int:
|
||
"""计算最长公共子序列长度"""
|
||
m, n = len(seq1), len(seq2)
|
||
dp = [[0] * (n + 1) for _ in range(m + 1)]
|
||
|
||
for i in range(1, m + 1):
|
||
for j in range(1, n + 1):
|
||
if seq1[i-1] == seq2[j-1]:
|
||
dp[i][j] = dp[i-1][j-1] + 1
|
||
else:
|
||
dp[i][j] = max(dp[i-1][j], dp[i][j-1])
|
||
|
||
return dp[m][n]
|
||
|
||
def calculate_character_overlap(self, reference: str, candidate: str) -> float:
|
||
"""计算字符重叠率"""
|
||
if not reference or not candidate:
|
||
return 0.0
|
||
|
||
ref_chars = set(reference)
|
||
cand_chars = set(candidate)
|
||
|
||
if not cand_chars:
|
||
return 0.0
|
||
|
||
overlap = len(ref_chars & cand_chars)
|
||
total_cand = len(cand_chars)
|
||
|
||
return overlap / total_cand if total_cand > 0 else 0.0
|
||
|
||
def calculate_length_similarity(self, reference: str, candidate: str) -> float:
|
||
"""计算长度相似度"""
|
||
if not reference or not candidate:
|
||
return 0.0
|
||
|
||
ref_len = len(reference)
|
||
cand_len = len(candidate)
|
||
|
||
if ref_len == 0 and cand_len == 0:
|
||
return 1.0
|
||
|
||
if ref_len == 0 or cand_len == 0:
|
||
return 0.0
|
||
|
||
# 使用相对差异计算相似度
|
||
similarity = 1 - abs(ref_len - cand_len) / max(ref_len, cand_len)
|
||
return max(0, similarity)
|
||
|
||
def calculate_exact_match_rate(self, reference: str, candidate: str) -> float:
|
||
"""计算完全匹配率"""
|
||
if not reference and not candidate:
|
||
return 1.0
|
||
|
||
if not reference or not candidate:
|
||
return 0.0
|
||
|
||
# 去除空白字符后比较
|
||
ref_clean = re.sub(r'\s+', '', reference.strip())
|
||
cand_clean = re.sub(r'\s+', '', candidate.strip())
|
||
|
||
return 1.0 if ref_clean == cand_clean else 0.0
|
||
|
||
def calculate_keyword_overlap(self, reference: str, candidate: str) -> float:
|
||
"""计算关键词重叠率"""
|
||
ref_keywords = set(self.get_keywords(reference))
|
||
cand_keywords = set(self.get_keywords(candidate))
|
||
|
||
if not cand_keywords:
|
||
return 0.0
|
||
|
||
overlap = len(ref_keywords & cand_keywords)
|
||
total_cand = len(cand_keywords)
|
||
|
||
return overlap / total_cand if total_cand > 0 else 0.0
|
||
|
||
def evaluate_all_metrics(self, reference: str, candidate: str) -> Dict[str, float]:
|
||
"""计算所有评估指标"""
|
||
return {
|
||
'bleu_score': self.calculate_bleu_score(reference, candidate),
|
||
'rouge_l_score': self.calculate_rouge_l(reference, candidate),
|
||
'character_overlap_rate': self.calculate_character_overlap(reference, candidate),
|
||
'length_similarity': self.calculate_length_similarity(reference, candidate),
|
||
'exact_match_rate': self.calculate_exact_match_rate(reference, candidate),
|
||
'keyword_overlap_rate': self.calculate_keyword_overlap(reference, candidate)
|
||
}
|
||
|
||
|
||
def get_cpu_count():
|
||
"""获取CPU核心数"""
|
||
try:
|
||
return mp.cpu_count()
|
||
except:
|
||
return 4 # 默认值
|
||
|
||
|
||
def evaluate_single_item(args):
|
||
"""单条数据评估函数（用于并发处理）"""
|
||
idx, item, evaluator, use_real_llm = args
|
||
|
||
# 支持多种字段名格式
|
||
input_text = item.get('question', item.get('Input', item.get('问题', '')))
|
||
output_text = item.get('output', item.get('Output', item.get('生成答案', '')))
|
||
answer_text = item.get('answer', item.get('Answer', item.get('参考答案', '')))
|
||
|
||
# 计算各项指标
|
||
metrics = evaluator.evaluate_all_metrics(answer_text, output_text)
|
||
|
||
# 获取LLM评估
|
||
if use_real_llm:
|
||
try:
|
||
prompt = evaluator.get_llm_evaluation_prompt(answer_text, output_text, input_text)
|
||
# 使用配置文件中的重试参数
|
||
llm_score, llm_reason = evaluator.call_llm_for_evaluation(
|
||
prompt,
|
||
max_retries=MAX_API_RETRIES,
|
||
retry_delay=RETRY_DELAY
|
||
)
|
||
except Exception as e:
|
||
# 静默处理错误，返回模拟评估结果
|
||
llm_score, llm_reason = evaluator.get_mock_llm_evaluation(answer_text, output_text, input_text)
|
||
else:
|
||
llm_score, llm_reason = evaluator.get_mock_llm_evaluation(answer_text, output_text, input_text)
|
||
|
||
# 添加原始数据
|
||
result = {
|
||
'index': idx + 1,
|
||
'Input': input_text,
|
||
'Output': output_text,
|
||
'Answer': answer_text,
|
||
**metrics,
|
||
'llm_score': llm_score,
|
||
'llm_reason': llm_reason
|
||
}
|
||
|
||
return result
|
||
|
||
|
||
def evaluate_dataset_parallel(data: List[Dict[str, Any]], evaluator: ModelEvaluator, use_real_llm: bool = False, max_workers: int = None) -> Tuple[List[Dict], Dict[str, float]]:
|
||
"""并发评估整个数据集
|
||
|
||
Args:
|
||
data: 数据列表
|
||
evaluator: 评估器实例
|
||
use_real_llm: 是否使用真实LLM评估
|
||
max_workers: 最大并发数，默认使用CPU核心数
|
||
"""
|
||
results = []
|
||
total_metrics = {
|
||
'bleu_score': 0.0,
|
||
'rouge_l_score': 0.0,
|
||
'character_overlap_rate': 0.0,
|
||
'length_similarity': 0.0,
|
||
'exact_match_rate': 0.0,
|
||
'keyword_overlap_rate': 0.0,
|
||
'llm_score': 0.0
|
||
}
|
||
|
||
# 获取并发数（优先级：参数 > 配置文件 > CPU核心数）
|
||
if max_workers is not None:
|
||
# 直接使用传入的参数
|
||
pass
|
||
elif MAX_CONCURRENT_WORKERS is not None:
|
||
# 使用配置文件中的设置
|
||
max_workers = MAX_CONCURRENT_WORKERS
|
||
else:
|
||
# 默认使用CPU核心数
|
||
max_workers = get_cpu_count()
|
||
|
||
print(f"\n开始并发评估 {len(data)} 条数据，使用 {max_workers} 个并发线程...")
|
||
|
||
if use_real_llm:
|
||
print("注意：LLM评分功能使用真实的大语言模型API")
|
||
print("配置来源：llm_config.py")
|
||
print(f"并发数: {max_workers}")
|
||
else:
|
||
print("注意：LLM评分功能使用模拟评估（基于传统指标的综合评分）")
|
||
print("配置来源：llm_config.py 中的 EVALUATION_WEIGHTS")
|
||
print(f"并发数: {max_workers}")
|
||
|
||
# 使用ThreadPoolExecutor进行并发评估（按批次处理）
|
||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||
# 创建总进度条
|
||
total_pbar = tqdm(total=len(data), desc="总进度", position=0, leave=True)
|
||
|
||
# 分批处理数据
|
||
batch_size = max_workers # 每批的大小等于并发数
|
||
batches = [data[i:i + batch_size] for i in range(0, len(data), batch_size)]
|
||
total_batches = len(batches)
|
||
|
||
# 处理每个批次（动态创建进度条）
|
||
for batch_idx, batch_data in enumerate(batches):
|
||
batch_num = batch_idx + 1
|
||
|
||
# 动态创建当前批次的进度条
|
||
current_batch_pbars = [] # 当前批次的进度条列表
|
||
if SHOW_DETAILED_PROGRESS:
|
||
for worker_idx in range(len(batch_data)):
|
||
pbar = tqdm(
|
||
total=1,
|
||
desc=f"批次{batch_num}-并发{worker_idx + 1}: 等待任务",
|
||
position=worker_idx + 1, # 从位置1开始
|
||
leave=False
|
||
)
|
||
current_batch_pbars.append(pbar)
|
||
|
||
# 提交当前批次的所有任务
|
||
future_to_info = {} # future -> (item_idx, worker_idx)
|
||
for item_idx, item in enumerate(batch_data):
|
||
worker_idx = item_idx # 当前批次内的worker索引
|
||
global_idx = batch_idx * batch_size + item_idx # 全局索引
|
||
future = executor.submit(evaluate_single_item, (global_idx, item, evaluator, use_real_llm))
|
||
future_to_info[future] = (global_idx, worker_idx)
|
||
|
||
# 更新批次进度条
|
||
if SHOW_DETAILED_PROGRESS:
|
||
pbar = current_batch_pbars[worker_idx]
|
||
pbar.set_description(f"批次{batch_num}-并发{worker_idx + 1}: 任务{global_idx + 1}")
|
||
pbar.refresh()
|
||
|
||
# 等待当前批次的所有任务完成
|
||
for future in as_completed(future_to_info):
|
||
global_idx, worker_idx = future_to_info[future]
|
||
|
||
try:
|
||
result = future.result()
|
||
results.append(result)
|
||
|
||
# 更新批次进度条状态
|
||
if SHOW_DETAILED_PROGRESS:
|
||
pbar = current_batch_pbars[worker_idx]
|
||
pbar.update(1)
|
||
pbar.set_description(f"批次{batch_num}-并发{worker_idx + 1}: 任务{global_idx + 1} [完成]")
|
||
pbar.refresh()
|
||
pbar.close()
|
||
|
||
# 累加指标
|
||
for key in total_metrics:
|
||
if key in result:
|
||
total_metrics[key] += result[key]
|
||
|
||
# 更新总进度条
|
||
total_pbar.update(1)
|
||
|
||
except Exception as e:
|
||
print(f" [Warning] 任务{global_idx + 1}评估失败: {e}")
|
||
if SHOW_DETAILED_PROGRESS:
|
||
pbar = current_batch_pbars[worker_idx]
|
||
pbar.set_description(f"批次{batch_num}-并发{worker_idx + 1}: 任务{global_idx + 1} [失败]")
|
||
pbar.refresh()
|
||
pbar.close()
|
||
total_pbar.update(1)
|
||
|
||
# 当前批次完成，关闭该批次的进度条
|
||
if SHOW_DETAILED_PROGRESS:
|
||
for pbar in current_batch_pbars:
|
||
pbar.close()
|
||
|
||
# 批次间隔（可选）
|
||
if batch_idx < total_batches - 1:
|
||
time.sleep(0.1) # 短暂间隔
|
||
|
||
# 关闭总进度条
|
||
total_pbar.close()
|
||
|
||
# 按原始顺序排序
|
||
results.sort(key=lambda x: x['index'])
|
||
|
||
# 计算平均值
|
||
num_samples = len(data)
|
||
for key in total_metrics:
|
||
total_metrics[key] /= num_samples if num_samples > 0 else 1
|
||
|
||
return results, total_metrics
|
||
|
||
|
||
def load_data(file_path: str) -> List[Dict[str, Any]]:
|
||
"""加载数据文件
|
||
|
||
支持 .jsonl (JSON Lines) 和 .json (JSON Array) 格式
|
||
"""
|
||
data = []
|
||
file_ext = os.path.splitext(file_path)[1].lower()
|
||
|
||
if file_ext == '.jsonl':
|
||
# JSON Lines格式：每行一个JSON对象
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
for line in f:
|
||
line = line.strip()
|
||
if line:
|
||
try:
|
||
data.append(json.loads(line))
|
||
except json.JSONDecodeError:
|
||
print(f"Warning: Failed to parse line: {line[:100]}")
|
||
elif file_ext == '.json':
|
||
# JSON Array格式：一个包含多个JSON对象的数组
|
||
with open(file_path, 'r', encoding='utf-8') as f:
|
||
try:
|
||
json_data = json.load(f)
|
||
if isinstance(json_data, list):
|
||
data = json_data
|
||
else:
|
||
print(f"Warning: JSON file does not contain an array: {file_path}")
|
||
except json.JSONDecodeError as e:
|
||
print(f"Warning: Failed to parse JSON file: {file_path}, error: {e}")
|
||
else:
|
||
print(f"Warning: Unsupported file format: {file_ext}, only .json and .jsonl are supported")
|
||
|
||
return data
|
||
|
||
|
||
def evaluate_dataset(data: List[Dict[str, Any]], evaluator: ModelEvaluator, use_real_llm: bool = False) -> Tuple[List[Dict], Dict[str, float]]:
|
||
"""评估整个数据集
|
||
|
||
Args:
|
||
data: 数据列表
|
||
evaluator: 评估器实例
|
||
use_real_llm: 是否使用真实LLM评估（默认False，使用模拟评估）
|
||
"""
|
||
results = []
|
||
total_metrics = {
|
||
'bleu_score': 0.0,
|
||
'rouge_l_score': 0.0,
|
||
'character_overlap_rate': 0.0,
|
||
'length_similarity': 0.0,
|
||
'exact_match_rate': 0.0,
|
||
'keyword_overlap_rate': 0.0,
|
||
'llm_score': 0.0
|
||
}
|
||
|
||
print(f"\n开始评估 {len(data)} 条数据...")
|
||
|
||
if use_real_llm:
|
||
print("注意：LLM评分功能使用真实的大语言模型API")
|
||
print("配置来源：llm_config.py")
|
||
else:
|
||
print("注意：LLM评分功能使用模拟评估（基于传统指标的综合评分）")
|
||
print("配置来源：llm_config.py 中的 EVALUATION_WEIGHTS")
|
||
|
||
for idx, item in enumerate(tqdm(data, desc="评估进度")):
|
||
# 支持多种字段名格式
|
||
input_text = item.get('question', item.get('Input', item.get('问题', '')))
|
||
output_text = item.get('output', item.get('Output', item.get('生成答案', '')))
|
||
answer_text = item.get('answer', item.get('Answer', item.get('参考答案', '')))
|
||
|
||
# 计算各项指标
|
||
metrics = evaluator.evaluate_all_metrics(answer_text, output_text)
|
||
|
||
# 获取LLM评估
|
||
if use_real_llm:
|
||
try:
|
||
prompt = evaluator.get_llm_evaluation_prompt(answer_text, output_text, input_text)
|
||
llm_score, llm_reason = evaluator.call_llm_for_evaluation(prompt)
|
||
except Exception as e:
|
||
print(f" [Warning] LLM API调用失败，使用模拟评估: {e}")
|
||
llm_score, llm_reason = evaluator.get_mock_llm_evaluation(answer_text, output_text, input_text)
|
||
else:
|
||
llm_score, llm_reason = evaluator.get_mock_llm_evaluation(answer_text, output_text, input_text)
|
||
|
||
# 添加原始数据
|
||
result = {
|
||
'index': idx + 1,
|
||
'Input': input_text,
|
||
'Output': output_text,
|
||
'Answer': answer_text,
|
||
**metrics,
|
||
'llm_score': llm_score,
|
||
'llm_reason': llm_reason
|
||
}
|
||
results.append(result)
|
||
|
||
# 累加指标
|
||
for key, value in metrics.items():
|
||
total_metrics[key] += value
|
||
total_metrics['llm_score'] += llm_score
|
||
|
||
# 计算平均值
|
||
num_samples = len(data)
|
||
for key in total_metrics:
|
||
total_metrics[key] /= num_samples if num_samples > 0 else 1
|
||
|
||
return results, total_metrics
|
||
|
||
|
||
def save_to_excel(results: List[Dict], total_metrics: Dict, output_dir: str, filename: str):
|
||
"""保存结果到Excel文件"""
|
||
if not os.path.exists(output_dir):
|
||
os.makedirs(output_dir)
|
||
|
||
# 创建DataFrame
|
||
df = pd.DataFrame(results)
|
||
|
||
# 重命名列名
|
||
column_mapping = {
|
||
'index': '序号',
|
||
'Input': '问题',
|
||
'Answer': '参考答案',
|
||
'Output': '生成答案',
|
||
'bleu_score': 'BLEU',
|
||
'rouge_l_score': 'ROUGE-L',
|
||
'character_overlap_rate': '字符重叠率',
|
||
'length_similarity': '长度相似度',
|
||
'exact_match_rate': '完全匹配率',
|
||
'keyword_overlap_rate': '关键词重叠率',
|
||
'llm_score': 'LLM评分',
|
||
'llm_reason': 'LLM评价理由'
|
||
}
|
||
df = df.rename(columns=column_mapping)
|
||
|
||
# 计算整体统计信息
|
||
stats_df = pd.DataFrame([total_metrics])
|
||
stats_column_mapping = {
|
||
'bleu_score': 'BLEU',
|
||
'rouge_l_score': 'ROUGE-L',
|
||
'character_overlap_rate': '字符重叠率',
|
||
'length_similarity': '长度相似度',
|
||
'exact_match_rate': '完全匹配率',
|
||
'keyword_overlap_rate': '关键词重叠率',
|
||
'llm_score': 'LLM评分'
|
||
}
|
||
stats_df = stats_df.rename(columns=stats_column_mapping)
|
||
|
||
# 保存到Excel
|
||
excel_path = os.path.join(output_dir, filename)
|
||
with pd.ExcelWriter(excel_path, engine='openpyxl') as writer:
|
||
# 写入详细结果
|
||
df.to_excel(writer, sheet_name='详细结果', index=False)
|
||
|
||
# 写入统计信息
|
||
stats_df.to_excel(writer, sheet_name='整体统计', index=False)
|
||
|
||
print(f"\n结果已保存到: {excel_path}")
|
||
return excel_path
|
||
|
||
|
||
def print_summary(total_metrics: Dict):
|
||
"""打印评估摘要"""
|
||
print("\n" + "="*60)
|
||
print("模型评估结果摘要")
|
||
print("="*60)
|
||
print(f"BLEU分数: {total_metrics['bleu_score']:.4f}")
|
||
print(f"ROUGE-L分数: {total_metrics['rouge_l_score']:.4f}")
|
||
print(f"字符重叠率: {total_metrics['character_overlap_rate']:.4f}")
|
||
print(f"长度相似度: {total_metrics['length_similarity']:.4f}")
|
||
print(f"完全匹配率: {total_metrics['exact_match_rate']:.4f}")
|
||
print(f"关键词重叠率: {total_metrics['keyword_overlap_rate']:.4f}")
|
||
print(f"LLM评分: {total_metrics['llm_score']:.4f}")
|
||
print("="*60)
|
||
|
||
# 计算综合评分（加权平均）
|
||
weights = {
|
||
'bleu_score': 0.2,
|
||
'rouge_l_score': 0.25,
|
||
'character_overlap_rate': 0.15,
|
||
'length_similarity': 0.1,
|
||
'exact_match_rate': 0.15,
|
||
'keyword_overlap_rate': 0.15
|
||
}
|
||
|
||
composite_score = sum(total_metrics[key] * weight for key, weight in weights.items())
|
||
print(f"综合评分: {composite_score:.4f}")
|
||
print("="*60)
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("大模型微调验证系统")
|
||
print("="*60)
|
||
|
||
# 获取CPU核心数
|
||
cpu_count = get_cpu_count()
|
||
print(f"检测到CPU核心数: {cpu_count}")
|
||
|
||
# 打印当前配置信息
|
||
print_config_info()
|
||
|
||
# 显示当前并发配置
|
||
if MAX_CONCURRENT_WORKERS is not None:
|
||
print(f"\n配置文件中的并发设置: {MAX_CONCURRENT_WORKERS}")
|
||
else:
|
||
print(f"\n未配置并发数，将使用CPU核心数: {cpu_count}")
|
||
|
||
# 询问是否使用并发
|
||
print("\n" + "="*60)
|
||
use_parallel = input("是否使用并发评估？(y/n，默认为y): ").strip().lower()
|
||
if not use_parallel or use_parallel == 'y':
|
||
use_parallel = True
|
||
# 使用配置文件中的设置，如果未配置则使用CPU核心数
|
||
if MAX_CONCURRENT_WORKERS is not None:
|
||
max_workers = MAX_CONCURRENT_WORKERS
|
||
print(f"将使用并发模式，并发数: {max_workers} (来自配置文件)")
|
||
else:
|
||
max_workers = cpu_count
|
||
print(f"将使用并发模式，并发数: {max_workers} (使用CPU核心数)")
|
||
else:
|
||
use_parallel = False
|
||
max_workers = None
|
||
print("将使用串行模式")
|
||
print("="*60)
|
||
|
||
# 数据目录
|
||
data_dir = "data"
|
||
output_dir = "outputs"
|
||
|
||
# 获取所有数据文件（支持 .jsonl 和 .json 格式）
|
||
data_files = [f for f in os.listdir(data_dir) if f.endswith(('.jsonl', '.json'))]
|
||
print(f"\n发现 {len(data_files)} 个数据文件:")
|
||
for file in data_files:
|
||
file_ext = os.path.splitext(file)[1]
|
||
print(f" - {file} ({file_ext})")
|
||
|
||
# 初始化评估器
|
||
evaluator = ModelEvaluator()
|
||
|
||
# 存储所有结果
|
||
all_results = []
|
||
all_stats = {}
|
||
|
||
# 逐个评估每个文件
|
||
for filename in data_files:
|
||
print(f"\n{'='*60}")
|
||
print(f"评估文件: {filename}")
|
||
print(f"{'='*60}")
|
||
|
||
file_path = os.path.join(data_dir, filename)
|
||
data = load_data(file_path)
|
||
print(f"加载数据: {len(data)} 条记录")
|
||
|
||
# 评估数据（根据用户选择使用并发或串行）
|
||
if use_parallel:
|
||
results, total_metrics = evaluate_dataset_parallel(
|
||
data,
|
||
evaluator,
|
||
use_real_llm=USE_REAL_LLM,
|
||
max_workers=max_workers
|
||
)
|
||
else:
|
||
results, total_metrics = evaluate_dataset(
|
||
data,
|
||
evaluator,
|
||
use_real_llm=USE_REAL_LLM
|
||
)
|
||
|
||
# 保存结果
|
||
base_name = os.path.splitext(filename)[0]
|
||
excel_filename = f"{base_name}_evaluation.xlsx"
|
||
excel_path = save_to_excel(results, total_metrics, output_dir, excel_filename)
|
||
|
||
# 存储结果
|
||
all_results.extend([{**r, 'file': filename} for r in results])
|
||
all_stats[filename] = total_metrics
|
||
|
||
# 打印摘要
|
||
print_summary(total_metrics)
|
||
|
||
# 生成汇总报告
|
||
print(f"\n{'='*60}")
|
||
print("生成汇总报告")
|
||
print(f"{'='*60}")
|
||
|
||
# 创建汇总DataFrame
|
||
summary_data = []
|
||
for filename, stats in all_stats.items():
|
||
summary_data.append({
|
||
'文件名': filename,
|
||
'BLEU分数': stats['bleu_score'],
|
||
'ROUGE-L分数': stats['rouge_l_score'],
|
||
'字符重叠率': stats['character_overlap_rate'],
|
||
'长度相似度': stats['length_similarity'],
|
||
'完全匹配率': stats['exact_match_rate'],
|
||
'关键词重叠率': stats['keyword_overlap_rate'],
|
||
'LLM评分': stats['llm_score']
|
||
})
|
||
|
||
summary_df = pd.DataFrame(summary_data)
|
||
|
||
# 计算所有文件的平均分
|
||
avg_scores = summary_df.select_dtypes(include=[np.number]).mean()
|
||
avg_df = pd.DataFrame([avg_scores])
|
||
avg_df.index = ['平均分']
|
||
|
||
# 保存汇总报告
|
||
summary_path = os.path.join(output_dir, "evaluation_summary.xlsx")
|
||
with pd.ExcelWriter(summary_path, engine='openpyxl') as writer:
|
||
summary_df.to_excel(writer, sheet_name='各文件评分', index=False)
|
||
avg_df.to_excel(writer, sheet_name='平均分')
|
||
|
||
print(f"汇总报告已保存到: {summary_path}")
|
||
|
||
# 打印最终汇总
|
||
print(f"\n{'='*60}")
|
||
print("所有文件评估结果汇总")
|
||
print(f"{'='*60}")
|
||
print(summary_df.to_string(index=False, float_format='%.4f'))
|
||
print(f"{'='*60}")
|
||
print("平均分:")
|
||
print(avg_df.to_string(float_format='%.4f'))
|
||
print(f"{'='*60}")
|
||
|
||
# 保存所有详细结果
|
||
all_results_path = os.path.join(output_dir, "all_detailed_results.xlsx")
|
||
all_results_df = pd.DataFrame(all_results)
|
||
# 重命名列名
|
||
all_column_mapping = {
|
||
'index': '序号',
|
||
'Input': '问题',
|
||
'Answer': '参考答案',
|
||
'Output': '生成答案',
|
||
'bleu_score': 'BLEU',
|
||
'rouge_l_score': 'ROUGE-L',
|
||
'character_overlap_rate': '字符重叠率',
|
||
'length_similarity': '长度相似度',
|
||
'exact_match_rate': '完全匹配率',
|
||
'keyword_overlap_rate': '关键词重叠率',
|
||
'llm_score': 'LLM评分',
|
||
'llm_reason': 'LLM评价理由'
|
||
}
|
||
all_results_df = all_results_df.rename(columns=all_column_mapping)
|
||
all_results_df.to_excel(all_results_path, index=False, engine='openpyxl')
|
||
print(f"\n所有详细结果已保存到: {all_results_path}")
|
||
|
||
|
||
# =============================================================================
|
||
# 测试函数
|
||
# =============================================================================
|
||
|
||
def test_single_evaluation():
|
||
"""测试单个评估"""
|
||
from llm_config import USE_REAL_LLM
|
||
|
||
evaluator = ModelEvaluator()
|
||
|
||
# 测试数据
|
||
question = "什么是合同？"
|
||
reference_answer = "合同是当事人之间设立、变更、终止民事法律关系的协议。"
|
||
candidate_answer = "合同是双方或多方之间达成的协议，用于约定权利和义务。"
|
||
|
||
print("="*60)
|
||
print("LLM评估测试")
|
||
print("="*60)
|
||
print(f"评估模式: {'真实LLM API' if USE_REAL_LLM else '模拟评估'}")
|
||
print(f"问题: {question}")
|
||
print(f"参考答案: {reference_answer}")
|
||
print(f"生成答案: {candidate_answer}")
|
||
print("-"*60)
|
||
|
||
try:
|
||
# 获取LLM评估
|
||
prompt = evaluator.get_llm_evaluation_prompt(reference_answer, candidate_answer, question)
|
||
|
||
if USE_REAL_LLM:
|
||
print("尝试调用真实LLM API...")
|
||
score, reason = evaluator.call_llm_for_evaluation(prompt)
|
||
|
||
# 同时计算传统指标
|
||
print("\n传统指标计算:")
|
||
all_metrics = evaluator.evaluate_all_metrics(reference_answer, candidate_answer)
|
||
for key, value in all_metrics.items():
|
||
print(f" {key}: {value:.4f}")
|
||
else:
|
||
print("使用模拟评估...")
|
||
score, reason = evaluator.get_mock_llm_evaluation(reference_answer, candidate_answer, question)
|
||
|
||
print(f"\n[SUCCESS] 评估成功!")
|
||
print(f"评分: {score}/10")
|
||
print(f"评价理由: {reason}")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"\n[ERROR] 评估失败: {e}")
|
||
print("\n详细错误信息:")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |