Spans

Spans represent individual operations within a trace.

What is a Span?

A span represents a single unit of work within a trace. Spans can be nested to form a tree structure that shows the execution flow and relationships between operations.

Common span types include:

llm - LLM API calls (OpenAI, Anthropic, etc.)
retrieval - Vector database or search operations
tool - External tool or function calls
plan - Agent planning steps
action - Agent actions
Custom types as needed

Creating Spans

Create spans within a trace context using the span method:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
import tracium
 
client = tracium.init()
 
with client.agent_trace(agent_name="my-agent") as trace:
    # Create an LLM span
    with trace.span(
        span_type="llm",
        name="generate_response",
        model_id="gpt-4"
    ) as span:
        span.record_input({"messages": [...]})
        response = call_openai()
        span.record_output({"content": response})
        span.set_token_usage(
            input_tokens=100,
            output_tokens=200
        )

Span Properties

Property	Type	Description
`span_type`	str	Required. Type of operation (llm, retrieval, tool, etc.)
`name`	str \| None	Human-readable name for the span
`input`	Any	Input data (can also use record_input)
`tags`	list[str] \| None	Tags for filtering spans
`metadata`	dict \| None	Additional contextual data
`model_id`	str \| None	Model used (for LLM spans)
`parent_span_id`	str \| None	ID of parent span (auto-detected if nested)

Span Handle Methods

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
with trace.span(span_type="llm", name="completion") as span:
    # Access span properties
    print(f"Span ID: {span.id}")
    print(f"Trace ID: {span.trace_id}")
    print(f"Parent: {span.parent_span_id}")
    
    # Record input data
    span.record_input({
        "messages": [...],
        "model": "gpt-4"
    })
    
    # Record output data
    span.record_output({
        "response": "...",
        "finish_reason": "stop"
    })
    
    # Set token usage
    span.set_token_usage(
        input_tokens=150,
        output_tokens=200,
        cached_input_tokens=50  # Optional
    )
    
    # Add tags
    span.add_tags(["streaming", "high-priority"])
    
    # Set status
    span.set_status("completed")  # or "failed"
    
    # Mark as failed with error message
    # span.mark_failed("Error description")

Nested Spans

Spans automatically form a hierarchy when nested. Child spans reference their parent:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
with client.agent_trace(agent_name="rag-agent") as trace:
    with trace.span(span_type="plan", name="analyze_query") as plan_span:
        plan_span.record_input({"query": user_query})
        
        # Nested span - automatically linked to parent
        with trace.span(span_type="retrieval", name="search_docs") as search_span:
            search_span.record_input({"query": processed_query})
            docs = search_documents(processed_query)
            search_span.record_output({"num_results": len(docs)})
        
        # Another nested span
        with trace.span(span_type="llm", name="generate") as llm_span:
            llm_span.record_input({"context": docs, "query": user_query})
            response = generate_response(docs, user_query)
            llm_span.record_output({"response": response})

Resulting span tree:

analyze_query (plan)

├── search_docs (retrieval)

└── generate (llm)

Parallel Spans

Tracium automatically detects parallel execution and groups concurrent spans:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import asyncio
import tracium
 
client = tracium.init()
 
async def fetch_data(source: str) -> dict:
    with tracium.span(span_type="retrieval", name=f"fetch_{source}") as span:
        span.record_input({"source": source})
        data = await fetch_from_source(source)
        span.record_output({"count": len(data)})
        return data
 
async def main():
    with client.agent_trace(agent_name="parallel-fetcher") as trace:
        # These spans run in parallel and are grouped together
        results = await asyncio.gather(
            fetch_data("database"),
            fetch_data("api"),
            fetch_data("cache")
        )

Using the span() Helper

For convenience, you can use the module-level span() helper when inside an active trace:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import tracium
from tracium import span
 
client = tracium.init()
 
def process_request():
    # This works when called within an active trace context
    with span(span_type="tool", name="process") as s:
        s.record_input({"data": ...})
        result = do_work()
        s.record_output({"result": result})
    return result
 
with client.agent_trace(agent_name="example") as trace:
    # The span() helper finds the current trace automatically
    process_request()

Record Span (One-Shot)

For simple cases, use record_span to create a completed span in one call:

1
2
3
4
5
6
7
8
9
10
with client.agent_trace(agent_name="example") as trace:
    # Record a complete span in one call
    trace.record_span(
        span_type="tool",
        name="lookup",
        input={"key": "user_123"},
        output={"name": "John", "email": "[email protected]"},
        latency_ms=45,
        status="completed"
    )

Error Handling in Spans

Spans automatically capture exceptions and mark themselves as failed:

1
2
3
4
5
6
7
8
9
with trace.span(span_type="tool", name="risky_operation") as span:
    try:
        result = might_fail()
        span.record_output({"result": result})
    except TimeoutError as e:
        # Manually mark failed with custom message
        span.mark_failed(f"Operation timed out: {e}")
        raise
    # If an exception escapes, span is auto-marked as failed

← Traces Auto-Instrumentation →