DEV Community: Maya Collins

A Practical Python Pipeline for Batch Image Generation

Maya Collins — Wed, 09 Sep 2026 02:22:27 +0000

I prefer treating image generation as a job-processing problem rather than embedding a provider SDK throughout an application.

Put each request into a queue, select a model at routing time, limit concurrency, retry only temporary failures, and save the resulting asset with a manifest record. With a unified API, compatible models can share the same authentication and request path; changing models then becomes a routing decision instead of a new integration.

This walkthrough builds that workflow in Python. It accepts JSON Lines jobs, routes them by type, handles URL and base64 responses, records estimated usage, and writes successful and failed jobs to a manifest.

The Pipeline

The basic flow is:

jobs.jsonl → bounded worker pool → /v1/images/generations → object storage → manifest.jsonl

The queue and storage remain under application control. In this example, the API base URL is https://api.cometapi.com/v1, using one server-side CometAPI key.

The implementation covers the production concerns I usually want in a first version:

Durable job IDs
Model selection and catalog validation
Bounded concurrency
Exponential backoff with jitter
URL and base64 image responses
Estimated per-job cost
A manifest containing success and failure records

Requirements and Model Selection

You need Python 3.10 or later, the requests package, a key, and a writable output directory.

pip install requests

Set the key on the server rather than putting it in browser code or a repository:

export COMETAPI_KEY="your-key"

The generation endpoint is POST /images/generations under the base URL https://api.cometapi.com/v1.

Before deploying, check the live model catalog. It returns the current model ID, supported endpoint, features, and pricing metadata without requiring an authorization header.

As of August 20, 2026, two useful routes in that catalog were:

Workload	Model ID	Notes
Product images with controlled settings	`gpt-image-2`	Returns usage data and base64 image content on the documented OpenAI-compatible route
High-volume advertising and content concepts	`doubao-seedream-4-5-251128`	Uses the same generation route and is listed with per-request pricing

These models should not be assumed to support identical options. Size, quality, format, reference-image handling, and response behavior can differ. I would check the individual model record before sending optional parameters.

Design the Input Around Stable IDs

JSON Lines keeps the input simple. A queue consumer, database export, or spreadsheet conversion can produce the same format:

{"id":"sku-1001","kind":"product","prompt":"Studio product photo of a ceramic coffee dripper on a warm neutral background"}
{"id":"campaign-204","kind":"ad","prompt":"Editorial summer travel image, vivid natural light, wide composition, no text"}
{"id":"blog-088","kind":"content","prompt":"Minimal illustration of a developer automating a creative workflow, no text"}

The id is used for the output filename and manifest key. In a real queue, I would also use it as the idempotency key and skip IDs already marked successful.

The sample routing is:

product → gpt-image-2
ad → doubao-seedream-4-5-251128
content → doubao-seedream-4-5-251128

A job can override the default with its own model field. The script checks the live catalog at startup so an obsolete model ID fails early instead of producing a series of invalid requests.

Concurrency and Retry Behavior

The default worker count is four. That is an application-level starting point, not a universal account limit. I would monitor latency and 429 responses before increasing MAX_WORKERS.

Only these responses are retried:

408
429
5xx

The retry loop uses exponential backoff and jitter. Authentication failures, invalid models, and unsupported parameters are not retried because the request itself must be fixed first.

Save the Actual Image, Not Just a Provider URL

The documented GPT Image response includes data[0].b64_json. Other compatible models may return data[0].url instead.

The worker supports both forms. It writes the result to a temporary path conceptually by completing the download or decode first, then stores the final file. For a production deployment, I would replace the local output/ directory with S3, R2, GCS, or another object store.

A provider-hosted URL should not be treated as permanent storage unless its retention policy explicitly guarantees that.

Complete Python Example

Save this as batch_image_pipeline.py. Put jobs.jsonl beside it, then run python3 batch_image_pipeline.py.

import base64, json, os, random, time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import requests

BASE_URL = "https://api.cometapi.com/v1"
KEY = os.environ["COMETAPI_KEY"]
WORKERS = int(os.getenv("MAX_WORKERS", "4"))
OUT = Path("output")
ROUTES = {
    "product": "gpt-image-2",
    "ad": "doubao-seedream-4-5-251128",
    "content": "doubao-seedream-4-5-251128",
}

