RAG自動評価〜Synthetic DataとLLMで効率化〜

def generate_qa_pairs(text_chunk):
  # LLMへの指示（プロンプト）
  prompt_template = """
あなたはRAGシステムの評価データ作成担当者です。
以下のテキストに基づいて、従業員が問い合わせそうな「質問」と、
それに対するテキストに基づいた正しい「回答」のペアを作成してください。

【重要】回答は必ずテキスト内の情報のみを使用すること。
"""
  # ... (Gemini APIの呼び出し処理)
  return json.loads(response.content)

{
"question": "カフェで仕事をしてもいいですか？",
"answer": "はい、事前にセキュリティ環境が確保されていることを条件とし、上長の許可を得れば可能です。"
}

import os
import json
import time
import ast
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from dotenv import load_dotenv

load_dotenv()

# ユーザーは環境設定ファイルまたは環境変数で認証キーを設定する必要があります
if not os.getenv("GOOGLE_API_KEY"):
    print("WARNING: GOOGLE_API_KEY not found. Please set it in .env")

# Gemini（生成モデル）を初期化
llm = ChatGoogleGenerativeAI(model="gemini-3-flash-preview", temperature=0.7)

def load_data(path):
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

def generate_qa_pairs(text_chunk):
    """
    指定されたテキストチャンクからGeminiを使って質問・回答ペアを生成します。
    """
    prompt_template = """
    あなたはRAGシステムの評価データ作成担当者です。
    以下のテキストテキストは社内規定の一部です。
    このテキストに基づいて、従業員が問い合わせそうな「質問」と、それに対するテキストに基づいた正しい「回答」のペアを2つ作成してください。
    
    【制約事項】
    - 回答は必ず提供されたテキストに含まれる情報のみに基づいて作成してください。
    - 出力は純粋なJSONリスト形式（Markdownコードブロックなし）にしてください。
    - キーと値は必ずダブルクォート(")で囲んでください。
    
    【出力フォーマット】
    [
      {{"question": "ここに質問を書く", "answer": "ここに回答を書く"}},
      {{"question": "ここに質問を書く", "answer": "ここに回答を書く"}}
    ]

    【対象テキスト】
    {text}
    """
    prompt = ChatPromptTemplate.from_template(prompt_template)
    chain = prompt | llm
    
    try:
        response = chain.invoke({"text": text_chunk})
        content = response.content
        if isinstance(content, list):
            content = "".join([str(item) for item in content])
            
        # マークダウンのコードブロックが混ざっていたら除去
        content = content.replace("```json", "").replace("```", "").strip()
        try:
            return json.loads(content)
        except json.JSONDecodeError:
            # シングルクォート等の緩いフォーマットに対するフォールバック
            try:
                return ast.literal_eval(content)
            except:
                print(f"Failed to parse JSON: {content[:100]}...")
                return []
    except Exception as e:
        print(f"Error generating QA for chunk: {e}")
        return []

def main():
    doc_path = "data/remote_work_policy.md"
    if not os.path.exists(doc_path):
        # ルート/ソース直下から実行した場合のフォールバック
        doc_path = "../data/remote_work_policy.md" 
        if not os.path.exists(doc_path):
             print("Error: Document not found.")
             return

    text = load_data(doc_path)
    
    # 見出し（##）で分割し、各条文を1チャンクとして扱う
    sections = text.split("\n## ")
    qa_dataset = []
    
    print(f"Found {len(sections)} sections. Generating QA pairs...")
    
    for i, section in enumerate(sections):
        if not section.strip(): continue
        
        # （先頭を除き）取り除かれた見出しマーカーを付け直す
        # 分割で区切り文字自体は消えるので、内容だけを使う
        # ただ、チャンクが何の条文か分かるようにしたい
        
        # 無料枠のレート制限対策（目安: 毎分約15回）。
        # 1セクションの処理で外部呼び出しは1回。
        # 念のため1秒スリープ。
        time.sleep(1) 
        
        print(f"Processing section {i+1}...")
        pairs = generate_qa_pairs(section)
        
        if isinstance(pairs, dict):
            pairs = [pairs]
        elif not isinstance(pairs, list):
            pairs = []
            
        qa_dataset.extend(pairs)
        
    output_path = "data/synthetic_qa.json"
    # ソース直下から実行している場合は出力パスを調整
    if not os.path.exists("data") and os.path.exists("../data"):
        output_path = "../data/synthetic_qa.json"

    with open(output_path, "w", encoding="utf-8") as f:
        json.dump(qa_dataset, f, ensure_ascii=False, indent=2)
    
    print(f"Successfully generated {len(qa_dataset)} QA pairs and saved to {output_path}")

