taskset.py

이 글의 원장 예제에 포함된 코드입니다. 압축을 푼 폴더의 rewrite-v2/examples/verifiers/ledger_reconciliation_v1/ledger_reconciliation_v1/taskset.py와 같습니다.

전체 예제 ZIP / 실행법

"""Original synthetic ledger task on the pinned native Verifiers v1 API."""

import json
from typing import Any

import verifiers.v1 as vf

from .contract import FIXTURES, expected_answer, grade_artifact, oracle, read_ledger


class LedgerReconciliationData(vf.TaskData):
    rows: list[dict[str, Any]]
    answer: dict[str, Any]


class LedgerReconciliationTask(vf.Task[LedgerReconciliationData]):
    @vf.stop
    async def single_turn(self, trace: vf.Trace) -> bool:
        return trace.num_turns >= 1

    @vf.reward(weight=1.0)
    async def correct(self, trace: vf.Trace) -> float:
        grade = grade_artifact(trace.last_reply, expected=self.data.answer)
        trace.info["grade"] = {"reward": grade.reward, "reason": grade.reason}
        return grade.reward

    async def validate(self, runtime: vf.Runtime) -> bool:
        # The checked-in gold is independent of this reference implementation.
        reference = oracle(self.data.rows)
        return reference == self.data.answer and grade_artifact(
            json.dumps(reference), expected=self.data.answer
        ).reward == 1.0


class LedgerReconciliationTaskset(
    vf.Taskset[LedgerReconciliationTask, vf.TasksetConfig]
):
    def load(self) -> list[LedgerReconciliationTask]:
        rows = read_ledger()
        instruction = (FIXTURES / "instruction.txt").read_text().strip()
        return [
            LedgerReconciliationTask(
                LedgerReconciliationData(
                    idx=0,
                    name="synthetic-ledger-v1",
                    prompt=instruction + "\n\nLedger:\n" + json.dumps(rows),
                    rows=rows,
                    answer=expected_answer(),
                ),
                self.config.task,
            )
        ]

SHA-256: d334a9294a55c8c38ef7f5fd2366ab41e85d18e025c28e7f02d19b0a2167bf27