diff --git a/docs/signals.md b/docs/signals.md new file mode 100644 index 000000000..4109ce00f --- /dev/null +++ b/docs/signals.md @@ -0,0 +1,116 @@ +# Signals + +A **signal** runs the same labeling job on a repeating schedule: bind a +[job definition](job_definition_parameters.md) to an [audience](audiences.md) +and an interval, and Rapidata creates a new [job](understanding_the_results.md) +on every tick. + +## What a signal is + +A signal ties together three things: + +- an **audience** — who labels the data, +- a **job definition** — the task that gets run, +- an **interval** — how often it fires, in hours. + +Each firing creates one `RapidataJob`, identical to a job you create directly. +A signal is just a scheduler that keeps producing those jobs; every job runs the +same job definition against the same audience. + +```mermaid +graph LR + S[Signal
audience + job definition + interval] -->|every interval| J1[Job 1] + S -->|every interval| J2[Job 2] + S -->|every interval| J3[Job 3] +``` + +## Creating a signal + +```py +from rapidata import RapidataClient + +client = RapidataClient() + +audience = client.audience.get_audience_by_id("aud_MU1GZYoESyO") + +job_definition = client.job.create_compare_job_definition( + name="Prompt Alignment Job", + instruction="Which image follows the prompt more accurately?", + datapoints=[ + ["https://assets.rapidata.ai/flux_book.jpg", + "https://assets.rapidata.ai/mj_book.jpg"] + ], + contexts=["A small blue book sitting on a large red book."], +) + +signal = client.signals.create_signal( + name="Daily prompt alignment", + audience=audience, + job_definition=job_definition, + interval_hours=24, +) +``` + +`audience` and `job_definition` also accept id strings. By default the signal +uses the latest revision of the job definition at fire time; pin one with +`revision_number=...`. Set `is_public=True` to let others in your organization +read the signal. + +## The jobs a signal creates + +Every firing creates a `RapidataJob`: + +```py +for job in signal.get_jobs(page_size=10): + print(job, job.get_status()) + +results = signal.get_jobs(page_size=1)[0].get_results() +``` + +A firing can be skipped (for example if the previous job hasn't finished). A +skipped firing creates no job and won't appear in `get_jobs()`. + +## Triggering a job on demand + +Fire one job immediately instead of waiting for the schedule: + +```py +signal.trigger() +job = signal.wait_for_next_job(timeout=600) +print(job.get_results()) +``` + +`trigger()` returns right away; the job is created in the background. +`wait_for_next_job()` blocks until the next firing has created its job and +returns it. + +## Managing a signal + +```py +signal.pause() +signal.resume() +signal.update(name="Hourly prompt alignment", interval_hours=1) +signal.delete() +``` + +Look signals up later: + +```py +signal = client.signals.get_signal_by_id("signal_id") +signals = client.signals.find_signals(name="alignment") +``` + +## Property reference + +| Property | Description | +|---|---| +| `id` | The signal's unique id. | +| `name` / `description` | Display name and optional description. | +| `audience_id` | The audience each job targets. | +| `job_definition_id` | The job definition each job is created from. | +| `revision_number` | Pinned job-definition revision, or `None` for "latest at fire time". | +| `interval_hours` | How often the signal fires, in hours. | +| `next_run_at` / `last_run_at` | Timestamps of the next and most recent firings. | +| `is_paused` | Whether the scheduler is currently skipping this signal. | +| `is_public` | Whether other users can discover and read it. | +| `created_at` | When the signal was created. | diff --git a/mkdocs.yml b/mkdocs.yml index c6dbcf34d..9a3e2117d 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -59,6 +59,7 @@ plugins: - starting_page.md - quickstart.md - audiences.md + - signals.md - job_definition_parameters.md - understanding_the_results.md - error_handling.md @@ -120,6 +121,7 @@ nav: - Overview: starting_page.md - Quick Start: quickstart.md - Custom Audiences: audiences.md + - Signals: signals.md - Parameter Reference: job_definition_parameters.md - Understanding Results: understanding_the_results.md - Early Stopping: confidence_stopping.md diff --git a/src/rapidata/__init__.py b/src/rapidata/__init__.py index c2fe4d086..76f2b32ca 100644 --- a/src/rapidata/__init__.py +++ b/src/rapidata/__init__.py @@ -11,6 +11,8 @@ RapidataJob, RapidataJobDefinition, RapidataJobManager, + RapidataSignal, + RapidataSignalManager, ValidationSetManager, RapidataValidationSet, Box, diff --git a/src/rapidata/rapidata_client/__init__.py b/src/rapidata/rapidata_client/__init__.py index 07b312c02..c375713ba 100644 --- a/src/rapidata/rapidata_client/__init__.py +++ b/src/rapidata/rapidata_client/__init__.py @@ -7,6 +7,7 @@ ) from .order import RapidataOrderManager, RapidataOrder from .job import RapidataJob, RapidataJobDefinition, RapidataJobManager +from .signal import RapidataSignal, RapidataSignalManager from .validation import ValidationSetManager, RapidataValidationSet, Box from .results import RapidataResults from .selection import ( diff --git a/src/rapidata/rapidata_client/rapidata_client.py b/src/rapidata/rapidata_client/rapidata_client.py index e41937b18..2546f50ad 100644 --- a/src/rapidata/rapidata_client/rapidata_client.py +++ b/src/rapidata/rapidata_client/rapidata_client.py @@ -37,6 +37,9 @@ from rapidata.rapidata_client.datapoints._asset_uploader import AssetUploader from rapidata.rapidata_client.job.rapidata_job_manager import RapidataJobManager from rapidata.rapidata_client.flow.rapidata_flow_manager import RapidataFlowManager +from rapidata.rapidata_client.signal.rapidata_signal_manager import ( + RapidataSignalManager, +) from rapidata.rapidata_client.api.rapidata_api_client import ( optional_api_call, mark_sdk_outdated, @@ -108,6 +111,8 @@ def __init__( audience (RapidataAudienceManager): The RapidataAudienceManager instance. job (JobManager): The JobManager instance. mri (RapidataBenchmarkManager): The RapidataBenchmarkManager instance. + signals (RapidataSignalManager): The RapidataSignalManager instance for managing + recurring audience-job schedules (signals) and observing their runs. context (ContextManager): The ContextManager instance for shortening datapoint contexts against a question. """ @@ -165,6 +170,11 @@ def __init__( logger.debug("Initializing RapidataBenchmarkManager") self.mri = RapidataBenchmarkManager(openapi_service=self._openapi_service) + logger.debug("Initializing RapidataSignalManager") + self.signals = RapidataSignalManager( + openapi_service=self._openapi_service + ) + logger.debug("Initializing RapidataAudienceManager") self.audience = RapidataAudienceManager( openapi_service=self._openapi_service diff --git a/src/rapidata/rapidata_client/signal/__init__.py b/src/rapidata/rapidata_client/signal/__init__.py new file mode 100644 index 000000000..948e8085f --- /dev/null +++ b/src/rapidata/rapidata_client/signal/__init__.py @@ -0,0 +1,2 @@ +from .rapidata_signal import RapidataSignal +from .rapidata_signal_manager import RapidataSignalManager diff --git a/src/rapidata/rapidata_client/signal/rapidata_signal.py b/src/rapidata/rapidata_client/signal/rapidata_signal.py new file mode 100644 index 000000000..a57e6d879 --- /dev/null +++ b/src/rapidata/rapidata_client/signal/rapidata_signal.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from time import monotonic, sleep +from typing import TYPE_CHECKING, Union + +from rapidata.rapidata_client.config import logger, managed_print, tracer + +if TYPE_CHECKING: + from rapidata.api_client.models.get_signal_by_id_endpoint_output import ( + GetSignalByIdEndpointOutput, + ) + from rapidata.api_client.models.query_signals_endpoint_output import ( + QuerySignalsEndpointOutput, + ) + from rapidata.rapidata_client.job.rapidata_job import RapidataJob + from rapidata.service.openapi_service import OpenAPIService + + SignalOutput = Union[GetSignalByIdEndpointOutput, QuerySignalsEndpointOutput] + + +class RapidataSignal: + """A live handle to a Rapidata signal. + + A signal creates one audience job on a recurring schedule — every interval tick + spawns a :class:`RapidataJob` built from the signal's job definition and audience. + Use the manager (:py:attr:`RapidataClient.signals`) to create or look up signals; + use the methods here to control a signal and reach the jobs it produces. + + This is a *live* handle: identity fields (``id``, ``audience_id``, + ``job_definition_id``, ``created_at`` …) are fixed at creation, while mutable state + (``name``, ``is_paused``, ``next_run_at`` …) is re-fetched from the server on every + access so it never goes stale. + """ + + def __init__( + self, + openapi_service: OpenAPIService, + data: SignalOutput, + ): + self._openapi_service = openapi_service + # Identity / immutable fields — safe to cache for the object's lifetime. + self.id: str = data.id + self.audience_id: str = data.audience_id + self.job_definition_id: str = data.job_definition_id + self.revision_number: int | None = data.revision_number + self.is_public: bool = data.is_public + self.created_at: datetime = data.created_at + # Last-known name, kept only as a display label for __str__/logging; the `name` + # property always reads the live value. + self._name: str = data.name + logger.debug("RapidataSignal initialized: %s", self.id) + + def _latest(self) -> GetSignalByIdEndpointOutput: + """Fetch the current server-side state of this signal.""" + return self._openapi_service.signal.signal_api.signal_signal_id_get(self.id) + + @property + def name(self) -> str: + return self._latest().name + + @property + def description(self) -> str | None: + return self._latest().description + + @property + def interval_hours(self) -> float: + return self._latest().interval_seconds / 3600 + + @property + def next_run_at(self) -> datetime: + return self._latest().next_run_at + + @property + def last_run_at(self) -> datetime | None: + return self._latest().last_run_at + + @property + def is_paused(self) -> bool: + return self._latest().is_paused + + def pause(self) -> RapidataSignal: + """Pause the signal. Scheduled jobs stop until :py:meth:`resume` is called. + + Returns: + RapidataSignal: ``self`` for chaining. + """ + with tracer.start_as_current_span("RapidataSignal.pause"): + logger.info("Pausing signal '%s'", self.id) + self._openapi_service.signal.signal_api.signal_signal_id_pause_post(self.id) + managed_print(f"Signal '{self}' has been paused.") + return self + + def resume(self) -> RapidataSignal: + """Resume a paused signal. Scheduled jobs start firing again at the configured interval. + + Returns: + RapidataSignal: ``self`` for chaining. + """ + with tracer.start_as_current_span("RapidataSignal.resume"): + logger.info("Resuming signal '%s'", self.id) + self._openapi_service.signal.signal_api.signal_signal_id_resume_post( + self.id + ) + managed_print(f"Signal '{self}' has been resumed.") + return self + + def delete(self) -> None: + """Delete the signal. Jobs it already created are unaffected.""" + with tracer.start_as_current_span("RapidataSignal.delete"): + logger.info("Deleting signal '%s'", self.id) + self._openapi_service.signal.signal_api.signal_signal_id_delete(self.id) + managed_print(f"Signal '{self}' has been deleted.") + + def trigger(self) -> None: + """Trigger the signal manually, creating one extra job outside the schedule. + + The job is created asynchronously; use :py:meth:`wait_for_next_job` to block + until it appears. + """ + with tracer.start_as_current_span("RapidataSignal.trigger"): + logger.info("Triggering signal '%s'", self.id) + self._openapi_service.signal.signal_api.signal_signal_id_trigger_post( + self.id + ) + + def update( + self, + *, + name: str | None = None, + description: str | None = None, + interval_hours: float | None = None, + ) -> RapidataSignal: + """Update mutable fields on the signal. + + Args: + name: New display name. Omit to leave unchanged. + description: New description. Omit to leave unchanged. + interval_hours: New scheduling interval, in hours. Omit to leave unchanged. + + Returns: + RapidataSignal: ``self`` for chaining. + """ + if name is None and description is None and interval_hours is None: + # Nothing requested — avoid a useless round trip. + return self + + from rapidata.api_client.models.update_signal_endpoint_input import ( + UpdateSignalEndpointInput, + ) + + with tracer.start_as_current_span("RapidataSignal.update"): + self._openapi_service.signal.signal_api.signal_signal_id_patch( + self.id, + UpdateSignalEndpointInput( + name=name, + description=description, + intervalSeconds=( + None if interval_hours is None else round(interval_hours * 3600) + ), + ), + ) + if name is not None: + self._name = name + return self + + def get_jobs( + self, + page: int = 1, + page_size: int = 20, + sort_descending: bool = True, + ) -> list[RapidataJob]: + """List the jobs this signal has created (newest first by default). + + Each scheduled or manual firing that spawned a job is returned as a live + :class:`RapidataJob`. Firings that were skipped without creating a job are + not included. + + Args: + page: 1-indexed page number. + page_size: Number of firings to inspect per page. + sort_descending: When ``True`` (default), newest jobs come first. + + Returns: + list[RapidataJob]: The jobs on the requested page. + """ + from rapidata.rapidata_client.job.rapidata_job_manager import ( + RapidataJobManager, + ) + + with tracer.start_as_current_span("RapidataSignal.get_jobs"): + runs = self._openapi_service.signal.signal_api.signal_signal_id_run_get( + self.id, + page=page, + page_size=page_size, + sort=["-started_at" if sort_descending else "started_at"], + ).items + job_manager = RapidataJobManager(openapi_service=self._openapi_service) + return [ + job_manager.get_job_by_id(run.audience_job_id) + for run in runs + if run.audience_job_id + ] + + def wait_for_next_job( + self, + timeout: float = 300, + poll_interval: float = 5.0, + ) -> RapidataJob: + """Block until the signal's next firing creates a job, then return it. + + Waits for a firing that started *after* this call and has spawned a job + (skipped firings are ignored). Handy right after :py:meth:`trigger`. The + returned :class:`RapidataJob` is live — call ``get_results()`` or + ``display_progress_bar()`` on it to follow the job to completion. + + Args: + timeout: Maximum seconds to wait. Raises :py:class:`TimeoutError` on expiry. + poll_interval: Seconds between polls. + + Returns: + RapidataJob: The job created by the next firing. + + Raises: + TimeoutError: If no new job is created within ``timeout`` seconds. + """ + if timeout <= 0: + raise ValueError("timeout must be positive") + if poll_interval <= 0: + raise ValueError("poll_interval must be positive") + + from rapidata.rapidata_client.job.rapidata_job_manager import ( + RapidataJobManager, + ) + + with tracer.start_as_current_span("RapidataSignal.wait_for_next_job"): + started_waiting = datetime.now(timezone.utc) + deadline = monotonic() + timeout + job_manager = RapidataJobManager(openapi_service=self._openapi_service) + + logger.info( + "Waiting up to %.0fs for the next job of signal '%s'", + timeout, + self.id, + ) + while True: + runs = self._openapi_service.signal.signal_api.signal_signal_id_run_get( + self.id, page=1, page_size=20, sort=["-started_at"] + ).items + # Newest-first, so the last match is the earliest new firing with a job. + fresh_job_ids = [ + run.audience_job_id + for run in runs + if run.started_at >= started_waiting and run.audience_job_id + ] + if fresh_job_ids: + return job_manager.get_job_by_id(fresh_job_ids[-1]) + + if monotonic() >= deadline: + raise TimeoutError( + f"No new job from signal '{self.id}' within {timeout} seconds." + ) + sleep(poll_interval) + + def __str__(self) -> str: + return f"RapidataSignal(name='{self._name}', id='{self.id}')" + + def __repr__(self) -> str: + return self.__str__() diff --git a/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py b/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py new file mode 100644 index 000000000..128c86c6b --- /dev/null +++ b/src/rapidata/rapidata_client/signal/rapidata_signal_manager.py @@ -0,0 +1,154 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +from rapidata.rapidata_client.config import logger, tracer +from rapidata.rapidata_client.signal.rapidata_signal import RapidataSignal + +if TYPE_CHECKING: + from rapidata.rapidata_client.audience._audience_base import RapidataAudienceBase + from rapidata.rapidata_client.job.rapidata_job_definition import ( + RapidataJobDefinition, + ) + from rapidata.service.openapi_service import OpenAPIService + + +class RapidataSignalManager: + """Manage signals: schedules that periodically create audience jobs. + + A signal binds an audience to a job definition and an interval. The signal service + creates one audience job per interval tick. Use this manager to create, look up, + and find signals; use methods on the returned :class:`RapidataSignal` to control a + specific signal and reach the jobs it produces. + + Access this manager via :py:attr:`RapidataClient.signals`. + """ + + def __init__(self, openapi_service: OpenAPIService): + self._openapi_service = openapi_service + logger.debug("RapidataSignalManager initialized") + + def create_signal( + self, + name: str, + audience: str | RapidataAudienceBase, + job_definition: str | RapidataJobDefinition, + interval_hours: float, + description: str | None = None, + revision_number: int | None = None, + is_public: bool = False, + ) -> RapidataSignal: + """Create a new signal. + + Args: + name: Display name for the signal. + audience: The audience each job targets — a :class:`RapidataAudience`, + a :class:`RapidataFilteredAudience`, or an audience id. + job_definition: The job definition each job is created from — a + :class:`RapidataJobDefinition` or a job definition id. + interval_hours: How often the signal fires, in hours. + description: Optional human-readable description. + revision_number: Optional explicit revision of ``job_definition`` to pin + this signal to. If omitted, the signal follows the latest revision. + is_public: Whether other users can discover and read this signal. + + Returns: + RapidataSignal: The created signal. + """ + if interval_hours <= 0: + raise ValueError("interval_hours must be positive") + + from rapidata.rapidata_client.audience._audience_base import ( + RapidataAudienceBase, + ) + from rapidata.rapidata_client.job.rapidata_job_definition import ( + RapidataJobDefinition, + ) + from rapidata.api_client.models.create_signal_endpoint_input import ( + CreateSignalEndpointInput, + ) + + audience_id = ( + audience.id if isinstance(audience, RapidataAudienceBase) else audience + ) + job_definition_id = ( + job_definition.id + if isinstance(job_definition, RapidataJobDefinition) + else job_definition + ) + + with tracer.start_as_current_span("RapidataSignalManager.create_signal"): + logger.debug( + "Creating signal '%s' (audience=%s, job_definition=%s, interval=%sh)", + name, + audience_id, + job_definition_id, + interval_hours, + ) + created = self._openapi_service.signal.signal_api.signal_post( + CreateSignalEndpointInput( + name=name, + description=description, + audienceId=audience_id, + jobDefinitionId=job_definition_id, + revisionNumber=revision_number, + intervalSeconds=round(interval_hours * 3600), + isPublic=is_public, + ) + ) + # The create endpoint only echoes the id; fetch the full signal to populate. + return self.get_signal_by_id(created.id) + + def get_signal_by_id(self, signal_id: str) -> RapidataSignal: + """Get a signal by ID. + + Args: + signal_id: The signal to fetch. + + Returns: + RapidataSignal: The signal. + """ + with tracer.start_as_current_span("RapidataSignalManager.get_signal_by_id"): + data = self._openapi_service.signal.signal_api.signal_signal_id_get( + signal_id + ) + return RapidataSignal(self._openapi_service, data) + + def find_signals( + self, + name: str = "", + amount: int = 10, + page: int = 1, + ) -> list[RapidataSignal]: + """Find signals visible to you (your own signals plus public ones). + + Args: + name: Filter signals by name (matching signals will contain this string). + Defaults to "" for any signal. + amount: The maximum number of signals to return. Defaults to 10. + page: The page of signals to return. Defaults to 1. + + Returns: + list[RapidataSignal]: The matching signals. + """ + from rapidata.api_client.models.audience_audience_id_jobs_get_job_id_parameter import ( + AudienceAudienceIdJobsGetJobIdParameter, + ) + + with tracer.start_as_current_span("RapidataSignalManager.find_signals"): + logger.debug("Finding signals: %s, %s", name, amount) + result = self._openapi_service.signal.signal_api.signal_get( + page=page, + page_size=amount, + name=AudienceAudienceIdJobsGetJobIdParameter(contains=name), + sort=["-created_at"], + ) + return [ + RapidataSignal(self._openapi_service, item) for item in result.items + ] + + def __str__(self) -> str: + return "RapidataSignalManager" + + def __repr__(self) -> str: + return self.__str__() diff --git a/src/rapidata/service/openapi_service.py b/src/rapidata/service/openapi_service.py index d62c04b0e..615408eff 100644 --- a/src/rapidata/service/openapi_service.py +++ b/src/rapidata/service/openapi_service.py @@ -22,6 +22,7 @@ from rapidata.service.services.workflow_service import WorkflowService from rapidata.service.services.leaderboard_service import LeaderboardService from rapidata.service.services.rapid_service import RapidService + from rapidata.service.services.signal_service import SignalService from rapidata.service.services.translation_service import TranslationService @@ -76,6 +77,7 @@ def __init__( self._workflow: WorkflowService | None = None self._leaderboard: LeaderboardService | None = None self._rapid: RapidService | None = None + self._signal: SignalService | None = None self._translation: TranslationService | None = None if token: @@ -213,6 +215,13 @@ def rapid(self) -> RapidService: self._rapid = RapidService(self.api_client) return self._rapid + @property + def signal(self) -> SignalService: + if self._signal is None: + from rapidata.service.services.signal_service import SignalService + self._signal = SignalService(self.api_client) + return self._signal + @property def translation(self) -> TranslationService: if self._translation is None: diff --git a/src/rapidata/service/services/__init__.py b/src/rapidata/service/services/__init__.py index 12c007108..0cf244dd3 100644 --- a/src/rapidata/service/services/__init__.py +++ b/src/rapidata/service/services/__init__.py @@ -7,6 +7,7 @@ from rapidata.service.services.order_service import OrderService from rapidata.service.services.pipeline_service import PipelineService from rapidata.service.services.rapid_service import RapidService +from rapidata.service.services.signal_service import SignalService from rapidata.service.services.validation_service import ValidationService from rapidata.service.services.workflow_service import WorkflowService @@ -20,6 +21,7 @@ "OrderService", "PipelineService", "RapidService", + "SignalService", "ValidationService", "WorkflowService", ] diff --git a/src/rapidata/service/services/signal_service.py b/src/rapidata/service/services/signal_service.py new file mode 100644 index 000000000..00de11327 --- /dev/null +++ b/src/rapidata/service/services/signal_service.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from rapidata.api_client.api.signal_api import SignalApi + from rapidata.rapidata_client.api.rapidata_api_client import RapidataApiClient + + +class SignalService: + def __init__(self, api_client: RapidataApiClient) -> None: + self._api_client = api_client + self._signal_api: SignalApi | None = None + + @property + def signal_api(self) -> SignalApi: + if self._signal_api is None: + from rapidata.api_client.api.signal_api import SignalApi + self._signal_api = SignalApi(self._api_client) + return self._signal_api diff --git a/uv.lock b/uv.lock index 23c812895..9f10b5e1b 100644 --- a/uv.lock +++ b/uv.lock @@ -2570,7 +2570,7 @@ wheels = [ [[package]] name = "rapidata" -version = "3.15.0" +version = "3.15.1" source = { editable = "." } dependencies = [ { name = "authlib" },