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

低レベルクライアント API (詳細)

MlflowClient APIs、トレースのライフサイクル管理を直接的かつきめ細かく制御できます。高レベルのAPIsはほとんどのユースケースをエレガントに処理しますが、トレースの作成、カスタムトレースID、または既存のオブザーバビリティシステムとの統合を明示的に制御する必要がある高度なシナリオでは、クライアントAPIsが不可欠です。

警告

始める前に : クライアント API は、高レベルの API 要件を満たさない場合にのみ使用することをお勧めします。

  • 親子関係の自動検出はありません
  • 手動の例外処理が必要
  • 自動トレース統合と互換性がありません
  • トレースのライフサイクルを完全に制御
  • カスタムトレースID管理
  • 既存のシステムとの統合

コアコンセプト​

トレースライフサイクル​

すべてのトレースは、明示的に管理する必要がある厳密なライフサイクルに従います。

Mermaid
graph LR
    A[Start Trace] --> B[Start Span 1]
    B --> C[Start Span 2]
    C --> D[End Span 2]
    D --> E[End Span 1]
    E --> F[End Trace]
important

ゴールデンルール :すべての start_trace または start_span コールには、対応する end_trace または end_span コールが必要です。スパンを閉じないと、トレースが不完全になります。

キー識別子​

これらの識別子を理解することは、クライアント API を使用するために重要です。

識別子

説明

使用量

request_id

一意のトレース識別子

トレース内のすべてのスパンをリンクします

span_id

一意のスパン識別子

特定のスパンから終了までを識別します

parent_id

親スパンの ID

スパン階層を作成します

はじめ​

クライアントの初期化​

Python
from mlflow import MlflowClient

# Initialize client with default tracking URI
client = MlflowClient()

# Or specify a custom tracking URI
client = MlflowClient(tracking_uri="databricks")

トレースの開始​

高レベルのAPIsとは異なり、 client.start_trace()を使用してスパンを追加する前に、明示的にトレースを開始する必要があります。

Python
# Start a new trace - this creates the root span
root_span = client.start_trace(
    name="my_application_flow",
    inputs={"user_id": "123", "action": "generate_report"},
    attributes={"environment": "production", "version": "1.0.0"}
)

# Extract the request_id for subsequent operations
request_id = root_span.request_id
print(f"Started trace with ID: {request_id}")

子スパンの追加​

アプリケーションのワークフローを表す client.start_span() を使用してスパンの階層を作成します。

Python
# Create a child span for data retrieval
data_span = client.start_span(
    name="fetch_user_data",
    request_id=request_id,  # Links to the trace
    parent_id=root_span.span_id,  # Creates parent-child relationship
    inputs={"user_id": "123"},
    attributes={"database": "users_db", "query_type": "select"}
)

# Create a sibling span for processing
process_span = client.start_span(
    name="process_data",
    request_id=request_id,
    parent_id=root_span.span_id,  # Same parent as data_span
    inputs={"data_size": "1024KB"},
    attributes={"processor": "gpu", "batch_size": 32}
)

エンディングスパン​

作成の逆の順序で client.end_span() を使用する終了スパン(LIFO - 後入れ先出し):

Python
# End the data retrieval span
client.end_span(
    request_id=data_span.request_id,
    span_id=data_span.span_id,
    outputs={"record_count": 42, "cache_hit": True},
    attributes={"duration_ms": 150}
)

# End the processing span
client.end_span(
    request_id=process_span.request_id,
    span_id=process_span.span_id,
    outputs={"processed_records": 42, "errors": 0},
    status="OK"
)

トレースの終了​

client.end_trace()を使用してルートスパンを終了してトレースを完了します。

Python
# End the root span (completes the trace)
client.end_trace(
    request_id=request_id,
    outputs={"report_url": "https://example.com/report/123"},
    attributes={"total_duration_ms": 1250, "status": "success"}
)

実用例​

例 1: エラー処理​

適切なエラー処理により、例外が発生した場合でもトレースが完了します。

