Przetwarzanie wsadowe z async i kolejkami

Dlaczego przetwarzanie wsadowe ma znaczenie

Podstawy asyncio

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}]
    )

Uruchamianie wielu operacji wydobywania danych za pomocą 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))

Kontrolowanie współbieżności za pomocą semaforów

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)

Wykładnicze zwiększanie opóźnienia po błędach limitu zapytań

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)

Korzystanie z kolejki zadań dla dużych partii

# 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)

Śledzenie postępu za pomocą bazy danych

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
            )

Wznawianie nieudanych zadań

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)

Przetwarzanie wsadowe za pomocą OpenAI Batch API

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)

Monitorowanie przepustowości i kosztów

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')

Przestrzeganie limitów tokenów na minutę

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

Szybkie sprawdzenie

Podsumowanie lekcji