"""Orquestrador do benchmark cego. Uso: python3 run.py # ensaio completo em modo mock (sem chaves) python3 run.py --live # rodada ao vivo (chaves em variáveis de ambiente) python3 run.py --limit 50 # apenas os N primeiros itens por participante python3 run.py --only-domain juridica """ from __future__ import annotations import argparse import json import os import re import sys import time sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import bank import blind import judge import report import stats from mock import MockJudge, MockProvider from providers import OpenAICompatProvider, SISTEMA RAIZ = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) def _letra(texto: str) -> str | None: t = (texto or "").strip() m = re.search(r"\b([A-Da-d])\s*[\)\.\:、）]", t) if m: return m.group(1).upper() m = re.search(r"(?:^|\s)([A-Da-d])(?:\s|$|[\)\.、，,])", t[:80]) if m: return m.group(1).upper() return None def _numero(texto: str) -> float | None: t = (texto or "").replace(",", ".") m = re.search(r"-?\d+(?:\.\d+)?", t) if not m: return None try: return float(m.group(0)) except ValueError: return None def construir_provedor(cfg: dict, aplicacao: dict, mock: bool, perfil_juiz: str = ""): if mock: c = dict(cfg) if perfil_juiz: return MockJudge(c) return MockProvider(c) return OpenAICompatProvider(cfg, aplicacao) def principal() -> int: ap = argparse.ArgumentParser(description="Benchmark cego multidomínio multilíngue") ap.add_argument("--config", default=os.path.join(RAIZ, "harness", "config.json")) ap.add_argument("--out", default=None) ap.add_argument("--seed", type=int, default=None) ap.add_argument("--limit", type=int, default=0, help="limita itens por participante (ensaio)") ap.add_argument("--only-domain", default=None, choices=bank.DOMINIOS) ap.add_argument("--live", action="store_true", help="força modo ao vivo") ap.add_argument("--mock", action="store_true", help="força modo mock") ap.add_argument("--sem-calibracao", action="store_true") args = ap.parse_args() with open(args.config, "r", encoding="utf-8") as fh: cfg = json.load(fh) seed = args.seed if args.seed is not None else cfg["seed"] aplicacao = cfg["aplicacao"] itens = bank.load_bank(os.path.join(RAIZ, cfg["banco"]["dir"])) problemas = bank.validate(itens) if problemas: print("BANCO INVÁLIDO — corrida abortada:") for p in problemas[:40]: print(" -", p) return 2 print(f"[ok] banco válido: {len(itens)} itens") calib, prova = bank.split_calibracao(itens, cfg["banco"]["calibracao_frac"], seed) alvo = calib + prova if args.sem_calibracao else prova if args.only_domain: alvo = [i for i in alvo if i["domain"] == args.only_domain] if args.limit: alvo = alvo[: args.limit] alvo = blind.ordem_apresentacao(alvo, seed) bank.injetar_canarios(alvo, cfg["rodada"]) participantes = cfg["participantes"] juizes_cfg = [j for j in cfg["jurados"] if j["perfil"] in cfg["juizes"]["perfis"]] adj_cfg = next((j for j in cfg["jurados"] if j["perfil"] == cfg["juizes"]["adjudicador"]), juizes_cfg[0]) limite = cfg["juizes"]["limite_divergencia"] modo_mock = args.mock or not args.live if args.live: sem_chave = [p["nome"] for p in participantes if not os.environ.get(p.get("key_env", ""), "")] if sem_chave: print(f"[aviso] sem chave de API para: {', '.join(sem_chave)}") print(" definina as variáveis de ambiente ou rode sem --live.") print(" prosseguindo em MOCK para que o pipeline possa ser ensaiado hoje.") modo_mock = True modo_rotulo = "mock (fallback: chaves ausentes)" else: modo_mock = False modo_rotulo = "ao vivo" else: modo_rotulo = "mock" provedores = {p["id"]: construir_provedor(p, aplicacao, modo_mock) for p in participantes} juizes = [construir_provedor(j, aplicacao, modo_mock, perfil_juiz=j["perfil"]) for j in juizes_cfg] adjudicador = construir_provedor(adj_cfg, aplicacao, modo_mock, perfil_juiz=adj_cfg["perfil"]) rotulos = blind.atribuir_rotulos([p["nome"] for p in participantes], seed) print(f"[ok] modo={'mock' if modo_mock else 'ao vivo'} · participantes={len(participantes)} · " f"itens por participante={len(alvo)} · juízes={len(juizes)}") detalhe: list[dict] = [] inicio = time.perf_counter() n_chamadas = 0 for pid, pcfg in zip([p["id"] for p in participantes], participantes): prov = provedores[pid] for item in alvo: lang = item["lang"] try: resp = prov.completa(SISTEMA[lang], item["q"], lang, item=item) except Exception as exc: # noqa: BLE001 print(f"[erro] {pcfg['nome']} / {item['id']}: {exc}") resp = {"text": "", "latency_ms": 0.0, "prompt_tokens": 0, "completion_tokens": 0, "refused": False, "raw_model": pcfg.get("model")} n_chamadas += 1 texto_cego = blind.anonimizar(resp["text"]) pontos, nota, extra = 0.0, None, {} if item["type"] == "mcq": letra = _letra(resp["text"]) acertou = letra is not None and letra == "ABCD"[item["a"]] pontos = item["pts"] if acertou else 0.0 nota = 1.0 if acertou else 0.0 extra = {"resposta_letra": letra} elif item["type"] == "calc": val = _numero(resp["text"]) ok = val is not None and abs(val - float(item["val"])) <= float(item["tol"]) pontos = item["pts"] if ok else 0.0 nota = 1.0 if ok else 0.0 extra = {"resposta_valor": val} else: if resp["refused"]: pontos, nota = 0.0, 0.0 extra = {"recusa": True} else: res = judge.pontuar_item(item, texto_cego, lang, juizes, adjudicador, limite) nota = stats.media(res["medias"]) pontos = round(nota, 3) extra = {"adjudicado": res["adjudicado"], "dispersao": res["dispersao"]} detalhe.append({ "id": item["id"], "domain": item["domain"], "lang": lang, "type": item["type"], "diff": item["diff"], "pts": item["pts"], "sistema": rotulos[pcfg["nome"]], "participante": pcfg["nome"], "participante_id": pid, "pontos": pontos, "nota": nota, "latencia_ms": resp["latency_ms"], "refused": bool(resp["refused"]), "canary": item.get("canary", ""), "model": resp.get("raw_model", pcfg.get("model")), "resposta": texto_cego, **extra, }) print(f"[ok] {pcfg['nome']}: {len(alvo)} itens aplicados") duracao = time.perf_counter() - inicio # ---------------- agregação ---------------- linhas = [] for p in participantes: sub = [d for d in detalhe if d["participante_id"] == p["id"]] if not sub: continue def recorte(chave: str) -> float: sel = [d for d in sub if chave(d)] poss = sum(d["pts"] for d in sel) if not poss: return 0.0 return 100.0 * sum(d["pontos"] for d in sel) / poss dominios = {dom: recorte(lambda d, dom=dom: d["domain"] == dom) for dom in bank.DOMINIOS} idiomas = {lg: recorte(lambda d, lg=lg: d["lang"] == lg) for lg in bank.IDIOMAS} global_nota = recorte(lambda d: True) notas = [d["pontos"] / d["pts"] * 100 for d in sub] ic = stats.ic_bootstrap(notas, seed=seed) lat = sorted(d["latencia_ms"] for d in sub) linhas.append({ "pos": 0, "sistema": rotulos[p["nome"]], "participante": p["nome"], "participante_id": p["id"], "nota_global": global_nota, "ic": ic, "n_itens": len(sub), "dominios": dominios, "idiomas": idiomas, "elo": 0.0, "latencia_p50": lat[len(lat) // 2] if lat else 0.0, "custo_relativo": 1.0, "recusa_pct": 100.0 * sum(1 for d in sub if d["refused"]) / len(sub), "adjudicados": sum(1 for d in sub if d.get("adjudicado")), "model": sub[0].get("model"), }) linhas.sort(key=lambda r: -r["nota_global"]) for i, r in enumerate(linhas, 1): r["pos"] = i # confronto direto (por item, vence quem pontuou mais) por_item: dict[tuple, dict[str, float]] = {} for d in detalhe: por_item.setdefault((d["domain"], d["lang"], d["id"]), {})[d["participante_id"]] = d["pontos"] vitorias: dict[tuple[str, str], int] = {} for _, placar in por_item.items(): ids = list(placar) for i in range(len(ids)): for j in range(i + 1, len(ids)): a, b = ids[i], ids[j] if abs(placar[a] - placar[b]) < 1e-9: continue vencedor = a if placar[a] > placar[b] else b perdedor = b if vencedor == a else a vitorias[(vencedor, perdedor)] = vitorias.get((vencedor, perdedor), 0) + 1 forca = stats.bradley_terry(vitorias) elos = stats.elo_de_forca(forca) for r in linhas: r["elo"] = elos.get(r["participante_id"], 1500.0) # itens mais difíceis med_por_item: dict[str, list[float]] = {} for d in detalhe: med_por_item.setdefault(d["id"], []).append(d["pontos"] / d["pts"]) info_item = {i["id"]: i for i in alvo} dificeis = sorted( ({"id": k, "domain": info_item[k]["domain"], "lang": info_item[k]["lang"], "type": info_item[k]["type"], "media": stats.media(v)} for k, v in med_por_item.items()), key=lambda x: x["media"], )[:15] abertos = [d for d in detalhe if d["type"] == "open"] sobrepos = [] canario_reproduzido = 0 for d in detalhe: enunciado = info_item[d["id"]]["q"] resposta_txt = str(d.get("resposta", "")) sobrepos.append(stats.sobreposicao_ngramas(enunciado, resposta_txt)) if d.get("canary") and d["canary"] in resposta_txt: canario_reproduzido += 1 ngram_media = stats.media(sobrepos) meta = { "rodada": cfg["rodada"], "versao_regulamento": cfg["versao_regulamento"], "seed": seed, "banco_hash": bank.bank_hash(itens), "n_itens": len(alvo), "n_participantes": len(linhas), "modo": modo_rotulo, "duracao_s": round(duracao, 1), "n_chamadas": n_chamadas, "por_idioma": {lg: sum(1 for i in alvo if i["lang"] == lg) for lg in bank.IDIOMAS}, "por_tipo": {t: sum(1 for i in alvo if i["type"] == t) for t in bank.TIPOS}, "por_dificuldade": {str(k): sum(1 for i in alvo if i["diff"] == k) for k in range(1, 6)}, "adjudicados": sum(1 for d in abertos if d.get("adjudicado")), "n_abertos": len(abertos), "recusa_media": stats.media([r["recusa_pct"] for r in linhas]), "canarios": len({d["canary"] for d in detalhe}), "canario_reproduzido": canario_reproduzido, "ngram_media": ngram_media, "itens_dificeis": dificeis, "detalhe_itens": detalhe, "calibracao_itens": len(calib), "prova_itens": len(prova), } destino = args.out or os.path.join(RAIZ, "resultados", f"{cfg['rodada']}-{'mock' if modo_mock else 'live'}") caminhos = report.escrever_relatorio(linhas, meta, destino) print(f"[ok] relatórios em {destino}") for k, v in caminhos.items(): print(f" - {k}: {os.path.relpath(v, RAIZ)}") print(f"[ok] duração {duracao:.1f}s · {n_chamadas} chamadas de modelo") return 0 if __name__ == "__main__": sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) sys.exit(principal())