Rapidata audience manager

RapidataAudienceManager

RapidataAudienceManager(openapi_service: OpenAPIService)

Handles everything regarding audiences from creation to retrieval.

A manager for creating, retrieving, and searching for audiences. Audiences are groups of annotators that can be recruited based on example tasks and assigned jobs.

Source code in src/rapidata/rapidata_client/audience/rapidata_audience_manager.py

def __init__(self, openapi_service: OpenAPIService):
    self._openapi_service = openapi_service

create_audience

create_audience(
    name: str,
    filters: list[RapidataFilter] | None = None,
    target_accuracy: float | None = None,
    min_tasks: int | None = None,
    max_tasks: int | None = None,
) -> RapidataAudience

Create a new audience.

Creates a new audience with the specified name and optional filters.

Annotators join an audience by passing a short admission trial: they must answer validation tasks well enough to be trusted. The two intuitive knobs below describe that bar as "solve N tasks at X accuracy":

Parameters:

Name	Type	Description	Default
name	str	The name of the audience.	required
filters	list[RapidataFilter]	The list of filters to apply to the audience. Defaults to None (no filters).	None
target_accuracy	float	X — the fraction of validation tasks (0..1) an annotator must get right to be admitted. Defaults to None, in which case the server default of 0.75 applies.	None
min_tasks	int	N — the number of validation tasks an annotator must complete before the accuracy verdict is trusted. Defaults to None, in which case the server default of 10 applies.	None
max_tasks	int	Upper bound on admission-trial tasks before a verdict is forced. Defaults to None (no cap).	None

Returns:

Name	Type	Description
RapidataAudience	RapidataAudience	The created audience instance.

Source code in src/rapidata/rapidata_client/audience/rapidata_audience_manager.py

def create_audience(
    self,
    name: str,
    filters: list[RapidataFilter] | None = None,
    target_accuracy: float | None = None,
    min_tasks: int | None = None,
    max_tasks: int | None = None,
) -> RapidataAudience:
    """Create a new audience.

    Creates a new audience with the specified name and optional filters.

    Annotators join an audience by passing a short admission trial: they must
    answer validation tasks well enough to be trusted. The two intuitive knobs
    below describe that bar as "solve N tasks at X accuracy":

    Args:
        name (str): The name of the audience.
        filters (list[RapidataFilter], optional): The list of filters to apply to the audience. Defaults to None (no filters).
        target_accuracy (float, optional): X — the fraction of validation tasks
            (0..1) an annotator must get right to be admitted. Defaults to None,
            in which case the server default of 0.75 applies.
        min_tasks (int, optional): N — the number of validation tasks an annotator
            must complete before the accuracy verdict is trusted. Defaults to None,
            in which case the server default of 10 applies.
        max_tasks (int, optional): Upper bound on admission-trial tasks before a
            verdict is forced. Defaults to None (no cap).

    Returns:
        RapidataAudience: The created audience instance.
    """
    with tracer.start_as_current_span("RapidataAudienceManager.create_audience"):
        from rapidata.rapidata_client.audience.rapidata_audience import (
            RapidataAudience,
        )
        from rapidata.api_client.models.create_audience_endpoint_input import (
            CreateAudienceEndpointInput,
        )

        logger.debug(f"Creating audience: {name}")
        if filters is None:
            filters = []

        graduation_rule = self._build_task_accuracy_rule(
            target_accuracy=target_accuracy,
            min_tasks=min_tasks,
            max_tasks=max_tasks,
        )
        response = self._openapi_service.audience.audience_api.audience_post(
            create_audience_endpoint_input=CreateAudienceEndpointInput(
                name=name,
                filters=[filter._to_audience_model() for filter in filters],
                graduationRule=graduation_rule,
            ),
        )
        audience = RapidataAudience(
            id=response.audience_id,
            name=name,
            filters=filters,
            openapi_service=self._openapi_service,
        )
        return audience

get_audience_by_id

get_audience_by_id(audience_id: str) -> RapidataAudience

Get an audience by its ID.

Parameters:

Name	Type	Description	Default
audience_id	str	The unique identifier of the audience.	required

Returns:

Name	Type	Description
RapidataAudience	RapidataAudience	The audience instance.

Source code in src/rapidata/rapidata_client/audience/rapidata_audience_manager.py

def get_audience_by_id(self, audience_id: str) -> RapidataAudience:
    """Get an audience by its ID.

    Args:
        audience_id (str): The unique identifier of the audience.

    Returns:
        RapidataAudience: The audience instance.
    """
    with tracer.start_as_current_span("RapidataAudienceManager.get_audience_by_id"):
        from rapidata.rapidata_client.filter._backend_filter_mapper import (
            BackendFilterMapper,
        )
        from rapidata.rapidata_client.audience.rapidata_audience import (
            RapidataAudience,
        )

        logger.debug(f"Getting audience by id: {audience_id}")
        response = (
            self._openapi_service.audience.audience_api.audience_audience_id_get(
                audience_id=audience_id,
            )
        )
        return RapidataAudience(
            id=audience_id,
            name=response.name,
            filters=[
                BackendFilterMapper.backend_filter_from_rapidata_filter(filter)
                for filter in response.filters
            ],
            openapi_service=self._openapi_service,
        )

find_audiences

find_audiences(
    name: str = "", amount: int = 10, page: int = 1
) -> list[RapidataAudience]

Find your audiences by name.

Parameters:

Name	Type	Description	Default
name	str	Filter audiences by name (matching audiences will contain this string). Defaults to "" for any audience.	''
amount	int	The maximum number of audiences to return. Defaults to 10.	10
page	int	The page of audiences to return. Defaults to 1.	1

Returns:

Type	Description
list[RapidataAudience]	list[RapidataAudience]: A list of RapidataAudience instances.

Source code in src/rapidata/rapidata_client/audience/rapidata_audience_manager.py

def find_audiences(
    self, name: str = "", amount: int = 10, page: int = 1
) -> list[RapidataAudience]:
    """Find your audiences by name.

    Args:
        name (str, optional): Filter audiences by name (matching audiences will contain this string). Defaults to "" for any audience.
        amount (int, optional): The maximum number of audiences to return. Defaults to 10.
        page (int, optional): The page of audiences to return. Defaults to 1.

    Returns:
        list[RapidataAudience]: A list of RapidataAudience instances.
    """
    with tracer.start_as_current_span("RapidataAudienceManager.find_audiences"):
        from rapidata.rapidata_client.filter._backend_filter_mapper import (
            BackendFilterMapper,
        )
        from rapidata.api_client.models.audience_audience_id_jobs_get_job_id_parameter import (
            AudienceAudienceIdJobsGetJobIdParameter,
        )
        from rapidata.rapidata_client.audience.rapidata_audience import (
            RapidataAudience,
        )

        logger.debug(f"Finding audiences: {name}, {amount}")
        response = self._openapi_service.audience.audience_api.audiences_get(
            page=page,
            page_size=amount,
            name=AudienceAudienceIdJobsGetJobIdParameter(contains=name),
            sort=["-created_at"],
        )
        audiences = []
        for item in response.items:
            audiences.append(
                RapidataAudience(
                    id=item.id,
                    name=item.name,
                    filters=[
                        BackendFilterMapper.backend_filter_from_rapidata_filter(
                            filter
                        )
                        for filter in item.filters
                    ],
                    openapi_service=self._openapi_service,
                )
            )

        return audiences