安定したAIエージェント開発・運用を実現するLangfuse活用方法

こちらは LayerX AI Agentブログリレー 3日目の記事です。

2日目の記事は Kenta さんによる手触り感のあるContext Engineering でした。社内でもトップクラスに手を動かしているKentaさんならではの記事でした。そちらもおすすめです！！

こんにちは。バクラク事業部エンジニアのomori（@onsd_）です。

最近は「AI申請レビュー」というAIエージェント機能の開発・運用に取り組んでいます。

この機能は、申請ルールに沿ってAIが入力内容をサポートしたり、ルール違反がないかチェックしてフィードバックを返し正しい申請を作成することで申請者・承認者双方の負担を減らす機能です。

bakuraku.jp

この機能の開発を進める中で、AIエージェント開発ならではの課題に直面しました。今回はその課題をどう解決したのかについてお話します。

AIエージェント開発で直面した3つの課題
課題解決のためのLLMOpsツール導入
- セルフホスト時の構成図
Langfuseを使った課題解決
これからの課題
- 評価周りの改善
- 可観測性の課題
おわりに

AIエージェント開発で直面した3つの課題

課題1: 挙動の可視化が難しい

AIエージェントは確率的に動作するため、同じ入力でも異なる結果が返ることがあります。そのため、「なぜこの結果になったのか」を後から追跡・分析する仕組みが不可欠でした。

従来のログでも詳細に記録すれば原因を追うことは可能ですが、膨大な情報の中から特定するのは容易ではありません。そこで、関数単位や処理単位で挙動をトレースする仕組みを導入することで、入力・出力の関係やプロンプトの影響が可視化され、直感的に把握できるようになります。

課題2: プロンプト更新のフローが未整備

当初、LLMのプロンプトは S3から取得するようにしていました。これは、リリースとは別タイミングでプロンプトの変更をできるようにするのが目的でしたが、誰が・いつ・なぜ変更したのかの履歴が残らないことが問題でした。また、S3のプロンプトの更新作業が属人化する問題もありました。

課題3: プロンプト変更による影響の不明確さ

AIエージェント機能はプロンプトを少し変更しただけでも、機能全体の精度に予期せぬ影響が出ることがあります。しかし、その影響を定量的に評価する仕組みがなく、改善したいポイントのみチェックし更新するしかありませんでした。これにより、良かれと思った変更が意図せず品質低下につながるリスクがありました。

課題解決のためのLLMOpsツール導入

これらの課題を解決するため、LLMOpsツールの導入を検討しました。

ツール選定にあたり、以下の要件を設定しました。

プロンプト管理機能
- バージョン管理や変更履歴の追跡ができること。
実験管理（精度検証）機能
- プロンプト変更による影響を定量的に評価できること。
Observability（可観測性)
- LLMの呼び出しをトレースし、監視できること。
- システムの挙動を関数単位や処理単位でトレースできる仕組みがあること
セルフホスト可能
- セキュリティ要件を満たすため、自社のインフラ上で運用できること。

複数のツールを比較検討した結果、私達は Langfuse の採用を決定しました。

langfuse.com

Langfuseは上記の要件を全て満たしていたことに加え、ユーザー数無制限でSSOにも対応したセルフホストが可能であった点が大きな決め手となりました。

セルフホスト時の構成図

Langfuse はオートスケールする複数台構成で、ストレージとして利用される ClickHouseを2台、 ClickHouse Keeperを3台で構築しています。

ClickHouse を AWS Fargate 上で冗長構成で構築する方法については次のブログをご覧ください。

tech.layerx.co.jp

Langfuseを使った課題解決

今回導入したLangfuseを使い、具体的にどのように3つの課題を解決したのかをご紹介します。

1. Trace 機能による挙動の可視化

Langfuseには、トレースデータを収集・可視化するObservabilityという機能があります。

LLM Observability & Application Tracing (open source) - Langfuse

これを利用してアプリケーションの挙動をわかりやすい形でチェックできるようになりました。

トレースの取得は @observe デコレータをつけるだけで行えます。

実装例

以下に、LLM呼び出し・関数呼び出しが混在するアプリケーションの例を示します。

from langfuse import observe, get_client
from langfuse.openai import OpenAI
import time
import random


@observe
def generate_product_description(product_name: str, features: list) -> str:
    """LLMを使って商品説明を生成"""
    features_text = ", ".join(features)

    response = OpenAI().responses.create(
        model="gpt-5-mini",
        instructions="あなたは優秀なマーケティング担当者です。魅力的な商品説明を日本語で作成してください。",
        input=f"""
商品名: {product_name}
特徴: {features_text}
        """

    )
    return response.output_text

@observe
def calculate_price(base_price: float, discount_rate: float = 0.0) -> float:
    """割引価格を計算する通常の関数"""
    time.sleep(0.1)  # 処理時間をシミュレート
    final_price = base_price * (1 - discount_rate)
    return round(final_price, 2)

