👻

Langfuseを用いた実験管理のメモ

2025/01/23に公開

Langfuseとは?

Langfuseとは、大規模言語モデル(LLM)を利用したアプリケーションなどの実験管理プラットフォームです。LLMでの処理のトラッキング(トレース)やメトリックでの評価などを行うことができ、プロンプトの管理や、評価用データセットの作成なども行うことができます。

似たようなものとして、LangSmithなどもありますが、Langfuseの方が手軽に利用できる(無料プランからセルフホストが可能)のが特徴です。
https://langfuse.com/

Langfuseの動かし方

クラウドでの利用方法

公式サイトからアカウント登録を行うことで始めることができます。無料プランからチームプランがあり、それぞれで利用範囲が異なります。
また、リージョンをEUかUSのどちらかを選択する必要があります。

アカウント作成後に、ホーム画面が出てくるので、organizationを新規作成します。ここでは、メンバーの登録やプロジェクトの作成などを行います。

セルフホストの場合

公式のgithubをcloneしてdockerを起動します。

git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up

起動して、http://localhost:3000にアクセスすると、アカウント作成に移行します。アカウント作成後はプロジェクトの作成を行います。

APIキーの発行

プロジェクトの設定画面からAPIの発行を行います。

Traceの仕方

Langfuseをインストールして使用していきます。

pip install langfuse openai

今回はlangchainを組み合わせたものを使用してみます。langchainを使用した場合には、Langfuseのコールバックを設定することでtraceが可能になります。

import os
from dotenv import load_dotenv
from langchain_openai import AzureChatOpenAI
from langchain import PromptTemplate, LLMChain
from langfuse.callback import CallbackHandler

load_dotenv()


template = """あなたはネーミングのプロです。
次の製品に関して、魅力的な名前を提案してください。

製品: {product}
色: {color}

提案する製品名:
"""

prompt = PromptTemplate(
    template=template,
    input_variables=["product", "color"],
)


llm = AzureChatOpenAI(
    model="gpt-4-0125-preview",
    temperature=1,
    azure_endpoint=os.environ["AZURE_OPENAI_ENDPOINT"],
    azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
    api_key=os.environ["AZURE_OPENAI_API_KEY"],
    api_version=os.environ["OPENAI_API_VERSION"],
)

# Callbackの設定
langfuse_handler = CallbackHandler(
    public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    host=os.environ["LANGFUSE_HOST"]
)

chain = LLMChain(
    prompt=prompt,
    llm=llm,
)

result = chain.invoke(
    {
        "product": "スマートスピーカー",
        "color": "ホワイト",
    }, 
    config={"callbacks": [langfuse_handler]}
)

print(result)

実行すると、UIのtrace画面から結果の確認ができるようになります。出力結果には、入力・出力以外に、レイテンシーやトークン数、コストなどが自動的に計算されて表示されています。また、各処理のタイムラインも確認することができ、どの処理にどのくらい時間がかかっているかなどを確認することもできます。

また、デコレータを使用することで、各処理を記録していくなども可能です。

@observe()
def run_azure_openai_chain(product_name: str, color: str):
    result = chain.invoke({
        "product": product_name,
        "color": color,
    },
    config={"callbacks": [langfuse_context.get_current_langchain_handler()]})
    return result


if __name__ == "__main__":
    response = run_azure_openai_chain(
        product_name="スマートスピーカー", 
        color="ホワイト"
    )
    print(response)

langfuse_context.get_current_langchain_handler()を使用することで、関数名としてtraceすることが可能です。関数内で複数回LLMの処理を行っていて、その最終結果をトラッキングしたい場合はこちらの方が便利だと思います。

Prompt

Lnagfuseでは、プロンプトも管理することが可能です。管理プロンプトはUI側もしくはコードで登録・更新が可能となっています。

# promptの登録
# 同じ名前のものが登録されている場合には、新しいバージョンとして更新が入る。
langfuse.create_prompt(
    name="movie-critic",
    type="text",
    prompt="あなたは{{criticlevel}}の映画評論家です。{{movie}}はお好きですか?",
    labels=["production"],  # プロダクトとして使用しているなどのラベル付けが可能。
    config={
        "model": "gpt-3.5-turbo",
        "temperature": 0.7,
        "supported_languages": ["en", "fr"],
    }
)

# chat形式も可能
langfuse.create_prompt(
    name="movie-critic-chat",
    type="chat",
    # promptを日本語に変更
    prompt=[
        { "role": "system", "content": "あなたは{{criticlevel}}の映画評論家です。" },
        { "role": "user", "content": "{{movie}}はお好きですか?" },
    ],
    labels=["production"],  # プロダクトとして使用しているなどのラベル付けが可能。
    config={
        "model": "gpt-3.5-turbo",
        "temperature": 0.7,
        "supported_languages": ["ja", "en", "fr"],
    },  # optionally, add configs or tags
)

登録したプロンプトは呼び出すことが可能です。

from langfuse import Langfuse
from langchain_core.prompts import ChatPromptTemplate
 
langfuse = Langfuse()
 
# 指定をしないと、最新が呼び出されます。
langfuse_prompt = langfuse.get_prompt("movie-critic")
 
