Pular para o conteúdo principal

APIs de cliente de baixo nível (avançado)

As APIs do MLflow Client fornecem controle direto e refinado sobre o gerenciamento do ciclo de vida dos traços. Embora as APIs de alto nível lidem com a maioria dos casos de uso de forma elegante, as APIs do cliente são essenciais para cenários avançados que exigem controle explícito sobre a criação de traços, IDs de traços personalizados ou integração com sistemas de observabilidade existentes.

atenção

Antes de começar : Recomendamos o uso de APIs de cliente somente quando as APIs de alto nível não atenderem às suas necessidades:

  • Sem detecção automática de relacionamento entre pais e filhos
  • É necessário o tratamento manual de exceções
  • Incompatível com integrações de rastreamento automático
  • Controle total sobre o ciclo de vida do rastreamento
  • Gerenciamento de ID de rastreamento personalizado
  • Integração com sistemas existentes

Conceitos principais​

Ciclo de vida do Trace​

Cada rastreamento segue um ciclo de vida rigoroso que deve ser gerenciado explicitamente:

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

Regra de ouro : cada chamada start_trace ou start_span deve ter uma chamada end_trace ou end_span correspondente. Não fechar os intervalos resultará em rastreamentos incompletos.

Identificadores-chave​

A compreensão desses identificadores é fundamental para o uso da API do cliente:

Identificador

Descrição

Uso

request_id

Identificador de rastreamento exclusivo

Vincula todas as extensões em um rastreamento

span_id

Identificador de extensão exclusivo

Identifica um intervalo específico de ponta a ponta

parent_id

ID do plano parental

Cria uma hierarquia de extensão

Como começar​

Inicializar o cliente​

Python
from mlflow import MlflowClient

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

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

Iniciando um rastreamento​

Ao contrário do APIs de alto nível, o senhor deve iniciar explicitamente um rastreamento antes de adicionar vãos:

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

Adicionando períodos infantis​

Crie uma hierarquia de vãos para representar o fluxo de trabalho do seu aplicativo:

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

Expansões finais​

Expansões finais na ordem inversa da criação (LIFO - Última entrada, primeira saída):

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

Terminando um rastro​

Complete o rastreamento terminando a extensão da raiz:

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

Exemplos práticos​

Exemplo 1: Tratamento de erros​

O tratamento adequado de erros garante que os rastreamentos sejam concluídos mesmo quando ocorrem exceções:

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

Exemplo 2: Gerenciamento de rastreamento personalizado​

Implemente a geração e o gerenciamento de IDs de rastreamento personalizados para integração com os sistemas existentes:

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

Exemplo 3: processamento de lotes com vãos aninhados​

Rastrear fluxos de trabalho complexos com vários níveis de aninhamento:

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

Melhores práticas​

1. Use gerenciadores de contexto para segurança​

Crie gerenciadores de contexto personalizados para garantir que os períodos estejam sempre fechados:

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. Implemente o gerenciamento do Trace State​

gerenciar o estado de rastreamento para aplicativos complexos:

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. Adicionar atributos significativos​

Enriqueça seus traços com contexto que ajuda na depuração:

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

Armadilhas comuns​

atenção

Evite esses erros comuns:

  1. Esquecendo de encerrar os períodos - sempre use try/finally ou gerenciadores de contexto
  2. Relacionamentos incorretos entre pais e filhos - Verifique novamente os IDs de intervalo
  3. Mistura de APIs de alto e baixo nível - Elas não são interoperáveis
  4. IDs de rastreamento codificados - Sempre gere IDs exclusivos
  5. Ignorando a segurança de thread - O cliente APIs não é thread-safe por default

Quando usar APIs de cliente​

Use as APIs do cliente para:

  • Esquemas personalizados de geração de ID de rastreamento
  • Integração com sistemas de rastreamento existentes
  • Gerenciamento complexo do ciclo de vida do rastreamento
  • Hierarquias avançadas de abrangência
  • Gerenciamento personalizado do estado de rastreamento

Evitar APIs de cliente para:

  • Rastreamento de função simples (use @mlflow.trace)
  • Aplicativos Python locais (use gerenciadores de contexto)
  • Prototipagem rápida (use APIs de alto nível)
  • Integração com rastreamento automático

Próximas etapas​

Continue sua jornada com estas ações recomendadas e o tutorial.

Guia de referência​

Explore a documentação detalhada dos conceitos e recursos mencionados neste guia.

Última atualização em