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:
- Records the start timestamp and perf counter
- Resolves project context from API key or decorator arguments
- Sets the
ContextVarfor child step collection - Calls
func(*args, **kwargs)synchronously - Computes latency, builds the trace payload, persists it, and resets the context
Example: Syncing a Classification Pipeline
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():
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
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.
@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
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"],
}