Observability & Resilience API

Observability

maticlib.observability.trace.StepTrace

Bases: BaseModel

Captures observability data for a single pipeline step.

Attributes:

Name	Type	Description
step_id	str	Unique identifier for the step.
step_name	str	Human-readable name of the step.
start_time	float	Unix timestamp when the step started.
end_time	Optional[float]	Unix timestamp when the step ended (or None if still running).
inputs	Dict[str, Any]	Arbitrary input data associated with the step.
outputs	Dict[str, Any]	Arbitrary output data associated with the step.
error	Optional[str]	Error message if the step failed.
prompt_tokens	int	Number of input/prompt tokens consumed.
completion_tokens	int	Number of completion tokens generated.
total_tokens	int	Total tokens used (prompt + completion).
model_name	Optional[str]	LLM or embedding model name used in this step.

maticlib.observability.trace.PipelineTrace

Bases: BaseModel

Container that aggregates StepTrace records for a full pipeline execution.

Attributes:

Name	Type	Description
trace_id	str	Unique identifier for this pipeline trace.
pipeline_name	str	Human-readable name for the pipeline.
start_time	float	Unix timestamp when the pipeline started.
end_time	Optional[float]	Unix timestamp when the pipeline ended (or None if still running).
steps	List[StepTrace]	Ordered list of StepTrace records.
metadata	Dict[str, Any]	Arbitrary metadata associated with this pipeline run.

add_step

add_step(step)

Appends a completed StepTrace to the pipeline trace.

Parameters:

Name	Type	Description	Default
step	StepTrace	A finished StepTrace to record.	required

Source code in maticlib/observability/trace.py

def add_step(self, step: StepTrace):
    """
    Appends a completed StepTrace to the pipeline trace.

    Args:
        step: A finished StepTrace to record.
    """
    self.steps.append(step)

maticlib.observability.callbacks.BaseCallbackHandler

Bases: ABC

Abstract base class for pipeline event callbacks.

on_pipeline_end abstractmethod

on_pipeline_end(trace)

Called when a pipeline completes.

Parameters:

Name	Type	Description	Default
trace	PipelineTrace	The finished PipelineTrace.	required

Source code in maticlib/observability/callbacks.py

@abstractmethod
def on_pipeline_end(self, trace: PipelineTrace) -> None:
    """
    Called when a pipeline completes.

    Args:
        trace: The finished PipelineTrace.
    """
    pass

on_pipeline_start abstractmethod

on_pipeline_start(trace)

Called when a pipeline starts.

Parameters:

Name	Type	Description	Default
trace	PipelineTrace	The PipelineTrace for the starting pipeline.	required

Source code in maticlib/observability/callbacks.py

@abstractmethod
def on_pipeline_start(self, trace: PipelineTrace) -> None:
    """
    Called when a pipeline starts.

    Args:
        trace: The PipelineTrace for the starting pipeline.
    """
    pass

on_step_end abstractmethod

on_step_end(step)

Called when a pipeline step ends.

Parameters:

Name	Type	Description	Default
step	StepTrace	The completed StepTrace.	required

Source code in maticlib/observability/callbacks.py

@abstractmethod
def on_step_end(self, step: StepTrace) -> None:
    """
    Called when a pipeline step ends.

    Args:
        step: The completed StepTrace.
    """
    pass

on_step_start abstractmethod

on_step_start(step)

Called when a pipeline step begins.

Parameters:

Name	Type	Description	Default
step	StepTrace	The StepTrace for the starting step.	required

Source code in maticlib/observability/callbacks.py

@abstractmethod
def on_step_start(self, step: StepTrace) -> None:
    """
    Called when a pipeline step begins.

    Args:
        step: The StepTrace for the starting step.
    """
    pass

maticlib.observability.callbacks.LoggingCallbackHandler

LoggingCallbackHandler(logger=None)

Bases: BaseCallbackHandler

A callback handler that logs pipeline and step events via Python's logging module.

Initializes the LoggingCallbackHandler.

Parameters:

Name	Type	Description	Default
logger	Logger	An optional Python Logger. Defaults to the module logger.	None

Source code in maticlib/observability/callbacks.py

def __init__(self, logger: logging.Logger = None):
    """
    Initializes the LoggingCallbackHandler.

    Args:
        logger: An optional Python Logger. Defaults to the module logger.
    """
    self.logger = logger or logging.getLogger(__name__)

Resilience

maticlib.resilience.retry.RetryPolicy

RetryPolicy(
    max_retries=3,
    backoff_factor=2.0,
    initial_delay=1.0,
    exceptions=(Exception,),
    logger=None,
)

