Tracing

The tracing module provides a protocol-based interface for distributed tracing integration. Implement the Tracer protocol to connect FastroAI with your observability platform, or use one of the built-in tracers.

Tracer

fastroai.tracing.Tracer

Protocol for distributed tracing implementations.

Implement this protocol to integrate FastroAI with your preferred observability platform (OpenTelemetry, Datadog, etc.).

FastroAI provides built-in implementations: - LogfireTracer: For Pydantic's Logfire platform - SimpleTracer: For logging-based tracing - NoOpTracer: For disabled tracing

Examples:

Using the built-in LogfireTracer:

from fastroai import FastroAgent, LogfireTracer

tracer = LogfireTracer()
agent = FastroAgent(model="openai:gpt-4o")
response = await agent.run("Hello!", tracer=tracer)

Custom implementation for OpenTelemetry:

from opentelemetry import trace as otel_trace

class OTelTracer:
    def __init__(self):
        self.tracer = otel_trace.get_tracer("fastroai")

    @asynccontextmanager
    async def span(self, name: str, **attrs):
        trace_id = str(uuid.uuid4())
        with self.tracer.start_as_current_span(name) as span:
            for key, value in attrs.items():
                span.set_attribute(key, value)
            yield trace_id

    def log_metric(self, trace_id: str, name: str, value):
        span = otel_trace.get_current_span()
        span.set_attribute(f"metric.{name}", value)

    def log_error(self, trace_id: str, error: Exception, context=None):
        span = otel_trace.get_current_span()
        span.record_exception(error)

log_error(trace_id, error, context=None)

Log an error associated with a trace.

Parameters:

Name	Type	Description	Default
trace_id	str	Trace ID to associate the error with.	required
error	Exception	The exception that occurred.	required
context	dict[str, Any] | None	Additional error context.	None

log_metric(trace_id, name, value)

Log a metric associated with a trace.

Parameters:

Name	Type	Description	Default
trace_id	str	Trace ID to associate the metric with.	required
name	str	Metric name.	required
value	Any	Metric value.	required

span(name, **attributes)

Create a traced span for an operation.

Parameters:

Name	Type	Description	Default
name	str	Name of the operation being traced.	required
**attributes	Any	Additional context to attach to the span.	{}

Returns:

Type	Description
AbstractAsyncContextManager[str]	Async context manager that yields a unique trace ID.

SimpleTracer

fastroai.tracing.SimpleTracer

Basic tracer implementation using Python's logging module.

Provides simple tracing functionality for development and debugging. For production use, consider implementing a Tracer for your observability platform.

Examples:

tracer = SimpleTracer()

async with tracer.span("my_operation", user_id="123") as trace_id:
    # Your operation here
    result = await do_something()
    tracer.log_metric(trace_id, "result_size", len(result))

# Logs:
# INFO [abc12345] Starting my_operation
# INFO [abc12345] Metric result_size=42
# INFO [abc12345] Completed my_operation in 0.123s

__init__(logger=None)

Initialize SimpleTracer.

Parameters:

Name	Type	Description	Default
logger	Logger | None	Logger to use. Defaults to 'fastroai.tracing'.	None

log_error(trace_id, error, context=None)

Log an error with trace correlation.

Parameters:

Name	Type	Description	Default
trace_id	str	Trace ID for correlation.	required
error	Exception	The exception that occurred.	required
context	dict[str, Any] | None	Additional error context.	None

log_metric(trace_id, name, value)

Log a metric with trace correlation.

Parameters:

Name	Type	Description	Default
trace_id	str	Trace ID for correlation.	required
name	str	Metric name.	required
value	Any	Metric value.	required

span(name, **attributes) async

Create a traced span with timing.

Parameters:

Name	Type	Description	Default
name	str	Name of the operation.	required
**attributes	Any	Additional context logged with the span.	{}

Yields:

Type	Description
AsyncIterator[str]	Unique trace ID (first 8 chars shown in logs for readability).

