Annotate an existing dataset¶

Start from a dataset that already contains in-phase/quadrature (IQ) samples, structured labels, and metadata, then produce text annotations such as captions, Q&A, reasoning, scene reports, or contrastive text. In rfgen terms, this runs Phase 2 annotation over records written by the Phase 1 storage layer.

The shipped Phase 2 paths are the rfgen annotate command, a synchronous local orchestrator, five managed batch orchestrators, and the lower-level Python annotator API.

Prerequisites¶

Install the inference-client dependency for the provider you use. Local vLLM uses the OpenAI-compatible HTTP surface.

pip install "rfgen[openai-batch]"
pip install "rfgen[anthropic-batch]"
pip install "rfgen[gcp]"

Set only the provider keys selected by your AnnotatorConfig.

export OPENAI_API_KEY=sk-...
export ANTHROPIC_API_KEY=sk-ant-...
export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account.json

Use an existing store as both input and append target. The storage API keeps the Phase 1 canonical record bytes immutable and appends Phase 2 text rows through StoreHandle.append_annotation_row.

Run the bundled local command¶

The bundled annotation config is a deterministic local example. It contains one job and one matching local_responses entry, so it runs without a provider SDK or credentials.

rfgen annotate

To append the result to an existing dataset, set its URI and storage backend:

rfgen annotate \
  store_uri=./out/phase1 \
  store_backend=zarr_local

The command composes annotation.yaml through Hydra, validates every job as a frozen AnnotationJob, and dispatches through the rfgen.commands and rfgen.annotation_orchestrators registries. Each job requires sample_id, annotation_type, template_id, run_id, prompt, and output_schema.

Submit a managed batch¶

Select a managed orchestrator and supply its typed resources. This Vertex example uses the selected stable Gemini model:

store_uri: gs://rfgen-datasets/phase1
store_backend: zarr_gcs
orchestrator_config:
  name: vertex_batch
  min_poll_interval_s: 10.0
  resources:
    project: ${oc.env:GCP_PROJECT}
    location: us-central1
    model: gemini-3.1-flash-lite
    input_uri: gs://rfgen-batch/input/
    output_uri: gs://rfgen-batch/output/
    retain_staging: false
jobs:
  - sample_id: sample-0001
    annotation_type: caption
    template_id: caption.dense.v3
    run_id: text-v1
    prompt: Describe the RF activity in this record.
    output_schema:
      type: object
      required: [caption]
      properties:
        caption: {type: string}
local_responses: {}

Save the file as managed-annotation.yaml, then run:

rfgen annotate --config-dir . --config-name managed-annotation

If the provider batch is still pending, the command prints a serialized handle and exits successfully. Store that JSON as durable run state. A later command must provide the JSON through resume_handle_json and leave jobs empty; supplying both is rejected.

# resume-annotation.yaml
store_uri: gs://rfgen-datasets/phase1
store_backend: zarr_gcs
orchestrator_config:
  name: vertex_batch
  min_poll_interval_s: 10.0
  resources:
    project: ${oc.env:GCP_PROJECT}
    location: us-central1
    model: gemini-3.1-flash-lite
    input_uri: gs://rfgen-batch/input/
    output_uri: gs://rfgen-batch/output/
    retain_staging: false
jobs: []
local_responses: {}
resume_handle_json: '<serialized handle JSON>'

rfgen annotate --config-dir . --config-name resume-annotation

Polling uses a monotonic clock and makes at most one provider poll request per min_poll_interval_s for each handle. Calls inside that interval return the cached status.

Retry persistence and cleanup safely¶

The stable job key is (sample_id, annotation_type, template_id, run_id). Submitting a key that is already durable reuses its stored result instead of submitting and billing it again.

Managed fetch is at-least-once. The handle checkpoints normalized provider results before persistence and records each key after its append-only store write succeeds. If persistence fails partway through a batch, resume with the latest serialized handle. The retry skips durable keys and continues with the first unpersisted result without refetching checkpointed provider output.

