You've built a pipeline that calls an LLM to classify customer support tickets. It works perfectly in your notebook — you run it on a sample of 50 tickets and the results look great. Then your product manager asks you to run it on the full backlog: 50,000 tickets. You fire it up, do some quick math, realize it's going to take 14 hours, and start wondering if there's a better way.
There is. The problem isn't your prompt or your model — it's that you're calling the API sequentially, one ticket at a time, waiting for each response before sending the next request. LLM APIs are inherently I/O-bound: the bulk of your wall-clock time is spent waiting for network round trips and model inference, not doing any CPU work on your side. That means concurrency — running many requests in flight simultaneously — can cut your runtime from hours to minutes without changing a single word of your prompt.
By the end of this lesson, you'll understand exactly why sequential LLM calls are a bottleneck, how async and concurrent patterns eliminate that bottleneck, and how to build a production-grade orchestration layer that handles rate limits, retries, error isolation, and throughput tuning. We'll build a real working system — a document enrichment pipeline — not a toy hello-world demo.
What you'll learn:
asyncio and aiohttp (plus the OpenAI async client) enable parallel LLM requestsYou should be comfortable with:
You do not need prior experience with production ML systems or distributed computing.
Before writing a single line of async code, let's understand the problem precisely, because understanding it will tell you exactly what the solution needs to look like.
When you call the OpenAI API synchronously in a loop, your timeline looks roughly like this:
Request 1: [---send---][==waiting for inference==][---receive---]
Request 2: [---send---][==waiting==][---receive---]
Request 3: [---send---]...
Each request takes somewhere between 1 and 10 seconds depending on model, prompt length, and output length. The entire time your Python process is blocked on that network call, doing absolutely nothing useful. If each call takes 3 seconds on average and you have 10,000 documents, that's 30,000 seconds — over 8 hours — of mostly idle waiting.
The fix is to have many requests in flight at the same time:
Request 1: [---send---][==waiting==][---receive---]
Request 2: [---send---][==waiting==][---receive---]
Request 3: [---send---][==waiting==][---receive---]
...
Request N: [---send---][==waiting==][---receive---]
If you can maintain 50 concurrent requests, you divide your total wall-clock time by roughly 50. That 8-hour job becomes about 10 minutes.
The constraint isn't your hardware — it's the API's rate limits. OpenAI, Anthropic, and similar providers enforce limits on both requests per minute (RPM) and tokens per minute (TPM). Your job is to push as close to those limits as possible without exceeding them, while also handling the inevitable errors gracefully.
This is exactly what a concurrency orchestrator does.
Python's asyncio library implements cooperative multitasking: a single-threaded event loop that can switch between coroutines whenever one is waiting on I/O. This is ideal for LLM calls because the bottleneck is I/O, not CPU.
The OpenAI Python SDK (v1.x) ships with a first-class async client. Here's the minimal async pattern:
import asyncio
from openai import AsyncOpenAI
client = AsyncOpenAI() # reads OPENAI_API_KEY from environment
async def classify_ticket(ticket_id: str, text: str) -> dict:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": "Classify this customer support ticket into one of: billing, technical, account, general. Respond with only the category name."
},
{
"role": "user",
"content": text
}
],
temperature=0,
max_tokens=10
)
return {
"ticket_id": ticket_id,
"category": response.choices[0].message.content.strip().lower(),
"tokens_used": response.usage.total_tokens
}
async def main():
# Simple demo: three tickets in parallel
tickets = [
("TKT-001", "I can't log into my account after resetting my password."),
("TKT-002", "I was charged twice for my subscription this month."),
("TKT-003", "How do I export my data to CSV?"),
]
tasks = [classify_ticket(tid, text) for tid, text in tickets]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
asyncio.run(main())
asyncio.gather() submits all three coroutines to the event loop simultaneously. While one is waiting for the API response, the loop is free to progress the others. For three tickets this feels trivial — for 10,000 it's the difference between feasibility and impossibility.
Important:
asyncio.gather()by default will raise immediately if any task raises an exception, canceling other pending tasks. For production pipelines, you almost always wantreturn_exceptions=Trueso failures are captured as values rather than propagated. We'll handle this properly in the full implementation below.
Running 10,000 requests simultaneously would immediately trigger rate limit errors (and probably get your API key suspended). You need to control how many requests are in flight at any moment. The right tool for this is asyncio.Semaphore.
A semaphore is a counter that allows at most N concurrent acquirers. Any coroutine that tries to acquire a full semaphore will wait until one is released. This naturally limits concurrency without complex queuing logic:
import asyncio
from openai import AsyncOpenAI
from typing import Any
client = AsyncOpenAI()
async def classify_ticket_with_limit(
semaphore: asyncio.Semaphore,
ticket_id: str,
text: str
) -> dict:
async with semaphore: # blocks until a slot is available
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Classify this support ticket: billing, technical, account, or general."},
{"role": "user", "content": text}
],
temperature=0,
max_tokens=10
)
return {
"ticket_id": ticket_id,
"category": response.choices[0].message.content.strip().lower(),
"tokens_used": response.usage.total_tokens
}
async def run_pipeline(tickets: list[tuple[str, str]], max_concurrent: int = 20) -> list[dict]:
semaphore = asyncio.Semaphore(max_concurrent)
tasks = [
classify_ticket_with_limit(semaphore, tid, text)
for tid, text in tickets
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
The max_concurrent parameter is your primary throughput dial. How do you pick the right number?
Start with your provider's rate limits. If you're on an OpenAI Tier 2 plan with 5,000 RPM for gpt-4o-mini, and each request takes about 2 seconds, you can sustain roughly 5000/60 * 2 ≈ 167 concurrent requests in theory. In practice, start at 20–50 and increase until you start seeing 429 (rate limit) errors, then back off 20%.
Token budget matters more than request count for expensive models. If you're using gpt-4o and your prompts are 2,000 tokens each, your TPM limit will constrain you before your RPM limit does. We'll address token-aware rate limiting later.
Even with a semaphore, you'll hit rate limits occasionally — especially during bursts when your in-flight requests all complete around the same time and trigger a new burst. You'll also encounter transient network errors, 500s from the provider, and timeouts. Production pipelines need retry logic baked in.
Here's a production-grade retry decorator that handles these cases:
import asyncio
import random
import logging
from functools import wraps
from openai import RateLimitError, APIConnectionError, APIStatusError
logger = logging.getLogger(__name__)
def async_retry(
max_attempts: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
jitter: bool = True,
retryable_exceptions: tuple = (RateLimitError, APIConnectionError)
):
"""
Decorator for async functions. Retries on specified exceptions
with exponential backoff and optional jitter.
"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
attempt = 0
while attempt < max_attempts:
try:
return await func(*args, **kwargs)
except retryable_exceptions as e:
attempt += 1
if attempt >= max_attempts:
logger.error(
f"{func.__name__} failed after {max_attempts} attempts: {e}"
)
raise
delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
if jitter:
# Add up to 25% random jitter to prevent thundering herd
delay *= (1 + random.uniform(0, 0.25))
logger.warning(
f"{func.__name__} attempt {attempt} failed ({type(e).__name__}). "
f"Retrying in {delay:.2f}s..."
)
await asyncio.sleep(delay)
except APIStatusError as e:
# Don't retry 4xx errors except 429
if e.status_code == 429:
attempt += 1
if attempt >= max_attempts:
raise
delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
if jitter:
delay *= (1 + random.uniform(0, 0.25))
await asyncio.sleep(delay)
else:
raise # Don't retry 400, 401, 403, etc.
return wrapper
return decorator
The jitter is important. Without it, if 50 requests all hit a rate limit at the same moment, they'll all retry after exactly the same delay and immediately overwhelm the API again — a "thundering herd." Jitter spreads them out across a random window.
Now apply this to your LLM call:
@async_retry(max_attempts=5, base_delay=2.0)
async def classify_ticket_with_retry(
semaphore: asyncio.Semaphore,
ticket_id: str,
text: str
) -> dict:
async with semaphore:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "Classify this support ticket: billing, technical, account, or general."},
{"role": "user", "content": text}
],
temperature=0,
max_tokens=10
)
return {
"ticket_id": ticket_id,
"category": response.choices[0].message.content.strip().lower(),
"tokens_used": response.usage.total_tokens
}
Warning: Put the
async with semaphoreinside the retry loop, not outside. If you hold the semaphore slot while waiting for a retry delay, you're blocking that concurrency slot from processing other documents. The retry delay should happen outside the semaphore context, which the decorator pattern above handles correctly since the semaphore is acquired on each attempt.
Concurrency and batching are complementary strategies that solve slightly different problems. Concurrency maximizes parallelism across independent requests. Batching reduces the overhead per item by grouping multiple items into a single request.
For short, uniform tasks (classification, extraction, translation of small text chunks), you can often pack multiple items into a single prompt and parse the response. This reduces:
Here's a batching implementation for ticket classification that processes items in groups:
import json
from typing import Any
async def classify_batch(
semaphore: asyncio.Semaphore,
batch: list[dict] # list of {"ticket_id": str, "text": str}
) -> list[dict]:
"""
Classify a batch of tickets in a single LLM call.
Returns results in the same order as the input batch.
"""
# Build a numbered list for the LLM to respond to
items_text = "\n\n".join(
f"[{i+1}] Ticket {item['ticket_id']}:\n{item['text']}"
for i, item in enumerate(batch)
)
prompt = f"""Classify each of the following customer support tickets.
For each ticket, respond with ONLY a JSON object on its own line in this format:
{{"index": <number>, "ticket_id": "<id>", "category": "<category>"}}
Valid categories: billing, technical, account, general
Tickets:
{items_text}"""
async with semaphore:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are a support ticket classifier. Follow the output format exactly."},
{"role": "user", "content": prompt}
],
temperature=0,
max_tokens=len(batch) * 60 # ~60 tokens per result
)
# Parse the response — each line should be a JSON object
raw_output = response.choices[0].message.content.strip()
results = []
for line in raw_output.split('\n'):
line = line.strip()
if not line:
continue
try:
parsed = json.loads(line)
results.append({
"ticket_id": parsed["ticket_id"],
"category": parsed["category"].lower(),
"batch_tokens_used": response.usage.total_tokens
})
except (json.JSONDecodeError, KeyError) as e:
logger.warning(f"Failed to parse batch result line: {line!r} — {e}")
return results
def chunk_list(lst: list, chunk_size: int) -> list[list]:
"""Split a list into chunks of at most chunk_size."""
return [lst[i:i + chunk_size] for i in range(0, len(lst), chunk_size)]
async def run_batched_pipeline(
tickets: list[dict],
batch_size: int = 10,
max_concurrent: int = 15
) -> list[dict]:
semaphore = asyncio.Semaphore(max_concurrent)
batches = chunk_list(tickets, batch_size)
tasks = [classify_batch(semaphore, batch) for batch in batches]
batch_results = await asyncio.gather(*tasks, return_exceptions=True)
# Flatten results, skip failed batches
all_results = []
for i, result in enumerate(batch_results):
if isinstance(result, Exception):
logger.error(f"Batch {i} failed: {result}")
# Mark individual items as failed for retry
for item in batches[i]:
all_results.append({
"ticket_id": item["ticket_id"],
"category": None,
"error": str(result)
})
else:
all_results.extend(result)
return all_results
Batching isn't always better. The downsides:
Practical guidance: For classification and short extraction tasks, batches of 5–15 items work well with modern models. For anything requiring careful reasoning or long outputs per item, skip batching entirely and use concurrency alone.
Let's put everything together into a real-world system. The scenario: you have 5,000 product reviews that need to be enriched with sentiment (positive/negative/neutral), a one-sentence summary, and a list of product aspects mentioned (battery life, design, customer service, etc.). This is a standard data enrichment pipeline that teams run regularly.
import asyncio
import json
import logging
import time
from dataclasses import dataclass, field, asdict
from typing import Optional
from openai import AsyncOpenAI, RateLimitError, APIConnectionError, APIStatusError
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(name)s — %(message)s"
)
logger = logging.getLogger("review_enricher")
client = AsyncOpenAI()
@dataclass
class ReviewInput:
review_id: str
product_id: str
text: str
@dataclass
class ReviewEnrichment:
review_id: str
product_id: str
sentiment: Optional[str] = None
summary: Optional[str] = None
aspects: list[str] = field(default_factory=list)
tokens_used: int = 0
error: Optional[str] = None
attempts: int = 0
ENRICHMENT_SCHEMA = {
"type": "object",
"properties": {
"sentiment": {
"type": "string",
"enum": ["positive", "negative", "neutral", "mixed"]
},
"summary": {
"type": "string",
"description": "One sentence summary of the review"
},
"aspects": {
"type": "array",
"items": {"type": "string"},
"description": "Product aspects mentioned (e.g. battery_life, design, price, support)"
}
},
"required": ["sentiment", "summary", "aspects"]
}
async def enrich_review(
semaphore: asyncio.Semaphore,
review: ReviewInput,
max_attempts: int = 4,
base_delay: float = 2.0
) -> ReviewEnrichment:
result = ReviewEnrichment(
review_id=review.review_id,
product_id=review.product_id
)
for attempt in range(1, max_attempts + 1):
result.attempts = attempt
try:
async with semaphore:
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{
"role": "system",
"content": (
"You are a product review analyst. "
"Analyze the review and return structured data about it."
)
},
{
"role": "user",
"content": f"Analyze this product review:\n\n{review.text}"
}
],
response_format={"type": "json_object"},
temperature=0,
max_tokens=200
)
raw = response.choices[0].message.content
data = json.loads(raw)
result.sentiment = data.get("sentiment")
result.summary = data.get("summary")
result.aspects = data.get("aspects", [])
result.tokens_used = response.usage.total_tokens
result.error = None
return result
except (RateLimitError, APIConnectionError) as e:
if attempt == max_attempts:
result.error = f"Max retries exceeded: {type(e).__name__}: {e}"
logger.error(f"Review {review.review_id} failed permanently: {result.error}")
return result
delay = min(base_delay * (2 ** (attempt - 1)), 60.0)
delay *= (1 + random.uniform(0, 0.25))
logger.warning(
f"Review {review.review_id} attempt {attempt} hit {type(e).__name__}. "
f"Retrying in {delay:.1f}s"
)
await asyncio.sleep(delay)
except APIStatusError as e:
if e.status_code == 429:
if attempt < max_attempts:
delay = min(base_delay * (2 ** (attempt - 1)), 60.0)
await asyncio.sleep(delay)
continue
result.error = f"API error {e.status_code}: {e.message}"
logger.error(f"Review {review.review_id} non-retryable error: {result.error}")
return result
except (json.JSONDecodeError, KeyError) as e:
# Parsing failure — likely a model output issue, don't retry
result.error = f"Parse error: {e}"
logger.error(f"Review {review.review_id} parse error: {result.error}")
return result
return result # shouldn't reach here, but satisfies type checker
async def run_enrichment_pipeline(
reviews: list[ReviewInput],
max_concurrent: int = 30,
progress_interval: int = 100
) -> list[ReviewEnrichment]:
semaphore = asyncio.Semaphore(max_concurrent)
total = len(reviews)
completed = 0
start_time = time.monotonic()
logger.info(f"Starting enrichment pipeline: {total} reviews, max_concurrent={max_concurrent}")
tasks = [enrich_review(semaphore, review) for review in reviews]
results = []
# Process with progress logging
for coro in asyncio.as_completed(tasks):
result = await coro
results.append(result)
completed += 1
if completed % progress_interval == 0 or completed == total:
elapsed = time.monotonic() - start_time
rate = completed / elapsed
eta = (total - completed) / rate if rate > 0 else 0
success_count = sum(1 for r in results if r.error is None)
logger.info(
f"Progress: {completed}/{total} ({completed/total*100:.1f}%) | "
f"Rate: {rate:.1f}/s | ETA: {eta:.0f}s | "
f"Success rate: {success_count/completed*100:.1f}%"
)
return results
async def main():
import random
# Simulate loading reviews from a database or file
sample_texts = [
"This laptop's battery life is incredible — 12 hours easily. The keyboard feels premium too. Only downside is it runs a bit hot under load.",
"Terrible customer service. Waited 3 weeks for a replacement and they kept closing my tickets. The product itself works fine.",
"Average product. Does what it says on the box. Nothing exceptional. Price seems about right for what you get.",
"Absolutely love the design and build quality. Feels like a premium device. App could use some work though.",
"Battery died after 6 months. Replacement process was smooth and support team was helpful. Mixed feelings overall.",
]
reviews = [
ReviewInput(
review_id=f"REV-{i:05d}",
product_id=f"PROD-{(i % 50):03d}",
text=random.choice(sample_texts)
)
for i in range(500) # Scale this to 5000+ in production
]
results = await run_enrichment_pipeline(reviews, max_concurrent=30)
# Summary stats
successful = [r for r in results if r.error is None]
failed = [r for r in results if r.error is not None]
total_tokens = sum(r.tokens_used for r in successful)
print(f"\n{'='*50}")
print(f"Pipeline complete")
print(f"Successful: {len(successful)}/{len(results)}")
print(f"Failed: {len(failed)}")
print(f"Total tokens used: {total_tokens:,}")
print(f"Avg tokens per review: {total_tokens/len(successful):.0f}" if successful else "N/A")
if failed:
print(f"\nFailed reviews:")
for r in failed[:5]: # Show first 5
print(f" {r.review_id}: {r.error}")
if __name__ == "__main__":
import random
asyncio.run(main())
The use of asyncio.as_completed() here instead of asyncio.gather() is intentional. as_completed() processes results as they arrive, which lets you log progress in real time and (if needed) stream results to a downstream sink without waiting for all tasks to finish. With asyncio.gather(), you'd collect everything in memory before processing any of it.
For high-volume pipelines with models like gpt-4o where TPM limits are often more constraining than RPM limits, a simple concurrency semaphore isn't enough. You need to track estimated token consumption and throttle when you're approaching your TPM ceiling.
Here's a lightweight token bucket implementation:
import asyncio
import time
from collections import deque
class TokenBucket:
"""
Rate limiter that enforces both requests-per-minute and tokens-per-minute limits.
Uses a sliding window approach.
"""
def __init__(self, max_rpm: int, max_tpm: int, window_seconds: int = 60):
self.max_rpm = max_rpm
self.max_tpm = max_tpm
self.window = window_seconds
self._request_times: deque = deque()
self._token_usage: deque = deque() # (timestamp, tokens) pairs
self._lock = asyncio.Lock()
async def acquire(self, estimated_tokens: int) -> None:
"""
Wait until there's capacity for a request with the given estimated token count.
"""
while True:
async with self._lock:
now = time.monotonic()
cutoff = now - self.window
# Evict old entries
while self._request_times and self._request_times[0] < cutoff:
self._request_times.popleft()
while self._token_usage and self._token_usage[0][0] < cutoff:
self._token_usage.popleft()
current_rpm = len(self._request_times)
current_tpm = sum(t for _, t in self._token_usage)
if current_rpm < self.max_rpm and (current_tpm + estimated_tokens) <= self.max_tpm:
self._request_times.append(now)
self._token_usage.append((now, estimated_tokens))
return # Cleared to proceed
# Not cleared — wait a bit and retry
await asyncio.sleep(0.1)
def record_actual_tokens(self, estimated_tokens: int, actual_tokens: int):
"""Adjust the last recorded entry with actual token count after response."""
# Simple approach: we pre-booked estimated_tokens, adjust if different
# In practice, for short-output tasks, estimates are close enough
pass # Production implementation would update the deque entry
# Usage in the pipeline:
async def enrich_review_with_token_limit(
semaphore: asyncio.Semaphore,
token_bucket: TokenBucket,
review: ReviewInput
) -> ReviewEnrichment:
# Estimate tokens: ~4 chars per token, plus ~100 for system + response overhead
estimated_tokens = len(review.text) // 4 + 150
await token_bucket.acquire(estimated_tokens)
async with semaphore:
# ... make the API call ...
pass
Tip: Don't over-engineer token tracking early on. For most workloads, a conservative
max_concurrentsemaphore combined with retry-on-429 will get you 80% of the way there. Add token-aware limiting when you have real data showing TPM is your actual constraint.
In any pipeline running thousands of LLM calls, some calls will fail — and that's fine, as long as you handle failures gracefully rather than letting them corrupt your results or crash your pipeline.
| Failure type | HTTP status | Should retry? | Action |
|---|---|---|---|
| Rate limit | 429 | Yes | Exponential backoff |
| Network timeout | — | Yes | Immediate retry, then backoff |
| Server error | 500, 502, 503 | Yes (limited) | Backoff, give up after 3-4 attempts |
| Bad request | 400 | No | Log, skip, flag for review |
| Content filter | 400 (specific) | No | Log, mark as filtered |
| Auth error | 401, 403 | No | Raise immediately — wrong key |
| Context length | 400 (specific) | Maybe | Truncate input, retry once |
Build your pipeline to output failed items to a separate list with their errors attached. After the main pipeline completes, you can inspect failures, fix systematic issues, and re-run just the failed subset:
async def retry_failed_items(
failed_results: list[ReviewEnrichment],
original_reviews: dict[str, ReviewInput], # keyed by review_id
max_concurrent: int = 10 # more conservative for retry runs
) -> list[ReviewEnrichment]:
"""Re-run only the failed items with a more conservative concurrency setting."""
failed_reviews = [
original_reviews[r.review_id]
for r in failed_results
if r.review_id in original_reviews
]
logger.info(f"Retrying {len(failed_reviews)} failed items with max_concurrent={max_concurrent}")
return await run_enrichment_pipeline(failed_reviews, max_concurrent=max_concurrent)
This pattern — run at high concurrency, collect failures, retry failures conservatively — is more reliable than trying to get everything right on the first pass. You'll typically find that 95%+ of items succeed on the first run and the stragglers succeed on retry.
Build a parallel LLM pipeline for a realistic scenario: extracting structured information from job postings.
Your task: You have 200 raw job posting texts (simulate these with varied samples). Build a pipeline that, for each posting, extracts:
role_level: one of intern, junior, mid, senior, staff, leadershipremote_policy: one of remote, hybrid, onsite, unspecified required_skills: list of technical skills mentionedsalary_mentioned: booleanRequirements:
AsyncOpenAI with a semaphore limiting to 25 concurrent requestsresponse_format={"type": "json_object"} for structured outputasyncio.as_completed() to log progress every 25 completionsremote_policy values across successful resultsjob_enrichments.jsonl), one result per lineStretch goal: Implement batching where you process 5 postings per API call to reduce total API calls by 5x. Compare total runtime and token usage between batched and non-batched approaches.
# WRONG — holds the semaphore slot during backoff sleep
async with semaphore:
for attempt in range(max_attempts):
try:
return await call_api()
except RateLimitError:
await asyncio.sleep(backoff_delay)
# RIGHT — releases the slot during backoff
for attempt in range(max_attempts):
try:
async with semaphore:
return await call_api()
except RateLimitError:
await asyncio.sleep(backoff_delay)
Holding the semaphore during sleep starves other coroutines of concurrency slots during the exact moment (API rate limit recovery) when they most need throughput.
Without it, the first exception from any task cancels all pending tasks and propagates up. You lose all in-flight work. Always use return_exceptions=True in production and check each result with isinstance(result, Exception).
Threading works, but it's heavier and harder to control. Each thread consumes memory and OS resources. Async coroutines are lightweight and explicitly designed for I/O-bound work. Migrate to the async client — it's not significantly more complex.
asyncio.as_completed() returns results in completion order, not submission order. If downstream processing requires ordered results (e.g., you're writing to a database keyed by row index), attach the original ID or index to every result, as shown in the pipeline above.
If your max_tokens is too tight, the model will truncate mid-response — sometimes in the middle of a JSON object, causing parse errors. For structured output, always add a buffer. If you're extracting a list of aspects and expect maybe 5 aspects at ~10 tokens each, don't set max_tokens=50. Set it to 200.
If you're seeing a high rate of 429 errors even with a semaphore:
await asyncio.sleep(0.05) after each acquire adds 50ms spacing without meaningfully hurting throughputYou now have a complete mental model and working implementation for high-throughput LLM pipelines. Let's recap the key ideas:
The core insight: LLM API calls are I/O-bound. Sequential loops waste nearly all of your time waiting. Async concurrency lets you maintain dozens of in-flight requests simultaneously, making jobs that would take hours take minutes.
The control knobs:
max_concurrent (semaphore): controls parallelism, tune based on your API tier's RPM limitbatch_size: reduces per-item overhead for uniform short tasks; adds parsing complexitymax_attempts + backoff: makes your pipeline resilient to transient failuresWhen to use which pattern:
Where to go from here:
response_format with Pydantic models for automatic validation of LLM responses before they enter your data pipelinestream=True lets you process partial responses as they arrive — explore client.chat.completions.create(stream=True) with async iterationThe pipeline pattern you built here — async coroutines, semaphore-controlled concurrency, per-item retry, progress tracking, failure collection — transfers directly to any LLM provider and any enrichment task. It's a production-ready template, not a toy.
Learning Path: Building with LLMs