if __name__ == "__main__":
    main()

from langchain_community.vectorstores import Chroma
from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.documents import Document
import os

class SimpleRAG:
    def __init__(self, persist_directory="./chroma_db"):
        self.embeddings = GoogleGenerativeAIEmbeddings(model="models/embedding-001")
        self.llm = ChatGoogleGenerativeAI(model="gemini-3-flash-preview", temperature=0)
        self.vectorstore = None
        self.persist_directory = persist_directory

    def index_documents(self, file_path):
        """マークダウンファイルを読み込み、チャンク化してクローマデータベースにインデックスします。"""
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"{file_path} not found.")

        with open(file_path, "r", encoding="utf-8") as f:
            text = f.read()
            
        # コンテキストを保つためのチャンク分割（規定の条文ごと）
        # チャンク内で文脈を保つため '##' を付け直す
        sections = text.split("\n## ")
        docs = []
        for s in sections:
            content = s.strip()
            if not content: continue
            if not content.startswith("#"): # 先頭が # でなければ、前文か分割時に削られた可能性がある
                content = "## " + content
            
            docs.append(Document(page_content=content, metadata={"source": file_path}))
            
        # ベクトルストアを作成して永続化
        self.vectorstore = Chroma.from_documents(
            documents=docs, 
            embedding=self.embeddings,
            persist_directory=self.persist_directory
        )
        # 新しいクローマは自動で永続化するが、明示的にしておいてもよい（または初期化に任せる）
        print(f"Indexed {len(docs)} documents into {self.persist_directory}")

    def query(self, question):
        """コンテキストを検索し、回答を生成します。"""
        if not self.vectorstore:
            # 既存データベースを読み込み
            self.vectorstore = Chroma(
                persist_directory=self.persist_directory, 
                embedding_function=self.embeddings
            )
            
        # 検索（取得）
        retriever = self.vectorstore.as_retriever(search_kwargs={"k": 2})
        docs = retriever.invoke(question)
        context = "\n\n".join([d.page_content for d in docs])
        
        # 生成
        prompt = f"""
        あなたは社内規定に関する質問に答えるアシスタントです。
        以下の「社内規定（コンテキスト）」のみに基づいて、質問に答えてください。
        コンテキストに答えが見つからない場合は、「規定に記載がありません」と答えてください。
        
        【コンテキスト】
        {context}
        
        【質問】: {question}
        """
        
        response = self.llm.invoke(prompt)
        return {
            "question": question,
            "answer": response.content.strip(),
            "context": context,
            "source_documents": docs
        }

import json
import os
import time
import pandas as pd
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from dotenv import load_dotenv
from rag_system import SimpleRAG

load_dotenv()

def evaluate_answer(question, truth, prediction, context):
    """
    Geminiを使って忠実性と正確性を採点します。
    (忠実性スコア, 正確性スコア) を返します。
    """
    llm = ChatGoogleGenerativeAI(model="gemini-3-flash-preview", temperature=0.0)
    
    # --- 指標1: 忠実性（ハルシネーションチェック） ---
    # 回答はコンテキストのみに基づいているか？
    faith_template = """
    あなたはRAGシステムの評価者です。
    以下の手順で「生成された回答」を評価してください。
    
    タスク: 「生成された回答」が「コンテキスト」に含まれる情報のみに基づいているか判定する。
    
    入力:
    - コンテキスト: {context}
    - 生成された回答: {prediction}
    
    判定基準:
    - コンテキストにある情報だけで回答が構成されていれば 1
    - コンテキストにない情報（幻覚や外部知識）が含まれていれば 0
    
    出力:
    数字（0 または 1）のみを出力してください。
    """
    
    # --- 指標2: 正確性（正解データとの整合性） ---
    # 回答は期待される正解（正解データ）と一致しているか？
    correct_template = """
    あなたはRAGシステムの評価者です。
    「生成された回答」と「正解（Ground Truth）」を比較して、その正確性を評価してください。
    
    入力:
    - 質問: {question}
    - 正解: {truth}
    - 生成された回答: {prediction}
    
    判定基準:
    - 生成された回答が、正解の意味内容と一致していれば 1
    - 矛盾している、または重要な情報が欠けていれば 0
    
    出力:
    数字（0 または 1）のみを出力してください。
    """
    
    try:
        faith_prompt = ChatPromptTemplate.from_template(faith_template)
        correct_prompt = ChatPromptTemplate.from_template(correct_template)
        
        faith_chain = faith_prompt | llm
        correct_chain = correct_prompt | llm
        
        faith_res = faith_chain.invoke({"context": context, "prediction": prediction})
        correct_res = correct_chain.invoke({"question": question, "truth": truth, "prediction": prediction})
        
        correct_res = correct_chain.invoke({"question": question, "truth": truth, "prediction": prediction})
        
        # 出力をパース
        f_content = faith_res.content
        c_content = correct_res.content
        
        if isinstance(f_content, list): f_content = "".join([str(x) for x in f_content])
        if isinstance(c_content, list): c_content = "".join([str(x) for x in c_content])
        
        f_score = int(f_content.strip().split('\n')[0].strip())
        c_score = int(c_content.strip().split('\n')[0].strip())
        
        return f_score, c_score
    except Exception as e:
        print(f"Error in evaluation: {e}")
        return 0, 0

