Examples

"""Basic example — fix a list of raw text samples using the SDVM API."""

import os

from dotenv import load_dotenv

from sdvm import Fixer
from sdvm.types import TextSample

load_dotenv()

api_key = os.environ["SDVM_API_KEY"]

data = [
    TextSample(text="the mitochondria is the powerhouse of the cell"),
    TextSample(text="water boils at 100 degrees"),
    TextSample(text="the earth orbits the sun once every 365 days approximately"),
]

fixer = Fixer(api_key=api_key, base_url="https://api.sdvm.ai")
result = fixer.run(data)

for item in result:
    print(item.text, item.fix)

"""Fix free text with SDVM.

Wrap each string as a ``TextSample``, pass the list to ``Fixer.run()``, and get fixed
``TextSample``s back (max 100 per request).

Run:
    export SDVM_API_KEY=sk-...
    python examples/fix_text.py
"""

import os

from sdvm import Fixer, TextSample

SAMPLES = [
    TextSample(text="The mitochondria is the powerhouse of the cell."),
    TextSample(text="water boils at 100 c at sea level"),
    TextSample(text="Photosynthesis are the process by which plants make food."),
]


def main() -> None:
    api_key = os.environ.get("SDVM_API_KEY")
    if not api_key:
        raise SystemExit("set SDVM_API_KEY to run this example")

    fixer = Fixer(api_key=api_key)
    fixed = fixer.run(SAMPLES)

    for before, after in zip(SAMPLES, fixed):
        print("before:", before.text)
        print("after :", after.text)
        print()


if __name__ == "__main__":
    main()

"""Audit then fix HellaSwag with SDVM.

HellaSwag is a sentence-completion multiple-choice task (the context is a STEM the correct choice
continues), so wrap each row as a ``MultipleChoiceSample`` — ``style`` defaults to
``"continuation"``, which is exactly right here. (See ``mmlu.py`` for the ``"qa"`` variant.)

  1. **Audit** (per-sample)       -> each sample comes back carrying its audit on ``.audit``.
     Pass ``config=AuditorConfig(votes=N)`` to audit each sample N times and take the majority
     verdict (denoise).
  2. **Aggregate** (cross-sample) -> a local dataset-level view: the choice-count
     distribution and an answer-position bias check (is the correct choice always in one slot?).
  3. **Fix** with the ``Fixer``  -> each sample carries a ``.fix`` block
     ``{"changes": [...], "flagged": bool, "attempts": int}``. The fix is never
     worse; ``flagged`` marks one it could not fully resolve, for you to review. Fix does NOT
     reaudit to check its own work.

(To run audit -> fix -> reaudit in one call — and see whether the fix actually helped — use the
``Refinery``; see ``pipeline.py``.)

HellaSwag is wikiHow text with formatting conventions (markup tokens, uniform lowercasing,
intentional truncation), so we pass ``conventions=...`` to both steps — otherwise the quality
audit reads that formatting as grammar defects (~40% false-alarm vs ~12% with the conventions).

Run:
    pip install "sdvm[examples]"         # datasets, python-dotenv
    export SDVM_API_KEY=sk-...           # for the audit/fix calls
    python examples/hellaswag.py
"""

import os

from datasets import load_dataset

from sdvm import Auditor, Fixer, MultipleChoiceSample

# The three fields that map onto the task's roles; everything else is passthrough metadata.
ROLE_FIELDS = {"ctx", "endings", "label"}

# HellaSwag's wikiHow text has DATASET CONVENTIONS that are not defects — declare them so the
# quality audit doesn't read formatting as grammar errors. Without this the grammatical
# false-alarm on clean items is ~40%; with it, ~12%.
CONVENTIONS = (
    "This text follows dataset conventions that are NOT errors: (1) bracketed markers like "
    "[header] [title] [step] [substeps] are structural section labels, not grammar errors or "
    "run-ons; (2) the text is uniformly LOWERCASED (including sentence starts and proper nouns) "
    "— do NOT flag lowercasing under grammatical; (3) the context may end mid-sentence by design "
    "— do NOT flag it as incomplete. Judge grammar/coherence on the underlying wording only."
)


