メインコンテンツまでスキップ

Claude Codeのトレース

MLflow Tracingは、Claude Agent SDK を使用して作成された Claude Code の会話とエージェントを自動的にトレースし、ユーザー プロンプト、AI 応答、ツールの使用状況、タイミング、セッション メタデータをキャプチャします。

MLflow は、Claude Code トレースの 2 つのアプローチをサポートしています。

  • CLI トレース :MLflow CLI を通じてトレースを設定し、対話型の Claude Code セッションを自動的にトレースします(MLflow 3.14 以降)。
  • SDKトレース : Claude Agent SDK(MLflow 3.14以降)を使用して、Pythonアプリケーションのトレースをプログラムで有効にします

要件

Claude Agent SDK トレースには以下が必要です:

  • Claude Agent SDK 0.1.0またはそれ以降
  • MLflow 3.14 以降(Databricks extra付き)
Bash
pip install --upgrade "mlflow[databricks]>=3.14.0" "claude-agent-sdk>=0.1.0"

Claude のコードを Databricks にトレースする

  1. Databricks と Anthropic の環境変数を設定します。

    Bash
    export DATABRICKS_HOST="https://your-workspace.cloud.databricks.com"
    export DATABRICKS_TOKEN="your-personal-access-token"
    export ANTHROPIC_API_KEY="your-anthropic-api-key"

    本番運用環境の場合は、安全なAPIキー管理のためにAI Gateway またはDatabricksシークレットを使用します。

  2. Claude Agent SDK の自動ログ記録を有効にして、すべての Claude Agent SDK のインタラクションを追跡します。

注記

MLflow はqueryへの直接呼び出しのトレースをサポートしていません。MLflow は、 ClaudeSDKClient使用するトレース インタラクションのみをサポートします。

Python
import asyncio
import mlflow.anthropic
from mlflow.entities.trace_location import UnityCatalog
from claude_agent_sdk import ClaudeSDKClient

# Enable autologging
mlflow.anthropic.autolog()

# Optionally configure MLflow experiment
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment(
experiment_name="my_claude_app",
trace_location=UnityCatalog(
catalog_name="<UC_CATALOG_NAME>",
schema_name="<UC_SCHEMA_NAME>",
table_prefix="<UC_TABLE_PREFIX>",
),
)


async def main():
async with ClaudeSDKClient() as client:
await client.query("What is the capital of France?")

async for message in client.receive_response():
print(message)


if __name__ == "__main__":
asyncio.run(main())

自動ログを無効にするには、 mlflow.anthropic.autolog(disable=True)を呼び出します。 3. DatabricksワークスペースのMLflowエクスペリメント UIでトレースを表示します。

上級: 評価付き SDK トレース

MLflow の GenAI 評価フレームワークで SDK トレースを使用できます。

Python
import asyncio
import os
import pandas as pd
from claude_agent_sdk import ClaudeSDKClient

import mlflow.anthropic
from mlflow.entities.trace_location import UnityCatalog
from mlflow.genai import evaluate, scorer
from mlflow.genai.judges import make_judge

os.environ["MLFLOW_TRACING_SQL_WAREHOUSE_ID"] = "<SQL_WAREHOUSE_ID>"

mlflow.anthropic.autolog()


async def run_agent(query: str) -> str:
"""Run Claude Agent SDK and return response"""
async with ClaudeSDKClient() as client:
await client.query(query)

response_text = ""
async for message in client.receive_response():
response_text += str(message) + "\n\n"

return response_text


def predict_fn(query: str) -> str:
"""Synchronous wrapper for evaluation"""
return asyncio.run(run_agent(query))


relevance = make_judge(
name="relevance",
instructions=(
"Evaluate if the response in {{ outputs }} is relevant to "
"the question in {{ inputs }}. Return either 'pass' or 'fail'."
),
model="openai:/gpt-4o",
)

# Create evaluation dataset
eval_data = pd.DataFrame(
[
{"inputs": {"query": "What is machine learning?"}},
{"inputs": {"query": "Explain neural networks"}},
]
)

# Run evaluation with automatic tracing
mlflow.set_tracking_uri("databricks")
mlflow.set_experiment(
experiment_name="claude_evaluation",
trace_location=UnityCatalog(
catalog_name="<UC_CATALOG_NAME>",
schema_name="<UC_SCHEMA_NAME>",
table_prefix="<UC_TABLE_PREFIX>",
),
)
evaluate(data=eval_data, predict_fn=predict_fn, scorers=[relevance])

トラブルシューティング

トレースが失われている:

  • 作成前にmlflow.anthropic.autolog()が呼び出されていることを確認してください ClaudeSDKClient
  • 環境変数( DATABRICKS_HOSTDATABRICKS_TOKEN )が正しく設定されていることを確認してください
  • Databricksの有効期限が切れていないことを確認してください