Decorators

from tracium import agent_trace

@agent_trace(
    client: TraciumClient,        # Required: Tracium client instance
    agent_name: str,              # Required: Name for the trace
    model_id: str | None = None,
    metadata: dict | None = None,
    tags: list[str] | None = None,
    trace_id: str | None = None,
    inject_trace_arg: str | None = None,  # Inject trace handle as kwarg
    auto_tag: bool = True,        # Auto-add @trace:func_name tag
)
def my_function():
    pass

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import tracium
from tracium import agent_trace
 
client = tracium.init()
 
@agent_trace(client=client, agent_name="support-bot")
def handle_support_request(request: str) -> str:
    """Each call to this function creates a new trace."""
    context = retrieve_context(request)
    response = generate_response(context)
    return response
 
# Create a trace automatically
result = handle_support_request("How do I reset my password?")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
@agent_trace(
    client=client,
    agent_name="analyzer",
    inject_trace_arg="trace"  # Inject as keyword argument
)
def analyze_data(data: dict, trace=None):
    """Access the trace handle inside the function."""
    if trace:
        trace.add_tags(["analysis", f"data_size:{len(data)}"])
        
        with trace.span(span_type="process", name="analyze") as span:
            span.record_input({"data": data})
            result = do_analysis(data)
            span.record_output({"result": result})
    
    return result

@agent_trace(client=client, agent_name="async-agent")
async def async_handler(request: str) -> str:
    """Works with async functions automatically."""
    result = await process_async(request)
    return result

from tracium import agent_span

@agent_span(
    span_type: str,               # Required: Type of span
    name: str | None = None,      # Defaults to function name
    metadata: dict | None = None,
    tags: list[str] | None = None,
    inject_span_arg: str | None = None,  # Inject span handle as kwarg
    capture_return: bool = False, # Auto-capture return value as output
    require_trace: bool = False,  # Raise error if no active trace
    auto_tag: bool = True,        # Auto-add @span:name tag
)
def my_function():
    pass

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import tracium
from tracium import agent_span
 
client = tracium.init()
 
@agent_span(span_type="retrieval")
def fetch_documents(query: str) -> list:
    """Creates a span when called within an active trace."""
    # Your retrieval logic
    return documents
 
@agent_span(span_type="llm", name="generate_response")
def generate(context: str) -> str:
    """Span is named 'generate_response' instead of 'generate'."""
    # Your LLM logic
    return response
 
# Use within a trace
with client.agent_trace(agent_name="rag-pipeline") as trace:
    docs = fetch_documents("query")  # Creates 'retrieval' span
    result = generate(docs)          # Creates 'llm' span

@agent_span(
    span_type="tool",
    capture_return=True  # Automatically record return as output
)
def lookup_user(user_id: str) -> dict:
    """Return value is automatically recorded as span output."""
    return {"id": user_id, "name": "John", "email": "[email protected]"}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
@agent_span(
    span_type="retrieval",
    inject_span_arg="span"
)
def search_database(query: str, span=None) -> list:
    """Access the span handle for detailed recording."""
    span.record_input({"query": query, "index": "main"})
    
    results = db.search(query)
    
    span.record_output({
        "count": len(results),
        "took_ms": results.took
    })
    span.set_token_usage(input_tokens=len(query.split()))
    
    return results

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# By default, @agent_span runs the function normally if no trace is active
@agent_span(span_type="tool")
def my_function():
    return "result"
 
# This works even without a trace (no span is created)
result = my_function()
 
# To require an active trace:
@agent_span(span_type="tool", require_trace=True)
def must_be_traced():
    return "result"
 
# This raises RuntimeError if no trace is active
must_be_traced()  # RuntimeError!

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
import tracium
from tracium import agent_trace, agent_span
 
client = tracium.init()
 
@agent_span(span_type="retrieval", capture_return=True)
def get_context(query: str) -> dict:
    """Creates a span within the parent trace."""
    return retrieve_relevant_docs(query)
 
@agent_span(span_type="plan")
def create_plan(context: dict, query: str) -> str:
    """Creates a planning span."""
    return analyze_and_plan(context, query)
 
@agent_span(span_type="llm", inject_span_arg="span")
def generate_response(plan: str, span=None) -> str:
    """Creates an LLM span with custom recording."""
    span.record_input({"plan": plan})
    response = call_llm(plan)
    span.record_output({"response": response})
    span.set_token_usage(input_tokens=100, output_tokens=200)
    return response
 
@agent_trace(client=client, agent_name="qa-agent")
def answer_question(question: str) -> str:
    """Entry point - creates the trace."""
    context = get_context(question)
    plan = create_plan(context, question)
    answer = generate_response(plan)
    return answer
 
# Run the pipeline - creates a trace with 3 nested spans
result = answer_question("What is machine learning?")

# If the Tracium API is unavailable, the function still executes
@agent_trace(client=client, agent_name="resilient")
def important_function():
    # This will always run, even if tracing fails
    return critical_business_logic()

# Exceptions in your code are still raised
@agent_trace(client=client, agent_name="example")
def might_fail():
    raise ValueError("Something went wrong")
    # The trace is automatically marked as failed
    # The exception propagates to your code