Review sync: pod-side helpers

Pod-side helpers implement the manual promotion path: call LangfuseReviewArtifactReader to read completed annotation-queue items, pass the results through a custom mapper with build_promoted_records(), and write the promoted records to Langfuse with publish_promoted_records(). Use this path when you need custom business logic between reading and writing — for example a one-off script, a migration, or a promotion flow with additional filtering that declarative bindings cannot express.

For the fully-managed declarative path (no pod code running, the service polls and publishes on a schedule), see 06-review-sync.md.

All types described here are exported from semantic_evaluation.orchestration.review_sync.

---

LangfuseReviewArtifactReader

Reads and normalizes completed items from a Langfuse annotation queue. Implements the ReviewArtifactReader Protocol.

from semantic_evaluation.orchestration.review_sync import LangfuseReviewArtifactReader

reader = LangfuseReviewArtifactReader(
    langfuse_public_key=os.environ["LANGFUSE_PUBLIC_KEY"],
    langfuse_secret_key=os.environ["LANGFUSE_SECRET_KEY"],
    langfuse_base_url=os.environ["LANGFUSE_BASE_URL"],
    queue_id="<queue-uuid>",
    page_size=50,
    max_items=None,   # None = read all
)

reviewed_records = await reader.load_reviewed_records()

Each ReviewedObservationRecord in the result carries:

Field	Description
record_id	Stable identifier (the Langfuse object ID)
trace_id, observation_id	Links to the Langfuse trace
original_input, original_output	Input/output of the traced call
corrected_output	Reviewer-supplied correction, or None if approved as-is
review_scores	Dict of score names to {value, comment, score_id, data_type}
reviewer_comments	Free-text comments from the reviewer
reviewed_at	ISO 8601 timestamp when the item was completed

A reviewed trace's input, output and metadata are its root observation's; an sdk-experiment span that an evaluation run added to the trace is ignored. A queued observation is read by id. Annotation and correction scores are read through Langfuse's v3 scores API and matched to the reviewed object by the score's subject.

The reader deduplicates multiple completed versions of the same object, keeping the most recently completed one.

---

build_promoted_records()

Converts a list of ReviewedObservationRecord objects into EvalCaseRecord objects ready for Langfuse publication. The caller supplies a record_mapper callback that maps each record to a PromotedRecordMapping describing which fields become input and expected_output.

from semantic_evaluation.orchestration.review_sync import (
    build_promoted_records, PromotedRecordMapping,
)

def my_mapper(reviewed, expected_output):
    return PromotedRecordMapping(
        dataset_scope="routing",
        input=reviewed.original_input,
        expected_output=expected_output,
        trace_id=reviewed.trace_id,
    )

promoted, approved_as_is_count, correction_count = build_promoted_records(
    reviewed_records,
    product_id="pathfinder",
    dataset_version="1.6",
    created_by="team-ai-evaluation",
    record_mapper=my_mapper,
)

When corrected_output is present on a record, the mapper receives it as expected_output; otherwise it receives original_output (approved as is). The function sets review_outcome (approved_as_is or explicit_correction) on each promoted record's metadata and leaves tags to the mapper's category labels.

---

publish_promoted_records()

Publishes a list of EvalCaseRecord objects to their target Langfuse datasets. Skips unchanged records by comparing content hashes before writing.

from semantic_evaluation.orchestration.review_sync import (
    publish_promoted_records, default_dataset_name,
)
from semantic_evaluation.orchestration.staging import LangfuseDatasetDefinition

dataset_definitions = {
    "routing": LangfuseDatasetDefinition(
        name=default_dataset_name("pathfinder", "routing", "1.6", schema_type="agent"),
        description="Pathfinder routing ground truth",
        input_schema={},
        expected_output_schema={},
    ),
}

result = publish_promoted_records(
    records=promoted,
    dataset_definitions=dataset_definitions,
)
print(result.published_items, result.skipped_unchanged_items, result.failed_publish_items)

---

persist_review_sync_artifacts()

Writes a set of local JSON files to an outputs_dir directory for later inspection. Called automatically by the service on each completed job (when persist_artifacts=True). Pod-side flows can call it directly.

Each artifact is written twice: as a last_*.json file (always overwritten, easy to inspect the most recent run) and as a timestamped *_{timestamp}.json file (for historical comparison):

Artifact	Content
last_reviewed_records.json	All normalized ReviewedObservationRecord objects
last_promoted_records.json	Records that were published or attempted
last_pending_reviewed_records.json	Records still awaiting review completion
last_handled_reviewed_records.json	Records skipped (already handled or policy-skipped)
last_sync_summary.json	JobSummary in full
last_cleanup_summary.json	Queue cleanup outcome summary
last_checkpoint_before.json, last_checkpoint_after.json	Checkpoint state snapshots