SDK reference

Everything on this page is importable from the top-level package:

from sdvm import Auditor, AuditorConfig, Fixer, FixerConfig, Refinery, TextSample, MultipleChoiceSample, ConversationSample

Clients

Auditor measures, Fixer fixes, Refinery runs the whole pipeline. Each has an async twin with the same run method, and each takes its stage config as an object: an AuditorConfig per audit stage, a FixerConfig for the fix.

Auditor

class · sdvm.auditor Synchronous client for the SDVM audit stage.

Auditor(
    *,
    api_key: str,
    config: AuditorConfig | None = None,
    base_url: str = 'https://api.sdvm.ai',
    timeout: float = 120.0,
)

Parameters

str

required

AuditorConfig | None

default:"None"

The AuditorConfig for every call — votes (odd) audits each sample N times and returns the majority verdict per dimension. None is AuditorConfig(votes=1).

str

default:"'https://api.sdvm.ai'"

float

default:"120.0"

from sdvm import Auditor, AuditorConfig
from sdvm.types import MultipleChoiceSample

auditor = Auditor(api_key="YOUR_API_KEY", config=AuditorConfig(votes=3))
audited = auditor.run([MultipleChoiceSample(...), ...])
# each returned sample carries its verdict on `.audit`; pass them to Fixer.run.

Auditor.aggregate()

static Cross-sample (dataset-level) audit over the output of run, computed locally. Returns per-field distributions and a per-check pass/fail rollup.

@staticmethod
def Auditor.aggregate(audited_samples: list[DataSample]) -> dict

Parameters

list[DataSample]

required

Auditor.run()

Audit a list of samples (max 100 per request).

def Auditor.run(
    data: list[DataSample],
    *,
    conventions: str | None = None,
    config: AuditorConfig | None = None,
) -> list[DataSample]

Parameters

list[DataSample]

required

Samples to audit.

str | None

default:"None"

Optional free-text description of dataset conventions the quality audit must not mistake for defects — e.g. a dataset that is uniformly lowercased, uses markup tokens, or truncates the context by design. Without it, such formatting is flagged as a grammatical defect.

AuditorConfig | None

default:"None"

Override the instance-level AuditorConfig for this request.

Returns The samples with their verdict attached on .audit.

AsyncAuditor

class · sdvm.auditor Async client for the SDVM audit stage.

AsyncAuditor(
    *,
    api_key: str,
    config: AuditorConfig | None = None,
    base_url: str = 'https://api.sdvm.ai',
    timeout: float = 120.0,
)

Parameters

str

required

AuditorConfig | None

default:"None"

str

default:"'https://api.sdvm.ai'"

float

default:"120.0"

AsyncAuditor.aggregate()

static Cross-sample (dataset-level) audit over already-audited samples — local, deterministic. See Auditor.aggregate.

@staticmethod
def AsyncAuditor.aggregate(audited_samples: list[DataSample]) -> dict

AsyncAuditor.run()

async Audit a list of samples asynchronously; see Auditor.run.

async def AsyncAuditor.run(
    data: list[DataSample],
    *,
    conventions: str | None = None,
    config: AuditorConfig | None = None,
) -> list[DataSample]

Parameters

list[DataSample]

required

str | None

default:"None"

AuditorConfig | None

default:"None"

Fixer

class · sdvm.fixer High-level synchronous client for the SDVM fix stage.

Fixer(
    *,
    api_key: str,
    config: FixerConfig | None = None,
    base_url: str = 'https://api.sdvm.ai',
    timeout: float = 120.0,
)

Fix routes each sample’s repair from an audit verdict — the .audit a sample already carries, or one routing audit run for it — and applies it with a never-worse guard. It does NOT reaudit to judge whether the fix helped; that is Refinery’s job (audit → fix → reaudit). Use Auditor for the verdict and Refinery when you want the whole pipeline. Parameters

str

required

FixerConfig | None

default:"None"

The FixerConfig for every call — max_attempts, how many tries the fixer gets at an accepted fix from one verdict. Left None, the server default of one attempt applies.

str

default:"'https://api.sdvm.ai'"