def to_sample(row: dict) -> MultipleChoiceSample:
    """Map a raw HellaSwag row onto the multiple-choice roles; keep the rest in ``extra``."""
    return MultipleChoiceSample(
        context=row["ctx"],
        choices=list(row["endings"]),
        answer_index=int(row["label"]),  # HellaSwag stores the label as a string
        style="continuation",  # the default — a stem the correct choice continues
        extra={k: v for k, v in row.items() if k not in ROLE_FIELDS},
    )


def main() -> None:
    ds = load_dataset("Rowan/hellaswag", split="validation")
    # The test split leaves `label` empty (no public answers); keep only labelled rows.
    samples = [to_sample(dict(r)) for r in ds if r["label"] != ""][:20]
    print(f"wrapped {len(samples)} HellaSwag rows as MultipleChoiceSample")

    api_key = os.environ.get("SDVM_API_KEY")
    if not api_key:
        print("set SDVM_API_KEY to run the audit/fix steps")
        return

    # 1. audit (per-sample) -> samples carry their audit info on `.audit`.
    #    Pass the wikiHow conventions so formatting isn't mistaken for grammar defects.
    #    (Add config=AuditorConfig(votes=3) to audit each sample 3x and majority-vote — denoise.)
    audited = Auditor(api_key=api_key).run(samples, conventions=CONVENTIONS)
    print("first sample's audit:", audited[0].audit)

    # 2. aggregate (cross-sample) -> local dataset-level view.
    agg = Auditor.aggregate(audited)
    choices = agg["fields"]["choices"]
    print("choice counts:", choices["length_counts"], "| uniform:", choices["uniform"])
    pos = agg["answer_position"]  # is the correct answer stuck in one position?
    if pos:
        print(
            "answer-position biased:",
            pos["biased"],
            "| shares:",
            pos["by_num_choices"].get(4, {}).get("position_shares"),
        )
    print("audit rollup:", agg["audit_rollup"])

    # 3. fix (per-sample; never worse). Same conventions so the routing audit
    #    doesn't trigger a needless grammar fix. Fix does not reaudit — for audit -> fix ->
    #    reaudit in one call, use the Refinery (pipeline.py).
    fixed = Fixer(api_key=api_key).run(samples, conventions=CONVENTIONS)
    print("first sample fix:", fixed[0].fix)  # {"changes": [...], "flagged": bool, "attempts": n}
    flagged = [s for s in fixed if s.fix and s.fix["flagged"]]
    print(f"{len(flagged)}/{len(fixed)} flagged (the fix could not fully resolve them)")


if __name__ == "__main__":
    main()

"""Audit then fix MMLU with SDVM — the question-answering counterpart to ``hellaswag.py``.

MMLU is a knowledge multiple-choice task: the context is a complete QUESTION the correct choice
answers (not a sentence stem). Same data shape as HellaSwag, one difference — set
``style="qa"`` so the audit reads the context as a question and judges completeness normally
(a QA question that is truncated IS a defect, unlike a HellaSwag continuation stem).

The flow is identical to ``hellaswag.py``:
  1. **Audit** (per-sample)       -> each sample carries its audit on ``.audit``.
  2. **Aggregate** (cross-sample) -> choice-count distribution + answer-position bias check.
  3. **Fix** with the ``Fixer``  -> each sample carries a ``.fix`` block
     ``{"changes": [...], "flagged": bool, "attempts": int}``; the fix never makes a sample worse.

Unlike ``hellaswag.py``, MMLU questions are standard-cased, complete prose, so we pass NO
``conventions`` — there is no dataset-specific formatting for the quality audit to misread.

Run:
    pip install "sdvm[examples]"         # datasets, python-dotenv
    export SDVM_API_KEY=sk-...           # for the audit/fix calls
    python examples/mmlu.py
"""

import os

from datasets import load_dataset

from sdvm import Auditor, Fixer, MultipleChoiceSample

# The three fields that map onto the task's roles; everything else is passthrough metadata.
ROLE_FIELDS = {"question", "choices", "answer"}


def to_sample(row: dict) -> MultipleChoiceSample:
    """Map a raw MMLU row onto the MC roles; the rest (e.g. subject) goes in ``extra``."""
    return MultipleChoiceSample(
        context=row["question"],  # a complete question, not a stem
        choices=list(row["choices"]),
        answer_index=int(row["answer"]),  # MMLU stores the label as an int (0-3)
        style="qa",  # <- the only difference from hellaswag.py
        extra={k: v for k, v in row.items() if k not in ROLE_FIELDS},
    )


