低级别客户端 API （高级）

2025-06-11

MLflow 客户端 API 提供对跟踪生命周期管理的直接精细控制。尽管高级 API 处理大多数用例很优雅，但客户端 API 对于需要显式控制跟踪创建、自定义跟踪 ID 或与现有可观测性系统的集成的高级方案至关重要。

警告

开始之前：建议仅在高级 API 不满足要求时使用客户端 API：

无自动父子关系检测
需要手动处理异常
与自动跟踪集成不兼容
完全控制跟踪生命周期
自定义追踪 ID 管理
与现有系统的集成

核心概念

跟踪生命周期过程

每个踪迹都遵循必须显式管理的严格生命周期：

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]

重要

黄金规则：每个 start_trace 或 start_span 呼叫都必须有相应的 end_trace 或 end_span 调用。未能关闭区段将导致跟踪不完整。

密钥标识符

了解这些标识符对于客户端 API 使用情况至关重要：

标识符	DESCRIPTION	用法
`request_id`	唯一跟踪标识符	链接跟踪中的所有区段
`span_id`	唯一范围标识符	确定特定范围直至结束
`parent_id`	父跨度 ID	创建跨度层次结构

入门指南

初始化客户端

from mlflow import MlflowClient

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

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

启动跟踪

与高级 API 不同，您必须在添加跨度之前显式启动跟踪。

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

添加子跨度

创建范围层次结构来表示应用程序的工作流：

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

终止范围

以相反顺序创建结束范围（LIFO - 最后一次传入，先出）：

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

结束跟踪

通过结束根跨度来完成追踪：

# 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：错误处理

正确的错误处理可确保即使发生异常，跟踪也已完成：

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 生成和管理，以便与现有系统集成：

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：使用嵌套跨度进行批处理

跟踪包含多个嵌套级别的复杂工作流：

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. 将上下文管理器用于安全

创建自定义上下文管理器，以确保始终关闭范围：

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. 实现跟踪状态管理

管理复杂应用程序的跟踪状态：

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. 添加有意义的属性

使用有助于调试的上下文来丰富跟踪数据：

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

常见陷阱

警告

避免以下常见错误：

忘记结束段落——始终使用 try/finally 或上下文管理器
父子关系不正确 - 仔细检查区段 ID
混合高级和低级别 API - 它们不会互作
硬编码跟踪 ID - 始终生成一个唯一 ID
忽略线程安全性 - 默认情况下客户端 API 不是线程安全的

何时使用客户端 API

将客户端 API 用于：

自定义跟踪 ID 生成方案
与现有跟踪系统的集成
复杂的跟踪生命周期管理
高级跨度层次结构
自定义跟踪状态管理

避免使用客户端 API 进行以下操作：

简单函数跟踪（使用 @mlflow.trace）
本地 Python 应用程序（使用上下文管理器）
快速原型制作（使用高级 API）
与自动跟踪集成

后续步骤

继续您的旅程，并参考这些推荐的行动和教程。

调试并观察应用 - 分析使用客户端 API 创建的跟踪
通过 SDK 查询跟踪 - 以编程方式访问跟踪数据
High-Level API - 大多数用例的更简单替代方法

参考指南

浏览本指南中提到的概念和功能的详细文档。

跟踪数据模型 - 深入了解跟踪和跨度结构
跟踪概念 - 了解分布式跟踪基础知识
常见问题解答 - 有关低级别 API 的常见问题