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(*, page: int | None = None, size: int | None = None, name: str | None = None, filters: str | None = None, sorting: str | None = None, request_options: RequestOptions | None = None) → AnnotationQueuePagePublic

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, 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])

  • 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(id: str, *, name: 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, request_options: RequestOptions | None = None) → None

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])

  • 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

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()