Lotes, roteamento de modelos e painéis de custos

Mais três estratégias para otimização de custos

API Batch da OpenAI: 50% de desconto para cargas de trabalho assíncronas

import json
from openai import OpenAI

client = OpenAI()

# Prepare batch file
requests = [
    {
        'custom_id': f'req_{i}',
        'method': 'POST',
        'url': '/v1/chat/completions',
        'body': {
            'model': 'gpt-4o-mini',
            'messages': [
                {'role': 'user', 'content': f'Summarize: {document}'}
            ],
            'max_tokens': 150,
        }
    }
    for i, document in enumerate(documents_to_process)
]

# Write JSONL batch file
with open('/tmp/batch_requests.jsonl', 'w') as f:
    for req in requests:
        f.write(json.dumps(req) + '\n')

# Upload and submit batch
with open('/tmp/batch_requests.jsonl', 'rb') as f:
    batch_file = client.files.create(file=f, purpose='batch')

batch = client.batches.create(
    input_file_id=batch_file.id,
    endpoint='/v1/chat/completions',
    completion_window='24h',
)
print(f'Batch {batch.id} submitted, status: {batch.status}')

Consultando os resultados do lote

import time

def wait_for_batch(batch_id: str, poll_interval: int = 60) -> str:
    while True:
        batch = client.batches.retrieve(batch_id)
        print(f'Status: {batch.status}, completed: {batch.request_counts.completed}')

        if batch.status == 'completed':
            return batch.output_file_id
        elif batch.status == 'failed':
            raise RuntimeError(f'Batch failed: {batch.errors}')

        time.sleep(poll_interval)

def download_batch_results(output_file_id: str) -> list[dict]:
    content = client.files.content(output_file_id)
    results = []
    for line in content.text.strip().split('\n'):
        results.append(json.loads(line))
    return results

output_file_id = wait_for_batch(batch.id)
results = download_batch_results(output_file_id)
for result in results[:3]:
    print(result['custom_id'], result.get('response', {}).get('body', {}).get('choices', [{}])[0])

Roteamento de modelos: associe a complexidade ao tamanho do modelo

CHEAP_MODEL = 'gpt-4o-mini'
POWERFUL_MODEL = 'gpt-4o'

def classify_query_complexity(query: str) -> str:
    # Heuristic-based routing (replace with ML classifier for production)
    words = query.split()
    has_code = any(c in query for c in ['```', 'def ', 'class ', 'SELECT ', 'function '])
    is_multi_step = any(w in query.lower() for w in ['compare', 'analyze', 'explain why', 'evaluate'])
    is_long = len(words) > 50

    if has_code or is_multi_step or is_long:
        return POWERFUL_MODEL
    return CHEAP_MODEL

def routed_completion(messages: list[dict]) -> str:
    user_query = messages[-1].get('content', '')
    model = classify_query_complexity(user_query)
    print(f'Routing to: {model}')
    response = client.chat.completions.create(model=model, messages=messages)
    return response.choices[0].message.content

Roteamento baseado em LLM para maior precisão

CLASSIFIER_SYSTEM = '''You are a query complexity classifier.
Classify the user query as SIMPLE or COMPLEX.
SIMPLE: factual lookup, extraction, classification with clear answer.
COMPLEX: multi-step reasoning, synthesis, comparison, code generation, long-form writing.
Reply with just SIMPLE or COMPLEX.'''

def llm_classify_complexity(query: str) -> str:
    response = client.chat.completions.create(
        model='gpt-4o-mini',  # use cheap model for routing
        messages=[
            {'role': 'system', 'content': CLASSIFIER_SYSTEM},
            {'role': 'user', 'content': query},
        ],
        max_tokens=10,
        temperature=0,
    )
    label = response.choices[0].message.content.strip()
    return POWERFUL_MODEL if label == 'COMPLEX' else CHEAP_MODEL

Acompanhamento do custo por funcionalidade

from collections import defaultdict

cost_tracker = defaultdict(lambda: {'prompt_tokens': 0, 'completion_tokens': 0, 'cost_usd': 0.0})

MODEL_PRICING = {
    'gpt-4o-mini': {'input': 0.15 / 1e6, 'output': 0.60 / 1e6},
    'gpt-4o':      {'input': 2.50 / 1e6, 'output': 10.00 / 1e6},
}

def tracked_completion(feature: str, messages: list[dict], model: str = 'gpt-4o-mini') -> str:
    response = client.chat.completions.create(model=model, messages=messages)
    usage = response.usage
    pricing = MODEL_PRICING.get(model, {'input': 0, 'output': 0})
    cost = usage.prompt_tokens * pricing['input'] + usage.completion_tokens * pricing['output']

    cost_tracker[feature]['prompt_tokens'] += usage.prompt_tokens
    cost_tracker[feature]['completion_tokens'] += usage.completion_tokens
    cost_tracker[feature]['cost_usd'] += cost

    return response.choices[0].message.content