catalog = requests.get("https://api.cometapi.com/api/models", timeout=30)
catalog.raise_for_status()
CATALOG = {model["id"]: model for model in catalog.json()["data"]}

def generate(job):
    model = job.get("model", ROUTES[job["kind"]])
    if model not in CATALOG:
        raise ValueError(f"Unknown model: {model}")
    payload = {"model": model, "prompt": job["prompt"], "n": 1}
    if model == "gpt-image-2":
        payload.update(quality="low", size="1024x1024", output_format="jpeg")

    for attempt in range(4):
        response = requests.post(
            f"{BASE_URL}/images/generations",
            headers={"Authorization": f"Bearer {KEY}"},
            json=payload,
            timeout=180,
        )
        if response.status_code not in {408, 429} and response.status_code &lt; 500:
            break
        time.sleep(2**attempt + random.random())
    response.raise_for_status()
    body = response.json()
    item = body["data"][0]

    if item.get("b64_json"):
        data = base64.b64decode(item["b64_json"])
        extension = body.get("output_format", "png")
    else:
        download = requests.get(item["url"], timeout=120)
        download.raise_for_status()
        data = download.content
        extension = {"image/png": "png", "image/webp": "webp"}.get(
            download.headers.get("content-type"), "jpg"
        )
    path = OUT / f"{job['id']}.{extension}"
    path.write_bytes(data)

    price, usage = CATALOG[model].get("pricing") or {}, body.get("usage", {})
    cost = price.get("per_request")
    if cost is None and price.get("input") is not None:
        cost = (usage.get("input_tokens", 0) * price["input"] +
                usage.get("output_tokens", 0) * price["output"]) / 1_000_000
    return {"id": job["id"], "model": model, "path": str(path),
            "estimated_usd": cost * price.get("ratio", 1) if cost is not None else None}

def safe_generate(job):
    try:
        return {"status": "success", **generate(job)}
    except Exception as error:
        return {"id": job["id"], "status": "failed", "error": str(error)}

OUT.mkdir(exist_ok=True)
jobs = [json.loads(line) for line in Path("jobs.jsonl").read_text().splitlines() if line]
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
    results = list(pool.map(safe_generate, jobs))
with (OUT / "manifest.jsonl").open("w") as manifest:
    manifest.writelines(json.dumps(result) + "\n" for result in results)

The catalog is fetched at runtime, but the fallback routes in ROUTES were verified on August 20, 2026. They should be checked again before deploying on another date.

Run a Small Smoke Test First

Start with one worker and one job:

MAX_WORKERS=1 python3 batch_image_pipeline.py

A successful GPT Image response has this shape:

{
  "created": 1776841943,
  "output_format": "jpeg",
  "quality": "low",
  "size": "1024x1024",
  "usage": {
    "input_tokens": 16,
    "output_tokens": 208,
    "total_tokens": 224
  },
  "data": [{"b64_json": ""}]
}

The script decodes this response and writes output/.jpeg. It then adds a success row to output/manifest.jsonl. URL-based responses are downloaded and represented in the same manifest format.

The code was syntax-checked locally, but a live generation request still requires a valid key. I would run this one-job test before increasing concurrency.

Estimating Cost

Pricing is time-sensitive. On August 20, 2026, the live catalog returned these base values and a 0.8 billing ratio:

gpt-image-2: $5 per 1M input tokens and $30 per 1M output tokens
Effective rates after the listed ratio: $4 per 1M input tokens and $24 per 1M output tokens
doubao-seedream-4-5-251128: $0.04 per request
Effective request price after the listed ratio: $0.032

The pricing guide describes token-based billing for models with official token pricing and request-based billing for models priced per call.

The script applies these calculations:

token cost = ratio × (input tokens × input rate + output tokens × output rate) / 1,000,000
request cost = ratio × per-request price

For the documented GPT Image response with 16 input tokens and 208 output tokens, the August 20 catalog values produce an illustrative estimate of about $0.005056.

The actual bill depends on the model, prompt, quality, size, response usage, and retries. Account usage and API responses should be treated as the billing record rather than a fixed per-image estimate.