# ChatPromptTemplateでの使用例
langchain_prompt = ChatPromptTemplate.from_template(langfuse_prompt.get_langchain_prompt())
 
# chat形式での例
langfuse_prompt = langfuse.get_prompt("movie-critic-chat", type="chat")
langchain_prompt = ChatPromptTemplate.from_messages(langfuse_prompt.get_langchain_prompt())

評価用データセットの登録

作成したプロンプトを改修など行う場合、過去のプロンプトとの出力の比較を行いたくなるかと思います。その際の評価用データセットをLangfuse上で登録することが可能です。
データセットの登録は、UI上からも可能ですし、python上からも可能です。

データセットには、入力に必要なデータと想定される出力を登録することができます。ただし、UI上からデータの登録を行う場合には、json形式でデータの登録を行う必要があるので、少し癖があります。(UI上で、json形式でデータを入力するのは少し骨が折れる感じもあるのでうまく登録できる機能があると、非エンジニアも利用しやすいなと思っています)

python上から一括で登録も可能なので、実際に評価用のデータセットを構築するのであればこちらの方が使用しやすいかなと思っています。

langfuse.create_dataset_item(
    dataset_name="<dataset_name>",
    # any python object or value, optional
    input={
        "text": "hello world"
    },
    # any python object or value, optional
    expected_output={
        "text": "hello world"
    },
    # metadata, optional
    metadata={
        "model": "llama3",
    }
)

実験

登録したデータセットに対して、プロンプトであったり、実際にプロダクトなどで組み込んでいる処理のを実験を行うことができ、これにより、プロンプトでの差分の確認であったり、処理全体での改善効果などを確認することができます。

登録したプロンプトに対して実験を行う場合(Prompt Experiment)はUI上から実施することができ非エンジニアでも簡単にプロンプトの効果が確認できます。また、それぞれの結果に関してもUI上から比較ができるので、短いサイクルで検証作業・記録・比較が行うことができるメリットがあります。

一方で、最終的な出力を生成する際に、複数のLLMの処理を行っている場合(例えば、要約をさせるときに、内容をトピックごとに分割して要約する機能と、その内容を清書する機能が二つある場合)はPrompt Experimentで行うことは難しいです(それぞれの機能ごとの評価は可能です)。このようなときに、最終的な出力をUIで確認・比較をしたい場合には、スクリプトベースで処理を作成したほうがいいです。

やり方としては、実験を行うRunnerを作成して、各データセットに処理をさせるだけです。処理部分にデコレータを付けるだけで良いので、LLMで実装しているものを実験用に改めて作成する必要もありません。

from langfuse.openai import openai
from langfuse.decorators import observe, langfuse_context


@observe()
def run_my_custom_llm_app(input, system_prompt):
    """
    実際にプロダクトや機能として評価したい部分になります。
    """
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": input["country"]}
    ]

    completion = openai.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=messages
    ).choices[0].message.content

    return completion

# 評価関数
def simple_evaluation(output, expected_output):
    """
    応答が期待された出力と完全一致するかを評価します。
    任意の評価関数を組み込むことが可能です。
    """
    return output == expected_output

# 実験を実行する関数
def run_experiment(experiment_name, system_prompt):
    """
    データセットに基づいて指定されたプロンプトで実験を実行します。
    各データ項目について、応答を生成し評価結果を記録します。
    """
    dataset = langfuse.get_dataset("capital_cities")  # 「capital_cities」データセットを取得

    for item in dataset.items:
        # トレースIDを作成し、観測を記録
        with item.observe(run_name=experiment_name) as trace_id:
            # アプリケーションを実行し、出力を取得
            output = run_my_custom_llm_app(item.input, system_prompt)

            # 評価結果をトレースに記録
            langfuse.score(
                trace_id=trace_id,
                name="exact_match",
                value=simple_evaluation(output, item.expected_output)
            )


def main():
    run_experiment(
        "有名な都市",
        "ユーザーが国名を入力します。その国で最も有名な都市を応答してください。"
    )

    run_experiment(
        "首都を直接尋ねる",
        "次の国の首都はどこですか?"
    )

    run_experiment(
        "首都を特定して尋ねる",
        "ユーザーが国名を入力します。その国の首都の名前だけを応答してください。"
    )

    run_experiment(
        "首都を特定して尋ねる(改良版)",
        "ユーザーが国名を入力します。その国の首都の名前だけを応答してください。都市名のみを述べてください。"
    )

    langfuse_context.flush()
    langfuse.flush()


if __name__ == "__main__":
    main()

また、LLM-as-a-judgeのようにLLMに結果の評価をさせることも可能であり、langfuse側でデフォルトで用意しているものもあります。(英語で記載されているので日本語で行っているものをそのまま使っていいのかというのはあるかもしれませんが)

また、RAGASなどとの連携も用意なので、様々な評価をトラッキングすることが可能です。
https://langfuse.com/guides/cookbook/evaluation_of_rag_with_ragas

その他

langfuseはgithub上で改善要望などを受け付けており、いいね数に応じて対応してくれるようです。またDiscordもあるので、気になる部分があればアクセスしてもいいかもしれません。
https://github.com/orgs/langfuse/discussions/categories/ideas
https://langfuse.com/discord

参考文献

https://langfuse.com/docs
https://github.com/langfuse/langfuse

Discussion