wehub-resource-sync
142 lines
5.0 KiB
Python
142 lines
5.0 KiB
Python
"""
|
|
RAG评估指标计算工具
|
|
简化版：只保留Recall/F1（检索）和 LLM Judge（答案准确性）
|
|
"""
|
|
|
|
import textwrap
|
|
from typing import Any
|
|
|
|
import json_repair
|
|
|
|
from yuxi.utils import logger
|
|
|
|
|
|
class RetrievalMetrics:
|
|
"""检索评估指标计算"""
|
|
|
|
@staticmethod
|
|
def precision_at_k(retrieved_ids: list[str], relevant_ids: list[str], k: int) -> float:
|
|
"""计算Precision@K"""
|
|
if not retrieved_ids[:k]:
|
|
return 0.0
|
|
retrieved_set = set(retrieved_ids[:k])
|
|
relevant_set = set(relevant_ids)
|
|
return len(retrieved_set & relevant_set) / k
|
|
|
|
@staticmethod
|
|
def recall_at_k(retrieved_ids: list[str], relevant_ids: list[str], k: int) -> float:
|
|
"""计算Recall@K"""
|
|
if not relevant_ids:
|
|
return 0.0
|
|
retrieved_set = set(retrieved_ids[:k])
|
|
relevant_set = set(relevant_ids)
|
|
return len(retrieved_set & relevant_set) / len(relevant_set)
|
|
|
|
@staticmethod
|
|
def f1_score_at_k(retrieved_ids: list[str], relevant_ids: list[str], k: int) -> float:
|
|
"""计算F1@K"""
|
|
precision = RetrievalMetrics.precision_at_k(retrieved_ids, relevant_ids, k)
|
|
recall = RetrievalMetrics.recall_at_k(retrieved_ids, relevant_ids, k)
|
|
if precision + recall == 0:
|
|
return 0.0
|
|
return 2 * precision * recall / (precision + recall)
|
|
|
|
|
|
class AnswerMetrics:
|
|
"""答案评估指标计算"""
|
|
|
|
@staticmethod
|
|
async def judge_correctness(query: str, generated_answer: str, gold_answer: str, judge_llm: Any) -> dict[str, Any]:
|
|
"""
|
|
使用LLM判断生成的答案是否正确
|
|
"""
|
|
if not generated_answer:
|
|
return {"score": 0.0, "reasoning": "未生成答案"}
|
|
if not gold_answer:
|
|
return {"score": 0.0, "reasoning": "无参考答案"}
|
|
|
|
prompt = textwrap.dedent(f"""你是一个公正的评判者，请评估AI生成的答案相对于标准答案的准确性。
|
|
|
|
问题：{query}
|
|
|
|
标准答案：
|
|
{gold_answer}
|
|
|
|
AI生成的答案：
|
|
{generated_answer}
|
|
|
|
请判断AI生成的答案是否在事实层面与标准答案一致。
|
|
忽略措辞、标点符号或格式上的细微差异。
|
|
只关注核心事实是否准确包含。
|
|
|
|
请返回以下JSON格式的结果（不要包含其他文本、Markdown 或注释）：
|
|
{{
|
|
"score": 1.0,
|
|
"reasoning": "简要说明判定理由"
|
|
}}
|
|
score 只能是 1.0 或 0.0。
|
|
""")
|
|
try:
|
|
response = await judge_llm.call(prompt, stream=False)
|
|
content = response.content.strip()
|
|
|
|
# 尝试清理可能的 markdown 代码块
|
|
if content.startswith("```json"):
|
|
content = content[7:]
|
|
if content.endswith("```"):
|
|
content = content[:-3]
|
|
content = content.strip()
|
|
|
|
result = json_repair.loads(content)
|
|
return {"score": float(result.get("score", 0.0)), "reasoning": result.get("reasoning", "")}
|
|
except Exception as e:
|
|
logger.error(f"LLM 评判失败: {e}")
|
|
return {"score": 0.0, "reasoning": f"评判出错: {str(e)}"}
|
|
|
|
|
|
class EvaluationMetricsCalculator:
|
|
"""综合评估指标计算器"""
|
|
|
|
@staticmethod
|
|
def calculate_retrieval_metrics(
|
|
retrieved_chunks: list[dict[str, Any]], gold_chunk_ids: list[str], k_values: list[int] = [1, 3, 5, 10]
|
|
) -> dict[str, float]:
|
|
"""计算检索指标 (Recall, F1)"""
|
|
if not retrieved_chunks or not gold_chunk_ids:
|
|
return {}
|
|
|
|
# 提取 ID
|
|
retrieved_ids = []
|
|
for chunk in retrieved_chunks:
|
|
chunk_id = chunk.get("chunk_id") or chunk.get("metadata", {}).get("chunk_id")
|
|
retrieved_ids.append(str(chunk_id) if chunk_id else "")
|
|
|
|
metrics = {}
|
|
for k in k_values:
|
|
metrics[f"recall@{k}"] = RetrievalMetrics.recall_at_k(retrieved_ids, gold_chunk_ids, k)
|
|
metrics[f"f1@{k}"] = RetrievalMetrics.f1_score_at_k(retrieved_ids, gold_chunk_ids, k)
|
|
|
|
return metrics
|
|
|
|
@staticmethod
|
|
async def calculate_answer_metrics(
|
|
query: str, generated_answer: str, gold_answer: str, judge_llm: Any = None
|
|
) -> dict[str, Any]:
|
|
"""计算答案指标 (LLM Judge)"""
|
|
if not judge_llm:
|
|
return {}
|
|
|
|
return await AnswerMetrics.judge_correctness(query, generated_answer, gold_answer, judge_llm)
|
|
|
|
@staticmethod
|
|
def calculate_overall_score(
|
|
retrieval_metrics_list: list[dict[str, float]], answer_metrics_list: list[dict[str, Any]]
|
|
) -> float | None:
|
|
"""综合得分：有答案准确率则用准确率，否则用 recall@10。"""
|
|
if answer_metrics_list:
|
|
scores = [m.get("score", 0.0) for m in answer_metrics_list]
|
|
return sum(scores) / len(scores) if scores else None
|
|
|
|
recalls = [m["recall@10"] for m in retrieval_metrics_list if m and "recall@10" in m]
|
|
return sum(recalls) / len(recalls) if recalls else None
|