Python
def traced_operation():
    client = MlflowClient()
    root_span = None

    try:
        # Start trace
        root_span = client.start_trace("risky_operation")

        # Start child span
        child_span = client.start_span(
            name="database_query",
            request_id=root_span.request_id,
            parent_id=root_span.span_id
        )

        try:
            # Risky operation
            result = perform_database_query()

            # End child span on success
            client.end_span(
                request_id=child_span.request_id,
                span_id=child_span.span_id,
                outputs={"result": result},
                status="OK"
            )
        except Exception as e:
            # End child span on error
            client.end_span(
                request_id=child_span.request_id,
                span_id=child_span.span_id,
                status="ERROR",
                attributes={"error": str(e)}
            )
            raise

    except Exception as e:
        # Log error to trace
        if root_span:
            client.end_trace(
                request_id=root_span.request_id,
                status="ERROR",
                attributes={"error_type": type(e).__name__, "error_message": str(e)}
            )
        raise
    else:
        # End trace on success
        client.end_trace(
            request_id=root_span.request_id,
            outputs={"status": "completed"},
            status="OK"
        )

例 2: カスタム・トレース管理​

既存のシステムと統合するためのカスタムトレースIDの生成と管理を実装します。

Python
import uuid
from datetime import datetime

class CustomTraceManager:
    """Custom trace manager with business-specific trace IDs"""

    def __init__(self):
        self.client = MlflowClient()
        self.active_traces = {}

    def generate_trace_id(self, user_id: str, operation: str) -> str:
        """Generate custom trace ID based on business logic"""
        timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
        return f"{user_id}_{operation}_{timestamp}_{uuid.uuid4().hex[:8]}"

    def start_custom_trace(self, user_id: str, operation: str, **kwargs):
        """Start trace with custom ID format"""
        trace_name = self.generate_trace_id(user_id, operation)

        root_span = self.client.start_trace(
            name=trace_name,
            attributes={
                "user_id": user_id,
                "operation": operation,
                "custom_trace_id": trace_name,
                **kwargs
            }
        )

        self.active_traces[trace_name] = root_span
        return root_span

    def get_active_trace(self, trace_name: str):
        """Retrieve active trace by custom name"""
        return self.active_traces.get(trace_name)

# Usage
manager = CustomTraceManager()
trace = manager.start_custom_trace(
    user_id="user123",
    operation="report_generation",
    report_type="quarterly"
)

例 3: ネストされたスパンを使用したバッチ処理​

複数のレベルのネストで複雑なワークフローを追跡します。

Python
def batch_processor(items):
    client = MlflowClient()

    # Start main trace
    root = client.start_trace(
        name="batch_processing",
        inputs={"batch_size": len(items)}
    )

    results = []

    # Process each item
    for i, item in enumerate(items):
        # Create span for each item
        item_span = client.start_span(
            name=f"process_item_{i}",
            request_id=root.request_id,
            parent_id=root.span_id,
            inputs={"item_id": item["id"]}
        )

        try:
            # Validation span
            validation_span = client.start_span(
                name="validate",
                request_id=root.request_id,
                parent_id=item_span.span_id
            )

            is_valid = validate_item(item)

            client.end_span(
                request_id=validation_span.request_id,
                span_id=validation_span.span_id,
                outputs={"is_valid": is_valid}
            )

            if is_valid:
                # Processing span
                process_span = client.start_span(
                    name="transform",
                    request_id=root.request_id,
                    parent_id=item_span.span_id
                )

                result = transform_item(item)
                results.append(result)

                client.end_span(
                    request_id=process_span.request_id,
                    span_id=process_span.span_id,
                    outputs={"transformed": result}
                )

            # End item span
            client.end_span(
                request_id=item_span.request_id,
                span_id=item_span.span_id,
                status="OK"
            )

        except Exception as e:
            # Handle errors gracefully
            client.end_span(
                request_id=item_span.request_id,
                span_id=item_span.span_id,
                status="ERROR",
                attributes={"error": str(e)}
            )

    # End main trace
    client.end_trace(
        request_id=root.request_id,
        outputs={
            "processed_count": len(results),
            "success_rate": len(results) / len(items)
        }
    )

    return results

