Langfuse + RagasでRAGの精度管理 - Re:ゼロから始めるML生活

最近RAGをいじっています。

www.nogawanogawa.com

大体評価周りに関して何したら良いかはわかってきた気がするので、今度は精度に関するトラッキングをしていきたいと思います。というわけで前回までのRagasやLlamaIndexに加えて、今回はLangfuseを使ってみたいと思います。

Langfuse
- LlamaIndexを使う際の他のObserbability系のツール
- お値段
使ってみる
参考文献
感想

Langfuse

Langfuseは、OSSとして開発されているLLMエンジニアリングプラットフォームです。

Langfuse is an open-source LLM engineering platform that helps teams collaboratively debug, analyze, and iterate on their LLM applications. Overview - Langfuse

提供されている主な機能としては、下記のようなものがあります。

Develop
- Observability: トラッキング
- Langfuse UI: トラッキングしたものの見る画面
- Prompt Management: Langfuse内のプロンプト管理
- Prompt Engineering: Langfuse内でのプロンプトエンジニアリング
Monitor
- Analytics: コストやレイテンシの分析
- Evals: 評価結果の記録
Test
- Experiments: バージョンごとの振る舞いの記録

LlamaIndexを使う際の他のObserbability系のツール

この記事、最初はLangSmithを使って書こうとおもったんですが、簡単なインテグレーションが提供されているサービスにLangSmithがありませんでした。 LlamaIndexを使ったときのObserbability系のツールは下記が紹介されていました。

Langfuse
DeepEval
Weights and Biases Prompts
OpenLLMetry
Arize Phoenix
OpenInference
TruEra TruLens
HoneyHive
PromptLayer

docs.llamaindex.ai

正直どれも機能にそこまで提供機能に差はないとは思っていますが、今回はLangfuseを使っていきます。

お値段

個人で試す分には基本無料で使用できます。トラッキング回数が多い場合やチームで使う場合や1ユーザー59$/month〜くらいになるみたいですね。

使ってみる

何はともあれ使ってみたいと思います。

まずはただ使ってみる

今回は大人しくSaaS版を使っていきます。

LangfuseのProjectを作成し、LANGFUSE_SECRET_KEYとLANGFUSE_PUBLIC_KEY、LANGFUSE_HOSTを設定します。

import os
from google.colab import userdata

os.environ["LANGFUSE_SECRET_KEY"] = userdata.get('LANGFUSE_SECRET_KEY')
os.environ["LANGFUSE_PUBLIC_KEY"] = userdata.get('LANGFUSE_PUBLIC_KEY')
os.environ["LANGFUSE_HOST"] = userdata.get('LANGFUSE_HOST')
os.environ["OPENAI_API_KEY"] = userdata.get('MY_OPENAI_API_KEY')

あとは設定をして終わりです。

from llama_index.core import Settings
from llama_index.core.callbacks import CallbackManager
from langfuse.llama_index import LlamaIndexCallbackHandler
 
# Instantiate a new LlamaIndexCallbackHandler and register it in the LlamaIndex Settings
langfuse_handler = LlamaIndexCallbackHandler()
Settings.callback_manager = CallbackManager([langfuse_handler])

その後は普通にRAGを動かすとこんな感じにトラッキングされた値を見ることができるようになります。

レスポンスタイムや処理したトークン数だけでなく、かかったコストも表示されるのは地味に嬉しいですね。

Ragasの出力を記録する

今度は前回やってみたRagasの出力について記録してみたいと思います。こちらを参考にやってみようと思います。

langfuse.com

一度トラッキングされたものを再度取得します。

def get_traces(name=None, limit=None, user_id=None):
    all_data = []
    page = 1

    while True:
        response = langfuse.client.trace.list(
            name=name, page=page, user_id=user_id
        )
        if not response.data:
            break
        page += 1
        all_data.extend(response.data)
        if len(all_data) > limit:
            break

    return all_data[:limit]

これを用いて、改めてragasのscoreを計算するようにします。

# run ragas evaluate
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

ds = Dataset.from_dict(evaluation_batch)
r = evaluate(ds, metrics=metrics)

これを再度langfuseにトラッキングさせています。

for _, row in df.iterrows():
    for metric in metrics:
        langfuse.score(
            name=metric.name,
            value=row[metric.name],
            trace_id=row["trace_id"]
        )

ここまでやるとこんな感じにragasの出力を記録することができました。