I also budget for failed or rejected work. A retry after an uncertain timeout may create a second billable result. A technically successful image can still fail review. A useful operational metric is:

effective cost per accepted image = total batch spend / approved images

Troubleshooting Guide

Symptom	Likely cause	Fix
401	Missing or invalid key	Check the server-side `COMETAPI_KEY`
400	Invalid model or unsupported option	Recheck the live catalog and remove model-specific fields
429	Excessive concurrency	Lower `MAX_WORKERS` and retain exponential backoff
Repeated 5xx	Temporary upstream failure	Retry with a cap, then use a dead-letter queue
No saved image	Different response container	Inspect `data[0]` and support `b64_json` or `url`
Duplicate spend	Replay after partial failure	Use durable IDs and acknowledge only after storage succeeds

A permanent 400 will not become valid through retries. Likewise, an unlimited 429 loop can turn a traffic spike into a growing backlog.

What I Would Change for Production

For multiple workers, I would replace JSON Lines with a durable queue. Set the visibility timeout longer than the maximum generation time, acknowledge only after both the image and manifest are stored, and send exhausted jobs to a dead-letter queue.

Keep optional parameters model-specific. The shared payload should contain only fields such as model, prompt, and n: 1. Add quality, size, or output_format only when the selected model documentation supports them. If fallback routing is added, rebuild the payload for the fallback model rather than blindly reusing provider-specific options.

Other practical safeguards include:

Store keys in a secret manager.
Restrict and validate prompt input.
Scan generated assets according to your policy.
Keep provider URLs out of long-term product records.
Log job ID, model ID, latency, attempts, usage, storage path, review result, and catalog snapshot date.
Set a maximum batch size and per-job retry limit.
Add daily spend alerts and approval-rate stop conditions.

The key metric is not headline price. It is accepted-image cost after retries, failures, post-processing, and review.

Further References

For endpoint and response details, use the Quick Start, model catalog documentation, image generation reference, and pricing guide.

GPT-6 Astra Is Expensive per Token. I’m Not Sure That’s the Right Metric

Maya Collins — Tue, 08 Sep 2026 03:10:12 +0000

The first thing that jumped out at me when GPT-6 Astra launched was the price.

Standard API pricing starts at $10 per million input tokens and $50 per million output tokens. That’s 2.5× the short-context token price of GPT-5.6 Sol. Once a request goes beyond 272K input tokens, Astra moves onto an even more expensive long-context schedule.

On paper, that makes the decision look pretty straightforward.

If two models can do roughly the same work and one costs 2.5× more per token, use the cheaper one.

The problem is that Astra doesn’t seem to be optimized around doing the same work slightly better.

Most of the interesting gains show up when the model has to keep acting until something is actually finished.

That changes the economics quite a bit.

The benchmark pattern is stranger than “Astra is smarter”

If you only look at broad intelligence scores, Astra doesn’t look like a huge generational jump.

On the Artificial Analysis Intelligence Index reported in OpenAI’s comparison, GPT-6 Astra scores 61.2, while GPT-5.6 Sol scores 60.9.

That’s basically flat.

Now look at some of the execution-heavy benchmarks.

On Terminal-Bench 4.0:

GPT-6 Astra: 57.9%
GPT-5.6 Sol: 37.3%

On AutomationBench:

GPT-6 Astra: 41.4%
GPT-5.6 Sol: 18.1%

And on very-long-context MRCR in the 512K–1M range:

GPT-6 Astra: 96.3%
GPT-5.6 Sol: 73.8%

That’s a very different story.

The improvement is concentrated around tasks where the model has to interact with an environment, use tools, keep track of state, recover from mistakes, and carry something through to completion.

That’s also much closer to the kind of work I’d consider paying premium-model prices for.

A failed cheap run is still expensive

This is where token pricing starts becoming a bad shortcut.

Imagine I give a coding agent a repository migration.

The cheaper model costs $2 for a run.

The expensive model costs $5.

If the $2 model fails twice before producing something I can accept, while the $5 model gets there on the first attempt, the supposedly cheaper model is no longer obviously cheaper.

And API spend is only part of it.

Retries can also mean:

