Zenn
セリオ株式会社 テックブログPublicationへの投稿
📐

LangfuseとRagasを活用したRAGシステムの定量評価

くじら
に公開
RAG
Langfuse
Ragas
tech

背景

社内ドキュメントや既存データを活用したRAGシステムを構築する際、その精度や品質を客観的に評価する手段が必要という話が上がりました。そこで、RAG全体を定量的かつ自動で評価する方法を調査していたところ、「Langfuse」や「Ragas」といった評価・可視化に特化したOSSがあったので、ツールの特徴、実装方法などを調べました。

対象読者

  • Python 初心者〜中級者
  • 社内 FAQ ボットを作りたい人

Ragasとは

Ragasは、Retrieval Augmented Generation（RAG）システムの評価を自動化するためのオープンソースのフレームワークです。
このツールは、RAGの4つのコンポーネント（Context、Question、Answer、GroundTruth）から、生成された回答の忠実性（Faithfulness）、質問との関連性（Answer Relevance）、および取得文脈の有用性（Context Metrics）など、複数の観点からRAGの性能を定量的に評価することができます[1]。

Langfuseとは

Langfuseは、LLMアプリケーションの開発、運用、評価のフェーズを支援するOSSです。運用方法としてCloud版とセルフホスト版があり、環境に合わせたデプロイが可能です。Langfuseの主な機能として以下のようなものが挙げられると思います。

  1. モニタリング機能
    LLMアプリケーションのすべてのLLM呼び出しや関連ロジックをトレースし、レイテンシ、コスト、エラー率などの詳細なメトリクスを提供し、最終的なレスポンスがどのように構築されたのかを把握できます。

  1. 評価機能
    Human Annotaion（人間による評価）やLLM-as-a-judge（LLMによる自動評価）などの評価手法に基づきLLMの出力を評価することができるみたいです。（無料のセルフホスト版だと評価機能が使えないので試せませんでした。）
  2. プロンプト管理
    Langfuseにプロンプトを登録、バージョン管理、およびアプリケーション側から取得することができる機能

プロンプト取得例

prompt = ChatPromptTemplate.from_template(
    langfuse.get_prompt("プロンプト名", label="ラベル名またはバージョン名").compile()
)

環境

項目 バージョン
OS Windows 11 Pro
ランタイム Python 3.10.17
主要ライブラリ langchain==0.3.25, langfuse==2.60.7, ragas==0.2.15
LLMのモデル gpt-4o-mini
埋め込みモデル text-embedding-3-large

事前準備

  1. GitHubリポジトリのクローンと起動（セルフホスト版）[2] 
git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up
  1. API Keyの取得
    ブラウザからアクセスし、Create API Keyよりシークレットキーとパブリックキーを取得します。

  1. Pythonライブラリのインストール

手順

1. プロンプトの登録

LLMアプリケーションののシステムプロンプトをLangfuseに登録します。
登録にはまずLangfuseを開き、Prompts->Create New promptからプロンプト名とプロンプトを入力し、Create promptをクリックすると登録できます。

サンプルとして以下のようなプロンプトを登録しました。

あなたは〇〇株式会社の知識検索エージェントです。  
ユーザーからの質問に対して、必要な場合は、ナレッジベクトルDBから検索した結果を基に回答し、コンテキストをsourceにすべて格納して答えてください。
もしコンテキスト内に答えが見つからない場合は、「提供された情報の中に該当する内容はありません」と答えてください。  
推測や事実に基づかない回答はしないでください。
質問でない限りは質問の回答はせず会話してください。
質問: {question}
{agent_scratchpad}
{output_parser}

2. Ragasの準備

Ragasの検証として正解データ（GrandTruth）が不要なFaithfulness、Answer Relevancy、Context Precisionの3つのメトリクスで評価しました。

from ragas.metrics import (
    Faithfulness,
    ResponseRelevancy,
    LLMContextPrecisionWithoutReference,
)
from ragas.run_config import RunConfig
from ragas.metrics.base import MetricWithLLM, MetricWithEmbeddings
from ragas.dataset_schema import SingleTurnSample


# 使うメトリクス
metrics = [
    Faithfulness(), # 忠実度
    ResponseRelevancy(), # 関連性
    LLMContextPrecisionWithoutReference(), # コンテキストの精度
]

# Ragasのメトリックスの初期化
def init_ragas_metrics(metrics, llm, embedding):
    for metric in metrics:
        if isinstance(metric, MetricWithLLM):
            metric.llm = llm
        if isinstance(metric, MetricWithEmbeddings):
            metric.embeddings = embedding
        run_config = RunConfig()
        metric.init(run_config)

# Ragasのメトリックスを使ってスコアを計算する関数
async def score_with_ragas(query, chunks, answer):
    scores = {}
    for metric in metrics:
        sample = SingleTurnSample(
            user_input = query,
            retrieved_contexts = chunks,
            response = answer
        )
        print(f"calculating {metric.name}")
        scores[metric.name] = await metric.single_turn_ascore(sample)
    
    return scores

3. 実行

以下のmain.pyを実行します。
ベクトルDBの内容は架空のソフトウェア名と仕様をAIに考えてもらいました。