Provider cleanup begins only after every result is durable. If cleanup fails, the error remains visible and the handle remains incomplete. Retry with the same handle; the orchestrator repeats cleanup without rewriting annotations or refetching provider output. retain_staging: false deletes managed staging on successful cleanup, while true preserves it.

Build the annotator¶

Create a shipped AnnotatorConfig and construct a MetadataAnnotator. The example enables all five implemented annotation types.

from rfgen.annotators import MetadataAnnotator
from rfgen.config.annotation import AnnotatorConfig, LLMConfig
from rfgen.enums import AnnotationType
from rfgen.inference.clients import OpenAIClient

config = AnnotatorConfig(
    enabled=True,
    types=[
        AnnotationType.CAPTION,
        AnnotationType.QA,
        AnnotationType.REASONING,
        AnnotationType.SCENE_REPORT,
        AnnotationType.CONTRASTIVE,
    ],
    bulk_llm=LLMConfig(
        provider="openai",
        model="gpt-4o-mini",
        temperature=0.2,
        max_tokens=512,
        json_schema_mode=True,
    ),
)

client = OpenAIClient.from_llm_config(config.bulk_llm)
annotator = MetadataAnnotator.from_config(config, bulk_client=client)

For one annotation type, restrict types:

config = AnnotatorConfig(
    enabled=True,
    types=[AnnotationType.REASONING],
    bulk_llm=LLMConfig(provider="openai", model="gpt-4o-mini"),
)

Run synchronously¶

The annotator receives a StoredRecord, which is a metadata-only view. For existing Phase 1 stores, call annotate_store(...): it reads the full record from an open StoreHandle, strips IQ through whitelist_filter, and appends the text row under the existing sample_id without rewriting IQ, scene, or labels.

from rfgen.annotators import MetadataAnnotator
from rfgen.enums import AnnotationType, StoreMode
from rfgen.storage import ZarrStore

dataset_uri = "./out/phase1"
run_id = "text-v1"

store = ZarrStore(uri=dataset_uri)

with store.open(dataset_uri, StoreMode.APPEND) as handle:
    for sample_id in handle.iter_sample_ids():
        for annotation_type in config.types:
            template_id = {
                AnnotationType.CAPTION: "caption.dense.v3",
                AnnotationType.QA: "qa.attribute.v1",
                AnnotationType.REASONING: "reasoning.physical.v1",
                AnnotationType.SCENE_REPORT: "scene_report.summary.v1",
                AnnotationType.CONTRASTIVE: "contrastive.hard_negative.v1",
            }[annotation_type]
            stored = annotator.annotate_store(
                handle,
                sample_id=sample_id,
                annotation_type=annotation_type,
                template_id=template_id,
                run_id=run_id,
            )

        assert stored.text is not None

Re-running annotate_store(...) with the same (sample_id, annotation_type, template_id, run_id) is idempotent. The store keeps the original sample bytes and appends only the new annotation row. Use a new run_id or template_id when you want a distinct text run.

Add verification¶

Set verifier_subset_pct=100.0 and provide verifier_llm plus a PAES metric registry when you want every record handed to annotate(...) verified. For partial verification, preselect sample IDs with select_paes_verifier_subset and construct the annotator through from_components(...) with verifier_subset_sample_ids.

config = AnnotatorConfig(
    enabled=True,
    types=[AnnotationType.CAPTION],
    bulk_llm=LLMConfig(provider="openai", model="gpt-4o-mini"),
    verifier_llm=LLMConfig(provider="anthropic", model="claude-sonnet-4-6"),
    verifier_subset_pct=100.0,
)

PAES support depends on a registered rfgen.metrics entry named paes; if the metric is not registered, annotate(...) raises an AnnotationError instead of silently skipping verification.

Inspect the result¶

Open the same dataset and read back the text field from a record:

with ZarrStore(uri=dataset_uri).open(dataset_uri, StoreMode.READ) as store:
    sample_id = next(store.iter_sample_ids())
    annotated = store.read(sample_id)
    print(annotated.text)

The text dictionary is keyed by annotation type, template id, and run id. Each row records the JSON output, model id, request id, token usage, optional PAES score, and schema version.