TraceModel¶

class opik.message_processing.emulation.models.TraceModel(id: str, start_time: ~datetime.datetime, name: str | None, project_name: str, input: ~typing.Dict[str, ~typing.Any] | None = None, output: ~typing.Dict[str, ~typing.Any] | None = None, tags: ~typing.List[str] | None = None, metadata: ~typing.Dict[str, ~typing.Any] | None = None, end_time: ~datetime.datetime | None = None, spans: ~typing.List[~opik.message_processing.emulation.models.SpanModel] = <factory>, feedback_scores: ~typing.List[~opik.message_processing.emulation.models.FeedbackScoreModel] = <factory>, error_info: ~opik.types.ErrorInfoDict | None = None, thread_id: str | None = None, last_updated_at: ~datetime.datetime | None = None)¶

Bases: object

Represents a trace model that encapsulates data about a trace, its related metadata, and associated spans. It is used for tracking and analyzing data during execution or processing tasks.

This class provides a structure to represent trace information, including the start and end times, associated project details, input/output data, feedback scores, error information, and thread association. It is designed to handle optional fields for flexible use across various scenarios.

id¶

Unique identifier for the trace.

Type:: str

start_time¶

Timestamp representing the start of the trace.

Type:: datetime.datetime

name¶

Optional name for the trace, which can provide a descriptive label.

Type:: str | None

project_name¶

Name of the project associated with the trace.

Type:: str

input¶

Optional dictionary containing the input data associated with the trace.

Type:: Dict[str, Any] | None

output¶

Optional dictionary containing the output data generated by the trace.

Type:: Dict[str, Any] | None

tags¶

Optional list of tags associated with the trace for classification or filtering purposes.

Type:: List[str] | None

metadata¶

Optional metadata providing additional information about the trace.

Type:: Dict[str, Any] | None

end_time¶

Timestamp representing the end of the trace.

Type:: datetime.datetime | None

spans¶

List of spans associated with the trace, representing individual processing parts or segments within the trace.

Type:: List[opik.message_processing.emulation.models.SpanModel]

feedback_scores¶

List of feedback scores associated with the trace.

Type:: List[opik.message_processing.emulation.models.FeedbackScoreModel]

error_info¶

Optional dictionary containing information about errors encountered during the trace.

Type:: opik.types.ErrorInfoDict | None

thread_id¶

Optional identifier of the thread associated with the trace.

Type:: str | None

last_updated_at¶

Timestamp for when the trace was last updated.

Type:: datetime.datetime | None

id: str¶

start_time: datetime¶

name: str | None¶

project_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¶

end_time: datetime | None = None¶

spans: List[SpanModel]¶

feedback_scores: List[FeedbackScoreModel]¶

error_info: ErrorInfoDict | None = None¶

thread_id: str | None = None¶

last_updated_at: datetime | None = None¶

__init__(id: str, start_time: ~datetime.datetime, name: str | None, project_name: str, input: ~typing.Dict[str, ~typing.Any] | None = None, output: ~typing.Dict[str, ~typing.Any] | None = None, tags: ~typing.List[str] | None = None, metadata: ~typing.Dict[str, ~typing.Any] | None = None, end_time: ~datetime.datetime | None = None, spans: ~typing.List[~opik.message_processing.emulation.models.SpanModel] = <factory>, feedback_scores: ~typing.List[~opik.message_processing.emulation.models.FeedbackScoreModel] = <factory>, error_info: ~opik.types.ErrorInfoDict | None = None, thread_id: str | None = None, last_updated_at: ~datetime.datetime | None = None) → None¶

Description¶

The TraceModel class represents a trace model that encapsulates data about a trace, its related metadata, and associated spans. It is used for tracking and analyzing data during execution or processing tasks.

A trace represents the complete execution path of a request or operation, containing one or more spans that represent individual steps or components within that execution.

Attributes¶

Required Attributes¶

opik.message_processing.emulation.models.id: str¶: Unique identifier for the trace.

opik.message_processing.emulation.models.start_time: datetime.datetime¶: Timestamp representing the start of the trace.

opik.message_processing.emulation.models.name: str | None¶: Optional name for the trace, which can provide a descriptive label for the operation being traced.

opik.message_processing.emulation.models.project_name: str¶: Name of the project associated with the trace.

Optional Attributes¶

opik.message_processing.emulation.models.input: Dict[str, Any] | None = None¶: Optional dictionary containing the input data associated with the trace. This represents the initial parameters or data that started the trace.

