Opik

class opik.Opik(project_name: str | None = None, workspace: str | None = None, host: str | None = None)

Bases: object

__init__(project_name: str | None = None, workspace: str | None = None, host: str | None = None) → None

Initialize an Opik object that can be used to log traces and spans manually to Opik server.

Parameters:

  • project_name – The name of the project. If not provided, traces and spans will be logged to the Default Project.

  • workspace – The name of the workspace. If not provided, default will be used.

  • host – The host URL for the Opik server. If not provided, it will default to https://www.comet.com/opik/api.

Returns:

None

trace(id: str | None = None, name: str | None = None, start_time: datetime | None = None, end_time: datetime | None = None, input: Dict[str, Any] | None = None, output: Dict[str, Any] | None = None, metadata: Dict[str, Any] | None = None, tags: List[str] | None = None, feedback_scores: List[FeedbackScoreDict] | None = None) → Trace

Create and log a new trace.

Parameters:

  • id – The unique identifier for the trace, if not provided a new ID will be generated. Must be a valid [UUIDv8](https://uuid.ramsey.dev/en/stable/rfc4122/version8.html) ID.

  • name – The name of the trace.

  • start_time – The start time of the trace. If not provided, the current local time will be used.

  • end_time – The end time of the trace.

  • input – The input data for the trace. This can be any valid JSON serializable object.

  • output – The output data for the trace. This can be any valid JSON serializable object.

  • metadata – Additional metadata for the trace. This can be any valid JSON serializable object.

  • tags – Tags associated with the trace.

  • feedback_scores – The list of feedback score dicts assosiated with the trace. Dicts don’t required to have an id value.

Returns:

The created trace object.

Return type:

trace.Trace

span(trace_id: str | None = None, id: str | None = None, parent_span_id: str | None = None, name: str | None = None, type: Literal['general', 'tool', 'llm'] = 'general', start_time: datetime | None = None, end_time: datetime | None = None, metadata: Dict[str, Any] | None = None, input: Dict[str, Any] | None = None, output: Dict[str, Any] | None = None, tags: List[str] | None = None, usage: UsageDict | None = None, feedback_scores: List[FeedbackScoreDict] | None = None) → Span

Create and log a new span.

Parameters:

  • trace_id – The unique identifier for the trace. If not provided, a new ID will be generated. Must be a valid [UUIDv8](https://uuid.ramsey.dev/en/stable/rfc4122/version8.html) ID.

  • id – The unique identifier for the span. If not provided, a new ID will be generated. Must be a valid [UUIDv8](https://uuid.ramsey.dev/en/stable/rfc4122/version8.html) ID.

  • parent_span_id – The unique identifier for the parent span.

  • name – The name of the span.

  • type – The type of the span. Default is “general”.

  • start_time – The start time of the span. If not provided, the current local time will be used.

  • end_time – The end time of the span.

  • metadata – Additional metadata for the span. This can be any valid JSON serializable object.

  • input – The input data for the span. This can be any valid JSON serializable object.

  • output – The output data for the span. This can be any valid JSON serializable object.

  • tags – Tags associated with the span.

  • usage – Usage data for the span.

  • feedback_scores – The list of feedback score dicts assosiated with the span. Dicts don’t required to have an id value.

Returns:

The created span object.

Return type:

span.Span

log_spans_feedback_scores(scores: List[FeedbackScoreDict]) → None

Log feedback scores for spans.

Parameters:

scores (List[FeedbackScoreDict]) – A list of feedback score dictionaries. Specifying a span id via id key for each score is mandatory.

Returns:

None

log_traces_feedback_scores(scores: List[FeedbackScoreDict]) → None

Log feedback scores for traces.

Parameters:

scores (List[FeedbackScoreDict]) – A list of feedback score dictionaries. Specifying a trace id via id key for each score is mandatory.

Returns:

None

get_dataset(name: str) → Dataset

Get dataset by name

Parameters:

name – The name of the dataset

Returns:

dataset object associated with the name passed.

Return type:

dataset.Dataset

delete_dataset(name: str) → None

Delete dataset by name

Parameters:

name – The name of the dataset

create_dataset(name: str, description: str | None = None) → Dataset

Create a new dataset.

Parameters:

  • name – The name of the dataset.

  • description – An optional description of the dataset.

Returns:

The created dataset object.

Return type:

dataset.Dataset

create_experiment(name: str, dataset_name: str, experiment_config: Dict[str, Any] | None = None) → Experiment
end(timeout: int | None = None) → None

End the Opik session and submit all pending messages.

Parameters:

timeout (Optional[int]) – The timeout for closing the streamer. Once the timeout is reached, the streamer will be closed regardless of whether all messages have been sent. If no timeout is set, the default value from the Opik configuration will be used.

Returns:

None

flush(timeout: int | None = None) → None

Flush the streamer to ensure all messages are sent.

Parameters:

timeout (Optional[int]) – The timeout for flushing the streamer. Once the timeout is reached, the flush method will return regardless of whether all messages have been sent.

Returns:

None

get_trace_content(id: str) → TracePublic
Parameters:

id (str) – trace id

Returns:

pydantic model object with all the data associated with the trace found. Raises an error if trace was not found.

Return type:

trace_public.TracePublic

get_span_content(id: str) → SpanPublic
Parameters:

id (str) – span id

Returns:

pydantic model object with all the data associated with the span found. Raises an error if span was not found.

Return type:

span_public.SpanPublic