start_trace

Context manager for creating traces with fine-grained control.

Returns

Yields a GalteaSpan object with helper methods for setting trace data dynamically.

Example

with start_trace("manual_span", type="SPAN") as span:
    # Do work
    span.update(metadata={"status": "done"})

Parameters

name · string · required

Name of the trace.

type · string

TraceType value: SPAN, GENERATION, EVENT, AGENT, TOOL, CHAIN, RETRIEVER, EVALUATOR, EMBEDDING, GUARDRAIL. See Trace Types for details. An unrecognized value emits a UserWarning and is still sent as-is, so values added in newer API versions can be used without upgrading the SDK.

description · string

Human-readable description of what this operation does. Useful for documentation and debugging. Maximum size: 1MB.

input · any

Input data for the trace. Accepts any value (non-serializable objects are converted to string). Maximum size: 10MB.

metadata · any

Metadata for the trace. Accepts any value (non-serializable objects are converted to string). Maximum size: 10MB.

attributes · dict

Custom OpenTelemetry attributes to add to the span.

GalteaSpan Methods

The yielded GalteaSpan object provides these methods:

update(input, output, metadata, type) · method

Update trace attributes. All parameters are optional and accept any value (non-serializable objects are converted to string).

set_attribute(key, value) · method

Set a custom attribute on the span.

record_exception(exception) · method

Manually record an exception on the span.

When to Use

Use start_trace() instead of @trace when you need:

1. Fine-grained control over specific code blocks rather than entire functions
2. Dynamic attributes that are only known at runtime
3. Conditional tracing based on runtime conditions
4. Tracing third-party code that you can’t decorate

Complete Example

def rag_pipeline(query: str, inference_result_id: str) -> str:
    token = set_context(inference_result_id=inference_result_id)

    try:
        # Retrieval step
        with start_trace(
            "retrieve_documents",
            type="RETRIEVER",
            description="Searches vector store for relevant documents",
            input={"query": query},
        ) as span:
            # Simulated vector store search
            docs = [
                {"id": "doc1", "content": "Paris is the capital of France."},
                {"id": "doc2", "content": "France is in Western Europe."},
            ]
            span.update(output={"doc_count": len(docs), "docs": docs})

        # Generation step
        with start_trace(
            "generate_response",
            type="GENERATION",
            description="Generates final response using retrieved context",
            input={"query": query},
        ) as span:
            # Simulated LLM response
            response_content = "Based on the documents, Paris is the capital of France."
            span.update(
                output={"response": response_content},
                metadata={"tokens_used": 42, "model": "gpt-4"},
            )

        return response_content
    finally:
        clear_context(token)


# Create an inference result to associate traces with
inference_result_for_rag = galtea.inference_results.create(
    session_id=session_start_trace.id,
    input="What is the capital of France?",
)
if inference_result_for_rag is None:
    raise ValueError("inference_result_for_rag is None")

result = rag_pipeline("What is the capital of France?", inference_result_for_rag.id)

Nested Traces

Traces automatically form a parent-child hierarchy when nested:

def process_with_nested_traces(inference_result_id: str) -> dict:
    token = set_context(inference_result_id=inference_result_id)

    try:
        with start_trace("parent_operation", type="CHAIN", input={"task": "process_all"}) as parent:
            # First child
            with start_trace("child_step_1", type="TOOL") as span:
                step1_result = {"processed": True, "items": 5}
                span.update(output=step1_result)

            # Second child
            with start_trace("child_step_2", type="TOOL") as span:
                step2_result = {"validated": True, "errors": 0}
                span.update(output=step2_result)

            parent.update(output={"total_steps": 2, "status": "completed"})

        return {"step1": step1_result, "step2": step2_result}
    finally:
        clear_context(token)


# Create another inference result for nested trace example
inference_result_for_nested = galtea.inference_results.create(
    session_id=session_start_trace.id,
    input="Process all items",
)
if inference_result_for_nested is None:
    raise ValueError("inference_result_for_nested is None")

nested_result = process_with_nested_traces(inference_result_for_nested.id)

Like @trace, traces created with start_trace() are automatically exported to Galtea API when clear_context() is called.