Annotation Queues Client¶

The Annotation Queues client provides methods for managing annotation queues in the Opik platform. Annotation queues enable human-in-the-loop workflows for reviewing and annotating traces or threads.

class opik.rest_api.annotation_queues.client.AnnotationQueuesClient(*, client_wrapper: SyncClientWrapper)¶

Bases: object

add_items_to_annotation_queue(id: str, *, ids: Sequence[str], request_options: RequestOptions | None = None) → None¶

Add traces or threads to annotation queue

Parameters:

id (str)
ids (Sequence[str])
request_options (Optional[RequestOptions]) – Request-specific configuration.

Return type:

None

Find annotation queues with filtering and sorting

Parameters:

page (Optional[int])
size (Optional[int])
name (Optional[str])
filters (Optional[str])
sorting (Optional[str])
request_options (Optional[RequestOptions]) – Request-specific configuration.

Returns:

Annotation queues page

Return type:

AnnotationQueuePagePublic

create_annotation_queue(*, project_id: str, name: str, scope: Literal['trace', 'thread'] | Any, id: str | None = OMIT, description: str | None = OMIT, instructions: str | None = OMIT, comments_enabled: bool | None = OMIT, feedback_definition_names: Sequence[str] | None = OMIT, annotators_per_item: int | None = OMIT, lock_timeout_seconds: int | None = OMIT, automation: AnnotationQueueAutomationWrite | None = OMIT, request_options: RequestOptions | None = None) → None¶

Create annotation queue for human annotation workflows

Parameters:

project_id (str)
name (str)
scope (AnnotationQueueWriteScope)
id (Optional[str])
description (Optional[str])
instructions (Optional[str])
comments_enabled (Optional[bool])
feedback_definition_names (Optional[Sequence[str]])
annotators_per_item (Optional[int])
lock_timeout_seconds (Optional[int])
automation (Optional[AnnotationQueueAutomationWrite])
request_options (Optional[RequestOptions]) – Request-specific configuration.

Return type:

None

create_annotation_queue_batch(*, annotation_queues: Sequence[AnnotationQueueWrite], request_options: RequestOptions | None = None) → None¶

Create multiple annotation queues for human annotation workflows

Parameters:

annotation_queues (Sequence[AnnotationQueueWrite]) – List of annotation queues to create
request_options (Optional[RequestOptions]) – Request-specific configuration.

Return type:

None

delete_annotation_queue_batch(*, ids: Sequence[str], request_options: RequestOptions | None = None) → None¶

Delete multiple annotation queues by their IDs

Parameters:

ids (Sequence[str])
request_options (Optional[RequestOptions]) – Request-specific configuration.

Return type:

None

get_annotation_queue_by_id(id: str, *, request_options: RequestOptions | None = None) → AnnotationQueuePublic¶

Get annotation queue by id

Parameters:

id (str)
request_options (Optional[RequestOptions]) – Request-specific configuration.

Returns:

Annotation queue resource

Return type:

AnnotationQueuePublic

Update annotation queue by id

Parameters:

id (str)
name (Optional[str])
description (Optional[str])
instructions (Optional[str])
comments_enabled (Optional[bool])
feedback_definition_names (Optional[Sequence[str]])
annotators_per_item (Optional[int])
lock_timeout_seconds (Optional[int])
automation (Optional[AnnotationQueueAutomation])
request_options (Optional[RequestOptions]) – Request-specific configuration.

Return type:

None

get_annotation_queue_locks(queue_id: str, *, request_options: RequestOptions | None = None) → LocksResponse¶

Returns lock status for all actively locked items in the queue

Parameters:

queue_id (str)
request_options (Optional[RequestOptions]) – Request-specific configuration.

Returns:

Queue locks

Return type:

LocksResponse

lock_annotation_queue_item(queue_id: str, item_id: str, *, request_options: RequestOptions | None = None) → LockResponse¶

Claim an annotation queue item for the current user, or extend an existing lock

Parameters:

queue_id (str)
item_id (str)
request_options (Optional[RequestOptions]) – Request-specific configuration.

Returns:

Lock result

Return type:

LockResponse

remove_items_from_annotation_queue(id: str, *, ids: Sequence[str], request_options: RequestOptions | None = None) → None¶

Remove items from annotation queue

Parameters:

id (str)
ids (Sequence[str])
request_options (Optional[RequestOptions]) – Request-specific configuration.

Return type:

None

search_annotation_queue_items(id: str, *, ids: Sequence[str], request_options: RequestOptions | None = None) → AnnotationQueueItemsPublic¶

Returns queue membership metadata — how each item got into the queue — for the given item ids. A lookup rather than a listing: the caller renders the items table from the traces or threads API with its own sort and filters, so it asks for exactly the ids it is displaying. Ids that are not in the queue are omitted.

Parameters:

id (str)
ids (Sequence[str])
request_options (Optional[RequestOptions]) – Request-specific configuration.

Returns:

Annotation queue items

Return type:

AnnotationQueueItemsPublic

Usage Example¶

import opik

client = opik.Opik()

# Create an annotation queue for traces
queue = client.create_annotation_queue(
    name="Review Queue",
    scope="trace",
    description="Queue for reviewing model outputs",
    instructions="Check for accuracy and relevance"
)

# Get traces and add them to the queue
traces = client.search_traces(project_name="my-project")
queue.add_traces(traces[:10])

# Fetch all the traces currently in the queue
items = queue.get_items()

# Get an existing queue by ID
existing_queue = client.get_annotation_queue("queue-id")

# List all annotation queues
queues = client.get_annotation_queues()

# Update queue properties
queue.update(description="Updated description")

# Remove traces from the queue
queue.remove_traces(traces[:5])

# Delete the queue
queue.delete()