Configurable retry policy with exponential backoff.

Initializes the RetryPolicy.

Parameters:

Name	Type	Description	Default
max_retries	int	Maximum number of retry attempts before raising.	3
backoff_factor	float	Multiplier applied to the delay after each failed attempt.	2.0
initial_delay	float	Initial delay in seconds before the first retry.	1.0
exceptions	Tuple[Type[Exception], ...]	Tuple of exception types to catch and retry on.	(Exception,)
logger	Logger	Optional Python Logger. Defaults to the module logger.	None

Source code in maticlib/resilience/retry.py

def __init__(
    self,
    max_retries: int = 3,
    backoff_factor: float = 2.0,
    initial_delay: float = 1.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,),
    logger: logging.Logger = None,
):
    """
    Initializes the RetryPolicy.

    Args:
        max_retries: Maximum number of retry attempts before raising.
        backoff_factor: Multiplier applied to the delay after each failed attempt.
        initial_delay: Initial delay in seconds before the first retry.
        exceptions: Tuple of exception types to catch and retry on.
        logger: Optional Python Logger. Defaults to the module logger.
    """
    self.max_retries = max_retries
    self.backoff_factor = backoff_factor
    self.initial_delay = initial_delay
    self.exceptions = exceptions
    self.logger = logger or logging.getLogger(__name__)

execute

execute(func, *args, **kwargs)

Executes a callable with the retry policy applied.

Parameters:

Name	Type	Description	Default
func	Callable[..., Any]	The callable function to execute with retries.	required
*args	Any	Positional arguments forwarded to func.	()
**kwargs	Any	Keyword arguments forwarded to func.	{}

Returns:

Type	Description
Any	The return value of func on success.

Raises:

Type	Description
RetryExhaustedError	When all retry attempts have failed.

Source code in maticlib/resilience/retry.py

def execute(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
    """
    Executes a callable with the retry policy applied.

    Args:
        func: The callable function to execute with retries.
        *args: Positional arguments forwarded to ``func``.
        **kwargs: Keyword arguments forwarded to ``func``.

    Returns:
        The return value of ``func`` on success.

    Raises:
        RetryExhaustedError: When all retry attempts have failed.
    """
    delay = self.initial_delay
    last_exception = None

    for attempt in range(1, self.max_retries + 1):
        try:
            return func(*args, **kwargs)
        except self.exceptions as e:
            last_exception = e
            if attempt == self.max_retries:
                self.logger.error(
                    f"Attempt {attempt}/{self.max_retries} failed for {func.__name__}. Exhausted."
                )
                break

            self.logger.warning(
                f"Attempt {attempt}/{self.max_retries} failed for {func.__name__}. Retrying in {delay}s..."
            )
            time.sleep(delay)
            delay *= self.backoff_factor

    raise RetryExhaustedError(
        f"Function {func.__name__} failed after {self.max_retries} attempts. Last error: {last_exception}"
    ) from last_exception

maticlib.resilience.retry.with_retry

with_retry(
    max_retries=3,
    backoff_factor=2.0,
    initial_delay=1.0,
    exceptions=(Exception,),
)

Decorator to apply a retry policy to any function.

Parameters:

Name	Type	Description	Default
max_retries	int	Maximum number of retry attempts.	3
backoff_factor	float	Multiplier applied to the delay after each failure.	2.0
initial_delay	float	Initial delay in seconds before the first retry.	1.0
exceptions	Tuple[Type[Exception], ...]	Tuple of exception types to catch and retry on.	(Exception,)

Returns:

Type	Description
Callable[[Callable[..., Any]], Callable[..., Any]]	A decorator that wraps the target function with the retry policy.

Source code in maticlib/resilience/retry.py

def with_retry(
    max_retries: int = 3,
    backoff_factor: float = 2.0,
    initial_delay: float = 1.0,
    exceptions: Tuple[Type[Exception], ...] = (Exception,),
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """
    Decorator to apply a retry policy to any function.

    Args:
        max_retries: Maximum number of retry attempts.
        backoff_factor: Multiplier applied to the delay after each failure.
        initial_delay: Initial delay in seconds before the first retry.
        exceptions: Tuple of exception types to catch and retry on.

    Returns:
        A decorator that wraps the target function with the retry policy.
    """

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        policy = RetryPolicy(
            max_retries=max_retries,
            backoff_factor=backoff_factor,
            initial_delay=initial_delay,
            exceptions=exceptions,
        )

        def wrapper(*args: Any, **kwargs: Any) -> Any:
            return policy.execute(func, *args, **kwargs)

        return wrapper

    return decorator