import os
import json
import asyncio
import vector_search
import langchain.agents
from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
from ragas_sample import score_with_ragas, init_ragas_metrics, metrics
from langfuse import Langfuse
from dotenv import load_dotenv
from langchain_openai import AzureOpenAIEmbeddings
from langchain_openai.chat_models import AzureChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from ragas.embeddings import LangchainEmbeddingsWrapper
from ragas.llms import LangchainLLMWrapper
from langfuse.callback import CallbackHandler

# 環境変数のロード
load_dotenv('.env')

# 出力モデル
class AnswerWithSourcesModel(BaseModel):
    answer: str = Field(..., description="回答")
    source: list[str] = Field(..., description="ベクトルDBの検索結果")

# モデル・エンベディング設定
model_name = "gpt-4o-mini"
deployment = "gpt-4o-mini"
api_version = "2024-12-01-preview"
embedding_model_name = "text-embedding-3-large"

model = AzureChatOpenAI(
    deployment_name=deployment,
    api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
    azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
    api_version=api_version,
    model_name=model_name
)

embeddings = AzureOpenAIEmbeddings(
    model=embedding_model_name,
    api_version="2024-12-01-preview",
    azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
    api_key=os.environ.get("AZURE_EMBEDDING_API_KEY")
)

# Langfuse設定
langfuse = Langfuse(
    public_key=os.environ.get("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.environ.get("LANGFUSE_SECRET_KEY"),
    host="http://localhost:3000"
)
langfuse.auth_check()
Langfuse_handler = CallbackHandler(
    public_key=os.environ.get("LANGFUSE_PUBLIC_KEY"),
    secret_key=os.environ.get("LANGFUSE_SECRET_KEY"),
    host="http://localhost:3000"
)

# Ragas初期化
init_ragas_metrics(
    metrics=metrics,
    llm=LangchainLLMWrapper(model),
    embedding=LangchainEmbeddingsWrapper(embeddings)
)

# プロンプト・ツール・エージェント設定
output_parser = PydanticOutputParser(pydantic_object=AnswerWithSourcesModel).get_format_instructions()
prompt = ChatPromptTemplate.from_template(
    langfuse.get_prompt("rag_agent_template", label="latest").compile()
).partial(output_parser=output_parser)

tools = [vector_search.search_vector_store]
agent = langchain.agents.create_tool_calling_agent(model, tools, prompt)
executer = langchain.agents.AgentExecutor(agent=agent, tools=tools)

def run_agent(question: str):
    """エージェント実行と出力パース"""
    output = executer.invoke({"question": question}, config={"callbacks": [Langfuse_handler]})
    print("output:", output)
    # output["output"]はJSON文字列
    parsed = json.loads(output.get("output", "{}"))
    answer = parsed.get("answer", "")
    retrieved_contexts = parsed.get("source", [])
    print("answer:", answer)
    print("retrieved_contexts:", retrieved_contexts)
    return question, answer, retrieved_contexts

async def calc_ragas(question, retrieved_contexts, answer):
    """Ragasスコア計算"""
    scores = await score_with_ragas(
        question,
        retrieved_contexts,
        answer
    )
    trace_id = Langfuse_handler.get_trace_id()
    for key, value in scores.items():
        langfuse.score(trace_id=trace_id, name=key, value=float(value))
    print("Ragasスコア:", scores)

if __name__ == "__main__":
    question = "EchoNexusってなに？"
    q, answer, retrieved_contexts = run_agent(question)
    asyncio.run(calc_ragas(q, retrieved_contexts, answer))

結果

上記のmain.pyの実行結果です。

output: {'question': 'EchoNexusってなに？', 'output': '{"answer":"EchoNexusは、企業向けの統合コミュニケーションプラットフォームです。チャット、ビデオ 会議、ファイル共有、及びプロジェクト管理機能を一体化し、シームレスな情報共有を実現します。主な機能には複数チャネル対応のコミュニケーションツール、シームレスなファイル共有とクラウド連携、
...
Ragasスコア: {'faithfulness': 1.0, 'answer_relevancy': np.float64(0.7704108952083059), 'llm_context_precision_without_reference': 0.9999999999}

Ragasによる評価が行われていることが分かります。
今回のサンプルでは、逐次評価を行っていますが、Ragasの処理に少し時間がかかったため、実際の運用にはまとめてバッチ処理などを検討したほうがいいかもしれません。
また、評価スコアに基づく回答の良し悪しを決める閾値などの設定も運用上の重要なポイントになりそうです。

LangfuseからもRagasのスコアが確認できます。

まとめ & 次のステップ

  • RAGシステムの評価ツールとして「Ragas」と「Langfuse」を調査し、各ツールの特徴や実装を解説しました。
  • 今後は、評価スコアの有効な活用方法の検証や、Langfuseをシステム監視の実践など、目的に合わせたた実験と検証を進める。

参考リンク

脚注

  1. Ragasのメトリクス一覧 ↩︎

  2. Langfuseのdockerセットアップ ↩︎

セリオ株式会社 テックブログ
セリオ株式会社 テックブログPublication

未来の技術に挑戦し続ける岡山のソフトウェア企業です。製造業をはじめ、物流・介護などさまざまな業界のお客様と共創しDXを実現します。

Discussion