@observe
def check_inventory(product_name: str) -> dict:
    """在庫状況をチェックする関数（外部API呼び出しをシミュレート）"""
    time.sleep(0.2)  # API呼び出し時間をシミュレート
    stock = random.randint(0, 100)
    return {
        "product": product_name,
        "stock": stock,
        "status": "available" if stock > 0 else "out_of_stock"
    }

@observe
def create_product_listing(product_name: str, base_price: float, features: list):
    """商品リスティングを作成するメイン関数"""
    print(f"=== {product_name} の商品リスティング作成開始 ===")

    # 在庫チェック
    inventory = check_inventory(product_name)
    print(f"在庫状況: {inventory['stock']}個")

    if inventory['status'] == 'out_of_stock':
        print("在庫切れのため、リスティング作成を中止します")
        return None

    # 価格計算（在庫が少ない場合は割引なし、多い場合は10%割引）
    discount = 0.1 if inventory['stock'] > 50 else 0.0
    final_price = calculate_price(base_price, discount)
    print(f"販売価格: ¥{final_price:,} (割引率: {discount*100}%)")

    # LLMで商品説明生成
    description = generate_product_description(product_name, features)
    print(f"商品説明:\n{description}")

    return {
        "name": product_name,
        "price": final_price,
        "description": description,
        "inventory": inventory
    }

# 実行例
if __name__ == "__main__":
    product_listing = create_product_listing(
        product_name="ワイヤレスイヤホン ProSound X1",
        base_price=15000,
        features=["ノイズキャンセリング", "24時間バッテリー", "防水機能", "高音質コーデック対応"]
    )

上のコードを実行すると、次のようにトレースが記録されます。

上の例だと、アプリケーション実行に全体で 14.23秒かかっており、そのうちLLM呼び出しは 13.99秒であることがわかります。また、LLM呼び出しのトークン数やコストなども表示されています。関数の入出力もそれぞれ追うことができるので、あとからなぜこの挙動になったのか？の調査も容易です。

2. プロンプトをコードと同じように管理して更新フローを標準化

Langfuse には Prompt Management 機能があり、プロンプトのバージョン管理が可能です。

ここでいうプロンプトはLLMへの指示だけでなく、推論モデルIDや各種パラメータも含まれます。

Open Source Prompt Management - Langfuse

Prompt Management は先述した Observability と連携しており、どのプロンプトでどのような結果が出たのかを簡単に追跡できます。

# サンプルコードの上部を変更

langfuse = get_client()  # Langfuseクライアントの初期化
langfuse.create_prompt( # Langfuse にプロンプトを登録
    name="generate_product_description",
    type="text",
    prompt="""商品名: {{product_name}}
特徴: {{features_text}}
""",
    labels=["production"],
    config={
        "model": "gpt-5-mini",
        "instructions":"あなたは優秀なマーケティング担当者です。魅力的な商品説明を日本語で作成してください。",
    }
)

@observe
def generate_product_description(product_name: str, features: list) -> str:
    """LLMを使って商品説明を生成"""
    features_text = ", ".join(features)

    # 事前に作成したプロンプトを取得してコンパイル
    prompt = langfuse.get_prompt("generate_product_description")
    compiled_prompt = prompt.compile(
        product_name=product_name,
        features_text=features_text
    )

    response = OpenAI().responses.create(
        input=compiled_prompt,
        **prompt.config,
        langfuse_prompt=prompt # トレースにプロンプト情報を含める
    )
    return response.output_text

このコードを実行すると、Langfuseからプロンプトを取得しLLM呼び出しが行われるようになります。

さらに、どのプロンプトがどれくらい使われたのかといった情報も連携されます。

プロンプトの更新フロー

プロンプトは一度登録すれば終わりではなく、改善のために継続的に更新していく必要があります。そのため、プロンプトをどのように管理し更新していくかというフローの設計が重要になります。

管理方法の比較

プロンプトの管理にはいくつかの選択肢が考えられます。

S3管理
- S3のようなオブジェクトストレージで管理する方法は手軽ですが、変更履歴やその理由を追跡するのが難しいという課題があります。
Git管理
- アプリケーションコードと同様にGitで管理すれば、変更履歴やレビューの背景が明確になるという大きなメリットがあります。
- しかし、プロンプトの変更を本番環境に反映させるには、アプリケーションのリリースサイクルに合わせる必要がありロールバックに時間がかかることが考えられます。
Langfuse管理
- Langfuse で管理することで、変更履歴の追跡やアプリケーションがどのバージョンを使うのかの切り替えが容易になります。
- アプリケーションが利用するプロンプトはラベルで管理され、デフォルトでは production ラベルがついたプロンプトが配信されます。
  - ラベルの変更はUIから行うことができるため、プロンプトに問題が見つかった場合でも即座にロールバックが可能です。
