はじめに
検索システムの評価、改善のため、検索結果が質問への回答根拠を与えているかどうかをLLMで自動評価したいです。
RAGASのcontext relevanceという指標が使えそうだったので試してみました。
過去に、RAGASを使わずにLLMと手書きのプロンプトで評価したことがあったので、その結果とも比べてみます。結果としては、RAGASのcontext relevanceはあまり良くなかった、という感じでした。
context relevanceとは
ragasに実装されたnvidia-metricsの１つで、質問と検索結果を与えて、関連性を評価する指標です。
２種類のプロンプトで、0（関連なし）、1（部分的に関連）、2（完全に関連）の３段階で評価し、その平均を取ります。
ソースコードによれば２種類のプロンプトは以下です。
1つ目のプロンプト：
You are a world class expert designed to evaluate the relevance score of a Context in order to answer the Question.
Your task is to determine if the Context contains proper information to answer the Question.
Do not rely on your previous knowledge about the Question.
Use only what is written in the Context and in the Question.
Follow the instructions below:
0. If the context does not contains any relevant information to answer the question, say 0.
1. If the context partially contains relevant information to answer the question, say 1.
2. If the context contains any relevant information to answer the question, say 2.
You must provide the relevance score of 0, 1, or 2, nothing else.
Do not explain.
Return your response as JSON in this format: {"rating": X} where X is 0, 1, or 2.
訳
あなたは、質問に答えるためのコンテキストの関連性スコアを評価するために設計された世界クラスの専門家です。
あなたの仕事は、コンテキストが質問に答えるための適切な情報を含んでいるかどうかを判断することです。
質問に関するあなたの以前の知識に頼らないでください。
コンテキストと質問に書かれていることだけを使用してください。
以下の指示に従ってください：
0. コンテキストが質問に答えるための関連情報を全く含んでいない場合は、0と言ってください。
1. コンテキストが質問に答えるための関連情報を部分的に含んでいる場合は、1と言ってください。
2. コンテキストが質問に答えるための関連情報を含んでいる場合は、2と言ってください。
関連性スコア0、1、または2のみを提供してください。
説明しないでください。
JSON形式で次のように応答を返してください：{"rating": X} ここでXは0、1、または2です。
2つ目のプロンプト：
As a specially designed expert to assess the relevance score of a given Context in relation to a Question, my task is to determine the extent to which the Context provides information necessary to answer the Question. I will rely solely on the information provided in the Context and Question, and not on any prior knowledge.
Here are the instructions I will follow:
* If the Context does not contain any relevant information to answer the Question, I will respond with a relevance score of 0.
* If the Context partially contains relevant information to answer the Question, I will respond with a relevance score of 1.
* If the Context contains any relevant information to answer the Question, I will respond with a relevance score of 2.
Return your response as JSON in this format: {"rating": X} where X is 0, 1, or 2.
訳
質問に関連するコンテキストの関連性スコアを評価するために特別に設計された専門家として、私の仕事は、コンテキストが質問に答えるために必要な情報をどの程度提供しているかを判断することです。私は、コンテキストと質問で提供された情報のみに頼り、以前の知識には頼りません。
以下の指示に従います：
* コンテキストが質問に答えるための関連情報を全く含んでいない場合、関連性スコア0で応答します。
* コンテキストが質問に答えるための関連情報を部分的に含んでいる場合、関連性スコア1で応答します。
* コンテキストが質問に答えるための関連情報を含んでいる場合、関連性スコア2で応答します。
JSON形式で次のように応答を返してください：{"rating": X} ここでXは0、1、または2です。
方針
過去記事に倣って、LLMの評価部分だけRAGASに置き換えてみます。
主な手順は以下です。
- MIRACL日本語版からクエリと正解文書、検索対象コーパスを取得
- 埋め込み検索で各クエリに対し上位10件程度を取得し、正解を含む検索結果と含まない検索結果を用意
- RAGASのcontext relevanceで各検索結果を評価
- 0をNG、1をPartial、2をOKとみなして、集計結果を算出
結果
評価に用いるLLMはgpt-4o-mini、またはgpt-4oとしました。
gpt-4o-miniで評価した結果は以下です。
=== 評価結果 ===
評価サンプル数: 200
正解を含むケース: 100
- 平均Context Relevance: 0.843
- ラベル分布:
- OK: 0.710
- Partial: 0.240
- NG: 0.050
正解を含まないケース: 100
- 平均Context Relevance: 0.158
- ラベル分布:
- OK: 0.060
- Partial: 0.210
- NG: 0.730
gpt-4oで評価した結果は以下です。
=== 評価結果 ===
評価サンプル数: 200
正解を含むケース: 100
- 平均Context Relevance: 0.792
- ラベル分布:
- OK: 0.650
- Partial: 0.250
- NG: 0.100
正解を含まないケース: 100
- 平均Context Relevance: 0.030
- ラベル分布:
- OK: 0.000
- Partial: 0.060
- NG: 0.940
考察
正解を含むケースではOKが1、含まないケースではNGが1に近いほど良い評価です。
gpt-4oの含まないケースだけいい感じですが、ほかは微妙でした。
gpt-4o-miniと4oでもあまり差はなかった印象です。
過去記事でRAGASを使わずに手書きのプロンプトで評価（gpt-4o使用）した結果（以下）のほうが、だいぶ良さそうに感じました。
"summary": {
"total_samples": 200,
"positive_cases": 100,
"positive_rates": {
"OK": 0.91,
"Partial": 0.04,
"NG": 0.05
},
"negative_cases": 100,
"negative_rates": {
"OK": 0.0,
"Partial": 0.03,
"NG": 0.97
}
}
事例は見れていないので、ひょっとしたら、RAGASの間違いも間違いとは言い切れないものもあるかもしれません。時間があったらまた分析してみたいです。
実装
過去記事のeval_llm.pyのみRAGASを用いたもの（eval_ragas.py）に変更しました。ほかは同じコードを使っています。
eval_ragas.py
import json
import pickle
import random
from pathlib import Path
from typing import Any, Dict, List
import numpy as np
from openai import AsyncOpenAI
from tqdm import tqdm
from ragas.llms import llm_factory
from ragas.metrics.collections import ContextRelevance
def score_to_label(score: float, *, eps: float = 1e-9) -> str:
"""Context Relevance スコアを3値ラベルに変換する。
仕様:
- score == 1 なら OK
- score == 0 なら NG
- 0 < score < 1 なら Partial
Args:
score: 0.0〜1.0 を想定
eps: 浮動小数の誤差吸収
"""
if score >= 1.0 - eps:
return "OK"
if score <= 0.0 + eps:
return "NG"
return "Partial"
# 設定
CACHE_DIR = Path("cache")
TOP_K = 10 # 各クエリについて評価する検索結果の数
NUM_SAMPLES = -1 # 評価するクエリの数
BATCH_SIZE = 10 # バッチサイズ
# Ragas / Context Relevance の設定
OPENAI_MODEL = "gpt-4o"
def get_search_results(
query_embeddings: np.ndarray,
corpus_embeddings: np.ndarray,
corpus: list[dict],
k: int = 10
) -> list[list[dict]]:
"""各クエリの検索結果を取得
Args:
query_embeddings: クエリの埋め込みベクトル。shape=(クエリ数, 埋め込み次元)
corpus_embeddings: コーパスの埋め込みベクトル。shape=(コーパス数, 埋め込み次元)
corpus: コーパスのリスト。各要素は以下のキーを含む辞書:
- docid: 文書ID
- text: 文書のテキスト
- title: 文書のタイトル（オプション）
k: 各クエリについて取得する検索結果の数
Returns:
各クエリの検索結果のリスト。shape=(クエリ数, k)
各検索結果は元のコーパスの辞書と同じ形式
"""
# コサイン類似度を計算
similarities = query_embeddings @ corpus_embeddings.T
# 各クエリについて上位k件を取得
results = []
for i in range(len(query_embeddings)):
top_k_indices = np.argsort(-similarities[i])[:k]
results.append([corpus[idx] for idx in top_k_indices])
return results
def create_positive_sample(
query: dict,
sorted_indices: np.ndarray,
pos_indices: list[int],
corpus: list[dict],
k: int
) -> dict:
"""正解を含む検索結果を作成
Args:
query: 質問。以下のキーを含む辞書:
- query: 質問文
- positive_passages: 正解文書のリスト
sorted_indices: 類似度でソートされた文書のインデックス
pos_indices: 正解文書のインデックスのリスト
corpus: コーパスのリスト
k: 検索結果に含める文書数
Returns:
評価用のサンプル。以下のキーを含む辞書:
- query: 質問文
- search_results: 検索結果の文書リスト
- has_positive: 正解文書を含むかどうか
- positive_passages: 正解文書リスト
"""
top_k_indices = sorted_indices[:k].tolist()
# sorted_indicesの順序で正解文書を取得
missing_positives = [idx for idx in sorted_indices if idx in pos_indices and idx not in top_k_indices]
if missing_positives:
# 含まれていない正解を、正解でない文書と入れ替え
# 後ろから順に処理することで、順序を維持
for pos_idx in reversed(missing_positives):
# 後ろから順に正解でない文書を探す
for idx in reversed(top_k_indices):
if idx not in pos_indices:
# 正解でない文書を見つけたら、それを正解と入れ替え
replace_idx = top_k_indices.index(idx)
top_k_indices[replace_idx] = pos_idx
break
results = [corpus[idx] for idx in top_k_indices]
return {
"query": query["query"],
"search_results": results,
"has_positive": True,
"positive_passages": query["positive_passages"]
}
def create_negative_sample(
query: dict,
sorted_indices: np.ndarray,
pos_indices: list[int],
corpus: list[dict],
k: int
) -> dict:
"""正解を含まない検索結果を作成
Args:
query: 質問。以下のキーを含む辞書:
- query: 質問文
- positive_passages: 正解文書のリスト
sorted_indices: 類似度でソートされた文書のインデックス
pos_indices: 正解文書のインデックスのリスト
corpus: コーパスのリスト
k: 検索結果に含める文書数
Returns:
評価用のサンプル。以下のキーを含む辞書:
- query: 質問文
- search_results: 検索結果の文書リスト
- has_positive: 正解文書を含むかどうか
- positive_passages: 正解文書リスト
"""
# sorted_indicesから正解でないものをk個取得
neg_indices = []
for idx in sorted_indices:
if idx not in pos_indices:
neg_indices.append(idx)
if len(neg_indices) == k:
break
results = [corpus[idx] for idx in neg_indices]
return {
"query": query["query"],
"search_results": results,
"has_positive": False,
"positive_passages": query["positive_passages"]
}
def create_positive_negative_samples(
query_embeddings: np.ndarray,
corpus_embeddings: np.ndarray,
corpus: list[dict],
queries: list[dict],
positive_indices: list[list[int]],
k: int = 10
) -> list[dict]:
"""各クエリに対して正解を含む/含まない検索結果を作成
Args:
query_embeddings: クエリの埋め込みベクトル。shape=(クエリ数, 埋め込み次元)
corpus_embeddings: コーパスの埋め込みベクトル。shape=(コーパス数, 埋め込み次元)
corpus: コーパスのリスト。各要素は以下のキーを含む辞書:
- docid: 文書ID
- text: 文書のテキスト
- title: 文書のタイトル（オプション）
queries: 質問のリスト。各質問は以下のキーを含む辞書:
- query: 質問文
- positive_passages: 正解文書のリスト
positive_indices: 各質問の正解文書のインデックスのリスト
k: 各検索結果に含める文書数
Returns:
評価用のサンプルのリスト。各サンプルは以下のキーを含む辞書:
- query: 質問文
- search_results: 検索結果の文書リスト
- has_positive: 正解文書を含むかどうか
- positive_passages: 正解文書リスト
"""
samples = []
# コサイン類似度を計算
similarities = query_embeddings @ corpus_embeddings.T
for i, (query, pos_indices) in enumerate(zip(queries, positive_indices)):
# 類似度でソートした全インデックス
sorted_indices = np.argsort(-similarities[i])
# 正解を含む検索結果を作成
if pos_indices: # 正解文書が存在する場合
samples.append(create_positive_sample(query, sorted_indices, pos_indices, corpus, k))
# 正解を含まない検索結果を作成
samples.append(create_negative_sample(query, sorted_indices, pos_indices, corpus, k))
return samples
def evaluate_with_llm(samples: list[dict]) -> list[dict[str, Any]]:
"""Ragas (NVIDIA Metrics) の Context Relevanceで評価を実行
Args:
samples: 評価用のサンプルのリスト。各サンプルは以下のキーを含む辞書:
- query: 質問文
- search_results: 検索結果の文書リスト
- has_positive: 正解文書を含むかどうか
- positive_passages: 正解文書リスト
Notes:
* この関数名は既存コードとの互換性のため `evaluate_with_llm` のままにしています。
Returns:
評価結果のリスト。各結果は以下のキーを含む辞書:
- query: 質問文
- has_positive: 正解文書を含むかどうか
- context_relevance: Context Relevance のスコア（0.0〜1.0）
- label: OK/Partial/NG
"""
client = AsyncOpenAI()
llm = llm_factory(OPENAI_MODEL, client=client)
scorer = ContextRelevance(llm=llm)
async def _score_one(sample: dict) -> float:
retrieved_contexts = [doc["text"] for doc in sample["search_results"]]
result = await scorer.ascore(
user_input=sample["query"],
retrieved_contexts=retrieved_contexts,
)
return float(result.value)
# バッチ処理（非同期）
import asyncio
async def _score_many(batch: list[dict]) -> list[float]:
return await asyncio.gather(*[_score_one(s) for s in batch])
results: list[dict[str, Any]] = []
with tqdm(total=len(samples), desc="Ragas Context Relevanceで評価中") as pbar:
for i in range(0, len(samples), BATCH_SIZE):
batch = samples[i : i + BATCH_SIZE]
scores = asyncio.run(_score_many(batch))
for sample, score in zip(batch, scores):
label = score_to_label(score)
results.append(
{
"query": sample["query"],
"has_positive": sample["has_positive"],
"context_relevance": score,
"label": label,
}
)
pbar.update(len(batch))
return results
def calculate_metrics(results: list[dict[str, Any]]) -> dict[str, Any]:
"""評価指標を計算（Context Relevance）
Args:
results: evaluate_with_llmの戻り値。各結果は以下のキーを含む辞書:
- has_positive: 正解文書を含むかどうか
- context_relevance: 0.0〜1.0
- label: OK/Partial/NG
Returns:
評価指標を含む辞書:
- positive_cases: 正解を含むケースの数
- negative_cases: 正解を含まないケースの数
- positive_mean: 正解を含むケースの平均スコア
- negative_mean: 正解を含まないケースの平均スコア
- positive_rates: 正解を含むケースのラベル分布
- negative_rates: 正解を含まないケースのラベル分布
"""
# 正解を含むケースと含まないケースを分ける
positive_cases = [r for r in results if r["has_positive"]]
negative_cases = [r for r in results if not r["has_positive"]]
def _mean(cases: list[dict[str, Any]]) -> float:
if not cases:
return 0.0
return float(sum(r["context_relevance"] for r in cases) / len(cases))
positive_mean = _mean(positive_cases)
negative_mean = _mean(negative_cases)
def _label_rates(cases: list[dict[str, Any]]) -> dict[str, float]:
total = len(cases)
if total == 0:
return {"OK": 0.0, "Partial": 0.0, "NG": 0.0}
return {
"OK": sum(1 for r in cases if r["label"] == "OK") / total,
"Partial": sum(1 for r in cases if r["label"] == "Partial") / total,
"NG": sum(1 for r in cases if r["label"] == "NG") / total,
}
positive_rates = _label_rates(positive_cases)
negative_rates = _label_rates(negative_cases)
return {
"positive_cases": len(positive_cases),
"negative_cases": len(negative_cases),
"positive_mean": positive_mean,
"negative_mean": negative_mean,
"positive_rates": positive_rates,
"negative_rates": negative_rates,
}
def save_evaluation_results(
results: List[dict],
metrics: Dict[str, Any],
samples: List[dict],
output_dir: Path
) -> None:
"""評価結果をJSONファイルとして保存
Args:
results: evaluate_with_llmの結果
metrics: calculate_metricsの結果
samples: 評価に使用したサンプル
output_dir: 出力先ディレクトリ
"""
# ドキュメントの辞書を作成
documents = {}
for sample in samples:
for doc in sample["search_results"]:
if doc["docid"] not in documents:
documents[doc["docid"]] = doc["text"]
# 詳細な評価結果を作成
details = []
for result, sample in zip(results, samples):
details.append({
"query": result["query"],
"has_positive": result["has_positive"],
"context_relevance": result["context_relevance"],
"label": result["label"],
"search_result_ids": [doc["docid"] for doc in sample["search_results"]],
"positive_passage_ids": [doc["docid"] for doc in sample["positive_passages"]]
})
# 評価結果全体を作成
evaluation_results = {
"summary": {
"total_samples": len(results),
"positive_cases": metrics["positive_cases"],
"negative_cases": metrics["negative_cases"],
"positive_mean": metrics["positive_mean"],
"negative_mean": metrics["negative_mean"],
"positive_rates": metrics["positive_rates"],
"negative_rates": metrics["negative_rates"],
},
"details": details,
"documents": documents
}
# 出力先ディレクトリを作成
output_dir.mkdir(parents=True, exist_ok=True)
# JSONファイルとして保存
output_file = output_dir / "evaluation_results.json"
with open(output_file, "w", encoding="utf-8") as f:
json.dump(evaluation_results, f, ensure_ascii=False, indent=2)
print(f"\n評価結果を保存しました: {output_file}")
def main():
# キャッシュされた埋め込みを読み込み
print("埋め込みを読み込み中...")
with open(CACHE_DIR / "embeddings_queries.pkl", "rb") as f:
query_embeddings = pickle.load(f)
with open(CACHE_DIR / "embeddings_corpus.pkl", "rb") as f:
corpus_embeddings = pickle.load(f)
# データセットを読み込み
print("データセットを読み込み中...")
with open(CACHE_DIR / "sample_queries_100.pkl", "rb") as f:
queries = pickle.load(f)
with open(CACHE_DIR / "sample_corpus_10000.pkl", "rb") as f:
corpus = list(pickle.load(f))
# 評価するクエリをランダムに選択
if NUM_SAMPLES > 0 and NUM_SAMPLES < len(queries):
print(f"ランダムに{NUM_SAMPLES}個のクエリを選択します...")
selected_indices = random.sample(range(len(queries)), NUM_SAMPLES)
queries = [queries[i] for i in selected_indices]
query_embeddings = query_embeddings[selected_indices]
all_docids = [doc["docid"] for doc in corpus]
# 正解文書のインデックスを取得
positive_indices = []
for query in queries:
pos_docids = [p["docid"] for p in query["positive_passages"]]
pos_indices = [
i for i, docid in enumerate(all_docids)
if docid in pos_docids
]
positive_indices.append(pos_indices)
# 評価用サンプルを作成
print("評価用サンプルを作成中...")
samples = create_positive_negative_samples(
query_embeddings,
corpus_embeddings,
corpus,
queries,
positive_indices,
k=TOP_K
)
# LLMで評価
evaluation_results = evaluate_with_llm(samples)
# 評価指標を計算
metrics = calculate_metrics(evaluation_results)
# 評価結果を表示
print("\n=== 評価結果 ===")
print(f"評価サンプル数: {len(evaluation_results)}")
print(f"正解を含むケース: {metrics['positive_cases']}")
print(f"- 平均Context Relevance: {metrics['positive_mean']:.3f}")
print("- ラベル分布:")
for label, rate in metrics["positive_rates"].items():
print(f" - {label}: {rate:.3f}")
print(f"\n正解を含まないケース: {metrics['negative_cases']}")
print(f"- 平均Context Relevance: {metrics['negative_mean']:.3f}")
print("- ラベル分布:")
for label, rate in metrics["negative_rates"].items():
print(f" - {label}: {rate:.3f}")
# 評価結果をファイルに保存
save_evaluation_results(
evaluation_results,
metrics,
samples,
output_dir=CACHE_DIR / "results"
)
if __name__ == "__main__":
main()