"""Benchmark pareado via OCI: 100 prompts idênticos, sem retry automático. Sem --execute, apenas valida dados e apresenta o plano. Resultados ficam locais. """ from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from decimal import Decimal import argparse import csv import fcntl import hashlib import json import math import os from pathlib import Path import statistics import time from urllib.parse import urlparse import requests ROOT = Path(__file__).resolve().parent MODELS = ['xai.grok-4.20-non-reasoning', 'xai.grok-4.3'] D = lambda value: Decimal(str(value)) SYSTEM = ('Você é um assistente em português brasileiro. Use web_search para verificar cada consulta, ' 'mesmo que conheça o assunto. A data de referência consta na pergunta. ' 'Responda em até 200 palavras, cite URLs e datas das fontes. ' 'Diferencie a data do fato e a data de publicação. Nunca invente preços, agendas, ' 'previsões ou disponibilidade. Informe explicitamente dados não confirmados.') def digest(value): return hashlib.sha256(json.dumps(value, ensure_ascii=False, sort_keys=True).encode()).hexdigest() def load_dataset(): dataset = json.loads((ROOT/'intents.json').read_text()) rows = dataset['intents'] assert len(rows) == 100 and len({r['id'] for r in rows}) == 100 assert len({r['prompt'] for r in rows}) == 100 for row in rows: assert hashlib.sha256(row['prompt'].encode()).hexdigest() == row['prompt_sha256'] assert all(sum(r['category'] == category for r in rows) == 25 for category in ['culinaria', 'esportes', 'clima', 'entretenimento']) return dataset def calculate_cost(usage, model, prices): """Saída já inclui raciocínio; cache é subconjunto da entrada. Sem duplicação.""" try: total_in = usage['input_tokens'] cached = usage['input_tokens_details']['cached_tokens'] output = usage['output_tokens'] calls = usage['server_side_tool_usage_details']['web_search_calls'] if any(not isinstance(v, int) or v < 0 for v in [total_in, cached, output, calls]) or cached > total_in: return {'status': 'unknown', 'reason': 'Invalid usage counters'} # A amostra tem prompts curtos. Bloquear inferência de faixa em contexto longo: # uso agregado de um loop de ferramentas não identifica o tamanho de cada passo. if total_in >= 200000: return {'status': 'unknown', 'reason': 'Long-context band needs per-step verification'} detail = usage['server_side_tool_usage_details'] if any(v for k, v in detail.items() if k != 'web_search_calls'): return {'status': 'unknown', 'reason': 'Other billable tools observed'} rate = prices['models']['grok-4.3' if model.endswith('4.3') else 'grok-4.20'] regular_cost = D(total_in-cached)*D(rate['input_short']['usd'])/D(1000000) cached_cost = D(cached)*D(rate['cached_short']['usd'])/D(1000000) output_cost = D(output)*D(rate['output_short']['usd'])/D(1000000) search_cost = D(calls)*D(prices['web_search']['usd_per_1000'])/D(1000) return {'status': 'computed_from_measured_usage', 'input_usd': str(regular_cost), 'cached_input_usd': str(cached_cost), 'output_usd': str(output_cost), 'web_search_usd': str(search_cost), 'total_usd': str(regular_cost+cached_cost+output_cost+search_cost), 'input_tokens': total_in, 'cached_input_tokens': cached, 'output_tokens': output, 'reasoning_tokens': usage.get('output_tokens_details', {}).get('reasoning_tokens'), 'web_search_calls': calls} except (KeyError, TypeError): return {'status': 'unknown', 'reason': 'Missing billing usage counters'} def append_event(path, event): with path.open('a') as handle: handle.write(json.dumps(event, ensure_ascii=False)+'\n') handle.flush() os.fsync(handle.fileno()) def invoke(model, intent, base, headers, prices, raw_folder): neutral_body = {'input': [{'role': 'user', 'content': intent['prompt']}], 'instructions': SYSTEM, 'tools': [{'type': 'web_search'}], 'store': False, 'max_output_tokens': 2400} body = {'model': model, **neutral_body} result = {'model': model, 'intent_id': intent['id'], 'category': intent['category'], 'prompt_sha256': intent['prompt_sha256'], 'shared_payload_sha256': digest(neutral_body), 'started_at': datetime.now(timezone.utc).isoformat()} started = time.monotonic() try: response = requests.post(base+'/responses', headers=headers, json=body, timeout=(10, 150)) result['http'] = response.status_code try: data = response.json() except ValueError: data = {} (raw_folder/f"{intent['id']}-{model}.json").write_text(json.dumps(data, ensure_ascii=False, indent=2)) output = data.get('output', []) events = [item for item in output if item.get('type') == 'web_search_call'] citations = [a for item in output for content in item.get('content', []) for a in content.get('annotations', []) if a.get('type') == 'url_citation'] sources = [s for item in events for s in item.get('action', {}).get('sources', [])] answer = '\n'.join(content.get('text', '') for item in output for content in item.get('content', [])) searches = [item for item in events if item.get('status') == 'completed' and item.get('action', {}).get('type') == 'search'] result.update(returned_model=data.get('model'), provider_response_id=data.get('id'), usage=data.get('usage'), tool_events=[{'action': e.get('action', {}).get('type'), 'status': e.get('status')} for e in events], citation_count=len(citations), source_count=len(sources), answer_chars=len(answer), response_status=data.get('status'), incomplete_details=data.get('incomplete_details'), websearch_confirmed=response.status_code == 200 and bool(searches) and bool(sources) and bool(citations), cost=calculate_cost(data.get('usage') or {}, model, prices)) if response.status_code != 200: result['failure'] = 'provider_http_error' except requests.RequestException as exc: # Timeout não prova ausência de cobrança. Não reenviar a interação. result.update(http=None, failure=type(exc).__name__, cost={'status': 'unknown', 'reason': 'Transport failure; billing may have occurred'}) result['seconds'] = round(time.monotonic()-started, 3) result['finished_at'] = datetime.now(timezone.utc).isoformat() return result def export_results(folder, rows, reservation): fields = ['intent_id','category','model','http','websearch_confirmed','seconds', 'input_tokens','cached_input_tokens','output_tokens','reasoning_tokens','web_search_calls', 'input_usd','cached_input_usd','output_usd','web_search_usd','total_usd','cost_status', 'citation_count','source_count','prompt_sha256','shared_payload_sha256'] with (folder/'interactions.csv').open('w') as handle: writer = csv.DictWriter(handle, fieldnames=fields, lineterminator="\n") writer.writeheader() for row in rows: flat = {**row, **row.get('cost', {}), 'cost_status': row.get('cost', {}).get('status')} writer.writerow({f: flat.get(f) for f in fields}) summaries = [] for model in MODELS: selected = [r for r in rows if r['model'] == model] known = [r for r in selected if r.get('cost', {}).get('total_usd') is not None] unknown = len(selected)-len(known) values = [D(r['cost']['total_usd']) for r in known] def total(key): return sum(D(r['cost'][key]) for r in known) summaries.append({'model': model, 'interactions': len(selected), 'http_200': sum(r.get('http') == 200 for r in selected), 'websearch_confirmed': sum(r.get('websearch_confirmed', False) for r in selected), 'cost_known_interactions': len(known), 'cost_unknown_interactions': unknown, 'total_usd_known': str(sum(values)), 'mean_usd_known': str(sum(values)/len(values)) if values else None, 'median_usd_known': str(statistics.median(values)) if values else None, 'p95_usd_known': str(sorted(values)[math.ceil(.95*len(values))-1]) if values else None, 'web_search_usd': str(total('web_search_usd')), 'input_tokens': int(total('input_tokens')), 'cached_input_tokens': int(total('cached_input_tokens')), 'output_tokens': int(total('output_tokens')), 'web_search_calls': int(total('web_search_calls')), 'median_seconds': statistics.median(r['seconds'] for r in selected) if selected else None}) paired = {} for row in rows: paired.setdefault(row['intent_id'], []).append(row) complete_pairs = [pair for pair in paired.values() if len(pair) == 2] assert all(pair[0]['shared_payload_sha256'] == pair[1]['shared_payload_sha256'] for pair in complete_pairs) report = {'generated_at': datetime.now(timezone.utc).isoformat(), 'models': summaries, 'completed_interactions': len(rows), 'expected_interactions': 200, 'paired_intents': len(complete_pairs), 'estimated_reserve_for_unknown_usd': str(sum(reservation for r in rows if r.get('cost', {}).get('total_usd') is None)), 'billing_note': 'Custo calculado do uso medido a preços de lista Oracle; não é fatura OCI. Sem impostos, descontos contratuais, VM, rede e armazenamento.'} (folder/'summary.json').write_text(json.dumps(report, ensure_ascii=False, indent=2)+'\n') return report def main(): parser = argparse.ArgumentParser() parser.add_argument('--execute', action='store_true') parser.add_argument('--budget-usd', type=Decimal) parser.add_argument('--run-id', default='comparison-2026-10-15') parser.add_argument('--pair-limit', type=int, default=100, help='Piloto retoma no mesmo run-id, sem repetir chamadas') args = parser.parse_args() assert 1 <= args.pair_limit <= 100 assert args.run_id and all(c.isalnum() or c in '-_' for c in args.run_id) dataset = load_dataset() prices = json.loads((ROOT/'oracle-prices.json').read_text()) print(f'100 intenções; {len(MODELS)} modelos; 200 interações; concorrência 2; sem retries.', flush=True) if not args.execute: print('Dry run: nenhuma chamada feita. Para executar, definir --execute --budget-usd LIMITE.') return assert args.budget_usd and args.budget_usd > 0, 'É necessário orçamento explícito' base = os.environ['OCI_BASE_URL'].rstrip('/') assert base == 'https://inference.generativeai.us-chicago-1.oci.oraclecloud.com/v1' headers = {'Authorization': 'Bearer '+os.environ['OCI_API_KEY'], 'OpenAI-Project': os.environ['OCI_PROJECT_ID']} folder = ROOT/'runs'/args.run_id folder.mkdir(parents=True, exist_ok=True) lockfile = (folder/'run.lock').open('w') fcntl.flock(lockfile, fcntl.LOCK_EX | fcntl.LOCK_NB) raw_folder = folder/'raw' raw_folder.mkdir(exist_ok=True) manifest = {'dataset_sha256': digest(dataset), 'prices_sha256': digest(prices), 'system_sha256': hashlib.sha256(SYSTEM.encode()).hexdigest(), 'models': MODELS, 'reference_date': dataset['reference_date'], 'region': 'us-chicago-1', 'concurrency': 2, 'max_output_tokens': 2400, 'retries': 0, 'transport_timeout_seconds': 150} manifest_file = folder/'manifest.json' if manifest_file.exists(): assert json.loads(manifest_file.read_text()) == manifest, 'Manifesto alterado; use outro run-id' else: manifest_file.write_text(json.dumps(manifest, ensure_ascii=False, indent=2)+'\n') ledger = folder/'events.jsonl' events = [json.loads(line) for line in ledger.read_text().splitlines()] if ledger.exists() else [] rows = [e['result'] for e in events if e['event'] == 'finished'] started = {(e['intent_id'],e['model']) for e in events if e['event'] == 'started'} finished = {(r['intent_id'],r['model']) for r in rows} assert started == finished, 'Há chamadas interrompidas sem resultado; revisar cobrança antes de continuar' assert len(finished) == len(rows), 'Interações duplicadas no registro' reservation = D('.25') def committed(): return sum(D(r['cost']['total_usd']) if r.get('cost', {}).get('total_usd') is not None else reservation for r in rows) # Um piloto de cada categoria; depois alternância de categorias ao longo do lote. order = [i+25*c for i in range(25) for c in range(4)] with ThreadPoolExecutor(max_workers=2) as pool: for index in order[:args.pair_limit]: intent = dataset['intents'][index] pending = [m for m in MODELS if (intent['id'],m) not in finished] if not pending: continue if committed()+reservation*len(pending) > args.budget_usd: print('STOP orçamento: sem saldo para reservar o próximo par.', flush=True) break jobs=[] for model in pending: append_event(ledger, {'event': 'started', 'intent_id': intent['id'], 'model': model, 'at': datetime.now(timezone.utc).isoformat(), 'reserved_usd': str(reservation)}) jobs.append(pool.submit(invoke, model, intent, base, headers, prices, raw_folder)) stop = False for job in jobs: result = job.result() append_event(ledger, {'event':'finished','result':result}) rows.append(result) finished.add((result['intent_id'],result['model'])) print(f"{len(rows)}/200 {result['intent_id']} {result['model']} HTTP={result.get('http')} busca={result.get('websearch_confirmed')} USD={result.get('cost',{}).get('total_usd','unknown')}", flush=True) # Falha exige diagnóstico; timeout não será repetido e custo não vira zero. stop |= result.get('http') != 200 or result.get('cost',{}).get('total_usd') is None export_results(folder, rows, reservation) if stop: print('STOP diagnóstico necessário; nenhuma nova interação será enviada.', flush=True) break report = export_results(folder, rows, reservation) print(json.dumps(report, ensure_ascii=False, indent=2), flush=True) if __name__ == '__main__': main()