使用异步处理和队列进行批处理
使用 asyncio 和作业队列构建异步提取流程，在遵守速率限制并跟踪进度的同时并行处理数千份文档。
使用异步处理和队列进行批处理 是 CoddyKit 上的免费 AI Engineering Academy 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Engineering Academy 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Engineering Academy 课程共包含 4 节课。
批处理为何重要
在生产环境中逐份处理数千份文档太慢了。按顺序依次调用 OpenAI API 的同步循环每秒可能只能处理 1 份文档，这意味着处理 10,000 份文档将近 3 小时。异步批处理可以同时并行处理数百个请求，将总耗时降低一个数量级。
asyncio 基础
Python 的 asyncio 事件循环可以让您在不使用线程的情况下并发运行许多受 I/O 限制的任务。当一个 API 调用正在等待网络响应时，事件循环会切换去处理另一个调用。您使用 async def 和 await 关键字编写代码，由运行时负责调度。对于大部分时间都在等待服务器的 LLM 调用而言，这种方式非常理想。
import asyncio
import instructor
from openai import AsyncOpenAI
async_client = instructor.from_openai(AsyncOpenAI())
async def extract_one(text: str) -> PersonExtract:
return await async_client.chat.completions.create(
model='gpt-4o-mini',
response_model=PersonExtract,
messages=[{'role': 'user', 'content': text}]
)使用 gather 运行多个提取任务
asyncio.gather 会并发运行一组协程，并在最后一个协程完成后返回所有结果。对于少量文档组成的批次，这已经足够。请使用列表推导式包装提取协程，然后将其传递给 gather。总耗时大致等于最慢的单次调用耗时，而不是所有调用耗时之和。
async def batch_extract(texts: list) -> list:
tasks = [extract_one(text) for text in texts]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Filter out exceptions
successes = [r for r in results if not isinstance(r, Exception)]
failures = [r for r in results if isinstance(r, Exception)]
print(f'Success: {len(successes)}, Failures: {len(failures)}')
return successes
results = asyncio.run(batch_extract(documents))使用 Semaphore 控制并发
同时发送数千个请求会触发速率限制，并导致 429 错误。请使用 asyncio.Semaphore 限制并发 API 调用的数量。值为 50 的 Semaphore 表示同一时刻最多有 50 个调用正在进行。请根据您的 OpenAI 层级针对目标模型设置的速率限制调整这个数值。
import asyncio
sem = asyncio.Semaphore(50) # max 50 concurrent calls
async def extract_with_limit(text: str, semaphore: asyncio.Semaphore):
async with semaphore:
return await extract_one(text)
async def batch_extract_limited(texts: list):
tasks = [extract_with_limit(t, sem) for t in texts]
return await asyncio.gather(*tasks, return_exceptions=True)速率限制错误时使用指数退避
即使使用了 Semaphore，在流量突发期间仍可能触发速率限制。请实现指数退避：重试前依次等待 1 秒、2 秒、4 秒和 8 秒。加入抖动（一个很小的随机偏移量），避免所有并发调用方恰好在同一时间重试，从而造成新一轮流量突发。使用 tenacity 库可以轻松实现这一点。
from tenacity import retry, wait_exponential, stop_after_attempt, retry_if_exception_type
from openai import RateLimitError
@retry(
wait=wait_exponential(multiplier=1, min=1, max=60),
stop=stop_after_attempt(5),
retry=retry_if_exception_type(RateLimitError)
)
async def extract_with_retry(text: str):
return await extract_one(text)使用作业队列处理大型批次
对于超过几千个项目的批次，持久化的作业队列比 asyncio.gather 更合适。Redis Queue（RQ）、Celery 或 Dramatiq 等队列可以在重启后保留作业，允许使用多个工作进程进行水平扩展，并让您了解作业状态和失败情况。工作进程从队列中提取作业，然后独立调用 API。
# With Redis Queue (RQ)
from rq import Queue
from redis import Redis
redis_conn = Redis()
q = Queue('extractions', connection=redis_conn)
def enqueue_documents(doc_ids: list):
for doc_id in doc_ids:
q.enqueue(
'workers.extract_document',
doc_id,
job_timeout=120,
result_ttl=3600
)
enqueue_documents(all_doc_ids)使用数据库跟踪进度
长时间运行的批处理作业需要进度跟踪，以便您监控状态、找出卡住的作业，并在失败后恢复。请在数据库中使用状态表，其中包含文档 ID、状态（待处理、处理中、已完成、失败）、时间戳和错误消息等字段。围绕每次提取调用以原子方式更新状态。
import asyncpg
async def process_document(pool, doc_id: str, text: str):
async with pool.acquire() as conn:
await conn.execute(
'UPDATE extractions SET status=$1, started_at=NOW() WHERE doc_id=$2',
'processing', doc_id
)
try:
result = await extract_one(text)
await conn.execute(
'UPDATE extractions SET status=$1, result=$2, completed_at=NOW() WHERE doc_id=$3',
'completed', result.model_dump_json(), doc_id
)
except Exception as e:
await conn.execute(
'UPDATE extractions SET status=$1, error=$2 WHERE doc_id=$3',
'failed', str(e), doc_id
)恢复失败的作业
批处理作业应当能够安全地重新启动。启动时，从数据库中查询状态为 pending 或 failed 的文档并重试。请为每个文档使用幂等键，这样如果同一文档意外入队两次，第二次尝试就能检测到已完成的结果并跳过重新处理。这可以防止向下游系统重复写入。
async def get_pending_docs(pool) -> list:
async with pool.acquire() as conn:
rows = await conn.fetch(
'SELECT doc_id, raw_text FROM extractions WHERE status IN ($1, $2)',
'pending', 'failed'
)
return [dict(row) for row in rows]
async def resume_batch(pool):
docs = await get_pending_docs(pool)
print(f'Resuming {len(docs)} unprocessed documents')
tasks = [process_document(pool, d['doc_id'], d['raw_text']) for d in docs]
await asyncio.gather(*tasks, return_exceptions=True)使用 OpenAI Batch API 批量处理
OpenAI 的Batch API 允许您在单个文件中提交最多 50,000 个请求，并在 24 小时内以五折价格获得结果。对于不紧急且成本比延迟更重要的提取处理流程，这非常理想。您可以上传包含请求的 JSONL 文件，轮询完成状态，然后下载结果文件。
from openai import OpenAI
import json
client = OpenAI()
# Build JSONL batch file
with open('/tmp/batch_requests.jsonl', 'w') as f:
for i, text in enumerate(documents):
request = {
'custom_id': f'doc_{i}',
'method': 'POST',
'url': '/v1/chat/completions',
'body': {
'model': 'gpt-4o-mini',
'messages': [{'role': 'user', 'content': text}]
}
}
f.write(json.dumps(request) + '\n')
# Upload and submit
batch_file = client.files.create(file=open('/tmp/batch_requests.jsonl', 'rb'), purpose='batch')
batch = client.batches.create(input_file_id=batch_file.id, endpoint='/v1/chat/completions', completion_window='24h')
print(batch.id)监控吞吐量和成本
在批处理运行期间跟踪提取吞吐量（每分钟处理的文档数）和每份文档的成本。用 API 总支出除以已处理的文档数，即可得到成本基线。扩展规模时，请留意成本是否线性增长——超线性增长通常说明您在不必要地过长的提示中浪费了令牌。简单的指标仪表板可以帮助您在低效问题不断累积前发现它们。
import time
class BatchMetrics:
def __init__(self):
self.start_time = time.time()
self.processed = 0
self.total_tokens = 0
self.cost = 0.0
def record(self, usage):
self.processed += 1
self.total_tokens += usage.total_tokens
self.cost += usage.prompt_tokens * 0.00000015 + usage.completion_tokens * 0.0000006
def report(self):
elapsed = time.time() - self.start_time
print(f'{self.processed} docs in {elapsed:.1f}s = {self.processed/elapsed:.1f} docs/sec')
print(f'Cost: ${self.cost:.4f} = ${self.cost/self.processed:.6f} per doc')遵守每分钟令牌数的速率限制
OpenAI 的速率限制同时适用于每分钟请求数（RPM）和每分钟令牌数（TPM）。对于 RPM，同时发送 50 个调用没有问题；但如果每个调用使用 2,000 个令牌，50 个调用就相当于每分钟 100,000 个令牌，很容易超过第 1 层级的限制。请使用 tiktoken 在提交前统计预计令牌数，并在并发 Semaphore 的基础上实现令牌预算。
import tiktoken
enc = tiktoken.encoding_for_model('gpt-4o-mini')
def estimate_tokens(text: str) -> int:
return len(enc.encode(text)) + 300 # +300 for schema + response
# TPM_LIMIT = 200_000 # Tier 2 limit
# Only submit a batch if estimated total tokens fits within budget
def fits_in_budget(texts: list, tpm_limit: int = 200_000) -> bool:
total = sum(estimate_tokens(t) for t in texts)
return total <= tpm_limit快速检查
测试您对文档提取异步批处理的理解。
课程回顾
在本课中，您学习了：asyncio 和 Semaphore 可以在遵守速率限制的同时并发调用 API；作业队列和状态表 可以让大型批处理作业支持恢复并便于观察；OpenAI Batch API 能为不紧急的工作负载节省 50% 的成本，但代价是 24 小时的延迟。接下来，我们将处理长时间运行的提取处理流程中的架构演进。
常见问题解答
「使用异步处理和队列进行批处理」课时是免费的吗？
是的 — 「使用异步处理和队列进行批处理」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Engineering Academy 课程的其余内容，请升级到 CoddyKit PRO。 AI Engineering Academy 课程共包含 4 节课。
「使用异步处理和队列进行批处理」这节课中我会学到什么？
使用 asyncio 和作业队列构建异步提取流程，在遵守速率限制并跟踪进度的同时并行处理数千份文档。 你通过在浏览器中直接运行的动手代码来练习 AI Engineering Academy，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Engineering Academy 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Engineering Academy 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。
「使用异步处理和队列进行批处理」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Engineering Academy 课中编写并运行代码吗？
能。每节 AI Engineering Academy 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。
此课程中的所有课时
- Instructor：使用 Pydantic 进行类型化提取
- 处理不完整和缺失数据
- 使用异步处理和队列进行批处理
- 模式演进与向后兼容