일괄 처리 및 비동기 실행
OpenAI Batch API, 비동기 Python, 동시 프롬프트 실행을 알아봅니다.
일괄 처리 및 비동기 실행은(는) CoddyKit의 무료 AI Prompt Engineering 강의입니다. 이것은 4개 중 2번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Prompt Engineering 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Prompt Engineering 강의에는 총 4개의 강의가 포함되어 있습니다.
배치 및 비동기 처리가 필요한 이유
수천 개의 LLM 요청을 순차적으로 처리하면 느리고 비용이 많이 듭니다. 배치 처리는 요청을 묶어 비용을 50% 절감합니다. 비동기 실행은 속도 제한 안에서 처리량을 극대화하도록 요청을 병렬화합니다. 두 방식을 함께 사용하면 비용과 실제 소요 시간을 모두 크게 줄일 수 있습니다.
OpenAI 배치 API: 비용 50% 절감
OpenAI 배치 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}')배치 작업 제출 및 상태 확인
입력 파일을 업로드한 후 배치 작업을 생성하고 완료될 때까지 상태를 확인합니다. 배치 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()를 사용한 파이썬 비동기 처리
실시간(배치가 아닌) 병렬 처리를 위해 파이썬의 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 배치 API
Anthropic도 OpenAI의 배치 API와 비슷한 비용 구조를 가진 메시지 배치 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 배치 API: 비용 50% 절감, 백그라운드 처리, 배치당 최대 50K개 요청
- asyncio.gather(): 실시간 요청을 동시에 처리하며 모든 요청을 동시에 실행하고 모든 결과를 기다립니다
- Semaphore: 속도 제한을 고려한 동시성 제어(일반적으로 10~50개의 동시 요청)
- 지수적 백오프: 속도 제한 또는 시간 초과 오류가 발생하면 대기 시간을 두 배로 늘려 재시도합니다
- 복원력 있는 gather: return_exceptions=True를 사용하면 하나의 실패로 전체 배치가 중단되는 것을 방지합니다
- 진행 상황 추적: 대규모 작업에 대해 처리 속도와 ETA를 포함해 완료 상황을 보고합니다
자주 묻는 질문
“일괄 처리 및 비동기 실행” 강의는 무료인가요?
네 — “일괄 처리 및 비동기 실행” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Prompt Engineering 강의 전체를 잠금 해제할 수 있습니다. AI Prompt Engineering 강의에는 총 4개의 강의가 포함되어 있습니다.
“일괄 처리 및 비동기 실행”에서 뭘 배우나요?
OpenAI Batch API, 비동기 Python, 동시 프롬프트 실행을 알아봅니다. 브라우저에서 직접 실행하는 실습 코드로 AI Prompt Engineering을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.
AI Prompt Engineering을(를) 시작하는 데 경험이 필요한가요?
사전 경험은 필요하지 않습니다. CoddyKit의 AI Prompt Engineering은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 2번째 강의입니다.
“일괄 처리 및 비동기 실행” 강의는 얼마나 걸리나요?
대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.
이 AI Prompt Engineering 강의에서 코드를 작성하고 실행할 수 있나요?
네. 모든 AI Prompt Engineering 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.
이 강의의 모든 강의
- 프롬프트 캐싱 전략
- 일괄 처리 및 비동기 실행
- 모델 간 부하 분산
- 프롬프트 파이프라인 모니터링 및 알림