float

default:"120.0"

from sdvm import Fixer, FixerConfig
from sdvm.types import MultipleChoiceSample

fixed = Fixer(api_key="YOUR_API_KEY", config=FixerConfig(max_attempts=2)).run(
    [MultipleChoiceSample(...), ...])
# each fixed sample carries a `.fix` block {"changes": [...], "flagged": bool, "attempts": int}

Fixer.run()

Fix a list of samples (max 100 per request).

def Fixer.run(
    data: list[DataSample],
    *,
    config: FixerConfig | None = None,
    conventions: str | None = None,
) -> list[DataSample]

Parameters

list[DataSample]

required

Samples to fix.

FixerConfig | None

default:"None"

Override the instance-level FixerConfig for this request; fields left None keep the instance’s values.

str | None

default:"None"

Optional dataset conventions for the routing audit (see Auditor.run) — declare formatting like uniform lowercasing or markup so it is not mistaken for a defect that triggers a needless fix.

Returns Fixed samples, reconstructed to their task’s type. A fixed sample carries a .fix block {"changes": [...], "flagged": bool, "attempts": int}: changes is what the fix did, flagged is true only for what it could not resolve (never made worse), attempts how many tries it got.

AsyncFixer

class · sdvm.fixer High-level async client for the SDVM fix stage; see Fixer.

AsyncFixer(
    *,
    api_key: str,
    config: FixerConfig | None = None,
    base_url: str = 'https://api.sdvm.ai',
    timeout: float = 120.0,
)

Parameters

str

required

FixerConfig | None

default:"None"

str

default:"'https://api.sdvm.ai'"

float

default:"120.0"

AsyncFixer.run()

async Fix a list of samples asynchronously; see Fixer.run.

async def AsyncFixer.run(
    data: list[DataSample],
    *,
    config: FixerConfig | None = None,
    conventions: str | None = None,
) -> list[DataSample]

Parameters

list[DataSample]

required

FixerConfig | None

default:"None"

str | None

default:"None"

Refinery

class · sdvm.refinery Run the full pipeline (audit → fix → reaudit) end to end.

Refinery(
    *,
    api_key: str,
    audit: AuditorConfig | None = None,
    fix: FixerConfig | None = None,
    reaudit: AuditorConfig | None = None,
    conventions: str | None = None,
    base_url: str = 'https://api.sdvm.ai',
    timeout: float = 120.0,
)

A Refinery runs audit -> fix -> reaudit, server-side (POST /refine), one pass: audit(votes) -> fix(max_attempts) -> reaudit(votes). Each returned sample carries one block per stage: .audit (the verdict BEFORE the fix — the one the fix was routed from, and the same block an Auditor returns), .fix (what the fix did: changes, flagged, attempts) and .reaudit (the verdict AFTER the fix, which is how you tell whether it helped; None for a sample the fix left alone, whose .audit still holds). Parameters

str

required

AuditorConfig | None

default:"None"

The audit before the fix — votes (odd) audits each sample N times and takes the majority verdict per dimension. None is one vote.

FixerConfig | None

default:"None"

The fix stage — max_attempts, tries at an accepted fix from the one verdict.

AuditorConfig | None

default:"None"

The audit after the fix, run on the samples the fix changed. None is one vote.

str | None

default:"None"

dataset conventions passed to every stage (see Auditor.run).

str

default:"'https://api.sdvm.ai'"

float

default:"120.0"

from sdvm import AuditorConfig, FixerConfig, Refinery
from sdvm.types import MultipleChoiceSample

refined = Refinery(
    api_key="YOUR_API_KEY",
    audit=AuditorConfig(votes=3),
    fix=FixerConfig(max_attempts=2),
    reaudit=AuditorConfig(votes=1),
).run([MultipleChoiceSample(...), ...])
# each returned sample carries `.audit` (before), `.fix` and `.reaudit` (after).

Refinery.run()

Execute the pipeline; returns the samples carrying .audit, .fix and .reaudit. Stage configs given here override the instance’s for this call.