ベストプラクティス​

1.安全のためにコンテキストマネージャーを使用する​

カスタムコンテキストマネージャーを作成して、スパンが常に閉じられるようにします。

Python
from contextlib import contextmanager

@contextmanager
def traced_span(client, name, request_id, parent_id=None, **kwargs):
    """Context manager for safe span management"""
    span = client.start_span(
        name=name,
        request_id=request_id,
        parent_id=parent_id,
        **kwargs
    )
    try:
        yield span
    except Exception as e:
        client.end_span(
            request_id=span.request_id,
            span_id=span.span_id,
            status="ERROR",
            attributes={"error": str(e)}
        )
        raise
    else:
        client.end_span(
            request_id=span.request_id,
            span_id=span.span_id,
            status="OK"
        )

# Usage
with traced_span(client, "my_operation", request_id, parent_id) as span:
    # Your code here
    result = perform_operation()

2. トレース状態管理の実装​

複雑なアプリケーションのトレース状態を管理します。

Python
class TraceStateManager:
    """Manage trace state across application components"""

    def __init__(self):
        self.client = MlflowClient()
        self._trace_stack = []

    @property
    def current_trace(self):
        """Get current active trace"""
        return self._trace_stack[-1] if self._trace_stack else None

    def push_trace(self, name: str, **kwargs):
        """Start a new trace and push to stack"""
        if self.current_trace:
            # Create child span if trace exists
            span = self.client.start_span(
                name=name,
                request_id=self.current_trace.request_id,
                parent_id=self.current_trace.span_id,
                **kwargs
            )
        else:
            # Create new trace
            span = self.client.start_trace(name=name, **kwargs)

        self._trace_stack.append(span)
        return span

    def pop_trace(self, **kwargs):
        """End current trace and pop from stack"""
        if not self._trace_stack:
            return

        span = self._trace_stack.pop()

        if self._trace_stack:
            # End child span
            self.client.end_span(
                request_id=span.request_id,
                span_id=span.span_id,
                **kwargs
            )
        else:
            # End root trace
            self.client.end_trace(
                request_id=span.request_id,
                **kwargs
            )

3. 意味のある属性を追加する​

デバッグを支援するコンテキストでトレースを充実させます。

Python
# Good: Specific, actionable attributes
client.start_span(
    name="llm_call",
    request_id=request_id,
    parent_id=parent_id,
    attributes={
        "model": "gpt-4",
        "temperature": 0.7,
        "max_tokens": 1000,
        "prompt_template": "rag_v2",
        "user_tier": "premium"
    }
)

# Bad: Generic, unhelpful attributes
client.start_span(
    name="process",
    request_id=request_id,
    parent_id=parent_id,
    attributes={"step": 1, "data": "some data"}
)

よくある落とし穴​

警告

次のよくある間違いを避けてください。

  1. スパンの終了を忘れる - 常に try/finally またはコンテキスト マネージャーを使用してください
  2. 親子関係が正しくない - スパン ID を再確認する
  3. 高レベルと低レベルのAPIの混在 - 相互運用できない
  4. トレース ID のハードコーディング - 常に一意の ID を生成します
  5. スレッドセーフを無視する - クライアント API はデフォルトによってスレッドセーフではありません

クライアントAPIを使用する場合​

Client APIs は、次の場合に使用します。

  • カスタムトレース ID 生成スキーム
  • 既存のトレースシステムとの統合
  • 複雑なトレースライフサイクル管理
  • 高度なスパン階層
  • カスタム・トレース・ステート管理

次の場合にクライアント API を避ける

  • 単純な関数トレース( @mlflow.traceを使用)
  • ローカル Python アプリケーション (コンテキスト マネージャーを使用)
  • クイックプロトタイピング(高レベル APIを使用)
  • オートトレースとの統合

次のステップ​

これらの推奨アクションとチュートリアルで旅を続けてください。

リファレンスガイド​

このガイドで説明されている概念と機能の詳細なドキュメントをご覧ください。

に最終更新