バッチ処理と非同期実行
OpenAI Batch API、非同期Python、プロンプトの並行実行を学びます。
「バッチ処理と非同期実行」はCoddyKit上の無料AI Prompt Engineeringレッスンです。 これはレッスン2/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Prompt Engineering学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Prompt Engineeringコースには全4レッスンが含まれています。
バッチ処理と非同期処理を使う理由
数千件のLLMリクエストを順番に処理すると、時間とコストがかかります。バッチ処理はリクエストをまとめ、コストを50%削減します。非同期実行はレート制限の範囲内でリクエストを並列化し、スループットを最大化します。これらを組み合わせることで、コストと実時間の両方を大幅に削減できます。
OpenAI Batch API：コストを50%削減
OpenAI Batch APIは、リクエストをバックグラウンドで非同期に（最大24時間）処理し、通常のAPI料金の50%で利用できます。評価の実行、データセットの処理、リアルタイム性が不要なワークロードに適しています。
import openai
import json
client = openai.OpenAI(api_key='YOUR_API_KEY')
# Step 1: Create batch input file (JSONL format)
batch_requests = [
{
'custom_id': f'request-{i}',
'method': 'POST',
'url': '/v1/chat/completions',
'body': {
'model': 'gpt-4o-mini',
'messages': [
{'role': 'user', 'content': f'Summarize this document: {doc}'}
],
'max_tokens': 200
}
}
for i, doc in enumerate(['Doc A text...', 'Doc B text...', 'Doc C text...'])
]
# Write to JSONL file
with open('batch_input.jsonl', 'w') as f:
for req in batch_requests:
f.write(json.dumps(req) + '\n')
# Step 2: Upload the file
batch_file = client.files.create(
file=open('batch_input.jsonl', 'rb'),
purpose='batch'
)
print(f'Batch file uploaded: {batch_file.id}')バッチジョブの送信とポーリング
入力ファイルをアップロードしたら、バッチジョブを作成し、完了するまでポーリングします。Batch APIは24時間以内にリクエストを処理します（通常、小規模なバッチならさらに短時間で完了します）。
import time
# Step 3: Create batch job
batch = client.batches.create(
input_file_id=batch_file.id,
endpoint='/v1/chat/completions',
completion_window='24h'
)
print(f'Batch created: {batch.id} | Status: {batch.status}')
# Step 4: Poll for completion
def wait_for_batch(batch_id, poll_interval=30, timeout=3600):
start = time.time()
while time.time() - start < timeout:
batch = client.batches.retrieve(batch_id)
print(f'Status: {batch.status} | '
f'Completed: {batch.request_counts.completed}/ '
f'{batch.request_counts.total}')
if batch.status == 'completed':
return batch
if batch.status in ('failed', 'expired', 'cancelling', 'cancelled'):
raise RuntimeError(f'Batch {batch_id} ended with status: {batch.status}')
time.sleep(poll_interval)
raise TimeoutError('Batch polling timed out')
# batch = wait_for_batch(batch.id)バッチ結果の取得
バッチが完了したら、出力ファイルをダウンロードし、JSONLの結果を解析して利用可能な形式に戻します。
def retrieve_batch_results(batch):
if not batch.output_file_id:
raise ValueError('No output file — batch may have failed')
# Download output file
content = client.files.content(batch.output_file_id).text
# Parse JSONL: one result per line
results = {}
for line in content.strip().split('\n'):
if not line:
continue
result = json.loads(line)
custom_id = result['custom_id']
if result.get('error'):
results[custom_id] = {'error': result['error']}
else:
response_body = result['response']['body']
text = response_body['choices'][0]['message']['content']
results[custom_id] = {'text': text}
# Report error rate
errors = sum(1 for r in results.values() if 'error' in r)
print(f'Retrieved {len(results)} results, {errors} errors')
return results
# results = retrieve_batch_results(batch)
# for req_id, result in results.items():
# print(req_id, result.get('text', result.get('error', ''))[:50])asyncio.gather()を使った非同期Python
リアルタイム（非バッチ）の並列処理では、Pythonのasyncioとasyncio.gather()を使うと、複数のAPI呼び出しを同時に開始し、すべての完了を待機できます。これにより、複数リクエストのワークロード全体にかかる実時間を大幅に短縮できます。
import asyncio
import openai
async_client = openai.AsyncOpenAI(api_key='YOUR_API_KEY')
async def async_completion(messages, model='gpt-4o-mini', max_tokens=200):
response = await async_client.chat.completions.create(
model=model,
messages=messages,
max_tokens=max_tokens
)
return response.choices[0].message.content
async def process_parallel(prompts):
tasks = [
async_completion([{'role': 'user', 'content': p}])
for p in prompts
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return results
# Usage
async def main():
prompts = ['Explain photosynthesis.', 'Explain gravity.', 'Explain evolution.']
results = await process_parallel(prompts)
for prompt, result in zip(prompts, results):
if isinstance(result, Exception):
print(f'ERROR: {result}')
else:
print(f'{prompt[:30]}... -> {result[:60]}...')
# asyncio.run(main())
print('asyncio.gather: all 3 requests fire simultaneously')レート制限を考慮した同時呼び出し
同時に送信するリクエストが多すぎると、レート制限エラーが発生します。セマフォで同時実行数を制限すれば、スループットを最大化しながらレート制限の範囲内に収められます。
import asyncio
# Rate limits (example for gpt-4o-mini):
# RPM (requests per minute): 500
# TPM (tokens per minute): 200,000
MAX_CONCURRENT = 20 # stay well below rate limit
async def process_with_rate_limit(prompts, max_concurrent=MAX_CONCURRENT):
semaphore = asyncio.Semaphore(max_concurrent)
results = [None] * len(prompts)
async def bounded_completion(i, prompt):
async with semaphore:
try:
result = await async_completion(
[{'role': 'user', 'content': prompt}]
)
results[i] = result
except openai.RateLimitError as e:
print(f'Rate limited on prompt {i}: {e}')
await asyncio.sleep(60) # back off and retry
result = await async_completion(
[{'role': 'user', 'content': prompt}]
)
results[i] = result
await asyncio.gather(*[
bounded_completion(i, p) for i, p in enumerate(prompts)
])
return results
print('Semaphore limits to', MAX_CONCURRENT, 'concurrent requests')Anthropic Batch API
Anthropicも、OpenAIのBatch APIと同様の料金体系でMessage Batches APIを提供しています。バッチは非同期に処理され、結果はポーリングまたはストリーミングで取得します。
import anthropic
client = anthropic.Anthropic(api_key='YOUR_API_KEY')
# Create a batch of messages
batch = client.messages.batches.create(
requests=[
{
'custom_id': f'doc-{i}',
'params': {
'model': 'claude-haiku-4-5',
'max_tokens': 200,
'messages': [{
'role': 'user',
'content': f'Classify the sentiment of: {text}'
}]
}
}
for i, text in enumerate([
'Amazing product, exceeded expectations!',
'Terrible quality, broke after one use.',
'It works as described.'
])
]
)
print(f'Batch created: {batch.id} | Status: {batch.processing_status}')
# Poll for completion
# while (batch := client.messages.batches.retrieve(batch.id)).processing_status != 'ended':
# time.sleep(30)
# Retrieve results
# for result in client.messages.batches.results(batch.id):
# print(result.custom_id, result.result.message.content[0].text[:50])スループットの最適化：バッチ戦略
レイテンシ要件とワークロードの特性に基づいて適切なバッチ戦略を選択し、スループットを最大化します。
throughput_strategies = {
'API Batch (OpenAI/Anthropic)': {
'cost': '50% of normal price',
'latency': 'Minutes to hours (background processing)',
'best_for': 'Offline workloads: eval runs, dataset labeling, report generation',
'max_batch_size': '50,000 requests per batch'
},
'asyncio.gather()': {
'cost': 'Normal price',
'latency': 'Same as slowest individual request',
'best_for': 'Real-time parallel enrichment, multi-step pipelines',
'max_concurrent': '10-50 depending on rate limits'
},
'Streaming + Concurrent': {
'cost': 'Normal price',
'latency': 'First token arrives faster, total similar',
'best_for': 'User-facing applications needing perceived speed',
'pattern': 'asyncio with stream=True per request'
},
'Worker Queue (Celery, RQ)': {
'cost': 'Normal price',
'latency': 'Variable (depends on queue depth)',
'best_for': 'High-volume production with auto-scaling workers',
'backends': 'Redis, RabbitMQ'
}
}
for strategy, details in throughput_strategies.items():
print(f'{strategy}: {details["best_for"][:60]}')バッチ／非同期処理のエラーハンドリングと再試行ロジック
同時実行やバッチのワークロードには、堅牢なエラーハンドリングが必要です。個々のリクエストの失敗でバッチ全体を停止させず、失敗をログに記録し、バックオフを挟んで再試行し、全体の成功率を報告します。
import asyncio
import random
async def resilient_completion(prompt, max_retries=3, base_delay=1.0):
for attempt in range(max_retries):
try:
return await async_completion(
[{'role': 'user', 'content': prompt}]
)
except openai.RateLimitError:
wait = base_delay * (2 ** attempt) + random.uniform(0, 1)
print(f'Rate limited. Waiting {wait:.1f}s (attempt {attempt+1})')
await asyncio.sleep(wait)
except openai.APITimeoutError:
print(f'Timeout on attempt {attempt+1}')
await asyncio.sleep(base_delay)
except openai.APIError as e:
if e.status_code >= 500:
await asyncio.sleep(base_delay * (attempt + 1))
else:
raise # Don't retry 4xx errors
raise RuntimeError(f'Failed after {max_retries} attempts')
async def batch_with_error_reporting(prompts):
tasks = [resilient_completion(p) for p in prompts]
results = await asyncio.gather(*tasks, return_exceptions=True)
successes = sum(1 for r in results if not isinstance(r, Exception))
print(f'Batch complete: {successes}/{len(prompts)} succeeded')
return resultsバッチ処理に向けた大きな入力の分割
モデルのコンテキストウィンドウを超えるドキュメントは、バッチ処理の前に分割する必要があります。各チャンクが個別のバッチリクエストになり、後で結果を結合または要約します。
def chunk_document(text, max_tokens=3000, overlap_tokens=200):
'''
Split a long document into overlapping chunks for batch processing.
Approximate: 1 token ~ 4 characters
'''
max_chars = max_tokens * 4
overlap_chars = overlap_tokens * 4
chunks = []
start = 0
while start < len(text):
end = min(start + max_chars, len(text))
# Try to break at a sentence boundary
if end < len(text):
last_period = text.rfind('.', start, end)
if last_period > start + max_chars // 2:
end = last_period + 1
chunks.append({'text': text[start:end], 'start': start, 'end': end})
start = end - overlap_chars # overlap for context continuity
return chunks
def batch_summarize_long_document(document_text, summary_prompt):
chunks = chunk_document(document_text)
print(f'Document split into {len(chunks)} chunks')
# Create one batch request per chunk
batch_inputs = [
{'custom_id': f'chunk-{i}',
'content': summary_prompt + '\n\n' + chunk['text']}
for i, chunk in enumerate(chunks)
]
# Submit all chunks as one batch job
return batch_inputs
long_doc = 'Lorem ipsum ' * 5000 # ~20K character document
chunks = chunk_document(long_doc)
print(f'Chunks: {len(chunks)}, first chunk length: {len(chunks[0]["text"])} chars')大規模バッチの進捗追跡
大規模なバッチジョブ（数千件のリクエスト）では、オペレーターがスループットを監視し、完了時間を見積もれるように、リアルタイムの進捗を表示します。
import asyncio
import time
async def batch_with_progress(prompts, max_concurrent=20):
semaphore = asyncio.Semaphore(max_concurrent)
completed = 0
total = len(prompts)
start_time = time.time()
results = [None] * total
async def process_one(i, prompt):
nonlocal completed
async with semaphore:
results[i] = await resilient_completion(prompt)
completed += 1
elapsed = time.time() - start_time
rate = completed / elapsed if elapsed > 0 else 0
eta = (total - completed) / rate if rate > 0 else float('inf')
if completed % 10 == 0 or completed == total:
print(f'Progress: {completed}/{total} '
f'({completed/total:.0%}) | '
f'{rate:.1f} req/s | '
f'ETA: {eta:.0f}s')
await asyncio.gather(*[
process_one(i, p) for i, p in enumerate(prompts)
])
return results
print('Progress tracking: reports every 10 completions with ETA.')理解度チェック
一晩で10,000件のドキュメントに感情分析のラベルを付ける必要があります。コストを最小限に抑えたい一方で、リアルタイムの結果は必要ありません。最適な方法はどれですか。
バッチ処理と非同期処理のまとめ
バッチ処理と非同期実行は、大規模なプロンプトエンジニアリングに不可欠です：
- OpenAI/Anthropic Batch API：コストを50%削減し、バックグラウンドで処理できます。1バッチあたり最大50Kリクエストに対応します
- asyncio.gather()：リアルタイムリクエストを同時実行し、すべてを一斉に開始して全結果を待機します
- セマフォ：レート制限を考慮した同時実行制御です（通常は10～50件の同時リクエスト）
- 指数バックオフ：レート制限エラーやタイムアウトエラーに対し、待機時間を倍増させながら再試行します
- 堅牢なgather：return_exceptions=Trueにより、1件の失敗でバッチ全体が停止するのを防ぎます
- 進捗追跡：大規模ジョブについて、完了数、処理速度、ETAを報告します
よくある質問
「バッチ処理と非同期実行」レッスンは無料ですか？
はい。「バッチ処理と非同期実行」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Prompt Engineeringコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Prompt Engineeringコースには全4レッスンが含まれています。
「バッチ処理と非同期実行」で何を学びますか？
OpenAI Batch API、非同期Python、プロンプトの並行実行を学びます。 ブラウザで直接実行するハンズオンコードでAI Prompt Engineeringを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Prompt Engineeringを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Prompt Engineeringは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン2/4です。
「バッチ処理と非同期実行」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Prompt Engineeringレッスンでコードを書いて実行できますか？
はい。すべてのAI Prompt Engineeringレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- プロンプトのキャッシュ戦略
- バッチ処理と非同期実行
- モデル間の負荷分散
- プロンプトパイプラインの監視とアラート