Examples

A/B Testing Setup

This example demonstrates how to set up a simple A/B test where you route traffic to two different variants and track them as separate experiments.

import random
from basalt.experiments import ExperimentsClient
from basalt.observability import start_observe, observe

# Initialize client
client = ExperimentsClient(api_key="your-api-key")

# Create experiments for two variants
variant_a = client.create_sync(feature_slug="checkout-flow", name="Variant A: Standard")
variant_b = client.create_sync(feature_slug="checkout-flow", name="Variant B: Streamlined")

def get_variant():
    return variant_a if random.random() < 0.5 else variant_b

@start_observe(feature_slug="checkout-flow", name="checkout_process")
def process_checkout(user_id):
    # Determine which experiment to run
    experiment = get_variant()
    
    # Start a span for the specific variant, attaching the experiment ID
    with start_observe(
        name=f"variant_{experiment.id}", 
        experiment=experiment.id,
        feature_slug=experiment.feature_slug
    ):
        if experiment.id == variant_a.id:
            # Logic for Variant A
            observe.set_input({"step": "standard_checkout"})
            pass
        else:
            # Logic for Variant B
            observe.set_input({"step": "streamlined_checkout"})
            pass

# Simulate traffic
process_checkout("user_123")

Experiment Loop (Batch Evaluation)

Process multiple items under the same experiment. Each start_observe() call creates a separate trace automatically — the context manager cleans up OpenTelemetry context on exit, so the next iteration starts with a fresh trace_id.

from basalt import Basalt
from basalt.observability import start_observe, observe, ObserveKind

basalt = Basalt(api_key="your-api-key")

# Create one experiment for the entire batch
experiment = basalt.experiments.create_sync(
    feature_slug="support-ticket",
    name="GPT-4o-mini Ticket Analysis",
)

tickets = [
    "I can't log in after resetting my password.",
    "Order #12345 hasn't arrived yet.",
    "I was billed twice this month.",
]

for ticket in tickets:
    # New trace per iteration — no client recreation needed
    with start_observe(
        name="analyse_ticket",
        feature_slug="support-ticket",
        experiment=experiment,
    ) as span:
        span.set_input({"ticket": ticket})
        result = analyse(ticket)
        observe.set_output({"result": result})

# Shutdown ONCE at the end
basalt.shutdown()

Do not call basalt.shutdown() inside the loop or recreate the Basalt client per iteration. shutdown() permanently kills the global TracerProvider — all subsequent traces would be silently dropped.

Context Manager Usage

You can also use the context manager to wrap specific blocks of code with experiment context.

from basalt.observability import start_observe

# ... assuming experiment_id is known ...
experiment_id = "exp-123"

with start_observe(
    feature_slug="search-ranking",
    name="search.v2",
    experiment=experiment_id,
):
    # All spans created within this block will be associated with the experiment
    results = run_search_v2("query")