opentrain-ai

Official Python SDK for the OpenTrain API — human verification, evaluation, and labeling for AI teams.

Push model outputs — synthetic rows, eval responses, RL episodes — and get human verdicts back. Use verify.reward as a human reward function inside a training loop.

pip install opentrain-ai   # imports as `opentrain`

• Python ≥ 3.10, fully typed (py.typed), Pydantic v2 models, httpx transport
• Sync (OpenTrain) and async (AsyncOpenTrain) clients with the same surface
• Apache-2.0

Quickstart

Create an API token at app.opentrain.ai → Settings → Developer, then:

from opentrain import OpenTrain

client = OpenTrain()  # reads OPENTRAIN_API_TOKEN

# Push a batch of model outputs for human verification
created = client.verify.batches.create(
    items=[
        {"prompt": "What is 2+2?", "response": "4"},
        {"prompt": "Capital of France?", "response": "Lyon"},
    ],
)
print(created.batch.batch_id, created.batch.status)

# Read verdicts back (ACCEPT / FIX / REJECT per item)
results = client.verify.batches.results(created.batch.batch_id)
print(results.summary)

Human reward function

# Synchronous LLM-judge scoring (bring your own OpenRouter key in Quality settings)
scored = client.verify.reward.judge_now(
    items=[{"prompt": "2+2?", "response": "4"}],
    rubric=["The response is factually correct."],
)
print(scored.rewards[0].score)  # 0..1

# Asynchronous human scoring — poll or subscribe to verify.reward.completed
queued = client.verify.reward.human_async(
    items=[{"prompt": "2+2?", "response": "5"}],
    rubric=["The response is factually correct."],
    training_run_id="run_...",  # optional: links verdicts to your training run
)
rewards = client.verify.reward.get(queued.batch_id)

Async

import asyncio
from opentrain import AsyncOpenTrain

async def main() -> None:
    async with AsyncOpenTrain() as client:
        created = await client.verify.batches.create(
            items=[{"prompt": "p", "response": "r"}]
        )
        print(created.batch.batch_id)

asyncio.run(main())

Webhooks

from opentrain import verify_webhook_signature

ok = verify_webhook_signature(
    secret="whsec_...",                      # returned once at webhook creation
    raw_body=request_body_text,              # verify BEFORE parsing JSON
    signature_header=headers["X-OpenTrain-Signature"],
)

Other surfaces

client.label.launch(...), client.evaluate.launch(...), client.synthetic_runs.create(...) / .push_rows(...), client.training_runs.register_external(...), client.benchmarks.list(), client.webhooks.create(...). These return dict payloads matching the OpenAPI document at /api/public/v1/openapi.

Errors

Non-2xx responses raise OpenTrainAPIError with status, code (e.g. PAYMENT_REQUIRED), request_id, and details.

Development

scripts/test.sh   # venv + mypy --strict + pytest