another terminal session
another browser run
more tool calls
duplicated infrastructure work
more time waiting
another human review pass

Once agents start doing real work, I care much more about what the entire task costs than what one token costs.

That’s why I’d rather track cost per accepted task.

It’s a boring metric, but it forces everything into the same number: token usage, retries, failed attempts, and whether the output was actually useful.

Astra’s token efficiency makes this less obvious than I expected

There’s another wrinkle here.

Independent testing from Artificial Analysis found that Astra used roughly three times fewer tokens than GPT-5.6 Sol at max effort in its coding-agent setup, while still scoring higher on its Coding Agent Index.

That meant Astra could end up around the same cost per coding-agent task despite having much more expensive individual tokens.

That does not mean Astra is suddenly cheap.

The same testing found a very different result on general-intelligence workloads. There, the higher per-token price outweighed the token savings, leaving Astra roughly 75% more expensive per task at max effort.

I actually like that result because it makes model selection less simplistic.

For coding agents, the premium may be recoverable.

For ordinary reasoning, maybe not.

Those are exactly the distinctions that disappear when everything gets reduced to "$10 input / $50 output."

I definitely wouldn’t use Astra everywhere

There are plenty of requests where paying for Astra makes little sense to me.

I wouldn’t reach for it first for basic classification, short rewriting, simple extraction, routine summaries, or other high-volume work that a cheaper model already handles reliably.

Even within coding, not every issue needs the strongest agent you can buy.

A one-file config change is very different from a migration that touches 30 files, runs tests, changes infrastructure, and needs to recover from failures along the way.

The harder it is to recover from a bad result, the easier Astra’s premium becomes to justify.

That’s the line I’d use.

Not “hard prompt versus easy prompt.”

More like cheap failure versus expensive failure.

The 1M context window has a price attached to it

Astra also comes with a roughly 1.05M-token context window and 128K maximum output.

That sounds great until it encourages people to stop managing context.

I wouldn’t.

Requests above 272K input tokens enter Astra’s long-context pricing tier, where Standard pricing rises to $20 per million input tokens and $75 per million output tokens.

So the million-token window is useful headroom, but it’s not free storage.

I’d still use retrieval.

I’d still trim irrelevant tool history.

I’d still summarize old state.

I’d still avoid dumping an entire repository into the prompt just because the API accepts it.

In fact, the bigger the available context becomes, the more important it is to know which parts are actually helping.

Where I’d pay the premium

The workloads that make Astra interesting to me are the ones where execution reliability matters more than raw generation price.

Things like:

repository-wide coding changes
terminal-heavy agents
browser and computer-use automation
SRE and debugging workflows
long technical investigations
workflows involving several tools
professional tasks where the final artifact has to be usable

OpenAI’s published results line up pretty closely with that. Astra’s biggest gains over Sol show up in computer use, terminal work, automation, science, SRE, and long-context retrieval rather than a giant jump in general intelligence.

That’s why I think treating Astra as “the new default GPT” misses the point.

It makes more sense to me as an escalation model.

Use something cheaper for the normal path.

Send Astra the jobs where retries, mistakes, or human intervention are actually expensive.

I’d route before I’d standardize

This is also why I’m increasingly skeptical of picking one model for an entire product.

Different requests have completely different economics.

A fast model might be perfect for 70% of the traffic.

A stronger mid-tier model can take another 20%.

Maybe Astra only sees the final 10% of tasks that are genuinely difficult.

If that 10% contains the work responsible for most of your failures, that can still be a very good trade.

I’ve been using CometAPI for this kind of comparison because I can keep the surrounding API layer mostly unchanged while swapping models and rerunning the same workload.

That’s the useful part for me.

I don’t want to compare a GPT-6 Astra demo against a completely different Gemini or Claude demo.

I want the same task, same tools, same success criteria, and different models.

Then I can compare completion rate, token use, latency, retries, and actual cost.

GPT-6 Astra is undeniably expensive per token.

For a lot of workloads, it will also simply be expensive per task.

But for the jobs where failed attempts are the expensive part, I’m not convinced token price is the number that matters most.

That’s what I’d benchmark before deciding whether the premium is worth paying.

Disclosure: This post is adapted from research originally published by the CometAPI team.