ractogateway.batch.openai_batch

OpenAI Batch API processor.

Submits large sets of chat-completion requests to OpenAI’s asynchronous Batch API, which processes them within 24 hours at ~50 % cost compared to the synchronous Chat Completions API.

Workflow:

upload JSONL file  →  create batch job  →  poll status
→  download results  →  parse into BatchResult list

Both synchronous and async variants are provided for every operation.

Usage:

from ractogateway import openai_developer_kit as gpt
from ractogateway.prompts.engine import RactoPrompt

prompt = RactoPrompt(role="assistant", aim="answer briefly",
                     constraints="", tone="", output_format="text")

processor = gpt.OpenAIBatchProcessor(model="gpt-4o-mini", default_prompt=prompt)

results = processor.submit_and_wait([
    gpt.BatchItem(custom_id="q1", user_message="What is 2+2?"),
    gpt.BatchItem(custom_id="q2", user_message="Capital of France?"),
])

for r in results:
    print(r.custom_id, r.response.content if r.ok else r.error)

class ractogateway.batch.openai_batch.OpenAIBatchProcessor(model='gpt-4o-mini', *, api_key=None, base_url=None, default_prompt=None)[source]

Bases: object

Submit thousands of chat-completion requests to OpenAI’s Batch API at ~50 % of standard API cost.

Parameters:

model (str) – Chat model to use for all items in a batch (e.g. "gpt-4o-mini").
api_key (str | None) – OpenAI API key. Falls back to OPENAI_API_KEY env var.
base_url (str | None) – Custom base URL (Azure OpenAI / proxy).
default_prompt (RactoPrompt | None) – RACTO prompt used as the system message for every batch item.

submit_batch / asubmit_batch:: Upload JSONL and create batch job → returns BatchJobInfo.

poll_status / apoll_status:: Fetch current job state → returns updated BatchJobInfo.

get_results / aget_results:: Download and parse completed job results → list[BatchResult].

submit_and_wait / asubmit_and_wait:: Convenience: submit + poll until done + return results.

provider: str = 'openai'

submit_batch(items, *, prompt=None, completion_window='24h')[source]

Upload items as a JSONL file and create an OpenAI batch job.

Returns immediately with a BatchJobInfo (status = IN_PROGRESS).

Return type:: BatchJobInfo

poll_status(job_id)[source]

Fetch the current status of batch job job_id.

Return type:: BatchJobInfo

get_results(job_id)[source]

Download and parse results for a completed batch job.

Raises:: RuntimeError – If the job is not yet completed.
Return type:: list[BatchResult]

submit_and_wait(items, *, prompt=None, completion_window='24h', poll_interval_s=60.0, max_wait_s=86400.0)[source]

Submit a batch and block until it completes, then return results.

Parameters:

poll_interval_s (float) – Seconds between status-poll API calls. Default 60.0.
max_wait_s (float) – Maximum total seconds to wait. Default 86400 (24 h).

Raises: