Overview

@traced(capture_input=False, capture_output=True)
def process_payment(card_number: str, amount: float) -> str:
    return "payment_processed"

from orq_ai_sdk.traced import current_span

@traced(type="llm", name="GPT Analysis")
def analyze_text(prompt: str) -> str:
    result = call_llm_api(prompt)
    
    current_span().log(
        input=[{"role": "user", "content": prompt}],
        output=result["content"],
        metrics={
            "prompt_tokens": result["usage"]["prompt_tokens"],
            "completion_tokens": result["usage"]["completion_tokens"],
            "tokens": result["usage"]["total_tokens"],
            "completion_tokens_details": {
                "accepted_prediction_tokens": 3,
                "reasoning_tokens": 2,
                "rejected_prediction_tokens": 1
            },
            "prompt_tokens_details": {
                "cached_tokens": 5
            }
        },
        metadata={"temperature": 0.7, "custom_field": "value"},
        request={
            "provider": "openai",
            "model": "gpt-3.5-turbo",
            "temperature": 0.7,
            "max_tokens": 1000
        }
    )
    
    return result["content"]

@traced
def data_pipeline(raw_data: list) -> dict:
    cleaned = clean_data(raw_data)
    processed = transform_data(cleaned)
    return analyze_results(processed)

@traced
def clean_data(data: list) -> list:
    return [item for item in data if item is not None]

@traced(type="tool")
def transform_data(data: list) -> list:
    return [item.upper() for item in data]

@traced(type="retrieval")
def analyze_results(data: list) -> dict:
    return {"count": len(data), "sample": data[:3]}

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# Set up OpenTelemetry with OTLP exporter
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

# Export to OpenTelemetry-compatible backend (e.g., Jaeger, Zipkin, etc.)
otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)

@traced
def query_database(table: str, filters: dict):
    return {"results": [{"id": 1, "name": "John"}], "count": 1}

@traced
def process_results(raw_data: dict):
    return [item["name"] for item in raw_data["results"]]

# Create parent span that will correlate with traced functions
with tracer.start_as_current_span("user_lookup_operation") as span:
    span.set_attribute("user_id", "12345")
    
    # These traced functions become child spans
    result = query_database("users", {"active": True})
    processed = process_results(result)
    
    span.set_attribute("result_count", len(processed))