def Refinery.run(
    data: list[DataSample],
    *,
    audit: AuditorConfig | None = None,
    fix: FixerConfig | None = None,
    reaudit: AuditorConfig | None = None,
    conventions: str | None = None,
) -> list[DataSample]

Parameters

list[DataSample]

required

AuditorConfig | None

default:"None"

FixerConfig | None

default:"None"

AuditorConfig | None

default:"None"

str | None

default:"None"

AsyncRefinery

class · sdvm.refinery Async Refinery; see it for details.

AsyncRefinery(
    *,
    api_key: str,
    audit: AuditorConfig | None = None,
    fix: FixerConfig | None = None,
    reaudit: AuditorConfig | None = None,
    conventions: str | None = None,
    base_url: str = 'https://api.sdvm.ai',
    timeout: float = 120.0,
)

Parameters

str

required

AuditorConfig | None

default:"None"

FixerConfig | None

default:"None"

AuditorConfig | None

default:"None"

str | None

default:"None"

str

default:"'https://api.sdvm.ai'"

float

default:"120.0"

AsyncRefinery.run()

async Execute the pipeline asynchronously; see Refinery.run.

async def AsyncRefinery.run(
    data: list[DataSample],
    *,
    audit: AuditorConfig | None = None,
    fix: FixerConfig | None = None,
    reaudit: AuditorConfig | None = None,
    conventions: str | None = None,
) -> list[DataSample]

Parameters

list[DataSample]

required

AuditorConfig | None

default:"None"

FixerConfig | None

default:"None"

AuditorConfig | None

default:"None"

str | None

default:"None"

Samples

The data types you pass in and get back. After a run, a sample carries .audit (Auditor, Refinery), .fix (Fixer, Refinery) and .reaudit (Refinery).

DataSample

class · sdvm.types · extends ABC Abstract base for a sample.

DataSample()

Every concrete sample declares a non-empty task_type (enforced at definition time) and serializes via to_dict, which emits task_type plus the sample’s fields (a field named extra is flattened in as passthrough metadata). Concrete classes are registered by task_type so from_dict can rebuild a sample from a server response.

DataSample.from_dict()

classmethod Rebuild a sample from a serialized dict, dispatching on its task_type.

@classmethod
def DataSample.from_dict(d: dict) -> "'DataSample'"

Any audit / fix / reaudit block on the dict is carried onto the rebuilt sample. Parameters

dict

required

DataSample.to_dict()

Serialize the sample for the API: task_type plus its fields, with extra flattened in and audit / fix / reaudit included only when present.

def DataSample.to_dict() -> dict

TextSample

class · sdvm.types · extends DataSample A single text sample, e.g. a pre-training document.

TextSample(text: str)

Fields

str

required

from sdvm import TextSample

sample = TextSample(text="The mitochondria is the powerhouse of the cell.")

TextSample.from_dict()

classmethod Rebuild a sample from a serialized dict, dispatching on its task_type.

@classmethod
def TextSample.from_dict(d: dict) -> "'DataSample'"

Any audit / fix / reaudit block on the dict is carried onto the rebuilt sample. Parameters

dict

required

TextSample.to_dict()

Serialize the sample for the API: task_type plus its fields, with extra flattened in and audit / fix / reaudit included only when present.

def TextSample.to_dict() -> dict

MultipleChoiceSample

class · sdvm.types · extends DataSample A single-answer multiple-choice item: a context, choices, and the answer_index of the one correct choice. Extra columns pass through via extra.

MultipleChoiceSample(
    context: str,
    choices: list[str],
    answer_index: int,
    style: str = 'continuation',
    extra: dict = <factory>,
)

style tells the audit how to read the context (see MC_STYLES): the same data shape covers sentence-completion ("continuation") and question-answering ("qa"). Fields

str

required

list[str]

required

int

required

str

default:"'continuation'"

dict

default:"dict()"

from sdvm import MultipleChoiceSample

sample = MultipleChoiceSample(
    context="A woman applies mascara. she",
    choices=["applies it to her lashes.", "drives off.", "mixes a bowl.", "exits."],
    answer_index=0,
)
qa = MultipleChoiceSample(
    context="What is the capital of France?",
    choices=["Berlin", "Paris", "Madrid", "Rome"],
    answer_index=1,
    style="qa",
)