def print_cost_report():
    print(f'{"Feature":<30} {"Prompt":<10} {"Completion":<12} {"Cost USD":<12}')
    for feature, stats in sorted(cost_tracker.items(), key=lambda x: -x[1]['cost_usd']):
        print(f'{feature:<30} {stats["prompt_tokens"]:<10} {stats["completion_tokens"]:<12} ${stats["cost_usd"]:.4f}')

Criação de um painel de custos simples

from fastapi import FastAPI
import datetime

app = FastAPI()

async def record_cost(feature: str, model: str, prompt_tokens: int, completion_tokens: int):
    pricing = MODEL_PRICING.get(model, {'input': 0, 'output': 0})
    cost = prompt_tokens * pricing['input'] + completion_tokens * pricing['output']
    today = datetime.date.today().isoformat()
    key = f'cost:{today}:{feature}:{model}'
    await async_r.incrbyfloat(key, cost)
    await async_r.expire(key, 86400 * 30)  # keep 30 days

@app.get('/dashboard/costs')
async def cost_dashboard():
    today = datetime.date.today().isoformat()
    pattern = f'cost:{today}:*'
    costs = {}
    async for key in async_r.scan_iter(match=pattern):
        value = await async_r.get(key)
        parts = key.split(':')
        feature_model = ':'.join(parts[2:])
        costs[feature_model] = float(value or 0)
    return {'date': today, 'costs': costs, 'total': sum(costs.values())}

Alertas de orçamento mensal

import datetime
import httpx

SLACK_WEBHOOK = 'https://hooks.slack.com/services/YOUR/WEBHOOK'
MONTHLY_BUDGET_USD = 500.0

async def check_budget_alert():
    today = datetime.date.today()
    days_in_month = 30
    day_of_month = today.day
    days_remaining = days_in_month - day_of_month

    # Sum today's costs
    today_total = sum(cost_tracker[f]['cost_usd'] for f in cost_tracker)
    avg_daily = today_total  # simplified: just today's spend
    projected_month = avg_daily * days_in_month

    if projected_month > MONTHLY_BUDGET_USD:
        message = (
            f'LLM Budget Alert: Projected monthly spend ${projected_month:.2f} '
            f'exceeds budget ${MONTHLY_BUDGET_USD:.2f}. '
            f'Today spend: ${today_total:.2f}'
        )
        async with httpx.AsyncClient() as client:
            await client.post(SLACK_WEBHOOK, json={'text': message})

Fila de solicitações para gerenciar limites de taxa

import asyncio
from asyncio import Queue

class RateLimitedLLMClient:
    def __init__(self, requests_per_minute: int = 500):
        self.rpm = requests_per_minute
        self.queue: Queue = Queue(maxsize=1000)
        self.interval = 60.0 / requests_per_minute

    async def start(self):
        asyncio.create_task(self._worker())

    async def _worker(self):
        while True:
            request_fn, future = await self.queue.get()
            try:
                result = await request_fn()
                future.set_result(result)
            except Exception as e:
                future.set_exception(e)
            await asyncio.sleep(self.interval)

    async def submit(self, request_fn) -> str:
        loop = asyncio.get_event_loop()
        future = loop.create_future()
        await self.queue.put((request_fn, future))
        return await future

Reunindo tudo: conjunto de otimização de custos

# Decision framework for cost optimization priority:
#
# 1. Enable prefix caching (free, zero effort, automatic)
# 2. Add exact caching (high hit rate for FAQ/support bots)
# 3. Add model routing (simple heuristics first, ML classifier later)
# 4. Add semantic caching (complex, high ROI for paraphrase-heavy use cases)
# 5. Enable batch API (only for non-real-time pipelines)
# 6. Build cost dashboard (essential for ongoing monitoring)
#
# Typical combined result in a customer support bot:
# Before: $1,000/month
# After step 1-2: $400/month (-60%)
# After step 3-4: $200/month (-50% of remaining)
# After step 5-6: $150/month and visible

Fallback em cascata quando um modelo barato falha

async def routing_with_fallback(messages: list[dict], min_length: int = 50) -> str:
    # Try cheap model first
    cheap_response = await async_client.chat.completions.create(
        model=CHEAP_MODEL, messages=messages, temperature=0.0
    )
    answer = cheap_response.choices[0].message.content

    # Quality check: response too short indicates poor answer
    if len(answer.strip()) < min_length:
        print(f'Cheap model answer too short ({len(answer)} chars), escalating...')
        powerful_response = await async_client.chat.completions.create(
            model=POWERFUL_MODEL, messages=messages, temperature=0.0
        )
        return powerful_response.choices[0].message.content

    return answer

Verificação rápida

Recapitulação da lição