Documentation

Sync Functions

Trace synchronous application code.

Overview

The @trace decorator works with synchronous functions out of the box. When applied to a def function (not async def), the decorator returns a synchronous wrapper that transparently captures the same trace lifecycle without any additional configuration.

How It Works

The decorator uses inspect.iscoroutinefunction(func) to detect whether the decorated function is async. For sync functions, it generates a plain def wrapper(*args, **kwargs) that:

  1. Records the start timestamp and perf counter
  2. Resolves project context from API key or decorator arguments
  3. Sets the ContextVar for child step collection
  4. Calls func(*args, **kwargs) synchronously
  5. Computes latency, builds the trace payload, persists it, and resets the context

Example: Syncing a Classification Pipeline

sync_function.pyCopy
python
from tracellm import trace

@trace(
    prompt="classify_invoice",
    model_name="gpt-4o",
    project="doc-processing",
    environment="production",
)
def classify_invoice(invoice_text: str) -> dict:
    # Simulate processing
    categories = ["receipt", "invoice", "purchase_order"]
    result = {"category": categories[hash(invoice_text) % 3], "confidence": 0.94}
    return result

# Usage
output = classify_invoice("INV-2026-05-31: Widget order #4412")
print(output["category"])  # "invoice"

Latency and Token Tracking

For sync functions, latency is measured with time.perf_counter() before and after the function call. The delta is converted to milliseconds and rounded to two decimal places.

Token counts are estimated heuristically from the prompt text, response text, and step inputs/outputs using estimate_tokens():

token estimationCopy
python
def estimate_tokens(*parts: Any) -> int:
    text = " ".join(str(part) for part in parts if part is not None)
    if not text.strip():
        return 0
    return max(1, len(text.split()) + len(text) // 4)

Info

When using the OpenAI integration, actual token counts from the API response (completion.usage.total_tokens) replace the heuristic estimate.

Error Handling

If the decorated function raises an exception, the trace is still finalized and persisted — but with status: "failed" and the exception message recorded as failure_reason. The exception is re-raised after the trace is saved, so existing error handling in your application continues to work.

error_handling.pyCopy
python
@trace(prompt="risky_operation")
def might_fail(value: int) -> int:
    if value < 0:
        raise ValueError("negative values not allowed")
    return value * 2

try:
    might_fail(-1)
except ValueError:
    pass  # Trace was still saved with status="failed"
           # and failure_reason="negative values not allowed"

Production Patterns

production.pyCopy
python
from tracellm import trace

# Batch processing with per-item tracing
@trace(prompt="process_batch", model_name="gpt-4o-mini")
def process_batch(items: list[dict]) -> list[dict]:
    results = []
    for item in items:
        results.append(transform(item))
    return results

# Multi-step business logic
@trace(
    prompt="fulfill_order",
    model_name="gpt-4o",
    project="order-service",
    environment="production",
)
def fulfill_order(order: dict) -> dict:
    validated = validate_order(order)
    priced = calculate_pricing(validated)
    confirmed = submit_to_warehouse(priced)
    return {
        "order_id": confirmed["id"],
        "status": "fulfilled",
        "total": priced["total"],
    }