def main():
    # パス設定（ソース直下/ルート実行に対応）
    if os.path.exists("data"):
        data_dir = "data"
    elif os.path.exists("../data"):
        data_dir = "../data"
    else:
        # ゼロから/ソース直下で実行する場合はデータ用ディレクトリを作成 
        # ただし、ここではデータ用ディレクトリが存在すると仮定。
        data_dir = "data"

    doc_path = os.path.join(data_dir, "remote_work_policy.md")
    qa_path = os.path.join(data_dir, "synthetic_qa.json")
    results_path = os.path.join(data_dir, "evaluation_results.csv")
    db_path = "./chroma_db" # カレントディレクトリ

    print("Initializing RAG System...")
    rag = SimpleRAG(persist_directory=db_path)
    
    print("Initializing RAG System...")
    rag = SimpleRAG(persist_directory=db_path)
    
    # 再インデックスを避けるためデータベースの存在を確認（クォータ節約）
    if os.path.exists(db_path) and os.listdir(db_path):
        print(f"Vector DB found at {db_path}, skipping re-indexing.")
    else:
        print(f"Indexing {doc_path}...")
        try:
            rag.index_documents(doc_path)
        except Exception as e:
            print(f"Error indexing: {e}. Waiting 60s to retry...")
            time.sleep(60)
            rag.index_documents(doc_path)
    
    if not os.path.exists(qa_path):
        print(f"QA data not found at {qa_path}. Please run generate_qa.py first.")
        return

    with open(qa_path, "r", encoding="utf-8") as f:
        qa_pairs = json.load(f)
        
    print(f"Starting evaluation of {len(qa_pairs)} questions...")
    results = []
    
    for i, item in enumerate(qa_pairs):
        q = item["question"]
        t = item["answer"]
        
        print(f"\nExample {i+1}: Q: {q}")
        
        # 1. 検索拡張生成を実行
        rag_res = rag.query(q)
        pred = rag_res["answer"]
        ctx = rag_res["context"]
        
        print(f"  Prediction: {pred[:50]}...")
        
        # 2. 判定（採点）
        f_score, c_score = evaluate_answer(q, t, pred, ctx)
        print(f"  Scores -> Faithfulness: {f_score}, Correctness: {c_score}")
        
        results.append({
            "question": q,
            "ground_truth": t,
            "prediction": pred,
            "context": ctx,
            "faithfulness": f_score,
            "correctness": c_score
        })
        
        # 検索+採点のレート制限対策
        print("Sleeping 15s to avoid rate limits...")
        time.sleep(15)
        
    # 集計
    df = pd.DataFrame(results)
    df.to_csv(results_path, index=False)
    
    print("\n" + "="*30)
    print("EVALUATION SUMMARY")
    print("="*30)
    print(f"Total Questions: {len(df)}")
    print(f"Faithfulness Rate: {df['faithfulness'].mean():.2f}")
    print(f"Correctness Rate:  {df['correctness'].mean():.2f}")
    print(f"Detailed results saved to {results_path}")

if __name__ == "__main__":
    main()