Databricksで本格運用されているAIエージェントの監視

Last updated at 2025-10-14Posted at 2025-10-14

以前こちらの記事を書いたのですが、本番運用のモニタリングがあまり腹落ちしてませんでした。

こちらの続きという形でマニュアル片手に再度動かしてみます。

こちらを書いた際には、GUIからスコアラーを登録しました。

今回はマニュアルに沿ってコードでスコアラーを追加します。

%pip install -q --upgrade "mlflow[databricks]>=3.1"
dbutils.library.restartPython()

そして、この時点で新たな気づきが。コードからスコアラーを追加する際、追加先はどうやって指定するのか？ということでした。これまで触ってきてMLflowエクスペリメントになることはわかっていたので、mlflow.set_experimentを使います。

LoggedModelについて調べた時も感じたのですが、これまではMLflowエクスペリメントって正直箱としての使い方だったんですが、それが生成AIアプリに関するメタデータをまるっと管理するハブになってきたんだなとしみじみ思いました。

import mlflow
mlflow.set_experiment("/Workspace/Users/takaaki.yayoi@databricks.com/20251010_model_serving_trace/simple-agent-mlflow3")

あと、自分が混乱したのは以前存在していたMLflow 2のモニターの存在です。ところどころで、MLflow 2のモニターの画面にアクセスしちゃうんですよね。モデルサービングエンドポイントのエージェントを監視はMLflow 2のモニターにアクセスしてしまうので注意してください。

登録されているスコアラーの一覧

この時点では何も表示されません。

from mlflow.genai.scorers import list_scorers

# List all registered scorers
scorers = list_scorers()
for scorer in scorers:
    print(f"Name: {scorer.name}")
    print(f"Sample rate: {scorer.sample_rate}")
    print(f"Filter: {scorer.filter_string}")
    print("---")

定義済みスコアラーを使用する

すぐに使えるビルトインのスコアラーがあります。

from mlflow.genai.scorers import Safety, ScorerSamplingConfig

# スコアラーを名前付きで登録し、モニタリングを開始
safety_scorer = Safety().register(name="my_safety_scorer")  # 名前はエクスペリメントごとにユニークである必要があります
safety_scorer = safety_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))

これで、エクスペリメントのスコアラーにアクセスするとスコアラーが作成されていることを確認できます。

コードからも確認できます。

from mlflow.genai.scorers import list_scorers

# List all registered scorers
scorers = list_scorers()
for scorer in scorers:
    print(f"Name: {scorer.name}")
    print(f"Sample rate: {scorer.sample_rate}")
    print(f"Filter: {scorer.filter_string}")
    print("---")

Name: my_safety_scorer
Sample rate: 1.0
Filter: None
---

ガイドラインベースのLLMスコアラーを使用する

日本語でガイドラインを指定して評価することもできます。

from mlflow.genai.scorers import Guidelines

# ガイドラインスコアラーを作成して登録
japanese_scorer = Guidelines(
  name="my_japanese_scorer",
  guidelines=["回答は日本語でなければなりません"]
).register(name="my_japanese_scorer")  # nameはエクスペリメントごとにユニークである必要があります

# 指定したサンプリングレートでモニタリングを開始
japanese_scorer = japanese_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))

プロンプトベースのスコアラーを使用する

より詳細な観点で評価するようにプロンプトを構成することもできます。

from mlflow.genai.scorers import scorer, ScorerSamplingConfig


@scorer
def formality(inputs, outputs, trace):
    # スコアラ関数本体内でインラインでインポートする必要があります
    from mlflow.genai.judges.databricks import custom_prompt_judge
    from mlflow.entities.assessment import DEFAULT_FEEDBACK_NAME

    formality_prompt = """
    レスポンスを確認し、その形式性を判定します。

    <request>{{request}}</request>
    <response>{{response}}</response>

    以下のカテゴリのいずれかを選択してください。

    [[formal]]: レスポンスは非常に形式的です。
    [[semi_formal]]: レスポンスはやや形式的です。友情などが言及されている場合はやや形式的とみなします。
    [[not_formal]]: レスポンスは形式的ではありません。
    """

    my_prompt_judge = custom_prompt_judge(
        name="formality",
        prompt_template=formality_prompt,
        numeric_values={
            "formal": 1,
            "semi_formal": 0.5,
            "not_formal": 0,
        },
        model="databricks:/databricks-gpt-oss-20b",  # 任意
    )

    result = my_prompt_judge(request=inputs, response=inputs)
    if hasattr(result, "name"):
        result.name = DEFAULT_FEEDBACK_NAME
    return result

# カスタムスコアラを登録し、モニタリングを開始します
formality_scorer = formality.register(name="my_formality_scorer")  # 名前はエクスペリメントごとにユニークである必要があります
formality_scorer = formality_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.1))

カスタムスコアラー関数を使用する

関数を定義することもできます。

from mlflow.genai.scorers import scorer, ScorerSamplingConfig

# カスタムメトリクス: レスポンスがDatabricksに言及しているかを判定
@scorer
def mentions_databricks(outputs):
    """レスポンスがDatabricksに言及しているかを判定"""
    return "databricks" in str(outputs.get("response", "")).lower()

# カスタムメトリクス: レスポンスの長さをチェック
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """レスポンスの文字数を測定"""
    return len(str(outputs.get("response", "")))

# 複数入力を受け取るカスタムメトリクス
@scorer
def response_relevance_score(inputs, outputs):
    """キーワードマッチングによる関連度スコア"""
    query = str(inputs.get("query", "")).lower()
    response = str(outputs.get("response", "")).lower()

    # シンプルなキーワードマッチング（必要に応じてロジックを変更）
    query_words = set(query.split())
    response_words = set(response.split())

    if not query_words:
        return 0.0

    overlap = len(query_words & response_words)
    return overlap / len(query_words)

# カスタムスコアラーを登録し、モニタリングを開始
databricks_scorer = mentions_databricks.register(name="databricks_mentions")
databricks_scorer = databricks_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

length_scorer = response_length.register(name="response_length")
length_scorer = length_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))

relevance_scorer = response_relevance_score.register(name="response_relevance_score")  # nameはエクスペリメントごとにユニークである必要があります
relevance_scorer = relevance_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))