MultipleChoiceSample.from_dict()

classmethod Rebuild a sample from a serialized dict, dispatching on its task_type.

@classmethod
def MultipleChoiceSample.from_dict(d: dict) -> "'DataSample'"

Any audit / fix / reaudit block on the dict is carried onto the rebuilt sample. Parameters

dict

required

MultipleChoiceSample.to_dict()

Serialize the sample for the API: task_type plus its fields, with extra flattened in and audit / fix / reaudit included only when present.

def MultipleChoiceSample.to_dict() -> dict

MultipleChoiceCompletionSample

class · sdvm.types · extends MultipleChoiceSample A multiple-choice item whose context is a sentence STEM the correct choice continues.

MultipleChoiceCompletionSample(
    context: str,
    choices: list[str],
    answer_index: int,
    style: str = 'continuation',
    extra: dict = <factory>,
)

The stem is truncated by design — it ends mid-sentence on a dangling word — so “is the context complete?” is not a question that applies here, and the auditor does not ask it. That is the point of having a distinct type: the guarantee is visible in the name rather than buried in a style flag, and you cannot accidentally get a completeness verdict on a stem. Fields

str

required

list[str]

required

int

required

str

default:"'continuation'"

dict

default:"dict()"

MultipleChoiceCompletionSample(
    context="A woman applies mascara. she",
    choices=["applies it to her lashes.", "drives off.", "mixes a bowl.", "exits."],
    answer_index=0,
)

MultipleChoiceCompletionSample.from_dict()

classmethod Rebuild a sample from a serialized dict, dispatching on its task_type.

@classmethod
def MultipleChoiceCompletionSample.from_dict(d: dict) -> "'DataSample'"

Any audit / fix / reaudit block on the dict is carried onto the rebuilt sample. Parameters

dict

required

MultipleChoiceCompletionSample.to_dict()

Serialize the sample for the API: task_type plus its fields, with extra flattened in and audit / fix / reaudit included only when present.

def MultipleChoiceCompletionSample.to_dict() -> dict

MultipleChoiceQuestionAnswerSample

class · sdvm.types · extends MultipleChoiceSample A multiple-choice item whose context is a complete QUESTION the correct choice answers.

MultipleChoiceQuestionAnswerSample(
    context: str,
    choices: list[str],
    answer_index: int,
    style: str = 'qa',
    extra: dict = <factory>,
)

Unlike a completion stem, the question is meant to stand on its own, so a truncated or malformed one IS a defect — the auditor checks completeness for this type. Fields

str

required

list[str]

required

int

required

str

default:"'qa'"

dict

default:"dict()"

MultipleChoiceQuestionAnswerSample(
    context="What is the capital of France?",
    choices=["Berlin", "Paris", "Madrid", "Rome"],
    answer_index=1,
)

MultipleChoiceQuestionAnswerSample.from_dict()

classmethod Rebuild a sample from a serialized dict, dispatching on its task_type.

@classmethod
def MultipleChoiceQuestionAnswerSample.from_dict(d: dict) -> "'DataSample'"

Any audit / fix / reaudit block on the dict is carried onto the rebuilt sample. Parameters

dict

required

MultipleChoiceQuestionAnswerSample.to_dict()

Serialize the sample for the API: task_type plus its fields, with extra flattened in and audit / fix / reaudit included only when present.

def MultipleChoiceQuestionAnswerSample.to_dict() -> dict

QuestionAnswerSample

class · sdvm.types · extends DataSample A question and its written answer — two fields, no choices to pick from.

QuestionAnswerSample(
    question: str,
    answer: str,
    extra: dict = <factory>,
)

answer is whatever the dataset publishes as the answer: a bare value, a sentence, or a worked solution. Nothing here is specific to one corpus — a dataset that separates a reasoning trace from the value it arrives at puts the value in answer and the trace in extra, where every other column already passes through untouched. Fields

str

required

str

required

dict

default:"dict()"

from sdvm import QuestionAnswerSample