- ただし、Gitのようなコードレビューの仕組みはなく、誰が・なぜ変更したのかの背景がわかりにくいという課題があります。

構築した更新フロー

私たちのチームの要件は次の2点でした。

プロンプトの変更もアプリケーションコードと同様にレビュープロセスを挟みたい。
プロンプトに予期せぬ問題があった場合に、アプリケーションの再デプロイを待たずに迅速かつ容易にロールバックできる仕組みが欲しい。

この2つの要件を満たすため、私たちは GitHubを使った変更時のレビューの担保と、Langfuseによる柔軟なデプロイ管理を組み合わせた、以下のフローを構築しました。

プロンプトをアプリケーションコードと同じGitリポジトリで管理
変更はPull Request経由でレビュー・マージ
デプロイ時にCI環境からSDKで自動更新
問題があった場合はLangfuseのUIからラベルを切り替えてロールバック

これにより、普段はアプリケーションと足並みを揃えつつ、何かあったときだけLangfuseのUIから即座に対応できるという、安全で効率的な更新サイクルが実現できました。

実装例

CI環境で実行されるスクリプトは以下のようになります。

import yaml
from langfuse import Langfuse

langfuse = Langfuse()

# 実際はファイルから読み込む
# prompts/generate_product_description.yaml のようなファイルで管理している
yaml_string = """
name: generate_product_description
type: text
prompt: |
  商品名: {{product_name}}
  特徴: {{features_text}}
labels:
  - production
config:
  model: gpt-5-mini
  instructions: あなたは優秀なマーケティング担当者です。魅力的な商品説明を日本語で作成してください。
"""
prompt_data = yaml.safe_load(yaml_string)

langfuse.create_prompt(
    **prompt_data
)

3. プロンプトの自動リグレッションテスト

Langfuseの Evaluation 機能を活用し、プロンプトのリグレッションテストを自動で行う仕組みを構築しました。

Evaluation 機能の中でも Dataset と Score を活用しています。

Dataset
- 代表的な入力と期待される出力をLangfuse上で管理する機能
Score
- 実際の出力を任意の基準で採点しLangfuseに記録できる機能

具体的には、事前に代表的な入力と期待される出力をDatasetとして用意し、プロンプト変更後の出力と突き合わせて正答率を計算します。採点は単純な一致判定だけでなく、独自の評価基準に基づいた柔軟な採点も可能です。

# https://langfuse.com/docs/evaluation/dataset-runs/remote-run を基にした例
from langfuse import get_client, observe
from langfuse.openai import OpenAI

@observe
def my_llm_function(question: str):
    response = OpenAI().chat.completions.create(
        model="gpt-4o", messages=[{"role": "user", "content": question}]
    )
    output = response.choices[0].message.content

    # Update trace input / output
    get_client().update_current_trace(input=question, output=output)

    return output

# Load the dataset
dataset = get_client().get_dataset("データセット名")

# Loop over the dataset items
for item in dataset.items:
    # Use the item.run() context manager for automatic trace linking
    with item.run(
        run_name="テスト #1",
        run_description="My first run",
    ) as root_span:
        # my_llm_application は、評価対象としたいLLMアプリケーション全体（例えば、この記事で紹介した create_product_listing のような関数）を指すものとします。
        output = my_llm_application.run(item.input)

        # Optionally: Add scores computed in your experiment runner, e.g. json equality check
        root_span.score_trace(
            name="<example_eval>",
            value=my_eval_fn(item.input, output, item.expected_output),
            comment="This is a comment",  # optional, useful to add reasoning
        )

プロンプトの「評価」は、日々の改善サイクルの中で行うものと、安全に本番リリースを行うための「ガードレール」という２つの側面があります。今回の仕組みが担うのは後者の「ガードレール」の部分です。あらかじめ設定されたベースラインを維持し、プロンプトの変更によって意図しない性能劣化が起きることを防ぐことが目的です。

プロンプト変更のPull Request が作成されると、GitHub Actions 上で先程のリグレッションテストが行われます。結果は Langfuse に記録され、サマリーが自動でコメントされます。

この機構を2. プロンプトをコードと同じように管理して更新フローを標準化 と組み合わせることで、データに基づいたプロンプトの改善サイクルを回せるようになりました。

最終的なフローは以下のとおりです。

これからの課題

Langfuse を導入したことで、実行結果やプロンプト管理、精度検証の基盤は整いつつありますが、まだ課題も残っています。

評価周りの改善

現在は完全一致で比較できる部分のみ自動化できていますが、「文章の自然さ」や「表示されるレビューメッセージとしての正しさ」の評価は依然として人間が行っています。この課題に対し、まずは人間による評価を重ねて評価指標を定義します。そして評価指標が固まった後には、Langfuseの LLM as a Judge 機能などを活用し、この定性評価自体の自動化を目指します。

langfuse.com