admin/PlantAI-rag
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
PlantAI-rag/evaluation/rag_metrics_evaluator.py
T
potaser 5bb98a937a 初始提交
2026-07-10 17:59:59 +08:00

812 lines
31 KiB
Python
Executable File
Raw Permalink Blame History

#!/usr/bin/env python3
"""
Gene Family RAG 评估Pipeline
============================
功能：
1. 检索性能评估 (Recall@K, Precision@K, MRR, NDCG)
2. 生成质量评估 (Faithfulness, Relevance, Citation Accuracy)
3. 幻觉检测 (Hallucination Rate, Factual Precision)
4. 基因家族特有指标 (Gene ID Recognition, Cross-species Consistency)
5. 生成对比报告
使用方法：
python rag_metrics_evaluator.py --testset rag_benchmark_testset.json --output results/
作者: xialab
日期: 2026-04-26
"""
import os
import sys
import json
import re
import math
from typing import Dict, List, Tuple, Optional, Any
from dataclasses import dataclass, asdict
from pathlib import Path
from collections import defaultdict
# ============== 数据模型 ==============
@dataclass
class RetrievalMetrics:
"""检索性能指标"""
recall_at_k: Dict[int, float] # {1: 0.8, 5: 0.9, 10: 0.95}
precision_at_k: Dict[int, float]
mrr: float # Mean Reciprocal Rank
ndcg_at_k: Dict[int, float] # Normalized DCG
hit_rate_at_k: Dict[int, float]
def to_dict(self) -> Dict:
return {
"recall@k": self.recall_at_k,
"precision@k": self.precision_at_k,
"mrr": round(self.mrr, 4),
"ndcg@k": self.ndcg_at_k,
"hit_rate@k": self.hit_rate_at_k
}
@dataclass
class GenerationMetrics:
"""生成质量指标"""
faithfulness: float # 0-1, 与检索文档的一致性
answer_relevance: float # 0-5, LLM评分
completeness: float # 0-1, 回答覆盖度
citation_accuracy: float # 0-1, 引用准确性
fluency: float # 0-5, 流畅度
def to_dict(self) -> Dict:
return {
"faithfulness": round(self.faithfulness, 4),
"answer_relevance": round(self.answer_relevance, 2),
"completeness": round(self.completeness, 4),
"citation_accuracy": round(self.citation_accuracy, 4),
"fluency": round(self.fluency, 2)
}
@dataclass
class HallucinationMetrics:
"""幻觉检测指标"""
hallucination_rate: float # 0-1, 包含幻觉的回答比例
factual_precision: float # 0-1, 事实正确的陈述比例
source_attribution: float # 0-1, 可追溯到来源的陈述比例
contradiction_rate: float # 0-1, 与文档矛盾的陈述比例
def to_dict(self) -> Dict:
return {
"hallucination_rate": round(self.hallucination_rate, 4),
"factual_precision": round(self.factual_precision, 4),
"source_attribution": round(self.source_attribution, 4),
"contradiction_rate": round(self.contradiction_rate, 4)
}
@dataclass
class GeneFamilyMetrics:
"""基因家族特有指标"""
gene_id_recognition_rate: float # 基因ID识别准确率
functional_annotation_accuracy: float # 功能注释准确率
cross_species_consistency: float # 跨物种一致性
ortholog_mapping_accuracy: float # 同源基因映射准确率
go_term_accuracy: float # GO term匹配准确率
def to_dict(self) -> Dict:
return {
"gene_id_recognition_rate": round(self.gene_id_recognition_rate, 4),
"functional_annotation_accuracy": round(self.functional_annotation_accuracy, 4),
"cross_species_consistency": round(self.cross_species_consistency, 4),
"ortholog_mapping_accuracy": round(self.ortholog_mapping_accuracy, 4),
"go_term_accuracy": round(self.go_term_accuracy, 4)
}
@dataclass
class EvaluationResult:
"""单个问题的评估结果"""
question_id: str
question: str
category: str
difficulty: str
# 检索结果
retrieved_docs: List[Dict]
relevant_doc_indices: List[int]
# 生成结果
generated_answer: str
citations: List[str]
# 各项指标
retrieval_metrics: Optional[RetrievalMetrics] = None
generation_metrics: Optional[GenerationMetrics] = None
hallucination_metrics: Optional[HallucinationMetrics] = None
genefamily_metrics: Optional[GeneFamilyMetrics] = None
def to_dict(self) -> Dict:
return {
"question_id": self.question_id,
"question": self.question,
"category": self.category,
"difficulty": self.difficulty,
"retrieved_docs_count": len(self.retrieved_docs),
"relevant_doc_indices": self.relevant_doc_indices,
"generated_answer": self.generated_answer[:500] + "..." if len(self.generated_answer) > 500 else self.generated_answer,
"citations": self.citations,
"retrieval_metrics": self.retrieval_metrics.to_dict() if self.retrieval_metrics else None,
"generation_metrics": self.generation_metrics.to_dict() if self.generation_metrics else None,
"hallucination_metrics": self.hallucination_metrics.to_dict() if self.hallucination_metrics else None,
"genefamily_metrics": self.genefamily_metrics.to_dict() if self.genefamily_metrics else None
}
# ============== 评估器实现 ==============
class RAGMetricsEvaluator:
"""RAG系统评估器"""
def __init__(self, k_values: List[int] = [1, 3, 5, 10]):
self.k_values = k_values
def evaluate_retrieval(
self,
retrieved_docs: List[Dict],
relevant_docs: List[str], # 相关文档ID列表
relevance_scores: Optional[Dict[str, int]] = None
) -> RetrievalMetrics:
"""
评估检索性能
Args:
retrieved_docs: 检索到的文档列表，每个文档包含id字段
relevant_docs: 相关文档ID列表
relevance_scores: 文档相关性评分 {doc_id: score}
"""
if relevance_scores is None:
relevance_scores = {doc_id: 1 for doc_id in relevant_docs}
retrieved_ids = [doc.get("id", doc.get("doc_id", str(i))) for i, doc in enumerate(retrieved_docs)]
relevant_set = set(relevant_docs)
# 计算Recall@K, Precision@K, HitRate@K
recall_at_k = {}
precision_at_k = {}
hit_rate_at_k = {}
for k in self.k_values:
retrieved_k = set(retrieved_ids[:k])
relevant_retrieved = len(retrieved_k & relevant_set)
recall_at_k[k] = relevant_retrieved / len(relevant_set) if relevant_set else 0
precision_at_k[k] = relevant_retrieved / min(k, len(retrieved_ids)) if retrieved_ids else 0
hit_rate_at_k[k] = 1 if relevant_retrieved > 0 else 0
# 计算MRR
mrr = 0
for i, doc_id in enumerate(retrieved_ids):
if doc_id in relevant_set:
mrr = 1 / (i + 1)
break
# 计算NDCG@K
ndcg_at_k = {}
for k in self.k_values:
retrieved_k = retrieved_ids[:k]
# 计算DCG
dcg = 0
for i, doc_id in enumerate(retrieved_k):
rel = relevance_scores.get(doc_id, 0)
dcg += (2 ** rel - 1) / math.log2(i + 2)
# 计算IDCG (理想情况)
ideal_rels = sorted(relevance_scores.values(), reverse=True)[:k]
idcg = sum((2 ** rel - 1) / math.log2(i + 2) for i, rel in enumerate(ideal_rels))
ndcg_at_k[k] = dcg / idcg if idcg > 0 else 0
return RetrievalMetrics(
recall_at_k=recall_at_k,
precision_at_k=precision_at_k,
mrr=mrr,
ndcg_at_k=ndcg_at_k,
hit_rate_at_k=hit_rate_at_k
)
def evaluate_generation(
self,
generated_answer: str,
question: str,
expected_answer: str,
citations: List[str],
retrieved_docs: List[Dict]
) -> GenerationMetrics:
"""
评估生成质量
注意：这里使用启发式方法，实际应用中可以使用LLM-as-judge
"""
# Faithfulness: 检查回答中的关键信息是否在检索文档中
faithfulness = self._calculate_faithfulness(generated_answer, retrieved_docs)
# Answer Relevance: 检查回答与问题的相关性
answer_relevance = self._calculate_relevance(generated_answer, question)
# Completeness: 检查回答覆盖预期答案的程度
completeness = self._calculate_completeness(generated_answer, expected_answer)
# Citation Accuracy: 检查引用的准确性
citation_accuracy = self._calculate_citation_accuracy(citations, retrieved_docs)
# Fluency: 简单检查回答长度和句子结构
fluency = self._calculate_fluency(generated_answer)
return GenerationMetrics(
faithfulness=faithfulness,
answer_relevance=answer_relevance,
completeness=completeness,
citation_accuracy=citation_accuracy,
fluency=fluency
)
def evaluate_hallucination(
self,
generated_answer: str,
retrieved_docs: List[Dict],
expected_answer: str
) -> HallucinationMetrics:
"""评估幻觉程度"""
# 简单启发式：检查回答中是否包含预期答案中没有的信息
# 实际应用中应使用NLI模型或LLM判断
# Hallucination Rate: 包含无法验证信息的概率
hallucination_rate = self._detect_hallucination(generated_answer, retrieved_docs)
# Factual Precision: 事实正确率
factual_precision = self._calculate_factual_precision(generated_answer, expected_answer)
# Source Attribution: 可溯源比例
source_attribution = self._calculate_source_attribution(generated_answer, retrieved_docs)
# Contradiction Rate: 矛盾率
contradiction_rate = self._detect_contradiction(generated_answer, retrieved_docs)
return HallucinationMetrics(
hallucination_rate=hallucination_rate,
factual_precision=factual_precision,
source_attribution=source_attribution,
contradiction_rate=contradiction_rate
)
def evaluate_genefamily_specific(
self,
generated_answer: str,
question_data: Dict
) -> GeneFamilyMetrics:
"""评估基因家族特有指标"""
# Gene ID Recognition Rate
gene_id_rate = self._check_gene_id_recognition(generated_answer, question_data)
# Functional Annotation Accuracy
func_accuracy = self._check_functional_annotation(generated_answer, question_data)
# Cross-species Consistency
cross_species = self._check_cross_species_consistency(generated_answer, question_data)
# Ortholog Mapping Accuracy
ortholog_accuracy = self._check_ortholog_mapping(generated_answer, question_data)
# GO Term Accuracy
go_accuracy = self._check_go_terms(generated_answer, question_data)
return GeneFamilyMetrics(
gene_id_recognition_rate=gene_id_rate,
functional_annotation_accuracy=func_accuracy,
cross_species_consistency=cross_species,
ortholog_mapping_accuracy=ortholog_accuracy,
go_term_accuracy=go_accuracy
)
# ============== 辅助计算方法 ==============
def _calculate_faithfulness(self, answer: str, docs: List[Dict]) -> float:
"""计算忠实度：回答内容有多少能在检索文档中找到支持"""
if not docs:
return 0.0
# 合并所有文档文本
doc_text = " ".join([doc.get("text", doc.get("content", "")) for doc in docs])
doc_text = doc_text.lower()
# 提取回答中的关键短语（简单实现：按句子分割）
sentences = re.split(r'[。！？.!?]', answer)
supported_count = 0
for sent in sentences:
sent = sent.strip().lower()
if len(sent) < 5: # 跳过太短的句子
continue
# 检查句子或其主要部分是否在文档中
if sent in doc_text or any(phrase in doc_text for phrase in self._extract_key_phrases(sent)):
supported_count += 1
total_sentences = len([s for s in sentences if len(s.strip()) >= 5])
return supported_count / total_sentences if total_sentences > 0 else 0.0
def _calculate_relevance(self, answer: str, question: str) -> float:
"""计算回答与问题的相关性 (0-5分)"""
# 简单启发式：检查问题关键词是否出现在回答中
question_keywords = self._extract_keywords(question)
answer_lower = answer.lower()
matched = sum(1 for kw in question_keywords if kw in answer_lower)
score = (matched / len(question_keywords)) * 5 if question_keywords else 3.0
return min(score, 5.0)
def _calculate_completeness(self, answer: str, expected: str) -> float:
"""计算回答覆盖预期答案的程度"""
expected_entities = self._extract_entities(expected)
answer_entities = self._extract_entities(answer)
if not expected_entities:
return 1.0
matched = len(set(expected_entities) & set(answer_entities))
return matched / len(expected_entities)
def _calculate_citation_accuracy(self, citations: List[str], docs: List[Dict]) -> float:
"""计算引用准确性"""
if not citations:
return 0.0
doc_ids = {doc.get("id", doc.get("doc_id", "")) for doc in docs}
valid_citations = sum(1 for c in citations if any(doc_id in c for doc_id in doc_ids))
return valid_citations / len(citations)
def _calculate_fluency(self, answer: str) -> float:
"""计算流畅度 (0-5分)"""
# 简单检查：句子数量、平均句子长度
sentences = re.split(r'[。！？.!?]', answer)
sentences = [s for s in sentences if len(s.strip()) > 5]
if not sentences:
return 1.0
avg_len = sum(len(s) for s in sentences) / len(sentences)
# 理想句子长度20-50字
if 20 <= avg_len <= 50:
return 5.0
elif 10 <= avg_len < 20 or 50 < avg_len <= 80:
return 4.0
else:
return 3.0
def _detect_hallucination(self, answer: str, docs: List[Dict]) -> float:
"""检测幻觉：回答中无法验证的信息比例"""
# 简化实现：检查数字、基因ID等具体信息
doc_text = " ".join([doc.get("text", "") for doc in docs]).lower()
# 提取回答中的数字和具体声明
numbers = re.findall(r'\d+', answer)
unverified = 0
for num in numbers:
if num not in doc_text:
unverified += 1
return unverified / len(numbers) if numbers else 0.0
def _calculate_factual_precision(self, answer: str, expected: str) -> float:
"""计算事实精确度"""
# 检查回答中的事实是否与预期答案一致
answer_facts = self._extract_facts(answer)
expected_facts = self._extract_facts(expected)
if not answer_facts:
return 1.0
correct = sum(1 for fact in answer_facts if any(self._fact_match(fact, ef) for ef in expected_facts))
return correct / len(answer_facts)
def _calculate_source_attribution(self, answer: str, docs: List[Dict]) -> float:
"""计算可溯源比例"""
sentences = re.split(r'[。！？.!?]', answer)
sentences = [s for s in sentences if len(s.strip()) > 10]
if not sentences:
return 0.0
doc_text = " ".join([doc.get("text", "") for doc in docs])
attributable = sum(1 for s in sentences if any(phrase in doc_text for phrase in self._extract_key_phrases(s)))
return attributable / len(sentences)
def _detect_contradiction(self, answer: str, docs: List[Dict]) -> float:
"""检测与文档矛盾的陈述比例"""
# 简化实现：检查否定词
doc_text = " ".join([doc.get("text", "") for doc in docs]).lower()
sentences = re.split(r'[。！？.!?]', answer)
contradictions = 0
for sent in sentences:
if "不" in sent or "没有" in sent or "not" in sent.lower():
# 检查否定是否与文档冲突
positive_version = sent.replace("不", "").replace("没有", "")
if positive_version in doc_text:
contradictions += 1
return contradictions / len(sentences) if sentences else 0.0
def _check_gene_id_recognition(self, answer: str, question_data: Dict) -> float:
"""检查基因ID识别准确率"""
key_ids = question_data.get("key_ids", [])
if not key_ids:
return 1.0
recognized = sum(1 for gid in key_ids if gid in answer)
return recognized / len(key_ids)
def _check_functional_annotation(self, answer: str, question_data: Dict) -> float:
"""检查功能注释准确性"""
key_concepts = question_data.get("key_concepts", [])
if not key_concepts:
return 1.0
answer_lower = answer.lower()
matched = sum(1 for concept in key_concepts if concept.lower() in answer_lower)
return matched / len(key_concepts)
def _check_cross_species_consistency(self, answer: str, question_data: Dict) -> float:
"""检查跨物种一致性"""
key_species = question_data.get("key_species", [])
if not key_species:
return 1.0
answer_lower = answer.lower()
mentioned = sum(1 for sp in key_species if sp.lower() in answer_lower)
return mentioned / len(key_species)
def _check_ortholog_mapping(self, answer: str, question_data: Dict) -> float:
"""检查同源基因映射准确性"""
# 简化：检查是否提到了同源关系
ortholog_terms = ["ortholog", "同源", "orthologous", "paralog"]
answer_lower = answer.lower()
return 1.0 if any(term in answer_lower for term in ortholog_terms) else 0.5
def _check_go_terms(self, answer: str, question_data: Dict) -> float:
"""检查GO term匹配"""
go_terms = question_data.get("go_terms", [])
if not go_terms:
return 1.0
matched = sum(1 for go in go_terms if go in answer)
return matched / len(go_terms)
# ============== 工具方法 ==============
def _extract_keywords(self, text: str) -> List[str]:
"""提取关键词"""
# 简单实现：提取长度>2的中文字符串或英文单词
words = re.findall(r'[\u4e00-\u9fa5]{2,}|[a-zA-Z]{3,}', text.lower())
return list(set(words))
def _extract_entities(self, text: str) -> List[str]:
"""提取实体（基因名、物种名等）"""
# 基因名模式：如 AtSUC1, WRKY33, CYP71
genes = re.findall(r'[A-Z][a-z]{1,4}[0-9]{1,3}', text)
# 物种名
species = re.findall(r'[A-Z][a-z]+ [a-z]+', text)
return genes + species
def _extract_key_phrases(self, sentence: str) -> List[str]:
"""提取关键短语"""
# 提取3-10个字符的子串
phrases = []
for i in range(len(sentence) - 2):
for j in range(3, min(11, len(sentence) - i + 1)):
phrases.append(sentence[i:i+j])
return phrases
def _extract_facts(self, text: str) -> List[str]:
"""提取事实陈述"""
# 提取包含数字的句子片段
return re.findall(r'[^。！？.!?]*\d+[^。！？.!?]*', text)
def _fact_match(self, fact1: str, fact2: str) -> bool:
"""判断两个事实是否匹配"""
# 简单实现：检查重叠度
return len(set(fact1) & set(fact2)) / max(len(fact1), len(fact2)) > 0.5
# ============== 批量评估 ==============
class BatchEvaluator:
"""批量评估器"""
def __init__(self, evaluator: RAGMetricsEvaluator):
self.evaluator = evaluator
self.results: List[EvaluationResult] = []
def evaluate_testset(
self,
testset_path: str,
rag_system_func, # 调用RAG系统的函数
output_dir: str
) -> Dict:
"""
评估整个测试集
Args:
testset_path: 测试集JSON文件路径
rag_system_func: 调用RAG系统的函数，接收question，返回(answer, retrieved_docs, citations)
output_dir: 输出目录
"""
# 加载测试集
with open(testset_path, 'r', encoding='utf-8') as f:
testset = json.load(f)
# 按类别评估
category_results = defaultdict(list)
for cat_key, cat_data in testset.get("categories", {}).items():
print(f"\n评估类别: {cat_data['name']}")
for q in cat_data.get("questions", []):
print(f" 问题 {q['id']}: {q['question'][:50]}...")
# 调用RAG系统
answer, retrieved_docs, citations = rag_system_func(q['question'])
# 评估检索
relevant_docs = q.get("pmid_reference", [])
rel_metrics = self.evaluator.evaluate_retrieval(
retrieved_docs, [relevant_docs] if isinstance(relevant_docs, str) else relevant_docs
)
# 评估生成
gen_metrics = self.evaluator.evaluate_generation(
answer, q['question'], q['expected_answer'], citations, retrieved_docs
)
# 评估幻觉
hal_metrics = self.evaluator.evaluate_hallucination(
answer, retrieved_docs, q['expected_answer']
)
# 评估基因家族特有指标
gf_metrics = self.evaluator.evaluate_genefamily_specific(answer, q)
# 保存结果
result = EvaluationResult(
question_id=q['id'],
question=q['question'],
category=cat_key,
difficulty=q.get('difficulty', 'medium'),
retrieved_docs=retrieved_docs,
relevant_doc_indices=[],
generated_answer=answer,
citations=citations,
retrieval_metrics=rel_metrics,
generation_metrics=gen_metrics,
hallucination_metrics=hal_metrics,
genefamily_metrics=gf_metrics
)
self.results.append(result)
category_results[cat_key].append(result)
# 生成汇总报告
report = self._generate_report(category_results)
# 保存结果
os.makedirs(output_dir, exist_ok=True)
# 保存详细结果
with open(os.path.join(output_dir, "detailed_results.json"), 'w', encoding='utf-8') as f:
json.dump([r.to_dict() for r in self.results], f, ensure_ascii=False, indent=2)
# 保存汇总报告
with open(os.path.join(output_dir, "evaluation_report.json"), 'w', encoding='utf-8') as f:
json.dump(report, f, ensure_ascii=False, indent=2)
# 保存Markdown报告
self._save_markdown_report(report, os.path.join(output_dir, "evaluation_report.md"))
return report
def _generate_report(self, category_results: Dict) -> Dict:
"""生成汇总报告"""
report = {
"summary": {},
"by_category": {},
"by_difficulty": {}
}
# 汇总所有结果
all_results = []
for cat_results in category_results.values():
all_results.extend(cat_results)
# 总体指标
report["summary"] = self._aggregate_metrics(all_results)
# 按类别汇总
for cat, results in category_results.items():
report["by_category"][cat] = self._aggregate_metrics(results)
# 按难度汇总
difficulty_groups = defaultdict(list)
for r in all_results:
difficulty_groups[r.difficulty].append(r)
for diff, results in difficulty_groups.items():
report["by_difficulty"][diff] = self._aggregate_metrics(results)
return report
def _aggregate_metrics(self, results: List[EvaluationResult]) -> Dict:
"""聚合多个结果的指标"""
if not results:
return {}
# 检索指标
avg_recall = {k: sum(r.retrieval_metrics.recall_at_k[k] for r in results if r.retrieval_metrics) / len(results)
for k in self.evaluator.k_values}
avg_precision = {k: sum(r.retrieval_metrics.precision_at_k[k] for r in results if r.retrieval_metrics) / len(results)
for k in self.evaluator.k_values}
avg_mrr = sum(r.retrieval_metrics.mrr for r in results if r.retrieval_metrics) / len(results)
# 生成指标
avg_faithfulness = sum(r.generation_metrics.faithfulness for r in results if r.generation_metrics) / len(results)
avg_relevance = sum(r.generation_metrics.answer_relevance for r in results if r.generation_metrics) / len(results)
avg_completeness = sum(r.generation_metrics.completeness for r in results if r.generation_metrics) / len(results)
# 幻觉指标
avg_hallucination = sum(r.hallucination_metrics.hallucination_rate for r in results if r.hallucination_metrics) / len(results)
avg_factual = sum(r.hallucination_metrics.factual_precision for r in results if r.hallucination_metrics) / len(results)
# 基因家族指标
avg_gene_id = sum(r.genefamily_metrics.gene_id_recognition_rate for r in results if r.genefamily_metrics) / len(results)
avg_func = sum(r.genefamily_metrics.functional_annotation_accuracy for r in results if r.genefamily_metrics) / len(results)
return {
"count": len(results),
"retrieval": {
"recall@5": round(avg_recall.get(5, 0), 4),
"precision@5": round(avg_precision.get(5, 0), 4),
"mrr": round(avg_mrr, 4)
},
"generation": {
"faithfulness": round(avg_faithfulness, 4),
"answer_relevance": round(avg_relevance, 2),
"completeness": round(avg_completeness, 4)
},
"hallucination": {
"hallucination_rate": round(avg_hallucination, 4),
"factual_precision": round(avg_factual, 4)
},
"genefamily": {
"gene_id_recognition": round(avg_gene_id, 4),
"functional_annotation": round(avg_func, 4)
},
"rag_score": round(
0.3 * avg_recall.get(5, 0) +
0.3 * avg_faithfulness +
0.2 * (1 - avg_hallucination) +
0.2 * avg_factual, 4
)
}
def _save_markdown_report(self, report: Dict, output_path: str):
"""保存Markdown格式报告"""
lines = []
lines.append("# Gene Family RAG 评估报告")
lines.append("")
lines.append("## 总体性能")
lines.append("")
summary = report.get("summary", {})
rag_score = summary.get("rag_score", 0)
lines.append(f"**RAG综合评分: {rag_score:.4f}**")
lines.append("")
# 检索性能
retrieval = summary.get("retrieval", {})
lines.append("### 检索性能")
lines.append(f"- Recall@5: {retrieval.get('recall@5', 0):.4f}")
lines.append(f"- Precision@5: {retrieval.get('precision@5', 0):.4f}")
lines.append(f"- MRR: {retrieval.get('mrr', 0):.4f}")
lines.append("")
# 生成质量
generation = summary.get("generation", {})
lines.append("### 生成质量")
lines.append(f"- Faithfulness: {generation.get('faithfulness', 0):.4f}")
lines.append(f"- Answer Relevance: {generation.get('answer_relevance', 0):.2f}/5")
lines.append(f"- Completeness: {generation.get('completeness', 0):.4f}")
lines.append("")
# 幻觉检测
hallucination = summary.get("hallucination", {})
lines.append("### 幻觉检测")
lines.append(f"- Hallucination Rate: {hallucination.get('hallucination_rate', 0):.4f}")
lines.append(f"- Factual Precision: {hallucination.get('factual_precision', 0):.4f}")
lines.append("")
# 基因家族特有
genefamily = summary.get("genefamily", {})
lines.append("### 基因家族特有指标")
lines.append(f"- Gene ID Recognition: {genefamily.get('gene_id_recognition', 0):.4f}")
lines.append(f"- Functional Annotation: {genefamily.get('functional_annotation', 0):.4f}")
lines.append("")
# 按类别详细
lines.append("## 按类别详细")
lines.append("")
for cat, metrics in report.get("by_category", {}).items():
lines.append(f"### {cat}")
lines.append(f"- RAG Score: {metrics.get('rag_score', 0):.4f}")
lines.append(f"- Faithfulness: {metrics.get('generation', {}).get('faithfulness', 0):.4f}")
lines.append(f"- Hallucination: {metrics.get('hallucination', {}).get('hallucination_rate', 0):.4f}")
lines.append("")
with open(output_path, 'w', encoding='utf-8') as f:
f.write("\n".join(lines))
print(f"\n报告已保存: {output_path}")
# ============== 主函数 ==============
def main():
"""主函数 - 示例用法"""
import argparse
parser = argparse.ArgumentParser(description='Gene Family RAG 评估')
parser.add_argument('--testset', '-t', required=True, help='测试集JSON文件路径')
parser.add_argument('--output', '-o', default='./evaluation_results', help='输出目录')
parser.add_argument('--k-values', '-k', default='1,3,5,10', help='K值列表，逗号分隔')
args = parser.parse_args()
# 解析K值
k_values = [int(k) for k in args.k_values.split(',')]
# 创建评估器
evaluator = RAGMetricsEvaluator(k_values=k_values)
batch_evaluator = BatchEvaluator(evaluator)
# 示例RAG系统函数（实际使用时替换为真实系统）
def mock_rag_system(question: str) -> Tuple[str, List[Dict], List[str]]:
"""模拟RAG系统调用"""
# 这里应该调用实际的RAG系统
return (
f"这是关于'{question}'的模拟回答。",
[{"id": "doc1", "text": "模拟文档内容"}],
["doc1"]
)
# 运行评估
print("=" * 60)
print("Gene Family RAG 系统评估")
print("=" * 60)
report = batch_evaluator.evaluate_testset(
testset_path=args.testset,
rag_system_func=mock_rag_system,
output_dir=args.output
)
# 打印汇总
print("\n" + "=" * 60)
print("评估完成!")
print(f"RAG综合评分: {report['summary'].get('rag_score', 0):.4f}")
print(f"详细结果保存至: {args.output}")
print("=" * 60)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink