Skip to main content

Observability SDK

Tracing Class​

The Tracing class provides methods for interacting with the current span in a trace. It allows you to store additional data such as internals, references, metadata, and metrics.

get_current_span​

get_current_span()

Returns the current active span.

current_span = ag.tracing.get_current_span()
  • Returns: CustomSpan instance representing the current span.
  • Behavior: If a span is active and recording, it returns a CustomSpan object; otherwise, it returns None.

store_internals​

store_internals(attributes, span=None)

Stores internal data within the current span.

ag.tracing.store_internals(attributes, span=None)
  • Parameters:
    • attributes (Dict[str, Any]): Dictionary of internal data to store.
    • span (Optional[Span]): Specific span to store the data in; defaults to the current span.
  • Behavior: Adds the provided attributes under the "internals" namespace within the span's data. Internals are shown as a collapsible section in the overview section of the tracing drawer.

store_refs​

store_refs(refs, span=None)

Stores references to Agenta resources in the current span.

ag.tracing.store_refs(refs, span=None)

  • Parameters:

    • refs (Dict[str, str]): Dictionary of reference keys and their corresponding values.
    • span (Optional[Span]): Specific span to store the references in; defaults to the current span.

  • Behavior: Sets reference attributes in the span and updates the tracer's reference context.

  • Valid Reference Keys:

    • "application.id"
    • "application.slug"
    • "application.version"
    • "variant.id"
    • "variant.slug"
    • "variant.version"
    • "environment.id"
    • "environment.slug"
    • "environment.version"

store_meta​

store_meta(meta, span=None)

Stores metadata in the current span.

ag.tracing.store_meta(meta, span=None)
  • Parameters:
    • meta (Dict[str, Any]): Dictionary of metadata to store.
    • span (Optional[Span]): Specific span to store the metadata in; defaults to the current span.
  • Behavior: Adds the provided metadata under the "meta" namespace within the span's attributes.

store_metrics​

store_metrics(metrics, span=None)

Stores custom metrics in the current span.

ag.tracing.store_metrics(metrics, span=None)
  • Parameters:
    • metrics (Dict[str, Any]): Dictionary of metrics to store.
    • span (Optional[Span]): Specific span to store the metrics in; defaults to the current span.
  • Behavior: Adds the provided metrics under the "metrics" namespace within the span's attributes.

CustomSpan Class​

The CustomSpan class extends the functionality of the standard Span class in the OpenTelemetry SDK, providing additional methods for setting attributes and recording events.

set_status​

set_status(status, description=None)

Sets the status of the span.

span.set_status(status, description=None)
  • Parameters:
    • status (Union[Status, StatusCode]): Status code or Status object.
    • description (Optional[str]): Description of the status.
  • Behavior: Updates the span's status with the provided code and description.

set_attributes​

set_attributes(attributes, namespace=None, max_depth=None)

Sets multiple attributes on the span.

span.set_attributes(attributes, namespace=None, max_depth=None)
  • Parameters:
    • attributes (Dict[str, Any]): Dictionary of attributes to set.
    • namespace (Optional[str]): Namespace for the attributes.
    • max_depth (Optional[int]): Maximum depth for nested attributes.
  • Behavior: Adds serialized attributes to the span, optionally under a namespace.

set_attribute​

set_attribute(key, value, namespace=None)

Sets a single attribute on the span.

span.set_attribute(key, value, namespace=None)
  • Parameters:
    • key (str): Attribute key.
    • value (Any): Attribute value.
    • namespace (Optional[str]): Namespace for the attribute.
  • Behavior: Adds the attribute to the span, optionally under a namespace.

add_event​

add_event(name, attributes=None, timestamp=None, namespace=None)

Adds an event to the span.

span.add_event(name, attributes=None, timestamp=None, namespace=None)
  • Parameters:
    • name (str): Event name.
    • attributes (Optional[Dict[str, Any]]): Event attributes.
    • timestamp (Optional[int]): Event timestamp.
    • namespace (Optional[str]): Namespace for the attributes.
  • Behavior: Records an event with the specified name and attributes.

add_link(context, attributes=None, namespace=None)

Adds a link to another span context.

