Function validate_trace_spans

agentops/validation.py:209–359

Validate that spans have been sent to AgentOps. Args: trace_id: Direct trace ID to validate trace_context: TraceContext object from start_trace (alternative to trace_id) max_retries: Maximum number of retries to wait for spans to appear retry_delay: Delay be

(
    trace_id: Optional[str] = None,
    trace_context: Optional[Any] = None,
    max_retries: int = 10,
    retry_delay: float = 1.0,
    check_llm: bool = True,
    min_spans: int = 1,
    api_key: Optional[str] = None,
)

def validate_trace_spans(
    trace_id: Optional[str] = None,
    trace_context: Optional[Any] = None,
    max_retries: int = 10,
    retry_delay: float = 1.0,
    check_llm: bool = True,
    min_spans: int = 1,
    api_key: Optional[str] = None,
) -> Dict[str, Any]:
    """
    Validate that spans have been sent to AgentOps.

    Args:
        trace_id: Direct trace ID to validate
        trace_context: TraceContext object from start_trace (alternative to trace_id)
        max_retries: Maximum number of retries to wait for spans to appear
        retry_delay: Delay between retries in seconds
        check_llm: Whether to specifically check for LLM spans
        min_spans: Minimum number of spans expected
        api_key: Optional API key (uses environment variable if not provided)

    Returns:
        Dictionary containing validation results and metrics

    Raises:
        ValidationError: If validation fails
        ValueError: If neither trace_id nor trace_context is provided
    """
    # Extract trace ID
    if trace_id is None and trace_context is None:
        # Try to get from current span
        try:
            from opentelemetry.trace import get_current_span

            current_span = get_current_span()
            if current_span and hasattr(current_span, "get_span_context"):
                span_context = current_span.get_span_context()
                if hasattr(span_context, "trace_id") and span_context.trace_id:
                    if isinstance(span_context.trace_id, int):
                        trace_id = format(span_context.trace_id, "032x")
                    else:
                        trace_id = str(span_context.trace_id)
        except ImportError:
            pass

    elif trace_context is not None and trace_id is None:
        # Extract from TraceContext
        if hasattr(trace_context, "span") and trace_context.span:
            span_context = trace_context.span.get_span_context()
            if hasattr(span_context, "trace_id"):
                if isinstance(span_context.trace_id, int):
                    trace_id = format(span_context.trace_id, "032x")
                else:
                    trace_id = str(span_context.trace_id)

    if trace_id is None:
        raise ValueError("No trace ID found. Provide either trace_id or trace_context parameter.")