Bulk embedding via Batch API

When you have thousands or millions of documents to embed (e.g. an existing knowledge base), the Batch API is the right tool: 50% cheaper than realtime, async.

Looking for fast realtime embedding of a handful of texts? This page is about the asynchronous Batch API (file upload, 24h SLA, discounted). If you have ~2 to 64 texts in memory at the same point in your pipeline and want them embedded immediately, send a single realtime /v1/embeddings call with input as an array — see Performance: batch texts in one request in the Embeddings reference (~3× faster than sequential single-text calls).

Generating the input file

JSONL, one request per line:

import json

with open("embed_input.jsonl", "w") as f:
    for doc in my_corpus:
        f.write(json.dumps({
            "custom_id": doc["id"],   # your internal ID for matching results back
            "method": "POST",
            "url": "/v1/embeddings",
            "body": {
                "model": "epithre-embed",
                "input": [doc["text"]],
                "dimensions": 1024,    # optional: smaller vectors for storage
            }
        }) + "\n")

Notes: - custom_id is required per line; the output preserves it. - Each body.input should be one text or a list of up to 64 (more efficient: batch into 64 per request to fit max 50K requests under the cap).

Submitting and polling

input_file = client.files.create(
    file=open("embed_input.jsonl", "rb"),
    purpose="batch_input",
)

batch = client.batches.create(
    input_file_id=input_file.id,
    endpoint="/v1/embeddings",
    completion_window="24h",
)

import time
while batch.status not in ("completed", "failed", "cancelled"):
    time.sleep(10)
    batch = client.batches.retrieve(batch.id)
    print(f"{batch.status}: {batch.request_counts.completed}/{batch.request_counts.total}")

For typical workloads, batches complete in minutes (not hours). The "24h" is just an SLA upper bound.

Downloading results

out = client.files.content(batch.output_file_id)
results = {}
for line in out.text.splitlines():
    r = json.loads(line)
    if r["error"] is None:
        emb = r["response"]["body"]["data"][0]["embedding"]
        results[r["custom_id"]] = emb
    else:
        print(f"Error on {r['custom_id']}: {r['error']['message']}")

Storing to pgvector

import psycopg2

conn = psycopg2.connect(...)
cur = conn.cursor()

for doc_id, vec in results.items():
    vec_str = "[" + ",".join(f"{v:.7f}" for v in vec) + "]"
    cur.execute("UPDATE documents SET embedding = %s::halfvec(1024) WHERE id = %s",
                (vec_str, doc_id))
conn.commit()

For very large inserts, use COPY FROM or batch executemany.

Webhook-driven (skip polling)

Register a webhook for batch.completed:

import httpx
wh = httpx.post(".../v1/webhooks", json={
    "url": "https://my-server.com/epithre-batch-done",
    "events": ["batch.completed", "batch.failed"],
}).json()
print(wh["secret"])  # save this

In your server, handle the POST:

@app.post("/epithre-batch-done")
def webhook(request):
    # verify signature first (see /reference/webhooks)
    payload = json.loads(request.body)
    if payload["event"] == "batch.completed":
        download_and_process(payload["data"]["id"])
    return {"ok": True}

Cost example

10K documents at average 500 tokens each = 5M tokens total.

Stacking with prompt cache doesn't apply to embeddings (no system prompt). But for chat batches with long stable prompts, cache + batch compounds to 0.05x base rate.

Handling errors

The output file's error field captures per-line failures. Common causes:

See also