Overview
The Hub Tracing module provides decorator-based tracing utilities for instrumenting guard calls, validator inference, and validator usage with OpenTelemetry spans in the Guardrails Hub telemetry pipeline.
Description
This module defines a set of decorators and helper functions that wrap synchronous, asynchronous, streaming, and async-streaming functions with OpenTelemetry tracing spans. The core decorators (trace, async_trace, trace_stream, async_trace_stream) create spans via the HubTelemetry singleton, inject context propagation headers, and attach rich attributes describing the traced operation. Helper functions extract domain-specific attributes for guard calls (e.g., guard ID, LLM API, output type), validator inference (e.g., whether remote or local inference was used), and validator usage (e.g., validator name, result outcome). When hub telemetry is disabled, the decorators fall through to the original function without overhead.
Usage
Use this module when you need to instrument Guardrails guard calls, validator inference endpoints, or validator usage with telemetry spans that are reported to the Guardrails Hub. The decorators are applied to methods on Guard, AsyncGuard, and validator service classes to automatically capture timing and metadata for each operation.
Code Reference
Source Location
- Repository: Guardrails
- File:
guardrails/hub_telemetry/hub_tracing.py
Signature
def get_guard_call_attributes(
attrs: Dict[str, Any], origin: str, *args, **kwargs
) -> Dict[str, Any]
def get_validator_inference_attributes(
attrs: Dict[str, Any], *args, **kwargs
) -> Dict[str, Any]
def get_validator_usage_attributes(
attrs: Dict[str, Any], response, *args, **kwargs
) -> Dict[str, Any]
def add_attributes(
span: Span,
attrs: Dict[str, Any],
name: str,
origin: str,
*args,
response=None,
**kwargs,
)
def trace(*, name: str, origin: Optional[str] = None, **attrs)
def async_trace(*, name: str, origin: Optional[str] = None)
def trace_stream(*, name: str, origin: Optional[str] = None, **attrs)
def async_trace_stream(*, name: str, origin: Optional[str] = None, **attrs)
Import
from guardrails.hub_telemetry.hub_tracing import trace, async_trace, trace_stream, async_trace_stream
from guardrails.hub_telemetry.hub_tracing import add_attributes
from guardrails.hub_telemetry.hub_tracing import get_guard_call_attributes
from guardrails.hub_telemetry.hub_tracing import get_validator_inference_attributes
from guardrails.hub_telemetry.hub_tracing import get_validator_usage_attributes
I/O Contract
trace Decorator
| Parameter |
Type |
Description
|
name |
str |
The span name (e.g., /guard_call, /validator_usage)
|
origin |
Optional[str] |
Origin identifier; defaults to name if not provided
|
**attrs |
Any |
Additional key-value attributes to attach to the span
|
| Returns |
Type |
Description
|
| Decorated function |
Callable |
A wrapper that creates a span, invokes the original function, attaches attributes, and returns the result
|
get_guard_call_attributes
| Parameter |
Type |
Description
|
attrs |
Dict[str, Any] |
Mutable attribute dictionary to populate
|
origin |
str |
The origin string identifying the calling context
|
*args |
Any |
Positional arguments from the decorated function (first is the guard instance)
|
**kwargs |
Any |
Keyword arguments from the decorated function
|
| Returns |
Type |
Description
|
| Enriched attributes |
Dict[str, Any] |
Dictionary with keys: stream, guard_id, user_id, custom_reask_messages, output_type, llm_api
|
get_validator_inference_attributes
| Parameter |
Type |
Description
|
attrs |
Dict[str, Any] |
Mutable attribute dictionary to populate
|
*args |
Any |
Positional arguments (first is the validator instance)
|
| Returns |
Type |
Description
|
| Enriched attributes |
Dict[str, Any] |
Dictionary with keys: validator_name, used_remote_inference, used_local_inference, used_guardrails_endpoint, used_custom_endpoint
|
get_validator_usage_attributes
| Parameter |
Type |
Description
|
attrs |
Dict[str, Any] |
Mutable attribute dictionary to populate
|
response |
Any |
The response returned by the validated function
|
*args |
Any |
Positional arguments (second is the validator instance)
|
| Returns |
Type |
Description
|
| Enriched attributes |
Dict[str, Any] |
Dictionary with keys: validator_name, validator_on_fail, validator_result
|
Usage Examples
from guardrails.hub_telemetry.hub_tracing import trace, async_trace
# Synchronous tracing of a guard call
@trace(name="/guard_call", origin="Guard.__call__")
def my_guard_call(self, llm_api=None, **kwargs):
# Guard execution logic
return result
# Asynchronous tracing
@async_trace(name="/guard_call", origin="AsyncGuard.__call__")
async def my_async_guard_call(self, llm_api=None, **kwargs):
return await some_async_operation()
# Streaming trace
from guardrails.hub_telemetry.hub_tracing import trace_stream
@trace_stream(name="/guard_call", origin="Guard.__call__")
def my_streaming_guard_call(self, **kwargs):
for chunk in stream:
yield chunk
Related Pages