sample = QuestionAnswerSample(
    question="Natalia sold clips to 48 friends in April, and half as many in May. How many clips did she sell altogether?",
    answer="She sold 48 / 2 = 24 clips in May, so 48 + 24 = 72 altogether.",
)

What sdvm-audit-1 returns for this type and what sdvm-fix-1 changes are on the model pages: sdvm-audit-1, sdvm-fix-1.

QuestionAnswerSample.from_dict()

classmethod Rebuild a sample from a serialized dict, dispatching on its task_type.

@classmethod
def QuestionAnswerSample.from_dict(d: dict) -> "'DataSample'"

Any audit / fix / reaudit block on the dict is carried onto the rebuilt sample. Parameters

dict

required

QuestionAnswerSample.to_dict()

Serialize the sample for the API: task_type plus its fields, with extra flattened in and audit / fix / reaudit included only when present.

def QuestionAnswerSample.to_dict() -> dict

ConversationSample

class · sdvm.types · extends DataSample A chat transcript: a list of messages in the OpenAI shape, as in an SFT corpus.

ConversationSample(
    messages: list[Message],
    turns: str = "last",
    extra: dict = <factory>,
)

At most one system message, and only first; at least one user and one assistant turn. The type does not require the transcript to alternate or to end on an assistant turn, and it accepts an empty turn: those are defects the audit reports and the fix repairs, so the type lets them through to be repaired. A tool or function turn is refused. Plain {"role", "content"} dicts are accepted anywhere a Message is and normalised. Fields

list[Message]

required

str

default:"last"

How much of the transcript the models read: "last", the final exchange, or "all", every assistant turn, reported per turn under audit["turns"]. Anything else is a ValueError.

dict

default:"dict()"

from sdvm import ConversationSample, Message

sample = ConversationSample(
    messages=[
        Message("system", "You are a concise assistant."),
        Message("user", "how do i reverse a list in python"),
        Message("assistant", "Use slicing: `xs[::-1]`, or `xs.reverse()` in place."),
    ],
)

What sdvm-audit-1 returns for this type and what sdvm-fix-1 changes are on the model pages: sdvm-audit-1, sdvm-fix-1.

ConversationSample.from_dict()

classmethod Rebuild a sample from a serialized dict, dispatching on its task_type.

def ConversationSample.from_dict(d: dict) -> "'DataSample'"

ConversationSample.to_dict()

Serialize for the API request: task_type, messages as plain dicts, turns, and the extra keys flattened in.

def ConversationSample.to_dict() -> dict

Message

class · sdvm.types One turn of a conversation.

Message(role: str, content: str)

Fields

str

required

"system", "user" or "assistant"; anything else is a ValueError.

str

required

The turn’s text. Must be a string; it may be empty (the audit reports it as an empty turn).

Message.to_dict()

def Message.to_dict() -> dict

Configuration

One object per pipeline stage. On the HTTP API the same objects are the config of each endpoint: {"votes"} on /audit, {"max_attempts"} on /fix, and {"audit": {...}, "fix": {...}, "reaudit": {...}} on /refine.

AuditorConfig

class · sdvm.types Configuration of one audit stage — the Auditor, or the audit / reaudit stage of a Refinery.

AuditorConfig(votes: int = 1)

Parameters

int

default:"1"

How many times to audit each sample; each field is the majority across them. 1 (default) is a single audit. Use an odd number: a tie has no majority. See Denoising with votes.

AuditorConfig.to_dict()

Serialize for the API request: {"votes": N}.

def AuditorConfig.to_dict() -> dict

FixerConfig

class · sdvm.types Configuration of the fix stage — the Fixer, or the fix stage of a Refinery. Fields left None mean “the server default” (or, on a per-call override, “the instance’s value”).

FixerConfig(max_attempts: int | None = None)

Parameters

int | None

default:"None"

How many tries the fixer gets at an ACCEPTED fix from the one audit verdict it was given. A sample that comes back flagged — a generation that failed its own verification, or an edit the never-worse guard reverted — is tried again on the original sample with what went wrong as feedback, stopping at the first accepted fix. A sample no verdict routed to a fix gets no attempt. 1 to 5; the server default is 1. See Attempts.

