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

As APIs MlflowClient APIs oferecem controle direto e detalhado sobre o gerenciamento do ciclo de vida das rastreabilidades. Embora as APIs de alto nível lidem com a maioria dos casos de uso de maneira elegante, as APIs de cliente são essenciais para cenários avançados que exigem controle explícito sobre a criação de rastreamentos, IDs de rastreamento 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 rastreamento de alto nível ( APIs), é necessário iniciar explicitamente o rastreamento antes de adicionar spans usando 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}")

Adicionando períodos infantis

Crie uma hierarquia de intervalos utilizando client.start_span() para representar o fluxo de trabalho da sua aplicação:

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

Encerre os períodos usando client.end_span() na ordem inversa da criação (LIFO - Último a entrar, primeiro a sair):

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

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:

Esquecendo de encerrar os períodos - sempre use try/finally ou gerenciadores de contexto
Relacionamentos incorretos entre pais e filhos - Verifique novamente os IDs de intervalo
Mistura de APIs de alto e baixo nível - Elas não são interoperáveis
IDs de rastreamento codificados - Sempre gere IDs exclusivos
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.

Depurar & observar seu aplicativo - Analisar traços criados com APIs de clientes
Consultar rastreamentos via SDK - Acesse programaticamente seus dados rastreados
APIs de alto nível - Alternativa mais simples para a maioria dos casos de uso

Guia de referência

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

Modelo de dados de rastreamento - Mergulhe profundamente na estrutura de rastreamento e abrangência
Conceitos de rastreamento - Entenda os fundamentos do rastreamento distribuído
FAQ - Perguntas comuns sobre APIs de baixo nível

Conceitos principais​

Ciclo de vida do Trace​

Identificadores-chave​

Como começar​

Inicializar o cliente​

Iniciando um rastreamento​

Adicionando períodos infantis​

Expansões finais​

Terminando um rastro​

Exemplos práticos​

Exemplo 1: Tratamento de erros​

Exemplo 2: Gerenciamento de rastreamento personalizado​

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

Melhores práticas​

1. Use gerenciadores de contexto para segurança​

2. Implemente o gerenciamento do Trace State​

3. Adicionar atributos significativos​

Armadilhas comuns​

Quando usar APIs de cliente​

Próximas etapas​

Guia de referência​