SpanModel¶

class opik.message_processing.emulation.models.SpanModel(id: str, start_time: ~datetime.datetime, name: str | None = None, 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, type: str = 'general', usage: ~typing.Dict[str, ~typing.Any] | None = None, end_time: ~datetime.datetime | None = None, project_name: str = 'Default Project', spans: ~typing.List[~opik.message_processing.emulation.models.SpanModel] = <factory>, feedback_scores: ~typing.List[~opik.message_processing.emulation.models.FeedbackScoreModel] = <factory>, model: str | None = None, provider: str | None = None, error_info: ~opik.types.ErrorInfoDict | None = None, total_cost: float | None = None, last_updated_at: ~datetime.datetime | None = None)¶

Bases: object

Represents a span model used to describe specific points in a process, their metadata, and associated data.

This class is used to store and manipulate structured data for events or spans, including metadata, time markers, associated input/output, tags, and additional properties. It serves as a representative structure for recording and organizing event-specific information, often used in applications like logging, distributed tracing, or data processing pipelines.

id¶

Unique identifier for the span.

Type:: str

start_time¶

Start time of the span.

Type:: datetime.datetime

name¶

Name of the span, if provided.

Type:: str | None

input¶

Input data associated with the span, if any.

Type:: Dict[str, Any] | None

output¶

Output data associated with the span, if any.

Type:: Dict[str, Any] | None

tags¶

List of tags linked to the span.

Type:: List[str] | None

metadata¶

Additional metadata for the span.

Type:: Dict[str, Any] | None

type¶

Type of the span, defaulting to “general”.

Type:: str

usage¶

Usage-related information for the span.

Type:: Dict[str, Any] | None

end_time¶

End time of the span, if available.

Type:: datetime.datetime | None

project_name¶

Name of the project the span is associated with, defaulting to a predefined project name.

Type:: str

spans¶

List of nested spans related to this span.

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

feedback_scores¶

List of feedback scores associated with the span.

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

model¶

Model identification used, if applicable.

Type:: str | None

provider¶

Provider of the span or associated services, if any.

Type:: str | None

error_info¶

Error information or diagnostics for the span, if applicable.

Type:: opik.types.ErrorInfoDict | None

total_cost¶

Total cost incurred associated with this span, if relevant.

Type:: float | None

last_updated_at¶

Timestamp of when the span was last updated, if available.

Type:: datetime.datetime | None

id: str¶

start_time: datetime¶

name: str | None = None¶

input: Dict[str, Any] | None = None¶

output: Dict[str, Any] | None = None¶

tags: List[str] | None = None¶

metadata: Dict[str, Any] | None = None¶

type: str = 'general'¶

usage: Dict[str, Any] | None = None¶

end_time: datetime | None = None¶

project_name: str = 'Default Project'¶

spans: List[SpanModel]¶

feedback_scores: List[FeedbackScoreModel]¶

model: str | None = None¶

provider: str | None = None¶

error_info: ErrorInfoDict | None = None¶

total_cost: float | None = None¶

last_updated_at: datetime | None = None¶

__init__(id: str, start_time: ~datetime.datetime, name: str | None = None, 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, type: str = 'general', usage: ~typing.Dict[str, ~typing.Any] | None = None, end_time: ~datetime.datetime | None = None, project_name: str = 'Default Project', spans: ~typing.List[~opik.message_processing.emulation.models.SpanModel] = <factory>, feedback_scores: ~typing.List[~opik.message_processing.emulation.models.FeedbackScoreModel] = <factory>, model: str | None = None, provider: str | None = None, error_info: ~opik.types.ErrorInfoDict | None = None, total_cost: float | None = None, last_updated_at: ~datetime.datetime | None = None) → None¶

Description¶

The SpanModel class represents a span model used to describe specific points in a process, their metadata, and associated data. This class is used to store and manipulate structured data for events or spans, including metadata, time markers, associated input/output, tags, and additional properties.

It serves as a representative structure for recording and organizing event-specific information, often used in applications like logging, distributed tracing, or data processing pipelines. In the context of Opik, spans represent individual operations or function calls within a larger trace.

Attributes¶

Required Attributes¶

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

opik.message_processing.emulation.models.start_time: datetime.datetime¶: Start time of the span, marking when the operation began.

