RAG評価フレームワーク「Ragas」の紹介

from ragas import Dataset
# csvファイルの読み込み
dataset_loaded = Dataset.load(
    name="{csvファイル名}",
    backend="local/csv",
    root_dir="{datasetsフォルダからのパス}",
)
    

from dotenv import load_dotenv
from metrics_with_reason import ContextPrecisionWithReason, ContextRecallWithReason
from openai import AsyncOpenAI
from ragas.llms import llm_factory
# 環境変数の読み込み
load_dotenv()
# 評価者LLMの設定
client = AsyncOpenAI()
llm = llm_factory("gpt-5.1", client=client)
# Ragasの潜在的なバグがあり、モデル名に"."が含まれている場合は
# max_tokens → max_completion_tokensへの変換を手動で行う必要がある
llm.model_args["max_completion_tokens"] = llm.model_args.pop("max_tokens", 1024)
llm.model_args["temperature"] = 0.0
# 評価指標の設定
precision_metric = ContextPrecisionWithReason(llm=llm)
recall_metric = ContextRecallWithReason(llm=llm)
    

class ContextPrecisionWithReason(ContextPrecision):
    async def ascore(self, user_input, reference, retrieved_contexts) -> MetricResult:
        #######################
        # Ragasの既存の処理
        #######################
        for idx, context in enumerate(retrieved_contexts):
            input_data = ContextPrecisionInput(
                question=user_input, context=context, answer=reference
            )
            prompt_string = self.prompt.to_string(input_data)
            result = await self.llm.agenerate(prompt_string, ContextPrecisionOutput)
            verdicts.append(result.verdict)

            # context_index, verdict, reasonを出力するよう追加実装
            items.append(
                {
                    "context_index": idx,
                    "verdict": result.verdict,
                    "reason": result.reason,
                }
            )
            
        score = self._calculate_average_precision(verdicts)
        reason_json = json.dumps(items, ensure_ascii=False)
        return MetricResult(value=float(score), reason=reason_json)


class ContextRecallWithReason(ContextRecall):
    async def ascore(self, user_input, retrieved_contexts, reference) -> MetricResult:
        #######################
        # Ragasの既存の処理
        #######################
        attributions = [c.attributed for c in result.classifications]
        score = sum(attributions) / len(attributions) if attributions else np.nan

        # statement, attributed, reasonを出力するよう追加実装
        items = [
            {
                "statement": c.statement,
                "attributed": c.attributed,
                "reason": c.reason,
            }
            for c in result.classifications
        ]
        reason_json = json.dumps(items, ensure_ascii=False)
        return MetricResult(value=float(score), reason=reason_json)

    

class RAGSystem:
    def __init__(self, index_dir: str = "faiss_index"):
        self.index_dir = index_dir
        self.embeddings = OpenAIEmbeddings()
        self.vectorstore = FAISS.load_local(
            self.index_dir,
            self.embeddings,
            allow_dangerous_deserialization=True,
        )

        self.llm = ChatOpenAI(model="gpt-4.1", temperature=0)
        # ベクトル検索結果を受け取り回答生成を行うチェーンの実装
        template = """以下のコンテキスト情報を基に質問に答えてください。
コンテキスト情報: {context}
質問: {question}
回答:"""
        prompt = ChatPromptTemplate.from_template(template)
        self._prompt_llm_chain = prompt | self.llm

    # RAGの実行
    def query(self, query):
        # ベクトル検索(検索件数は3件)
        source_documents = self.vectorstore.similarity_search_with_score(query, k=3)
        retrieved_contexts = [doc.page_content for doc, _ in source_documents]
        context = "\n\n".join(retrieved_contexts)
        # 回答生成
        result = self._prompt_llm_chain.invoke({"context": context, "question": query})
        answer = result.content

        return answer, retrieved_contexts
    

from pydantic import BaseModel

# 自前のRAGシステムをインポート
from rag_system import RAGSystem

rag_system = RAGSystem()

# 評価結果の型定義
# ここで定義した項目をカラム名としたcsvファイルが自動で出力される
class EvalResult(BaseModel):
    user_input: str
    reference: str
    context_precision_value: float
    context_precision_reason: Optional[str] = None
    context_recall_value: float
    context_recall_reason: Optional[str] = None

# 評価実行関数の定義
@experiment(EvalResult)
async def run_evaluation(row):
    response, retrieved_contexts = await asyncio.to_thread(
        rag_system.query, row["user_input"]
    )
    precision_result = await precision_metric.ascore(
        user_input=row["user_input"],
        retrieved_contexts=retrieved_contexts,
        reference=row["reference"],
    )
    recall_result = await recall_metric.ascore(
        user_input=row["user_input"],
        retrieved_contexts=retrieved_contexts,
        reference=row["reference"],
    )
    return EvalResult(
        **row,
        context_precision_value=precision_result.value,
        context_precision_reason=precision_result.reason,
        context_recall_value=recall_result.value,
        context_recall_reason=recall_result.reason,
    )

# 評価の実行
await run_evaluation.arun(
    dataset_loaded, name=f"{datetime.now().strftime('%Y%m%d%H%M%S')}"
)