Skip to main content

task

Prefect task for model inference (unified for fovea and GA models).

This single task replaces the separate fovea and GA inference ops. The flow invokes it twice with different configs.

Module​

Functions​

model_inference_task​

async def model_inference_task(    datasource: BaseSource,    model_resource: ModelResourceProtocol,    config: ModelInferenceConfig,    schema: BitfountSchema,    datastructure: DataStructure,    cache: CacheProtocol,    task_hash: str,    project_id: str,    run_id: str | None = None,    lifecycle_notifier: LifecycleNotifier | None = None,) ‑> ModelInferenceResult:

Run model inference on the datasource and persist results to cache.

RAW inference results are written directly to the model_inferences cache table via bulk_insert_inference_results — no postprocessing is applied here. Postprocessing is format-specific to each consumer, so it runs just-in-time inside the downstream calc step (see bitfount.preprocessing) rather than at cache time. Caching the raw output lets any task on the same datasource share these rows regardless of how it later postprocesses them. Downstream interactive steps access the rows through a CacheAccessor (<step_name>.cache) rather than receiving an in-memory DataFrame.

Arguments

  • datasource: The datasource to run inference on.
  • model_resource: Model resource used to load the model.
  • config: Model inference configuration (model ref, version, batch size).
  • schema: Datasource schema.
  • datastructure: Datasource data structure.
  • cache: Cache instance to persist inference results into.
  • task_hash: Pipeline task hash identifying this run; used as the partition key in the model_inferences table.
  • project_id: Provenance only — the project that triggered this run. NOT part of the cache key: rows are keyed by (task_hash, model_ref, model_version, file_id) so two projects on the same datasource share them.
  • run_id: The run_id for this phase of execution.
  • lifecycle_notifier: Optional notifier used to signal the batch count (NUMBER_OF_BATCHES) and per-batch ids (CURRENT_BATCH_ID) to the task initiator. Each CURRENT_BATCH_ID(n) is sent right before the n-th file chunk is inferred (mirroring the v8 worker batch loop), so the ids track actual progress. The phase-4 marker (PREPARING_DATA_BATCHES) and the batches-complete boundary are emitted once per DAG by the executor, not here. None (the default) disables signalling — e.g. background runs, which have no initiator mailbox.

Returns ModelInferenceResult with the number of records stored.