Add a custom annotator¶

Goal¶

Add a Python annotator class that produces inference-grounded text from already-labeled Phase 1 records. The shipped extension path is a MetadataAnnotator subclass with a custom Templater and optional package entry point.

When to use this¶

Use this when you need a new prompt family, provider wiring, or packaging unit around the implemented AnnotationType values. The current enum values are CAPTION, QA, REASONING, SCENE_REPORT, and CONTRASTIVE. Adding a new top-level annotation type is a framework enum change, not a plugin-only change.

If you only need a new task string for an existing annotation type, see Add a custom annotation template. That path can pass a Templater directly without declaring a new class.

Prerequisites¶

Read Concepts / Annotations, particularly the Grounding Contract. The annotator may only consume metadata fields present on the StoredRecord. It must not read IQ, spectrograms, or raw samples.

Inference calls go through BaseInferenceClient, which abstracts Gemini, Anthropic, OpenAI, vLLM, and offline test clients behind one interface. Do not instantiate provider SDKs inside an annotator.

Minimal implementation¶

Subclass MetadataAnnotator and declare a stable name. The inherited annotate(...) method already matches the shipped public contract: annotate(record: StoredRecord, *, annotation_type: AnnotationType, template_id: str, run_id: str) -> StoredRecord.

from typing import ClassVar

from rfgen.annotators import MetadataAnnotator, Templater
from rfgen.enums import AnnotationType


class EventSummaryAnnotator(MetadataAnnotator):
    name: ClassVar[str] = "event_summary"


templater = Templater(
    templates={
        "scene_report.event_summary.v1": (
            "Write a short scene report from the whitelisted FACTS. "
            "Summarize emitter count, class names, occupied bandwidth, and "
            "SNR only when those facts are present."
        )
    }
)

annotator = EventSummaryAnnotator.from_components(
    client=client,
    templater=templater,
    allowed_annotation_types=[AnnotationType.SCENE_REPORT],
)

Run it on an IQ-stripped stored record:

from rfgen.enums import AnnotationType

annotated = annotator.annotate(
    stored_record,
    annotation_type=AnnotationType.SCENE_REPORT,
    template_id="scene_report.event_summary.v1",
    run_id="event-summary-v1",
)

annotated is a new StoredRecord whose text field contains the annotation row. The row is keyed by annotation_type, template_id, and run_id; it includes the validated JSON output, model id, request id, token usage, optional PAES score, and schema version.

Register¶

For an in-tree call path, import the class directly and construct it with from_components(...) or from_config(...).

For a third-party package, declare the entry-point group rfgen.annotators in your package metadata. The value points at the annotator class.

[project.entry-points."rfgen.annotators"]
event_summary = "my_pkg.annotators:EventSummaryAnnotator"

The package can also declare a top-level PLUGIN: PluginMetadata attribute with plugin_kind="annotator"; see Reference / Plugin Metadata / Plugin registration for the full schema.

Configure¶

The shipped AnnotatorConfig schema accepts enabled, types, bulk_llm, verifier_llm, and verifier_subset_pct. Template routing and plugin-specific construction are passed through Python.

from rfgen.config.annotation import AnnotatorConfig, LLMConfig
from rfgen.enums import AnnotationType

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

annotator = EventSummaryAnnotator.from_config(
    config,
    bulk_client=client,
    templater=templater,
)

Verify¶

Confirm the behavior through public outputs:

annotator.name == "event_summary".
Calling annotate(...) returns a StoredRecord, not a dict.
The returned StoredRecord.text["scene_report"]["scene_report.event_summary.v1"]["event-summary-v1"] row has annotation_type == "scene_report".
The prompt contains only whitelisted metadata facts from the StoredRecord.
Re-running with the same (annotation_type, template_id, run_id) returns the existing row unchanged.

For plugin discovery, construct an EntryPointRegistry("rfgen.annotators"), call discover(), then get("event_summary") and instantiate the returned class through from_config(...) or from_components(...).

Troubleshoot¶

Symptom	Fix
Plugin discovery does not find the annotator	Check the `rfgen.annotators` entry point and ensure it points at the class.
Inference response fails to parse	Use `json_schema_mode=True` and keep the prompt aligned with the built-in annotation-type output schema.
Annotator references unwhitelisted fields	Restrict the template to fields present after `whitelist_filter(...)`.
Re-runs duplicate text outputs	Reuse the same `run_id` and `template_id`; use a new value only for a distinct text run.