| |
| """ |
| servir_decisor.py — API local do decisor System 1 (v3), zero dependências |
| além do stack de treino (stdlib http.server + torch). |
| |
| Endpoints: |
| GET /saude → status do modelo, temperaturas, versão |
| POST /decidir → {"estado": "<texto do chamado>", "limiar": 0.8 (opcional)} |
| responde as 3 perguntas (categoria, decisão, urgência) |
| com probabilidades calibradas e roteamento S1/S2. |
| |
| Roteamento: a confiança de roteamento é o MÍNIMO entre as três perguntas |
| (conservador: um chamado só vai 100% automático se categoria, decisão e |
| urgência forem todas confiáveis). Abaixo do limiar → "system2". |
| |
| Uso (host GPU, dentro do container de treino): |
| docker compose run --rm -T --service-ports treino \ |
| python3 scripts/servir_decisor.py --checkpoint models/decisor-v3 \ |
| --porta 8100 |
| |
| Teste: |
| curl -s localhost:8100/decidir -H 'Content-Type: application/json' \ |
| -d '{"estado": "Cliente: meu whatsapp parou de enviar mensagens"}' |
| """ |
| import argparse |
| import json |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
|
|
| import torch |
| from transformers import AutoTokenizer |
|
|
| 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}, |
| } |
|
|
| DECISOR: DecisorCPU | None = None |
|
|
|
|
| def decidir(estado: str, limiar: float) -> dict: |
| respostas = {} |
| confs = [] |
| for nome, pergunta in PERGUNTAS.items(): |
| r = DECISOR.prever(estado, pergunta) |
| respostas[nome] = r |
| confs.append(r["confianca"]) |
| conf_roteamento = min(confs) |
| automatico = conf_roteamento >= limiar |
| return { |
| "respostas": respostas, |
| "confianca_roteamento": round(conf_roteamento, 4), |
| "limiar": limiar, |
| "destino": "automatico" if automatico else "system2", |
| } |
|
|
|
|
| class Handler(BaseHTTPRequestHandler): |
| def _json(self, obj: dict, status: int = 200): |
| corpo = json.dumps(obj, ensure_ascii=False).encode("utf-8") |
| 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", |
| "checkpoint": DECISOR.model.encoder.name_or_path, |
| "calibrado": bool(DECISOR.temps), |
| "temperaturas": {k: round(v, 3) |
| for k, v in DECISOR.temps.items()}, |
| "categorias": len(CATEGORIAS), |
| }) |
| return |
| self._json({"erro": "rota desconhecida — 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 do chamado", |
| "limiar": "0.8 (opcional)"}}, 400) |
| return |
| self._json(decidir(estado, limiar)) |
|
|
|
|
| def main(): |
| global DECISOR |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--checkpoint", default="models/decisor-v3") |
| ap.add_argument("--porta", type=int, default=8100) |
| ap.add_argument("--host", default="0.0.0.0") |
| ap.add_argument("--threads", type=int, default=0, |
| help="threads CPU (0 = torch decide)") |
| ap.add_argument("--device", default="auto", |
| choices=["auto", "cuda", "mps", "xpu", "cpu"], |
| help="default: auto (cuda → mps → cpu)") |
| ap.add_argument("--test", action="store_true", |
| help="roda um caso de teste e sai (sem subir o servidor)") |
| args = ap.parse_args() |
|
|
| device = args.device |
| if device == "auto": |
| import torch |
| if torch.cuda.is_available(): |
| device = "cuda" |
| elif getattr(torch.backends, "mps", None) \ |
| and torch.backends.mps.is_available(): |
| device = "mps" |
| else: |
| device = "cpu" |
| print(f"device auto: {device}") |
|
|
| DECISOR = DecisorCPU(args.checkpoint, threads=args.threads, |
| device=device) |
|
|
| if args.test: |
| exemplo = ("[Caso #1]\nCliente: meu whatsapp parou de enviar mensagens " |
| "para os clientes, mensagem nao enviada\nCliente: urgente, " |
| "a equipe de vendas esta parada") |
| print(json.dumps(decidir(exemplo, 0.8), ensure_ascii=False, indent=2)) |
| return |
|
|
| srv = ThreadingHTTPServer((args.host, args.porta), Handler) |
| print(f"✔ decisor servindo em http://{args.host}:{args.porta} " |
| f"(checkpoint {args.checkpoint}) — POST /decidir, GET /saude") |
| srv.serve_forever() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
