| |
| """ |
| servir_ensemble.py — API do ensemble (média das probs calibradas de N decisors). |
| |
| Mesmos endpoints do servir_decisor (GET /saude, POST /decidir); carrega 2+ |
| checkpoints, roda cada pergunta em cada modelo e devolve a probabilidade |
| MÉDIA (calibrada por modelo antes da média — média de distribuições |
| calibradas continua calibrada). Roteamento: mínima confiança da média. |
| |
| Uso (host .144, GPU 1, porta de produção ou teste): |
| python3 scripts/servir_ensemble.py \ |
| --checkpoints models/decisor-v10 models/decisor-v11_cons \ |
| --device cuda --porta 8101 |
| """ |
| import argparse |
| import json |
| import sys |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
| from pathlib import Path |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| from chamados_para_decisor import CATEGORIAS, DECISOES, URGENCIA |
| from inferencia_decisor import DecisorCPU |
|
|
| PERGUNTAS = { |
| "categoria": {"tipo": "choice", |
| "instrucao": "Qual a categoria principal do chamado?", |
| "opcoes": CATEGORIAS}, |
| "decisao": {"tipo": "choice", |
| "instrucao": "Qual decisão tomar para este caso?", |
| "opcoes": DECISOES}, |
| "urgencia": {"tipo": "score", |
| "instrucao": "Qual a urgência? (0=baixa, 3=crítica)", |
| "opcoes": URGENCIA}, |
| } |
|
|
| DECISORES = [] |
|
|
|
|
| def decidir(estado: str, limiar: float) -> dict: |
| respostas, confs = {}, [] |
| for nome, pergunta in PERGUNTAS.items(): |
| ops = pergunta["opcoes"] |
| media = None |
| votos = [] |
| for d in DECISORES: |
| r = d["modelo"].prever(estado, pergunta) |
| probs = [r["distribuicao"].get(op, 0.0) for op in ops] |
| votos.append(probs) |
| votos.append(None) |
| votos.pop() |
| media = [sum(v[k] for v in votos) / len(votos) for k in range(len(ops))] |
| i = max(range(len(ops)), key=lambda k: media[k]) |
| respostas[nome] = { |
| "resposta": ops[i], |
| "confianca": round(media[i], 4), |
| "distribuicao": {op: round(media[k], 4) |
| for op, k in zip(ops, range(len(ops)))}, |
| } |
| confs.append(media[i]) |
| conf = min(confs) |
| return {"respostas": respostas, |
| "confianca_roteamento": round(conf, 4), |
| "limiar": limiar, |
| "destino": "automatico" if conf >= limiar else "system2"} |
|
|
|
|
| class Handler(BaseHTTPRequestHandler): |
| def _json(self, obj, status=200): |
| corpo = json.dumps(obj, ensure_ascii=False).encode() |
| self.send_response(status) |
| self.send_header("Content-Type", "application/json; charset=utf-8") |
| self.send_header("Content-Length", str(len(corpo))) |
| self.end_headers() |
| self.wfile.write(corpo) |
|
|
| def log_message(self, *a): |
| pass |
|
|
| def do_GET(self): |
| if self.path == "/saude": |
| self._json({ |
| "status": "ok", |
| "ensemble": [d["nome"] for d in DECISORES], |
| "n_modelos": len(DECISORES), |
| "calibrado": all(bool(d["modelo"].temps) for d in DECISORES), |
| "categorias": len(CATEGORIAS), |
| }) |
| return |
| self._json({"erro": "use GET /saude ou POST /decidir"}, 404) |
|
|
| def do_POST(self): |
| if self.path != "/decidir": |
| self._json({"erro": "rota desconhecida"}, 404) |
| return |
| try: |
| corpo = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) |
| estado = str(corpo["estado"]).strip() |
| if not estado: |
| raise ValueError("estado vazio") |
| limiar = float(corpo.get("limiar", 0.8)) |
| if not 0.0 <= limiar <= 1.0: |
| raise ValueError("limiar fora de [0,1]") |
| except (KeyError, ValueError, TypeError) as e: |
| self._json({"erro": f"payload inválido: {e}", |
| "formato": {"estado": "texto", "limiar": 0.8}}, 400) |
| return |
| self._json(decidir(estado, limiar)) |
|
|
|
|
| def _device_auto() -> str: |
| """Escolhe o melhor device disponível: cuda → mps → cpu.""" |
| import torch |
| if torch.cuda.is_available(): |
| return "cuda" |
| if getattr(torch.backends, "mps", None) and torch.backends.mps.is_available(): |
| return "mps" |
| return "cpu" |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--checkpoints", nargs="+", required=True) |
| ap.add_argument("--device", default="auto", |
| choices=["auto", "cuda", "mps", "xpu", "cpu"], |
| help="default: auto (cuda → mps → cpu)") |
| ap.add_argument("--porta", type=int, default=8101) |
| args = ap.parse_args() |
| device = _device_auto() if args.device == "auto" else args.device |
| print(f"device: {device}") |
| for ck in args.checkpoints: |
| DECISORES.append({"nome": ck.rstrip("/").split("/")[-1], |
| "modelo": DecisorCPU(ck, device=device)}) |
| print(f"✓ carregado: {ck}") |
| print(f"ensemble de {len(DECISORES)} modelos na porta {args.porta}") |
| ThreadingHTTPServer(("0.0.0.0", args.porta), Handler).serve_forever() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