def main() -> None:
    # "all" merges every subject; MMLU answers are public on every split.
    ds = load_dataset("cais/mmlu", "all", split="validation")
    samples = [to_sample(dict(r)) for r in ds][:20]
    print(f"wrapped {len(samples)} MMLU rows as MultipleChoiceSample (style='qa')")

    api_key = os.environ.get("SDVM_API_KEY")
    if not api_key:
        print("set SDVM_API_KEY to run the audit/fix steps")
        return

    # 1. audit (per-sample) -> samples carry their audit info on `.audit`
    audited = Auditor(api_key=api_key).run(samples)
    print("first sample's audit:", audited[0].audit)

    # 2. aggregate (cross-sample) -> local dataset-level view.
    agg = Auditor.aggregate(audited)
    choices = agg["fields"]["choices"]
    print("choice counts:", choices["length_counts"], "| uniform:", choices["uniform"])
    pos = agg["answer_position"]  # is the correct answer stuck in one position?
    if pos:
        print(
            "answer-position biased:",
            pos["biased"],
            "| shares:",
            pos["by_num_choices"].get(4, {}).get("position_shares"),
        )
    print("audit rollup:", agg["audit_rollup"])

    # 3. fix (never worse). Fix does not reaudit — for audit -> fix -> reaudit
    #    in one call, use the Refinery (see pipeline.py).
    fixed = Fixer(api_key=api_key).run(samples)
    print("first sample fix:", fixed[0].fix)  # {"changes": [...], "flagged": bool, "attempts": n}
    flagged = [s for s in fixed if s.fix and s.fix["flagged"]]
    print(f"{len(flagged)}/{len(fixed)} flagged (the fix could not fully resolve them)")


if __name__ == "__main__":
    main()

"""Audit then fix a chat SFT corpus with SDVM — the conversation counterpart to ``mmlu.py``.

UltraChat-200k is instruction-tuning data in the OpenAI messages shape: a list of
``{"role", "content"}`` turns per row. That is exactly what ``ConversationSample`` takes, so
a row maps onto the type with no reshaping; anything else on the row rides along in ``extra``.

The flow is the same as the other examples:
  1. **Audit** (per-sample) -> each sample carries its audit on ``.audit``: the structural
     checks (roles, order, empty / repeated / looping turns, open code fences, assistant
     boilerplate, template placeholders) and the model's verdicts on the final exchange.
  2. **Fix** with the ``Fixer`` -> each sample carries a ``.fix`` block
     ``{"changes": [...], "flagged": bool, "attempts": int}``; the fix never makes a sample worse.

A sample's ``turns="all"`` widens the model's verdicts from the final exchange to every
assistant turn (reported per turn under ``audit["turns"]``); the default ``"last"`` keeps the
cost of a conversation independent of its length.

Run:
    pip install "sdvm[examples]"         # datasets, python-dotenv
    export SDVM_API_KEY=sk-...           # for the audit/fix calls
    python examples/conversation.py
"""

import os

from datasets import load_dataset

from sdvm import Auditor, ConversationSample, Fixer


def to_sample(row: dict) -> ConversationSample:
    """Map a raw UltraChat row onto the type; ``prompt_id`` and the rest go in ``extra``."""
    return ConversationSample(
        messages=row["messages"],  # already [{"role": ..., "content": ...}, ...]
        extra={k: v for k, v in row.items() if k != "messages"},
    )


