Setup

import zeroeval as ze

# Option 1: ZEROEVAL_API_KEY in your environment variable file
ze.init()

# Option 2: Provide API key directly from
#           https://app.zeroeval.com/settings?tab=api-keys
ze.init(api_key="YOUR_API_KEY")

import zeroeval as ze

@ze.span(name="fetch_data")
def fetch_data(user_id: str):
    # Function arguments are automatically captured as inputs
    # Return values are automatically captured as outputs
    return {"user_id": user_id, "name": "John Doe"}

@ze.span(name="process_data", attributes={"version": "1.0"})
def process_data(data: dict):
    # Add custom attributes for better filtering
    return f"Welcome, {data['name']}!"

import zeroeval as ze

def complex_workflow():
    with ze.span(name="data_pipeline") as pipeline_span:
        # Fetch stage
        with ze.span(name="fetch_stage") as fetch_span:
            data = fetch_external_data()
            fetch_span.set_io(output_data=str(data))

        # Process stage
        with ze.span(name="process_stage") as process_span:
            processed = transform_data(data)
            process_span.set_io(
                input_data=str(data),
                output_data=str(processed)
            )

        # Save stage
        with ze.span(name="save_stage") as save_span:
            result = save_to_database(processed)
            save_span.set_io(output_data=f"Saved {result} records")

import zeroeval as ze

def resolve_ticket(ticket):
    with ze.span(name="resolve-ticket"):
        system_prompt = ze.prompt(name="support-copilot", content="...")
        decision = run_agent(system_prompt, ticket)

        with ze.artifact_span(
            name="final-decision",
            artifact_type="final_decision",
            role="primary",
            label="Final Decision",
            tags={"judge_target": "support_ops_final_decision"},
        ) as s:
            s.set_io(input_data=ticket.body, output_data=decision.json())

        with ze.artifact_span(
            name="customer-card",
            artifact_type="customer_card",
            role="secondary",
            label="Customer Card",
            tags={"has_customer_card": "true"},
        ) as card:
            card.set_io(input_data=ticket.summary, output_data=decision.json())
            card.add_image(base64_data=render_card(decision))

from zeroeval.observability.tracer import tracer

# Configure tracer settings
tracer.configure(
    flush_interval=5.0,        # Flush every 5 seconds
    max_spans=200,             # Buffer up to 200 spans
    collect_code_details=True  # Capture source code context
)

# Get the current span
current_span = ze.get_current_span()

# Get the current trace ID
trace_id = ze.get_current_trace()

# Get the current session ID
session_id = ze.get_current_session()

import uuid
import zeroeval as ze

session_id = str(uuid.uuid4())

@ze.span(name="process_request", session=session_id)
def process_request(data):
    return transform_data(data)

@ze.span(
    name="user_interaction",
    session={
        "id": session_id,
        "name": "Customer Support Chat - User #12345"
    }
)
def handle_support_chat(user_id, message):
    return generate_response(message)

session_info = {
    "id": str(uuid.uuid4()),
    "name": "Order Processing Pipeline"
}

@ze.span(name="process_order", session=session_info)
def process_order(order_id):
    validate_order(order_id)
    charge_payment(order_id)
    fulfill_order(order_id)

@ze.span(name="validate_order")
def validate_order(order_id):
    return check_inventory(order_id)

@ze.span(name="charge_payment")
def charge_payment(order_id):
    return process_payment(order_id)

session_info = {
    "id": str(uuid.uuid4()),
    "name": "Data Pipeline Run"
}

with ze.span(name="etl_pipeline", session=session_info) as pipeline_span:
    with ze.span(name="extract_data") as extract_span:
        raw_data = fetch_from_source()
        extract_span.set_io(output_data=f"Extracted {len(raw_data)} records")

    with ze.span(name="transform_data") as transform_span:
        clean_data = transform_records(raw_data)
        transform_span.set_io(
            input_data=f"{len(raw_data)} raw records",
            output_data=f"{len(clean_data)} clean records"
        )

@ze.span(
    name="handle_request",
    tags={
        "user_id": "42",
        "tenant": "acme-corp",
        "plan": "enterprise"
    }
)
def handle_request():
    with ze.span(name="fetch_data"):
        ...

    with ze.span(name="process", tags={"stage": "post"}):
        ...

@ze.span(name="top_level")
def top_level():
    with ze.span(name="db_call", tags={"table": "customers", "operation": "SELECT"}):
        query_database()

    with ze.span(name="render"):
        render_template()

with ze.span(name="root_invoke", session=session_info, tags={"run": "invoke"}):
    current_span = ze.get_current_span()
    ze.set_tag(current_span, {"phase": "pre-run"})

    current_trace = ze.get_current_trace()
    ze.set_tag(current_trace, {"run_mode": "invoke"})

    current_session = ze.get_current_session()
    ze.set_tag(current_session, {"env": "local"})

# Save your API key securely
zeroeval setup

# Run scripts with automatic tracing
zeroeval run my_script.py