FixerConfig.merge()

Return a new config with overrides taking priority over self.

def FixerConfig.merge(overrides: FixerConfig | None) -> FixerConfig

Parameters

FixerConfig | None

required

FixerConfig.to_dict()

Serialize non-None fields to a dict for the API request.

def FixerConfig.to_dict() -> dict

Exceptions

All inherit from SDVMError.

SDVMError

exception · sdvm.exceptions · extends Exception Base exception for all SDVM SDK errors.

SDVMError(...)

AuthenticationError

exception · sdvm.exceptions · extends SDVMError Raised when authentication fails (invalid or revoked API key).

AuthenticationError(...)

InsufficientCreditsError

exception · sdvm.exceptions · extends SDVMError Raised when the account has insufficient credits (HTTP 402).

InsufficientCreditsError(...)

RateLimitError

exception · sdvm.exceptions · extends SDVMError Raised when the rate limit is exceeded (HTTP 429).

RateLimitError(...)

APIError

exception · sdvm.exceptions · extends SDVMError Raised for unexpected API errors.

APIError(status_code: int, message: str)

Parameters

int

required

str

required

​Clients

​Auditor

​Auditor.aggregate()

​Auditor.run()

​AsyncAuditor

​AsyncAuditor.aggregate()

​AsyncAuditor.run()

​Fixer

​Fixer.run()

​AsyncFixer

​AsyncFixer.run()

​Refinery

​Refinery.run()

​AsyncRefinery

​AsyncRefinery.run()

​Samples

​DataSample

​DataSample.from_dict()

​DataSample.to_dict()

​TextSample

​TextSample.from_dict()

​TextSample.to_dict()

​MultipleChoiceSample

​MultipleChoiceSample.from_dict()

​MultipleChoiceSample.to_dict()

​MultipleChoiceCompletionSample

​MultipleChoiceCompletionSample.from_dict()

​MultipleChoiceCompletionSample.to_dict()

​MultipleChoiceQuestionAnswerSample

​MultipleChoiceQuestionAnswerSample.from_dict()

​MultipleChoiceQuestionAnswerSample.to_dict()

​QuestionAnswerSample

​QuestionAnswerSample.from_dict()

​QuestionAnswerSample.to_dict()

​ConversationSample

​ConversationSample.from_dict()

​ConversationSample.to_dict()

​Message

​Message.to_dict()

​Configuration

​AuditorConfig

​AuditorConfig.to_dict()

​FixerConfig

​FixerConfig.merge()

​FixerConfig.to_dict()

​Exceptions

​SDVMError

​AuthenticationError

​InsufficientCreditsError

​RateLimitError

​APIError

Clients

Auditor

Auditor.aggregate()

Auditor.run()

AsyncAuditor

AsyncAuditor.aggregate()

AsyncAuditor.run()

Fixer

Fixer.run()

AsyncFixer

AsyncFixer.run()

Refinery

Refinery.run()

AsyncRefinery

AsyncRefinery.run()

Samples

DataSample

DataSample.from_dict()

DataSample.to_dict()

TextSample

TextSample.from_dict()

TextSample.to_dict()

MultipleChoiceSample

MultipleChoiceSample.from_dict()

MultipleChoiceSample.to_dict()

MultipleChoiceCompletionSample

MultipleChoiceCompletionSample.from_dict()

MultipleChoiceCompletionSample.to_dict()

MultipleChoiceQuestionAnswerSample

MultipleChoiceQuestionAnswerSample.from_dict()

MultipleChoiceQuestionAnswerSample.to_dict()

QuestionAnswerSample

QuestionAnswerSample.from_dict()

QuestionAnswerSample.to_dict()

ConversationSample

ConversationSample.from_dict()

ConversationSample.to_dict()

Message

Message.to_dict()

Configuration

AuditorConfig

AuditorConfig.to_dict()

FixerConfig

FixerConfig.merge()

FixerConfig.to_dict()

Exceptions

SDVMError

AuthenticationError

InsufficientCreditsError

RateLimitError

APIError