LogfireTracer

fastroai.tracing.LogfireTracer

Tracer implementation for Pydantic's Logfire observability platform.

Integrates FastroAI with Logfire for production-grade observability, including distributed tracing, metrics, and error tracking. Requires the logfire package to be installed.

Note

Install logfire with: pip install logfire Configure logfire before use: logfire.configure()

Examples:

import logfire
from fastroai import FastroAgent, LogfireTracer

# Configure logfire (typically done once at startup)
logfire.configure()

tracer = LogfireTracer()
agent = FastroAgent(model="openai:gpt-4o")
response = await agent.run("Hello!", tracer=tracer)

# View traces in Logfire dashboard at https://logfire.pydantic.dev

With pipelines:

from fastroai import Pipeline, LogfireTracer

tracer = LogfireTracer()
result = await pipeline.execute(
    {"document": doc},
    deps=my_deps,
    tracer=tracer,
)

__init__()

Initialize LogfireTracer.

Raises:

Type	Description
ImportError	If the logfire package is not installed.

log_error(trace_id, error, context=None)

Log an error to Logfire with trace correlation.

Records the error with full exception information for debugging in the Logfire dashboard.

Parameters:

Name	Type	Description	Default
trace_id	str	Trace ID for correlation.	required
error	Exception	The exception that occurred.	required
context	dict[str, Any] | None	Additional error context (e.g., step_id, operation).	None

Examples:

try:
    result = await risky_operation()
except Exception as e:
    tracer.log_error(trace_id, e, {"step": "data_processing"})
    raise

log_metric(trace_id, name, value)

Log a metric to Logfire with trace correlation.

Metrics are logged as info-level spans with the metric name and value as attributes, allowing them to be queried and visualized in Logfire.

Parameters:

Name	Type	Description	Default
trace_id	str	Trace ID for correlation.	required
name	str	Metric name (e.g., "input_tokens", "cost_microcents").	required
value	Any	Metric value.	required

Examples:

tracer.log_metric(trace_id, "input_tokens", 150)
tracer.log_metric(trace_id, "cost_microcents", 2500)

span(name, **attributes) async

Create a traced span using Logfire.

Wraps Logfire's span context manager and generates a unique trace ID for correlation across FastroAI operations.

Parameters:

Name	Type	Description	Default
name	str	Name of the operation being traced.	required
**attributes	Any	Additional context to attach to the span. These appear as attributes in the Logfire dashboard.	{}

Yields:

Type	Description
AsyncIterator[str]	Unique trace ID for correlating metrics and errors.

Examples:

async with tracer.span("my_operation", user_id="123") as trace_id:
    result = await do_something()
    tracer.log_metric(trace_id, "result_size", len(result))

NoOpTracer

fastroai.tracing.NoOpTracer

Tracer that does nothing. Use when tracing is disabled.

This tracer satisfies the Tracer protocol but performs no operations, making it suitable for testing or when tracing overhead is undesirable.

Examples:

tracer = NoOpTracer()

async with tracer.span("operation") as trace_id:
    # trace_id is still generated for compatibility
    result = await do_something()

log_error(trace_id, error, context=None)

No-op error logging.

Parameters:

Name	Type	Description	Default
trace_id	str	Ignored.	required
error	Exception	Ignored.	required
context	dict[str, Any] | None	Ignored.	None

log_metric(trace_id, name, value)

No-op metric logging.

Parameters:

Name	Type	Description	Default
trace_id	str	Ignored.	required
name	str	Ignored.	required
value	Any	Ignored.	required

span(name, **attributes) async

Create a no-op span that just yields a trace ID.

Parameters:

Name	Type	Description	Default
name	str	Ignored.	required
**attributes	Any	Ignored.	{}

Yields:

Type	Description
AsyncIterator[str]	Unique trace ID (still generated for compatibility).

---

← Usage API Overview →