span.add_link(context, attributes=None, namespace=None)
  • Parameters:
    • context (SpanContext): The span context to link to.
    • attributes (Optional[Dict[str, Any]]): Link attributes.
    • namespace (Optional[str]): Namespace for the attributes.
  • Behavior: Creates a link to another span, useful for associating related traces.

record_exception​

record_exception(exception, attributes=None, timestamp=None, escaped=False, namespace=None)

Records an exception in the span.

span.record_exception(exception, attributes=None, timestamp=None, escaped=False, namespace=None)
  • Parameters:
    • exception (BaseException): The exception to record.
    • attributes (Optional[Dict[str, Any]]): Exception attributes.
    • timestamp (Optional[int]): Timestamp of the exception.
    • escaped (bool): Whether the exception escaped.
    • namespace (Optional[str]): Namespace for the attributes.
  • Behavior: Captures exception details in the span for error tracking.

Reference Enum​

The Reference enumeration defines valid keys for referencing Agenta resources in spans.

from enum import Enum

class Reference(str, Enum):
    VARIANT_ID = "variant.id"
    VARIANT_SLUG = "variant.slug"
    VARIANT_VERSION = "variant.version"
    ENVIRONMENT_ID = "environment.id"
    ENVIRONMENT_SLUG = "environment.slug"
    ENVIRONMENT_VERSION = "environment.version"
    APPLICATION_ID = "application.id"
    APPLICATION_SLUG = "application.slug"
  • Usage: Use these keys with ag.tracing.store_refs() to link spans to specific resources.

Utility Functions​

is_valid_attribute_key​

is_valid_attribute_key(string)

Validates whether a string is a valid attribute key.

from agenta.tracing.utils import is_valid_attribute_key

is_valid = is_valid_attribute_key("attribute_key")
  • Parameters:
    • string (str): The attribute key to validate.
  • Returns: True if valid; False otherwise.
  • Behavior: Checks if the string matches the pattern [A-Za-z0-9._-]+.

parse_span_kind​

parse_span_kind(type)

Parses a string to determine the corresponding SpanKind.

from agenta.tracing.utils import parse_span_kind

span_kind = parse_span_kind("client")
  • Parameters:
    • type (str): The span kind as a string.
  • Returns: SpanKind corresponding to the input string.
  • Valid Inputs:
    • "agent", "chain", "workflow": Returns SpanKind.SERVER.
    • "tool", "embedding", "query", "completion", "chat", "rerank": Returns SpanKind.CLIENT.
    • Others: Returns SpanKind.INTERNAL.

Notes​

  • Decorator Usage: The @ag.instrument() decorator should be applied directly above the function definition to ensure proper instrumentation.
  • Span Context: The ag.tracing methods operate on the current span within the execution context.
  • Error Handling: The methods are designed to suppress exceptions silently to avoid disrupting the application's flow.
  • Namespace Usage: Namespaces in attributes help organize data within spans, preventing key collisions.

Example Usage​

Instrumenting a Function​

import agenta as ag

@ag.instrument(spankind="workflow")
def process_data(input_data):
    # Function implementation
    result = compute_result(input_data)
    return result

Storing Metadata and Internals​

@ag.instrument()
def compute_result(data):
    ag.tracing.store_meta({"data_size": len(data)})
    intermediate = data_processing_step(data)
    ag.tracing.store_internals({"intermediate_result": intermediate})
    final_result = finalize(intermediate)
    return final_result

Linking to Agenta Resources​

@ag.instrument()
def execute_task():
    ag.tracing.store_refs({
        Reference.APPLICATION_SLUG.value: "my-app",
        Reference.VARIANT_SLUG.value: "v1",
        Reference.ENVIRONMENT_SLUG.value: "production",
    })
    # Function implementation

Recording Metrics​

import time

@ag.instrument()
def perform_operation():
    start_time = time.time()
    # Operation logic
    end_time = time.time()
    ag.tracing.store_metrics({"execution_time_ms": (end_time - start_time) * 1000})

Handling Exceptions​

@ag.instrument()
def risky_function():
    try:
        # Risky operation
        pass
    except Exception as e:
        ag.tracing.get_current_span().record_exception(e)
        raise