Module 4: Image Generation
7. Pipeline Integration
Description
Generating an image in isolation is useful for prototypes. But in production, image generation is a step within a larger pipeline: a system that generates an article and needs illustrations, an e-commerce site that creates automatic product listings, a flow that generates and validates images before publishing. In this capsule you'll build 4 concrete pipelines that combine LLMs, image generation and Vision APIs.
Why it matters: Generation without verification is a risk in production. DALL-E may generate an image that doesn't match the prompt. Without a verification step with the Vision API, you publish images you don't know are correct. The pipelines you build here include verification, fallback, cost tracking and error handling.
Connection with the module: This capsule integrates everything before it: generation with DALL-E (capsule 02) and SD (capsule 03), optimized prompts (capsule 05), editing (capsule 06). The final project (capsule 08) is a productized pipeline with fallback.
Reusable Cost Tracking
Before the pipelines, we define the cost tracker we'll use in all of them:
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class CostTracker:
items: list[dict] = field(default_factory=list)
def add(self, step: str, model: str, cost: float):
self.items.append({"step": step, "model": model, "cost_usd": cost})
@property
def total(self) -> float:
return sum(item["cost_usd"] for item in self.items)
def summary(self) -> str:
lines = [f" {i['step']} ({i['model']}): ${i['cost_usd']:.4f}" for i in self.items]
lines.append(f" TOTAL: ${self.total:.4f}")
return "\n".join(lines)
COST_TABLE = {
"dall-e-3-standard-1024": 0.040,
"dall-e-3-hd-1024": 0.080,
"gpt-4o-vision": 0.005,
"gpt-4o-mini-per-1k": 0.000150,
"sdxl-replicate": 0.004,
}
Pipeline 1: Text → Image → Verification with Vision
Generates an image and uses GPT-4o Vision to verify that the result matches the prompt.
import json
import time
from dataclasses import dataclass, field
from openai import OpenAI
client = OpenAI()
@dataclass
class VerifiedImageResult:
success: bool
image_b64: str = ""
score: float = 0.0
verification: dict = field(default_factory=dict)
prompt_used: str = ""
revised_prompt: str = ""
cost: float = 0.0
latency_seconds: float = 0.0
def pipeline_generate_and_verify(
prompt: str,
min_score: float = 0.7,
max_retries: int = 2,
) -> VerifiedImageResult:
costs = CostTracker()
start = time.time()
best_result = None
for attempt in range(max_retries + 1):
try:
gen = client.images.generate(
model="dall-e-3", prompt=prompt, size="1024x1024",
quality="standard", response_format="b64_json", n=1,
)
b64 = gen.data[0].b64_json
costs.add(f"generation_{attempt}", "dall-e-3", COST_TABLE["dall-e-3-standard-1024"])
verify = client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{
"type": "text",
"text": (
f"Evaluate whether this image corresponds to: '{prompt}'\n"
'JSON: {{"score": 0-1, "matches_subject": true/false, '
'"issues": [...], "description": "..."}}'
),
},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}},
],
}],
max_tokens=200,
)
costs.add(f"verification_{attempt}", "gpt-4o", COST_TABLE["gpt-4o-vision"])
try:
eval_data = json.loads(verify.choices[0].message.content)
except json.JSONDecodeError:
eval_data = {"score": 0.5, "issues": ["Parse error"]}
score = eval_data.get("score", 0)
result = VerifiedImageResult(
success=score >= min_score, image_b64=b64, score=score,
verification=eval_data, prompt_used=prompt,
revised_prompt=gen.data[0].revised_prompt,
cost=costs.total, latency_seconds=round(time.time() - start, 2),
)
if best_result is None or score > best_result.score:
best_result = result
if score >= min_score:
return result
except Exception as e:
best_result = VerifiedImageResult(
success=False, cost=costs.total,
latency_seconds=round(time.time() - start, 2),
)
return best_result
result = pipeline_generate_and_verify(
"Minimalist logo for an AI company, blue and white, geometric shapes",
min_score=0.75, max_retries=2,
)
print(f"Score: {result.score}, Cost: ${result.cost:.4f}, Latency: {result.latency_seconds}s")
Pipeline 2: Document → Summary → Illustration
Takes a document, generates a visual summary with an LLM, converts the summary into an image prompt, and generates the illustration.
import json
import time
from openai import OpenAI
client = OpenAI()
def pipeline_document_to_illustration(
document_text: str,
style: str = "modern flat illustration",
) -> dict:
costs = CostTracker()
start = time.time()
truncated = document_text[:4000]
summary_resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
"Given a document, generate:\n"
"1. A 2-3 sentence summary\n"
"2. A visual concept that represents the central idea\n"
'JSON: {"summary": "...", "visual_concept": "..."}'
),
},
{"role": "user", "content": truncated},
],
max_tokens=200,
)
costs.add("summarization", "gpt-4o-mini", 0.000300)
try:
summary = json.loads(summary_resp.choices[0].message.content)
except json.JSONDecodeError:
summary = {"summary": summary_resp.choices[0].message.content, "visual_concept": "abstract representation"}
prompt_resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": f"Convert into a prompt for DALL-E 3. Style: {style}. English, 2-3 sentences. Only the prompt.",
},
{"role": "user", "content": summary["visual_concept"]},
],
max_tokens=150,
)
costs.add("prompt_generation", "gpt-4o-mini", 0.000150)
image_prompt = prompt_resp.choices[0].message.content.strip()
gen = client.images.generate(
model="dall-e-3", prompt=image_prompt, size="1024x1024",
quality="standard", response_format="b64_json", n=1,
)
costs.add("image_generation", "dall-e-3", COST_TABLE["dall-e-3-standard-1024"])
return {
"summary": summary["summary"],
"visual_concept": summary["visual_concept"],
"image_prompt": image_prompt,
"image_b64": gen.data[0].b64_json,
"total_cost": costs.total,
"latency": round(time.time() - start, 2),
"cost_breakdown": costs.summary(),
}
document = """
Artificial intelligence is transforming the way companies manage their data.
Language models can analyze documents, extract key information and generate
automatic reports. Companies that adopt these technologies report a 40%
reduction in processing time.
"""
illust = pipeline_document_to_illustration(document)
print(f"Summary: {illust['summary']}")
print(f"Prompt: {illust['image_prompt']}")
print(f"Cost: ${illust['total_cost']:.4f}")
Pipeline 3: Product Description → Image → Quality Check
Pipeline for e-commerce: takes a description, generates a professional photo, validates quality before publishing.
import json
import time
from openai import OpenAI
client = OpenAI()
QUALITY_CRITERIA = {
"product_visible": "Product clearly visible and recognizable",
"background_clean": "Clean background that doesn't distract",
"lighting_pro": "Professional and uniform lighting",
"no_artifacts": "No artifacts, distortions or generated text",
"proportions": "Correct product proportions",
}
def pipeline_product_image(
product_description: str,
product_name: str = "",
approval_threshold: float = 0.75,
) -> dict:
costs = CostTracker()
start = time.time()
prompt_resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
"Generate a professional product photography prompt in English. "
"Include: centered product, white background, studio lighting, 4K. Only the prompt."
),
},
{"role": "user", "content": product_description},
],
max_tokens=150,
)
costs.add("prompt", "gpt-4o-mini", 0.000150)
img_prompt = prompt_resp.choices[0].message.content.strip()
gen = client.images.generate(
model="dall-e-3", prompt=img_prompt, size="1024x1024",
quality="hd", response_format="b64_json", n=1,
)
costs.add("generation", "dall-e-3", COST_TABLE["dall-e-3-hd-1024"])
b64 = gen.data[0].b64_json
criteria_text = "\n".join(f"- {k}: {v}" for k, v in QUALITY_CRITERIA.items())
qc_resp = client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{
"type": "text",
"text": (
f"Evaluate this product image ({product_description}).\n"
f"Criteria:\n{criteria_text}\n"
'JSON: {{"scores": {{...}}, "overall": 0-1, "issues": [...], '
'"ecommerce_ready": true/false}}'
),
},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{b64}"}},
],
}],
max_tokens=300,
)
costs.add("quality_check", "gpt-4o", COST_TABLE["gpt-4o-vision"])
try:
qc = json.loads(qc_resp.choices[0].message.content)
except json.JSONDecodeError:
qc = {"overall": 0.5, "issues": ["Parse error"], "ecommerce_ready": False}
approved = qc.get("overall", 0) >= approval_threshold and qc.get("ecommerce_ready", False)
return {
"product": product_name or product_description[:50],
"approved": approved,
"quality_score": qc.get("overall", 0),
"issues": qc.get("issues", []),
"image_prompt": img_prompt,
"total_cost": costs.total,
"latency": round(time.time() - start, 2),
}
product = pipeline_product_image(
"Over-ear wireless headphones in matte black",
product_name="WH-1000XM5",
)
print(f"Approved: {'YES' if product['approved'] else 'NO'} (score: {product['quality_score']})")
print(f"Cost: ${product['total_cost']:.4f}")
Batch of products
def batch_product_images(products: list[dict], threshold: float = 0.75) -> dict:
results = [
pipeline_product_image(p["description"], p.get("name", ""), threshold)
for p in products
]
approved = [r for r in results if r["approved"]]
return {
"total": len(results),
"approved": len(approved),
"approval_rate": len(approved) / len(results) if results else 0,
"total_cost": sum(r["total_cost"] for r in results),
"results": results,
}
Pipeline 4: Input → Generate → Edit → Verify → Final
The most complete pipeline: generates a base image, edits it (inpainting), verifies with Vision, and produces the final deliverable.
import json
import time
import base64
import io
import tempfile
import urllib.request
from PIL import Image, ImageDraw
from openai import OpenAI
client = OpenAI()
def pipeline_full_generation(
user_input: str,
edit_instruction: str = "",
min_quality: float = 0.7,
) -> dict:
costs = CostTracker()
start = time.time()
steps_done = []
prompt_resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Convert into a professional prompt for DALL-E 3. English, 2-3 sentences."},
{"role": "user", "content": user_input},
],
max_tokens=200,
)
costs.add("prompt_optimization", "gpt-4o-mini", 0.000150)
optimized = prompt_resp.choices[0].message.content.strip()
steps_done.append("prompt_optimization")
gen = client.images.generate(
model="dall-e-3", prompt=optimized, size="1024x1024",
quality="standard", response_format="b64_json", n=1,
)
costs.add("generation", "dall-e-3", COST_TABLE["dall-e-3-standard-1024"])
current_b64 = gen.data[0].b64_json
steps_done.append("generation")
if edit_instruction:
img_bytes = base64.b64decode(current_b64)
img = Image.open(io.BytesIO(img_bytes)).convert("RGBA")
w, h = img.size
mask = Image.new("RGBA", (w, h), (0, 0, 0, 255))
draw = ImageDraw.Draw(mask)
draw.rectangle([0, int(h * 0.6), w, h], fill=(0, 0, 0, 0))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as fi:
img.save(fi, "PNG")
img_path = fi.name
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as fm:
mask.save(fm, "PNG")
mask_path = fm.name
with open(img_path, "rb") as fi, open(mask_path, "rb") as fm:
edit_resp = client.images.edit(image=fi, mask=fm, prompt=edit_instruction, n=1, size="1024x1024")
costs.add("editing", "dall-e-2", 0.020)
edited_bytes = urllib.request.urlopen(edit_resp.data[0].url).read()
current_b64 = base64.b64encode(edited_bytes).decode("utf-8")
steps_done.append("editing")
verify = client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{
"type": "text",
"text": (
f"Evaluate for: '{user_input}'\n"
'JSON: {{"quality": 0-1, "relevance": 0-1, "production_ready": true/false}}'
),
},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{current_b64}"}},
],
}],
max_tokens=100,
)
costs.add("verification", "gpt-4o", COST_TABLE["gpt-4o-vision"])
steps_done.append("verification")
try:
v_data = json.loads(verify.choices[0].message.content)
except json.JSONDecodeError:
v_data = {"quality": 0.5, "relevance": 0.5, "production_ready": False}
avg = (v_data.get("quality", 0) + v_data.get("relevance", 0)) / 2
return {
"steps": steps_done,
"approved": avg >= min_quality and v_data.get("production_ready", False),
"verification": v_data,
"total_cost": costs.total,
"cost_breakdown": costs.summary(),
"latency": round(time.time() - start, 2),
}
final = pipeline_full_generation(
"Banner for a blog post about machine learning in healthcare",
edit_instruction="add text space on the left side with subtle gradient overlay",
)
print(f"Steps: {final['steps']}")
print(f"Approved: {final['approved']}")
print(f"Cost: ${final['total_cost']:.4f}")
Troubleshooting
Rate limit in batch
Symptom: Error 429 when generating images in batch.
import time
def rate_limited_generation(prompts: list[str], delay: float = 15.0) -> list[dict]:
results = []
for i, prompt in enumerate(prompts):
for attempt in range(3):
try:
resp = client.images.generate(model="dall-e-3", prompt=prompt, size="1024x1024", n=1)
results.append({"index": i, "url": resp.data[0].url, "success": True})
break
except Exception as e:
if "rate" in str(e).lower() or "429" in str(e):
time.sleep(delay * (attempt + 1))
else:
results.append({"index": i, "success": False, "error": str(e)})
break
if i < len(prompts) - 1:
time.sleep(delay)
return results
Inconsistent verification with Vision
Cause: Inherent variability in subjective evaluations.
Solution: Use multiple specific criteria (subject_match, quality, coherence) instead of a general score, and average the results. This reduces the variance between evaluations.
Pipeline cost spikes
Cause: Excessive retries, unnecessary HD, redundant verifications.
Solution: Use CostTracker to identify the most expensive step. Lower quality to "standard" in test iterations, reduce max_retries, verify only in the final step.
Exercises
Exercise 1: Multi-language pipeline with illustrations
Create a pipeline that takes an article in Spanish, summarizes it, generates an illustration prompt, generates the image, and verifies relevance. Return summary, illustration and score.
See solution
import json
from openai import OpenAI
client = OpenAI()
def pipeline_article_illustration(article_text: str, style: str = "modern editorial illustration") -> dict:
costs = CostTracker()
summary_resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": 'Summarize the article and identify the visual theme. JSON: {"summary_es": "...", "visual_theme": "..."}'},
{"role": "user", "content": article_text[:4000]},
],
max_tokens=200,
)
costs.add("summary", "gpt-4o-mini", 0.000300)
summary = json.loads(summary_resp.choices[0].message.content)
prompt_resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": f"Create a DALL-E 3 prompt based on the visual theme. Style: {style}. English. Only the prompt."},
{"role": "user", "content": summary["visual_theme"]},
],
max_tokens=150,
)
costs.add("prompt", "gpt-4o-mini", 0.000150)
img_prompt = prompt_resp.choices[0].message.content.strip()
gen = client.images.generate(
model="dall-e-3", prompt=img_prompt, size="1024x1024", response_format="b64_json", n=1,
)
costs.add("generation", "dall-e-3", 0.040)
verify = client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{"type": "text", "text": f"Article: {summary['summary_es']}\nIs the illustration relevant? JSON: {{\"relevance\": 0-1}}"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{gen.data[0].b64_json}"}},
],
}],
max_tokens=50,
)
costs.add("verification", "gpt-4o", 0.005)
relevance = json.loads(verify.choices[0].message.content).get("relevance", 0)
return {
"summary": summary["summary_es"],
"image_prompt": img_prompt,
"relevance": relevance,
"approved": relevance >= 0.7,
"cost_breakdown": costs.summary(),
}
r = pipeline_article_illustration(
"Quantum computing promises to revolutionize cryptography and molecular simulation."
)
print(f"Relevance: {r['relevance']}, Approved: {r['approved']}")
Exercise 2: Catalog pipeline with multi-provider fallback
Generate images for 3 products. For each one: try DALL-E 3 first; if it fails, use SD as fallback. Include quality check and cost tracking.
See solution
import json
import time
import base64
import urllib.request
from openai import OpenAI
import replicate
client = OpenAI()
def catalog_with_fallback(products: list[dict], delay: float = 5.0) -> dict:
costs = CostTracker()
results = []
for i, prod in enumerate(products):
name = prod.get("name", f"Product {i}")
image_b64 = None
provider = None
prompt_resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Professional product photo prompt in English. Only the prompt."},
{"role": "user", "content": prod["description"]},
],
max_tokens=100,
)
costs.add(f"{name}_prompt", "gpt-4o-mini", 0.000150)
img_prompt = prompt_resp.choices[0].message.content.strip()
try:
gen = client.images.generate(
model="dall-e-3", prompt=img_prompt, size="1024x1024",
response_format="b64_json", n=1,
)
image_b64 = gen.data[0].b64_json
provider = "dall-e-3"
costs.add(f"{name}_gen", "dall-e-3", 0.040)
except Exception:
try:
sd_out = replicate.run(
"stability-ai/sdxl:39a52a2a03a4faf0651640ac8a542059c52f2d04fc26c8b83e22b0a957ffedd3",
input={"prompt": img_prompt, "negative_prompt": "blurry, low quality",
"width": 1024, "height": 1024},
)
url = sd_out[0] if isinstance(sd_out, list) else str(sd_out)
image_b64 = base64.b64encode(urllib.request.urlopen(url).read()).decode("utf-8")
provider = "sd"
costs.add(f"{name}_gen", "sdxl", 0.004)
except Exception as e:
results.append({"name": name, "success": False, "error": str(e)})
continue
qc = client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{"type": "text", "text": f"Product photo '{prod['description']}'. JSON: {{\"score\": 0-1}}"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_b64}"}},
],
}],
max_tokens=50,
)
costs.add(f"{name}_qc", "gpt-4o", 0.005)
score = json.loads(qc.choices[0].message.content).get("score", 0)
results.append({"name": name, "provider": provider, "score": score, "approved": score >= 0.7, "success": True})
if i < len(products) - 1:
time.sleep(delay)
return {
"approved": sum(1 for r in results if r.get("approved")),
"total": len(results),
"results": results,
"cost_breakdown": costs.summary(),
}
cat = catalog_with_fallback([
{"name": "Headphones", "description": "Premium wireless headphones, matte black"},
{"name": "Watch", "description": "Smartwatch with AMOLED display, blue strap"},
{"name": "Lamp", "description": "LED desk lamp, articulated arm"},
])
print(f"Approved: {cat['approved']}/{cat['total']}")
print(cat["cost_breakdown"])
Exercise 3: Iterative generation pipeline with editing
Create a pipeline that: (1) generates a base image, (2) evaluates it with Vision, (3) if score < 0.8 identifies the problem area, (4) uses inpainting to fix it, (5) verifies again. Maximum 2 cycles.
See solution
import json
import time
import base64
import io
import tempfile
import urllib.request
from PIL import Image, ImageDraw
from openai import OpenAI
client = OpenAI()
def iterative_generate_and_fix(prompt: str, threshold: float = 0.8, max_fixes: int = 2) -> dict:
costs = CostTracker()
history = []
gen = client.images.generate(
model="dall-e-3", prompt=prompt, size="1024x1024", response_format="b64_json", n=1,
)
costs.add("initial_gen", "dall-e-3", 0.040)
current_b64 = gen.data[0].b64_json
for fix_round in range(max_fixes + 1):
ev = client.chat.completions.create(
model="gpt-4o",
messages=[{
"role": "user",
"content": [
{"type": "text", "text": (
f"Evaluate for: '{prompt}'\n"
'JSON: {{"score": 0-1, "issues": [{{"area": "top/center/bottom", "description": "..."}}], '
'"fix_prompt": "prompt to fix it"}}'
)},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{current_b64}"}},
],
}],
max_tokens=300,
)
costs.add(f"eval_{fix_round}", "gpt-4o", 0.005)
try:
eval_data = json.loads(ev.choices[0].message.content)
except json.JSONDecodeError:
eval_data = {"score": 0.5, "issues": [], "fix_prompt": ""}
score = eval_data.get("score", 0)
history.append({"round": fix_round, "score": score, "issues": eval_data.get("issues", [])})
if score >= threshold or fix_round == max_fixes or not eval_data.get("issues"):
break
area = eval_data["issues"][0].get("area", "center")
img = Image.open(io.BytesIO(base64.b64decode(current_b64))).convert("RGBA")
w, h = img.size
mask = Image.new("RGBA", (w, h), (0, 0, 0, 255))
draw = ImageDraw.Draw(mask)
coords = {"top": (0, 0, w, h//3), "center": (w//4, h//4, 3*w//4, 3*h//4), "bottom": (0, 2*h//3, w, h)}
draw.rectangle(coords.get(area, coords["center"]), fill=(0, 0, 0, 0))
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as fi:
img.save(fi, "PNG"); img_path = fi.name
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as fm:
mask.save(fm, "PNG"); mask_path = fm.name
with open(img_path, "rb") as fi, open(mask_path, "rb") as fm:
edit = client.images.edit(image=fi, mask=fm, prompt=eval_data.get("fix_prompt", prompt), n=1, size="1024x1024")
costs.add(f"fix_{fix_round}", "dall-e-2", 0.020)
current_b64 = base64.b64encode(urllib.request.urlopen(edit.data[0].url).read()).decode("utf-8")
return {
"final_score": history[-1]["score"],
"fix_rounds": len(history) - 1,
"history": history,
"approved": history[-1]["score"] >= threshold,
"cost_breakdown": costs.summary(),
}
r = iterative_generate_and_fix(
"Professional team meeting in modern glass conference room", threshold=0.8, max_fixes=2,
)
print(f"Final score: {r['final_score']}, Rounds: {r['fix_rounds']}, Approved: {r['approved']}")
Additional Resources
- OpenAI Vision API — Image verification
- OpenAI Image Generation — Generation and editing
- OpenAI Pricing — Prices for cost calculation
- Replicate Pricing — SD costs via Replicate
- Building Multimodal Pipelines — OpenAI Cookbook