Langfuseを導入してLLMアプリケーション開発を劇的に進化させる

• ツール導入前の課題

1. LLMアプリケーション（サービス）の開発速度向上

AIエージェント開発の際、ユーザーの質問に対して思うような回答をしてない場合その原因がプロンプトにあるのか、中間で呼び出されるツールの結果によるのかがわからず、開発難易度が上がっていました。

特にLLMを多段で呼び出すようなアプリケーション開発だと、中間で呼び出されるLLMの回答内容が可視化できる必要がありますが、LLMOpsツールを使わないとその可視化が難しい現状があります。

2. プロンプト管理

運用上変更の多いプロンプトをLLMアプリケーションのソースコードから切り離す形で管理し、デプロイを伴わず迅速なデリバリーができる必要がありました。

3. フィードバックループを回す

ユーザーフィードバックやドメインエキスパートによる回答の評価をLLMの回答と紐づけながら格納し、分析することで今後の改善につなげる必要がありました。

4. コスト管理

サービスの機能ごとに生成にかかったコストを算出しにくく、予算を適切に設定する材料が不足してました。

• どのような状態を目指していたか

上記を達成することでLLMアプリケーションの開発の細かな不満を取り除き開発速度を向上させることができることと、サービスの品質向上を期待しておりました。

• Langsmith
• PromptLayer

• AIエージェントのLLMOpsが実施したかったため、LLMのINPUT/OUTPUTだけでなくツールのINPUT/OUTPUTも記録できること
• モデルの生成コストを可視化できること
• ユーザーフィードバック・ドメインエキスパートによる評価・LLM as a judgeができる基盤であること
• セキュリティ要件をクリアするためにセルフホステッドができること
• セルフホステッドを選択してもライセンス費が跳ね上がらないこと
• Langsmithなど主要なLLMOpsツールと比較して遜色ない機能であること（アップデートによる追従が頻繁であること）

• セルフホステッド

一番のポイントはセルフホステッドで運用できる点です。 もちろんLangsmithもセルフホステッド版はあるものの、エンタープライズライセンスからの提供のためOSS版が使えるLangfuseに軍配が上がりました。

セルフホステッドにすることで、構築の手間はありますがLLMアプリケーションと同一のクラウド環境にホストすることでセキュリティ要件の調整をスキップすることが可能という背景がございました。

• 可観測実装の容易さ

LangfuseSDKにはObserveデコレーターが用意されており、観測したい関数に対してデコレーターをつけることで比較的手軽に可視化を実施することができます。

from langfuse import observe

@observe() # INPUT: name, OUTPUT: "xxxxx"がLangfuseのSpanとして記録される
def hoge_output(name: str) -> str:
    """(LLMな処理...)"""
    return "xxxxx"

ただし、Observeデコレータはあくまでも関数のIN/OUTを引数・戻りから記録し、可視化するのみなので中間イベントの記録、Generationの細かな記録(First Tokenのレイテンシーなど)、メタデータの記録をするにはLangfuseインスタンスから直接記録することをおすすめします。

@observe()
def complex_logic(tool_input: str) -> str:
    
    langfuse.create_event(
        name="external_tool_call",
        metadata={
            "tool_name": "SampleTool",
            "input": xxxxxxxx,
        }
    )

例えばAmazon BedrockのConverse APIをBoto3から利用する場合、質問に対して 生成 -> Tool -> 生成とサンドイッチ構造になりますが、それぞれをLangfuseのGeneration(LLMの応答を管理する特別なObservation) -> Span(時間軸を持ったObservation) -> Generation(LLMの応答を管理する特別なObservation) と記録して、全体を親Observation or Trace(observeデコレーター)で紐づける、ということをするときれいに記録されます。

