Online RL with Streams

A stream is a job that takes its items one push at a time. Your training loop pushes rollouts as the policy produces them, real people judge them, and each item lands on an ordered results feed once it finalizes. Use a stream when the judgments feed back into the model that produced the rollouts; use a job for a fixed batch.

Using online RL for your business? Contact us before you start, so we can allocate rater bandwidth for your training runs.

Every endpoint is in the Streams API reference.

The loop

Feedback is asynchronous. A push returns as soon as its items are stored, and an item's result arrives once it has collected its votes or reached its deadline, usually a few training steps later. Apply each result at the next update after it lands.

Every item has a deadline: the stream's item_ttl_seconds (30 seconds to 1 hour, default 10 minutes), or ttl_seconds on a push. At its deadline an item finalizes with the votes it has, so the deadline caps how stale feedback can get.

1. Create the stream once. Creating it again with the same name and settings returns the existing stream, even after it has closed, and the stream keeps its own item_ttl_seconds, max_pending_items and budget_credits (change them with PATCH). A different task_type, instruction, response_options, max_responses_per_item, annotator_filter or serving_environment is refused with 409 config_mismatch. A closed stream takes no new items, so give each training run its own stream name.
2. Push items as your policy produces them, each with an external_id you choose.
3. Read finalized items from the results feed with a cursor.
4. Expire items you no longer need, so their unused credits come back sooner.
5. Close the stream, then read the feed until it is drained.

Choosing a task type

Every stream has exactly one task type, fixed when you create it.

Streams also take i2v_comparison and text_answer. An item's media takes the same shape as a POST /jobs datapoint of that task type, with a few extra item rules. Streams don't take chain steps, dimensions or custom task UIs.

Item lifecycle

An item is pending from its push until it finalizes with one of four outcomes:

A failed or blocked item can still carry the votes it received before it failed or was blocked. It is never billed, and you should leave it out of training (see Quality).

• Expire named items to finalize them now, by external_id or by index.
• Close the stream to stop it taking new items. With drain: true (the default) pending items keep collecting votes until they complete or reach their deadlines; with drain: false they expire now.
• Pausing a stream, with POST /jobs/{stream_id}/pause or the Pause button on its dashboard page, stops serving it and refuses pushes with new items until you resume it with POST /jobs/{stream_id}/resume. Our team can pause a stream too, and GET /streams/{stream_id} then reports is_paused: true. Pending items still finalize at their deadlines.
• A stream closes itself 14 days after it was created, without draining.

Reading results

GET /streams/{stream_id}/results returns finalized items in the order they finalized. Each finalized item gets the next seq, with no gaps, so passing after set to the last seq you processed returns every later item exactly once.

• At-least-once. Save your cursor after you process an item. After a restart, resume from the saved cursor: you see again at most the items you hadn't saved yet.
• Long-poll. With wait_seconds (up to 20) the request returns once min_items results are ready (default 1), once a page is full when a full page holds fewer than min_items, or when the wait runs out. A reader looping with wait_seconds=20 makes at most 3 requests a minute while no results arrive, but while items keep finalizing each request returns with the results ready so far, so the loop costs a request per batch. Set min_items (at most limit) to take results in larger batches, or wait a few seconds between requests.
• Paging. Keep reading with after set to next_after while has_more is true. A page holds at most limit items, and fewer on a stream with a high max_responses_per_item.
• Several readers. The feed keeps no reader state. Each reader keeps its own cursor and sees every item.
• An item's result never changes once it has finalized.

Quality

• Results carry reliability-weighted fields beside the plain counts, such as weighted_consensus and weighted_scores. Each vote counts in proportion to its rater's reliability, so low-reliability votes count for less. See Weighted fields.
• include_responses=true adds each vote, with the rater's trust estimate and any free_text they wrote. To ask raters to explain their choice, create the stream with response_options.free_text set to {"enabled": true}.
• Filter on outcome first: train on completed and expired items only. A failed or blocked item can carry votes it received before it failed or was blocked, but it isn't billed, and a blocked item's content didn't pass review.
• total_responses is the number of votes an item finalized with. Drop thin items, such as expired ones with few votes, before you train on them. A completed item can also count fewer than max_responses_per_item votes, when quality checks discarded some of its answers.
• annotator_filter chooses who judges, as on POST /jobs. See Targeting.
• Attention checks are available on request: contact us.

Billing

• A stream charges a fixed rate per counted vote (credits.rate on the stream): your account's rate plus any targeting surcharges, frozen when you create the stream. Sandbox streams have a rate of 0 and are free.
• Each push holds max_responses_per_item × credits.rate credits per new item from your balance. Pushing an external_id the stream already stores holds nothing.
• When an item finalizes, the credits it didn't use come back.
• budget_credits caps what the stream may hold plus spend. A push past it is refused with 402 stream_budget_exceeded; raise the cap with PATCH /streams/{stream_id}. Finalizing items gives back what they held but never lowers what was spent, so a cap below the stream's spend stays exceeded until you raise it.
• Failed and blocked items are never billed.
• The stream's final charge is recorded once, after it has closed and its last item has finalized.
• All amounts are in credits. See Billing.

