@traced Decorator

Automatically trace function calls with OpenTelemetry.

Returns

Returns the decorated function with automatic tracing enabled.

Example

@traced(type=SpanType.TOOL, name="calculate_sum")
def calculate(a, b):
    return a + b


def my_maths_agent(input_data: AgentInput) -> AgentResponse:
    message = input_data.last_user_message_str() or ""
    a, b = map(int, message.split(","))
    return AgentResponse(content=f"Result: {calculate(a, b)}")


trace = galtea.traces.generate(
    agent=my_maths_agent,
    session=session,
    input="5,3",
)

Parameters

name · string

Custom span name. Defaults to the function name.

type · string

SpanType value: SPAN, GENERATION, EVENT, AGENT, TOOL, CHAIN, RETRIEVER, EVALUATOR, EMBEDDING, GUARDRAIL. See Span Types for details. An unrecognized value emits a UserWarning and is sent as SPAN, so your original label is not stored. Pass the label to spans.create instead when you need Galtea to keep it.

log_args · bool

Whether to log function arguments as input data. Default: True.

log_results · bool

Whether to log return value as output data. Default: True.

attributes · dict

Custom attributes to add to the span (e.g., model name, configuration).

include_docstring · bool

If True, the function’s docstring is automatically used as the span description (max 1MB). Default: False.

Features

Automatic Exception Recording

Exceptions are always recorded in spans for debugging, regardless of log_args and log_results settings:

@traced(type=SpanType.TOOL)
def risky_operation() -> str:
    # Exceptions are always recorded in spans for debugging
    # even with log_args=False and log_results=False
    return "Success"


def risky_agent(input_data: AgentInput) -> AgentResponse:
    result = risky_operation()
    return AgentResponse(content=result)


# The span will include error details if an exception occurs
trace_risky = galtea.traces.generate(
    agent=risky_agent,
    session=session_decorator,
    input="test",
)

Input/Output Serialization

Function arguments and return values are automatically serialized to JSON. Non-serializable objects are converted to string representation:

@traced(type=SpanType.TOOL)
def process_data(user_id: str, config: dict) -> dict:
    # Function arguments are automatically serialized to JSON
    # Non-serializable objects are converted to string representation
    return {"status": "processed", "user_id": user_id}


def data_agent(input_data: AgentInput) -> AgentResponse:
    result = process_data("user_123", {"setting": "value"})
    return AgentResponse(content=str(result))


session_serialization = galtea.sessions.create(version_id=version_id, is_production=True)
if session_serialization is None:
    raise ValueError("session_serialization is None")

trace_data = galtea.traces.generate(
    agent=data_agent,
    session=session_serialization,
    input="process",
)

Context Propagation

Spans automatically inherit the context set by set_context():

@traced(type=SpanType.AGENT)
def agent_workflow() -> str:
    # This span is automatically linked to the trace
    # when set_context() has been called with trace_id
    return "workflow completed"


# Create a trace for context propagation example
session_context = galtea.sessions.create(version_id=version_id, is_production=True)
if session_context is None:
    raise ValueError("session_context is None")

trace_context = galtea.traces.create(
    session_id=session_context.id,
    input="Run workflow",
)
if trace_context is None:
    raise ValueError("trace_context is None")

# Set context before running traced functions
token = set_context(trace_id=trace_context.id)

try:
    result = agent_workflow()
finally:
    clear_context(token)

The @traced decorator uses OpenTelemetry under the hood. Spans are automatically exported to Galtea API when clear_context() is called or when the batch processor flushes.