def main() -> None:
    # streaming: the split is large, and twenty rows are enough to see the shape of the audit
    ds = load_dataset("HuggingFaceH4/ultrachat_200k", split="train_sft", streaming=True)
    samples = [to_sample(dict(r)) for _, r in zip(range(20), ds)]
    print(f"wrapped {len(samples)} UltraChat rows as ConversationSample")

    api_key = os.environ.get("SDVM_API_KEY")
    if not api_key:
        print("set SDVM_API_KEY to run the audit/fix steps")
        return

    # 1. audit (per-sample) -> samples carry their audit on `.audit`
    audited = Auditor(api_key=api_key).run(samples)
    print("first sample's audit:", audited[0].audit)

    # the structural checks are the first cut on a chat corpus
    broken = [
        s
        for s in audited
        if s.audit
        and not all(
            s.audit.get(k) is not False
            for k in (
                "ends_on_assistant",
                "no_empty_turns",
                "no_duplicate_turns",
                "fences_balanced",
            )
        )
    ]
    leaky = [s for s in audited if s.audit and "identity_leak" in s.audit]
    print(f"{len(broken)} structurally broken, {len(leaky)} with assistant boilerplate")

    # every assistant turn, not just the last one — the sample says so, and pays per turn
    deep = Auditor(api_key=api_key).run(
        [ConversationSample(messages=s.messages, turns="all", extra=s.extra) for s in samples[:3]]
    )
    print("per-turn verdicts on the first sample:", deep[0].audit and deep[0].audit.get("turns"))

    # 2. fix (never worse). Fix does not reaudit — for audit -> fix -> reaudit
    #    in one call, use the Refinery (see pipeline.py).
    fixed = Fixer(api_key=api_key).run(audited)
    print("first sample fix:", fixed[0].fix)  # {"changes": [...], "flagged": bool, "attempts": n}
    flagged = [s for s in fixed if s.fix and s.fix["flagged"]]
    print(f"{len(flagged)}/{len(fixed)} flagged (the fix could not fully resolve them)")


if __name__ == "__main__":
    main()

"""One-call audit -> fix -> reaudit on MMLU with the ``Refinery`` pipeline.

The :class:`~sdvm.Refinery` runs the whole pipeline server-side in a single ``run()`` call:
``audit(votes) -> fix(max_attempts) -> reaudit(votes)``. Each returned sample carries one block
per stage: ``.audit`` (the verdict BEFORE the fix), ``.fix`` (what the fix did) and ``.reaudit``
(the verdict AFTER the fix — did it work?). ``votes`` audits each sample N times and majority-votes
the verdict (denoise); ``max_attempts`` is how many tries the fixer gets at an accepted fix.

Compare with ``mmlu.py``, which drives the two steps by hand (``Auditor`` then ``Fixer``); this is
the same work — plus the reaudit — in one call.

Run:
    pip install "sdvm[examples]"         # datasets, python-dotenv
    export SDVM_API_KEY=sk-...           # for the audit/fix calls
    python examples/pipeline.py
"""

import os

from datasets import load_dataset

from sdvm import AuditorConfig, FixerConfig, MultipleChoiceSample, Refinery

# The three fields that map onto the task's roles; everything else is passthrough metadata.
ROLE_FIELDS = {"question", "choices", "answer"}


def to_sample(row: dict) -> MultipleChoiceSample:
    return MultipleChoiceSample(
        context=row["question"],  # a complete question (question-answering task)
        choices=list(row["choices"]),
        answer_index=int(row["answer"]),
        style="qa",
        extra={k: v for k, v in row.items() if k not in ROLE_FIELDS},
    )


def main() -> None:
    ds = load_dataset("cais/mmlu", "all", split="validation")
    samples = [to_sample(dict(r)) for r in ds][:1]
    print(f"wrapped {len(samples)} MMLU rows as MultipleChoiceSample (style='qa')")

    api_key = os.environ.get("SDVM_API_KEY")
    if not api_key:
        print("set SDVM_API_KEY to run the pipeline")
        return

    # audit -> fix -> reaudit, in ONE call, one config object per stage: 3 votes on the audit
    # that routes the fix, up to 2 tries at an accepted fix, 1 vote on the reaudit. (MMLU is
    # standard-cased prose, so no `conventions` needed; for a formatted dataset like HellaSwag
    # pass conventions=... here — forwarded to every stage.)
    refinery = Refinery(
        api_key=api_key,
        audit=AuditorConfig(votes=3),
        fix=FixerConfig(max_attempts=2),
        reaudit=AuditorConfig(votes=1),
    )
    result = refinery.run(samples)

    # each result carries the verdict BEFORE the fix (`.audit`), what the fix did (`.fix`) and
    # the verdict AFTER it (`.reaudit`; None for a sample the fix left alone)
    first = result[0]
    print("first sample -> audit (before fix):", first.audit)
    print("             -> fix               :", first.fix)
    print("             -> reaudit (after)   :", first.reaudit)
    flagged = [s for s in result if s.fix and s.fix["flagged"]]
    print(f"{len(flagged)}/{len(result)} flagged (the fix could not fully resolve them)")


if __name__ == "__main__":
    main()