start_as_current_trace¶

opik.start_as_current_trace(name: str, input: Dict[str, Any] | None = None, output: Dict[str, Any] | None = None, tags: List[str] | None = None, metadata: Dict[str, Any] | None = None, project_name: str | None = None, thread_id: str | None = None, flush: bool = False) → Generator[TraceData, Any, None]¶

Starts a trace context manager to collect and manage tracing data during the execution of a code block. This function initializes a trace, allows for modifications within the context, and finishes the trace, sending data to the Opik tracing infrastructure.

Parameters:

name – The name of the trace for identification purposes.
input – Optional input data associated with the trace.
output – Optional output data expected or associated with the trace.
tags – Optional list of string tags for labeling or describing the trace.
metadata – Optional dictionary containing additional information about the trace.
project_name – Optional name of the project under which the trace belongs.
thread_id – Optional thread identifier to associate the trace with a specific thread.
flush – A boolean indicating whether to flush the trace data immediately after finishing the trace context.

Yields:

Provides the initialized trace data for manipulation during execution in the context.

Examples¶

Basic usage¶

import opik

with opik.start_as_current_trace("my_trace") as trace:
    # Your code here
    trace.metadata["custom_key"] = "custom_value"
    print("Executing trace...")

With input and output data¶

import opik

with opik.start_as_current_trace(
    name="user_query_processing",
    input={"user_question": "What is machine learning?"},
    output={"answer": "Machine learning is a subset of AI..."},
    tags=["user_query", "ai"],
    metadata={"session_id": "abc123"}
) as trace:
    # Your processing code here
    pass

With conversational threads support using thread_id identifier¶

import opik
import threading

with opik.start_as_current_trace(
    "chatbot_conversation",
    project_name="my_project",
    thread_id="00f067aa0ba902b7",
) as trace:
    # Your processing code here
    pass

Error handling¶

import opik

try:
    with opik.start_as_current_trace("risky_trace") as trace:
        # Code that might fail
        raise RuntimeError("Something went wrong")
except RuntimeError as e:
    # The trace will automatically capture error information
    print(f"Trace failed: {e}")
    # Error details are stored in trace.error_info

Nested spans within a trace¶

import opik

with opik.start_as_current_trace("main_workflow") as trace:
    # Main workflow code

    with opik.start_as_current_span("sub_operation_1") as span1:
        # First sub-operation
        pass

    with opik.start_as_current_span("sub_operation_2") as span2:
        # Second sub-operation
        pass