Optional Attributes¶

opik.message_processing.emulation.models.name: str | None = None¶: Name of the span, if provided. This typically describes the operation being performed.

opik.message_processing.emulation.models.input: Dict[str, Any] | None = None¶: Input data associated with the span, if any. This contains the parameters or data passed to the operation.

opik.message_processing.emulation.models.output: Dict[str, Any] | None = None¶: Output data associated with the span, if any. This contains the results or return values from the operation.

opik.message_processing.emulation.models.tags: List[str] | None = None¶: List of tags linked to the span. Tags are used for categorization and filtering.

opik.message_processing.emulation.models.metadata: Dict[str, Any] | None = None¶: Additional metadata for the span. This can contain any custom information about the operation.

opik.message_processing.emulation.models.type: str = "general"¶: Type of the span, defaulting to “general”. Common types include “llm”, “general”, “tool”, etc.

opik.message_processing.emulation.models.usage: Dict[str, Any] | None = None¶: Usage-related information for the span, such as token counts, API usage statistics, etc.

opik.message_processing.emulation.models.end_time: datetime.datetime | None = None¶: End time of the span, if available. This marks when the operation completed.

opik.message_processing.emulation.models.project_name: str = OPIK_PROJECT_DEFAULT_NAME¶: Name of the project the span is associated with, defaulting to a predefined project name.

opik.message_processing.emulation.models.model: str | None = None¶: Model identification used, if applicable. This is commonly used for LLM spans to track which model was used.

opik.message_processing.emulation.models.provider: str | None = None¶: Provider of the span or associated services, if any. Examples include “openai”, “anthropic”, etc.

opik.message_processing.emulation.models.error_info: ErrorInfoDict | None = None¶: Error information or diagnostics for the span, if applicable. Contains details about any errors that occurred.

opik.message_processing.emulation.models.total_cost: float | None = None¶: Total cost incurred associated with this span, if relevant. This is useful for tracking API costs.

opik.message_processing.emulation.models.last_updated_at: datetime.datetime | None = None¶: Timestamp of when the span was last updated, if available.

Collection Attributes¶

opik.message_processing.emulation.models.spans: List[SpanModel]¶: List of nested spans related to this span. This creates a hierarchical structure where spans can contain child spans.

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

Examples¶

Creating a basic span:

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

# Create a simple span
span = SpanModel(
    id="span_123",
    start_time=datetime.datetime.now(),
    name="llm_call",
    type="llm",
    input={"prompt": "What is the capital of France?"},
    output={"response": "Paris is the capital of France."},
    model="gpt-4",
    provider="openai"
)

Creating a span with nested spans:

# Create a parent span with child spans
parent_span = SpanModel(
    id="parent_123",
    start_time=datetime.datetime.now(),
    name="complex_operation"
)

child_span = SpanModel(
    id="child_456",
    start_time=datetime.datetime.now(),
    name="preprocessing_step"
)

parent_span.spans.append(child_span)

Adding feedback scores to a span:

from opik.message_processing.emulation.models import FeedbackScoreModel

# Add evaluation scores to the span
quality_score = FeedbackScoreModel(
    id="score_789",
    name="response_quality",
    value=0.92,
    reason="High quality response with accurate information"
)

span.feedback_scores.append(quality_score)

Usage in Task Span Evaluation¶

SpanModel objects are particularly important in task span evaluation, where custom metrics can analyze the span data:

from opik.evaluation.metrics import BaseMetric, score_result

class CustomSpanMetric(BaseMetric):
    def score(self, task_span: SpanModel) -> score_result.ScoreResult:
        # Access span properties for evaluation
        input_data = task_span.input
        output_data = task_span.output

        # Perform custom evaluation logic
        score_value = self.evaluate_span_quality(input_data, output_data)

        return score_result.ScoreResult(
            value=score_value,
            name=self.name,
            reason=f"Evaluated span '{task_span.name}'"
        )

Common Use Cases¶

SpanModel is commonly used for:

Function Tracking: Recording individual function or method calls
LLM Operations: Tracking language model API calls with usage and cost information
Pipeline Steps: Representing steps in data processing pipelines
Evaluation: Providing detailed execution data for custom evaluation metrics
Debugging: Analyzing the structure and performance of complex operations