opik.message_processing.emulation.models.output: Dict[str, Any] | None = None¶: Optional dictionary containing the output data generated by the trace. This represents the final results or return values.

opik.message_processing.emulation.models.tags: List[str] | None = None¶: Optional list of tags associated with the trace for classification or filtering purposes.

opik.message_processing.emulation.models.metadata: Dict[str, Any] | None = None¶: Optional metadata providing additional information about the trace.

opik.message_processing.emulation.models.end_time: datetime.datetime | None = None¶: Timestamp representing the end of the trace. When set, this marks when the operation completed.

opik.message_processing.emulation.models.error_info: ErrorInfoDict | None = None¶: Optional dictionary containing information about errors encountered during the trace.

opik.message_processing.emulation.models.thread_id: str | None = None¶: Optional identifier of the thread associated with the trace. Useful for concurrent operations.

opik.message_processing.emulation.models.last_updated_at: datetime.datetime | None = None¶: Timestamp for when the trace was last updated.

Collection Attributes¶

opik.message_processing.emulation.models.spans: List[SpanModel]¶: List of spans associated with the trace, representing individual processing parts or segments within the trace. Each span represents a specific operation or step in the overall execution.

opik.message_processing.emulation.models.feedback_scores: List[FeedbackScoreModel]¶: List of feedback scores associated with the trace. These are used for overall trace evaluation and quality assessment.

Examples¶

Creating a basic trace:

import datetime
from opik.message_processing.emulation.models import TraceModel

# Create a simple trace
trace = TraceModel(
    id="trace_123",
    start_time=datetime.datetime.now(),
    name="user_query_processing",
    project_name="my_project",
    input={"user_query": "What is machine learning?"},
    output={"response": "Machine learning is a subset of AI..."}
)

Creating a trace with spans:

from opik.message_processing.emulation.models import SpanModel

# Create a trace with associated spans
trace = TraceModel(
    id="trace_456",
    start_time=datetime.datetime.now(),
    name="complex_operation",
    project_name="ai_project"
)

# Add spans to represent different steps
preprocessing_span = SpanModel(
    id="span_1",
    start_time=datetime.datetime.now(),
    name="data_preprocessing"
)

llm_span = SpanModel(
    id="span_2",
    start_time=datetime.datetime.now(),
    name="llm_call",
    type="llm"
)

trace.spans.extend([preprocessing_span, llm_span])

Adding feedback scores to a trace:

from opik.message_processing.emulation.models import FeedbackScoreModel

# Add overall evaluation scores to the trace
overall_quality = FeedbackScoreModel(
    id="score_123",
    name="overall_quality",
    value=0.88,
    reason="Good response quality with minor improvements needed"
)

trace.feedback_scores.append(overall_quality)

Working with trace hierarchies:

# Access nested spans within a trace
for span in trace.spans:
    print(f"Span: {span.name}")

    # Each span can have nested spans too
    for nested_span in span.spans:
        print(f"  Nested: {nested_span.name}")

Usage in Evaluation Context¶

TraceModel objects are commonly used in evaluation scenarios where you need to analyze the complete execution:

# Example of accessing trace data in evaluation
def analyze_trace(trace: TraceModel):
    # Analyze overall trace performance
    duration = trace.end_time - trace.start_time if trace.end_time else None

    # Count different types of spans
    llm_spans = [s for s in trace.spans if s.type == "llm"]

    # Analyze input/output
    input_complexity = len(str(trace.input)) if trace.input else 0
    output_quality = evaluate_output_quality(trace.output)

    return {
        "duration": duration,
        "llm_calls": len(llm_spans),
        "complexity": input_complexity,
        "quality": output_quality
    }

Common Use Cases¶

TraceModel is commonly used for:

Request Tracking: Tracking complete user requests from start to finish
Performance Analysis: Analyzing the performance of complex operations
Evaluation: Providing complete context for evaluation metrics
Debugging: Understanding the full execution path and identifying issues
Cost Tracking: Aggregating costs across all spans in a trace
A/B Testing: Comparing different execution paths and their outcomes

Relationship with Spans¶

A trace acts as a container for spans, creating a hierarchical structure:

Trace: The top-level container representing the complete operation
Spans: Individual steps or operations within the trace
Nested Spans: Spans can contain other spans, creating a tree structure

This hierarchy allows for detailed tracking of complex operations while maintaining the overall context.