@observe
def use_tool():
    # Langfuse SDKのセットアップ
    langfuse = Langfuse()
    trace = langfuse.trace(id=trace_id)
    prompt_name = "hogehoge" # Langfuseに登録しているプロンプト名
    prompt_version = 1 # Langfuseに登録しているプロンプトのバージョン

    # 最初の生成処理のスパン作成
    initial_generation = trace.generation(
        name="invoke_converse_stream_use_tool_start",
        input=message,
    )

    # Bedrock ConverseAPIの呼び出しを実施
    messages = invoke_converse(messages, tool_config)

    # 生成処理の結果を更新
    initial_generation.end(
        output=messages,
    )

    # (中略..)ツール呼び出し判定処理
    # ツール使用のスパン作成
    tool_span = trace.span(
        name=tool_config["tools"][0]["toolSpec"]["name"],
        input=tool["input"][tool_arg_key],
        start_time=start_time
    )

    # ツールの実行
    tool_output = tool_function(tool_arg)

    # ツール使用のスパン終了
    tool_span.end(output=tool_output, level="DEBUG")
    message = invoke_converse(tool_output)

    # ツール実行後の生成処理のスパン作成
    final_generation = trace.generation(
        name="invoke_converse_stream_use_tool_start",
        input=message,
    )
    final_generation.end(
        output=message,
        metadata={
            "tool_use": "end",
            "context": tool_output,
            "stop_reason": stop_reason,
            "region": region,
            "index_model": index_model
        }
    )
    return message

キャプチャのように@observe単位で（赤と緑）括られ、その配下に子ObservationとしてGenerationやToolのSpanを保つ構造になってます。



もし、LangChain、LangGraphをお使いでしたら専用のCallbackも提供されているためさらに実装が簡単になります。

# LangChainのCallbackを使う場合
from langfuse.callback import CallbackHandler
langfuse_handler = CallbackHandler(
    secret_key="sk-lf-...",
    public_key="pk-lf-...",
    host="https://xxx"
)

chain.invoke({"input": "<user_input>"}, config={"callbacks": [langfuse_handler]})

# LangGraphのCallbackを使う場合
from langfuse.callback import CallbackHandler
 
langfuse_handler = CallbackHandler(
    secret_key="sk-lf-...",
    public_key="pk-lf-...",
    host="https://xxx"
)

for s in graph.stream({"messages": [HumanMessage(content = "<user_input>")]}, config={"callbacks": [langfuse_handler]}):

コードを実行するとこんな感じで良しなに記録されます。



今更ですが、LangfuseのObservation種別については次のドキュメントを参照してください。

• 改善したかった課題はどれくらい解決されたか

すべて解決しています。

新機能開発でも利用しない日はないですし、本番運用の監視・分析用途でも毎日利用してます。

ただし、Langfuseのプランをあげないと使えない機能がいくつかあったのでその点はCI/CDと絡めて運用したり、自作で評価システムを作ることで対応しております。

• どのような成果が得られたか

新機能開発時、AIエージェントのワークフローを構築する際のデバッグに利用したり、コスト含めモデルの挙動を細かく確認することができるようになったおかげで、開発速度があがったと思います。（プロジェクト初期から利用しているため、定量的な効果は出せませんが、肌感で工数が1/3〜2/3くらい減っていると思います。）

また、プロダクトオーナーやステークホルダーがアプリケーションの利用状況やレイテンシー、運用コストを確認する起点になっているため、サービスの提供機能のブラッシュアップ、予算・プライシング検討などに生かされております。

AWSによるコンテナアプリケーション構築の経験があれば、Langfuse v2の導入時はIaC化含めて苦労することはあまりないですが、Langfuseのバージョンをv3にする際に作成するコンポーネントが一気に増えるためインフラ構築の難易度が上がります。

また、本番・開発などの環境ごとにプロジェクトを切ってしまうと作成したプロンプトの引き継ぎができない（コピーできない）という細かいながら重要な課題がありました。

このあたりはアーキテクチャーにも記載しておりますが、CI/CDを活用してデプロイパイプラインとしてコピースクリプトを作成して解決しております。

LLMを用いたアプリケーション開発、特にAIエージェントなどツールや生成が入り交じるようなアプリケーションの開発においてはLangfuseのようなLLMOpsツールの導入をしないとかなり開発が困難です。

LangfuseはOSSかつセルフホステッドができるメリットを持ちつつ、他ツールと遜色ないレベルの機能が揃っている点がとても良いです。

昨年末(2024年末)からアップデートのスピードもかなり早くなっている印象があるので、新機能をほぼ毎週体験できるというワクワク感も魅力だと思います。

完全に余談なのですが、Langfuse開発者はかなりこまめにDiscussionを返信してくれるので、質問がある場合は気軽に聞いてみると良いかもしれません。

プロジェクトの垣根を超えて、弊社内全体でLangfuseを活用するにはまだ至っていないので、さらに普及させていきたいです。

またアーキテクチャ解説でもお話した通り、RagasとLangfuseを組み合わせた自働評価にもチャレンジしておりますが、まだ本格運用できてないのでこちらも活用に向けて検証進めていきたいと思います。