Rate budget

Stream requests count against your key's per-minute limit (60 by default) together with uploads and every other call. See Rate limits.

Push rollouts in batches rather than one request per rollout, and read results every few seconds rather than on every training step.

Quick start (Python)

This loop needs only the requests package and your API key in DATAPOINT_KEY. It creates a stream, pushes a group of four replies for each prompt at every step, reads whatever has finalized without blocking, then closes the stream and reads the feed until it has drained. It keeps two files per run: the cursor, saved after each result, and the last push, saved before it is sent. A restarted run with the same name sends that push again unchanged, carries on from the next step and reads the feed from where it stopped.

• Any call can be answered 429 (your key's rate limit, or a push that would pass max_pending_items) or 503 (such as stream_busy, when another update held the stream too long) with a Retry-After header. The request stored and changed nothing: wait that many seconds and send it again, as call does below.
• If a push gets no answer, for example after a timeout, send the same items again. Items already stored come back with duplicate: true and aren't charged again. Newly sampled replies under a stored external_id are refused with 409 external_id_conflict, so a restart must resend the saved push, not sample the step again.
• A sandbox stream is free and is served only by a test pool, so its items can finish expired with few or no votes. Set serving_environment to prod for real judgments; that spends credits.

import json
import os
import time
from pathlib import Path
 
import requests
 
API = "https://api.trydatapoint.com/data-labelling/v1"
RUN = "grpo-run-7"               # one stream per training run
CURSOR = Path(f"{RUN}.cursor")   # the last seq this run has processed
LAST_PUSH = Path(f"{RUN}.push")  # the last push this run sent, saved before sending it
PROMPTS = ["How do I keep basil alive indoors?", "What makes a good cover letter?"]
 
http = requests.Session()
http.headers["X-API-Key"] = os.environ["DATAPOINT_KEY"]
 
 
def call(method, path, **kwargs):
    """Send a request. On 429 or 503, wait Retry-After seconds and send it again."""
    for _ in range(10):
        r = http.request(method, API + path, timeout=60, **kwargs)
        if r.status_code not in (429, 503):
            break
        time.sleep(float(r.headers.get("Retry-After", 5)))
    if not r.ok:
        raise RuntimeError(f"{method} {path}: {r.status_code} {r.text}")
    return r.json()
 
 
def save(path, text):
    """Replace a file in one step, so a crash never leaves it half written."""
    tmp = path.with_name(path.name + ".tmp")
    tmp.write_text(text)
    tmp.replace(path)
 
 
def push(step, items):
    """Save a push, then send it, so a restart can send exactly these items again."""
    save(LAST_PUSH, json.dumps({"step": step, "items": items}))
    call("POST", f"/streams/{stream_id}/items", json={"items": items})
 
 
def sample_replies(prompt, step):
    """Stand-in for your policy: four replies to one prompt."""
    return [f"Reply {k} at step {step} to: {prompt}" for k in range(4)]
 
 
def train_on(result):
    """Stand-in for your update: one score per reply, keyed by its position in candidates."""
    print(result["external_id"], result["result"]["weighted_scores"])
 
 
def read_results(**params):
    """Process every result past the saved cursor, saving the cursor after each one."""
    while True:
        after = int(CURSOR.read_text()) if CURSOR.exists() else 0
        page = call("GET", f"/streams/{stream_id}/results", params={"after": after, **params})
        for result in page["results"]:
            if result["outcome"] in ("completed", "expired") and result["result"]["total_responses"]:
                train_on(result)
            save(CURSOR, str(result["seq"]))
        if not page["has_more"]:
            return page
 
 
stream_id = call("POST", "/streams", json={
    "name": RUN,
    "instruction": "Which reply better answers the prompt? {context}",
    "task_type": "pairwise_ranking",
    "max_responses_per_item": 6,  # 4 replies make 6 pairs: one vote each
    "item_ttl_seconds": 120,      # every item finalizes within 2 minutes of its push
    "serving_environment": "sandbox",
})["stream_id"]
 
first_step = 0
if LAST_PUSH.exists():  # a restart: send the last push again unchanged, then carry on after it
    last = json.loads(LAST_PUSH.read_text())
    push(last["step"], last["items"])
    first_step = last["step"] + 1
 
for step in range(first_step, 10):
    push(step, [
        {
            "external_id": f"{RUN}-{step}-{i}",  # one id per prompt per step
            "context": prompt,
            "media": {"candidates": [{"text": reply} for reply in sample_replies(prompt, step)]},
        }
        for i, prompt in enumerate(PROMPTS)
    ])
    read_results()  # returns at once with whatever has finalized
    time.sleep(10)  # stand-in for a training step
 
call("POST", f"/streams/{stream_id}/close", json={"drain": True})
while True:  # pending items keep collecting votes until their deadlines
    page = read_results(wait_seconds=20, min_items=2)
    if not page["results"]:
        stream = call("GET", f"/streams/{stream_id}")
        if stream["status"] == "closed" and page["next_after"] >= stream["last_seq"]:
            break

Next

• Streams API reference
• Pairwise ranking
• Media handling
• Online RL with Datapoint: talk to us about panel specs and pricing for a training run