Aller au contenu principal
Dernière mise à jour le

APIs client de bas niveau (avancé)

Les MlflowClient APIs offrent un contrôle direct et précis sur la gestion du cycle de vie des traces. Bien que les APIs du décorateur de fonction gèrent élégamment la plupart des cas d'utilisation, les APIs clientes sont essentielles pour les scénarios avancés nécessitant un contrôle explicite sur la création de traces, les ID de traces personnalisés ou l'intégration avec les systèmes d'observabilité existants.

Quand utiliser les APIs client​

Utilisez les APIs clientes pour :

  • Schémas de génération d'ID de trace personnalisés
  • Intégration avec les systèmes de traçage existants
  • Gestion complexe du cycle de vie de la trace
  • Hiérarchies de spans avancées.
  • Gestion de l'état des traces personnalisées

Évitez les APIs clientes pour :

  • Traçage simple de fonction (utilisez @mlflow.trace)
  • Applications Python locales (utilisez des gestionnaires de contexte)
  • Prototypage rapide (utiliser des APIs de haut niveau)
  • Intégration avec traçage automatique

Concepts clés​

Cycle de vie du traçage​

Chaque trace suit un cycle de vie strict qui doit être géré explicitement :

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

Chaque appel start_trace ou start_span doit avoir un appel end_trace ou end_span correspondant. Le fait de ne pas fermer les spans entraînera des traces incomplètes.

Identificateurs clés​

Comprendre ces identifiants est crucial pour l'utilisation de l'API client :

Identifiant

Description

Utilisation

request_id

Identifiant unique de la trace

Link toutes les étendues dans une trace

span_id

Identifiant d'étendue unique

Identifie un span spécifique à terminer.

parent_id

ID de la portée parente

Crée une hiérarchie de portées.

Identifiant

Description

Utilisation

request_id

Identifiant unique de la trace

Link toutes les étendues dans une trace

span_id

Identifiant d'étendue unique

Identifie un span spécifique à terminer.

parent_id

ID de la portée parente

Crée une hiérarchie de portées.

Démarrer​

Initialisez le client​

Python
from mlflow import MlflowClient

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

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

start a trace​

Contrairement aux APIs de haut niveau, vous devez explicitement démarrer un suivi avant d'ajouter des spans en utilisant 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}")

Ajouter des portées enfants​

Créez une hiérarchie d'intervalles à l'aide de client.start_span() pour représenter le workflow de votre application :

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}
)

Fin des spans​

Terminez les étendues à l'aide de client.end_span() dans l'ordre inverse de création (LIFO - dernier entré, premier sorti) :

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"
)

Terminer une trace​

Terminez la trace en terminant l'étendue racine à l'aide de 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"}
)

Exemples pratiques​

Exemple 1 : Gestion des erreurs​

Une gestion appropriée des erreurs garantit que les traces sont complétées même en cas d'exception :

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"
        )

Exemple 2 : Gestion personnalisée des traces​

Implémenter la génération et la gestion personnalisées des identifiants de trace pour l'intégration avec les systèmes existants :

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"
)

Exemple 3 : Traitement par batch avec des portées imbriquées​

Suivre les workflows complexes avec plusieurs niveaux d'imbrication :

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

Bonnes pratiques​

Utilisez des gestionnaires de contexte pour la sécurité​

Créez des gestionnaires de contexte personnalisés pour garantir que les étendues sont toujours fermées :

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()

Implémenter la gestion de l'état de trace​

Gérer l'état de la trace pour les applications complexes :

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
            )

Ajouter des attributs pertinents​

Enrichissez vos traces avec un contexte qui facilite le debugging :

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"}
)

Écueils courants​

Évitez ces erreurs courantes :

  1. **Oublier de terminer les spans** – Toujours utiliser try/finally ou des gestionnaires de contexte
  2. Relations parent-enfant incorrectes — Vérifiez à nouveau les ID d'étendue
  3. Mélanger les APIs de haut niveau et de bas niveau – Ils ne sont pas interopérables
  4. **Identifiants de trace codés en dur** - Générez toujours des identifiants uniques
  5. Ignorer la sécurité des threads - Les API clientes ne sont pas thread-safe par default

Ressources supplémentaires​

Sur cette page