veredicto / scripts /servir_veredicto_multi.py

Upload folder using huggingface_hub

8249eac verified 3 days ago

10.9 kB

#!/usr/bin/env python3
"""servir_veredicto_multi.py — API MULTI-ESPECIALISTA do Veredicto ("Frankenstein").
Roteamento por tarefa sobre UM backbone clef-flash (bf16, congelado):
  categoria ← cabeça do B2 (híbrido α=0,5)     — melhor cat da régua: 0,7012
  decisão   ← cabeça do E1 (híbrido α=margem)  — melhor dec: 0,6892
  urgência  ← adapter LoRA v4 + cabeça release — melhor urg: 0,7450
B2 e E1 são frozen_base (só a cabeça muda); o único LoRA é o v4. Logo não é
multi-LoRA de verdade: é 1 backbone + 3 cabeças + 1 adapter, 3 forwards/caso.
Régua esperada (teste de mesa 22/out, preds compostas nos 753 casos seed 42):
cat 0,7012 / dec 0,6892 / urg 0,7450 / geral 0,7118 — empate estatístico com
o juiz de fronteira (0,7127). Este servidor deve reproduzir esses números.
Mesmo contrato da produção (servir_ensemble.py): GET /saude,
POST /decidir {"estado", "limiar"} → {"respostas", "confianca_roteamento", ...}.
Uso (container treino, GPU 0, porta 8200):
  python3 scripts/servir_veredicto_multi.py \
    --checkpoint-cat models/veredicto_b2_step-2208 \
    --checkpoint-dec models/veredicto_e1_step-2576 \
    --base-path /workspace/.cache/huggingface/hub/models--Cloudflare--clef-flash/snapshots/8b2e5fd17c09fd49bc2880b805d5323ec7ff4ff3 \
    --adapter models/veredicto-lora-v4 \
    --porta 8200 --device cuda
"""
import argparse
import json
import sys
import threading
import time
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  # noqa: E402

OPCOES = {"categoria": list(CATEGORIAS), "decisao": list(DECISOES),
          "urgencia": list(URGENCIA)}
INSTRUCOES = {
    "categoria": "Qual a categoria principal do chamado?",
    "decisao": "Qual decisão tomar para este caso?",
    "urgencia": "Qual a urgência? (0=baixa, 3=crítica)",
}

LOCK = threading.Lock()  # troca de cabeça + toggle de adapter não são thread-safe
MODEL = {}


def _linha(tarefa: str, estado: str) -> dict:
    ops = OPCOES[tarefa]
    return {"estado": estado, "tarefa": tarefa,
            "pergunta": {"opcoes": ops, "instrucao": INSTRUCOES[tarefa]},
            "rotulo": 0}  # rótulo dummy (índice): exigido pelo contrato, ignorado


def _rotulos_codificados(qenc, tokenizer, canonicas: list[str]) -> list[str]:
    """Posição codificada → texto da opção, NA ORDEM dos option_ids.
    A ordem canônica de negócio (DECISOES/CATEGORIAS) NÃO coincide com a
    ordem codificada (alfabética) — devolver list(canonicas) troca os
    rótulos (bug pego na paridade 22/out: probs certas, nomes trocados;
    urgência escapava porque lá as duas ordens coincidem). Fallback:
    ordem alfabética, a convenção da régua.
    """
    try:
        decod = [tokenizer.decode([oid]).strip() for oid in qenc.option_ids]
        if sorted(decod) == sorted(o.strip() for o in canonicas):
            return decod
    except Exception:
        pass
    return sorted(o.strip() for o in canonicas)


def _forward(head: str, usar_adapter: bool, batch):
    """Um forward com a cabeça e o estado de adapter pedidos. Chamar sob LOCK.
    O retorno é CLONADO tensor a tensor: o clef reutiliza o buffer de saída
    entre forwards, então sem clone os logits do 1º forward seriam
    sobrescritos pelo 3º (bug pego na paridade 22/out: urgência 25/25 —
    era sempre a saída do ÚLTIMO forward — e cat/dec erradas).
    """
    import torch
    m = MODEL
    model = m["model"]
    model.head = m["heads"][head]
    lm = model.language_model
    if not usar_adapter:
        with lm.disable_adapter(), torch.inference_mode():
            return [t.detach().clone() for t in model(batch)[0]]
    with torch.inference_mode():
        return [t.detach().clone() for t in model(batch)[0]]


def decidir(estado: str, limiar: float) -> dict:
    import torch
    from veredicto_contrato import construir_caso, alinhar_perguntas
    m = MODEL
    clef, processor = m["clef"], m["processor"]
    tarefas = ["categoria", "decisao", "urgencia"]
    caso = construir_caso([_linha(t, estado) for t in tarefas],
                          contexto=m["meta"]["contexto"])
    enc = clef.encode_record(processor.tokenizer, caso["record"],
                             processor=processor, **caso["contexto"])
    batch = clef.collate_records([enc], processor.tokenizer.pad_token_id,
                                 torch.device(m["device"]))
    aligned = alinhar_perguntas(caso, enc.questions)

    t0 = time.time()
    with LOCK:
        logits_cat = _forward("b2", False, batch)        # cat ← cabeça B2
        logits_dec = _forward("e1", False, batch)        # dec ← cabeça E1
        logits_urg = _forward("release", True, batch)    # urg ← v4 + release
    latencia_ms = round((time.time() - t0) * 1000)

    fonte = {"categoria": logits_cat, "decisao": logits_dec, "urgencia": logits_urg}
    respostas, confs = {}, []
    for k, (qenc, _, _) in enumerate(aligned):
        tarefa = qenc.question_id
        lg = fonte[tarefa][k]
        probs = lg.float().softmax(-1).cpu().tolist()
        rotulos = _rotulos_codificados(qenc, processor.tokenizer,
                                       caso["alvos"][tarefa]["opcoes_originais"])
        i = max(range(len(probs)), key=probs.__getitem__)
        respostas[tarefa] = {
            "resposta": rotulos[i],
            "confianca": round(probs[i], 4),
            "distribuicao": {r: round(p, 4) for r, p in zip(rotulos, probs)},
        }
        confs.append(probs[i])
    conf = min(confs)
    return {"respostas": respostas,
            "confianca_roteamento": round(conf, 4),
            "limiar": limiar,
            "destino": "automatico" if conf >= limiar else "system2",
            "latencia_ms": latencia_ms,
            "shadow": {"modo": m["modo"], "especialistas": m["especialistas"]}}


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", "shadow": True,
                        "modo": MODEL["modo"],
                        "especialistas": MODEL["especialistas"],
                        "calibrado": False})
            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
        try:
            self._json(decidir(estado, limiar))
        except Exception as e:  # erro vira 500 isolado, nunca derruba o servidor
            self._json({"erro": f"falha de inferência no multi: {e}"}, 500)


def _carregar_head(clef, ckpt_dir: str, device, dtype):
    """Instancia uma JointSchemaHead de um checkpoint frozen_base e carrega os pesos."""
    import torch
    d = Path(ckpt_dir)
    cfg = json.loads((d / "joint_head_config.json").read_text(encoding="utf-8"))
    head = clef.JointSchemaHead(**cfg)
    head.load_state_dict(
        torch.load(d / "joint_head.pt", map_location=device, weights_only=True),
        strict=True)
    return head.to(device=device, dtype=dtype).eval()


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--checkpoint-cat", required=True,
                    help="checkpoint frozen_base cuja cabeça responde categoria (B2)")
    ap.add_argument("--checkpoint-dec", required=True,
                    help="checkpoint frozen_base cuja cabeça responde decisão (E1)")
    ap.add_argument("--base-path", required=True)
    ap.add_argument("--adapter", required=True,
                    help="adapter LoRA de urgência (v4) — usado com a cabeça de release")
    ap.add_argument("--device", default="cuda")
    ap.add_argument("--porta", type=int, default=8200)
    args = ap.parse_args()

    import torch
    from safetensors.torch import load_file
    from veredicto_contrato import carregar_checkpoint, carregar_codigo_clef

    # O checkpoint do B2 fornece backbone + processor + meta + cabeça de categoria.
    model, processor, meta = carregar_checkpoint(
        args.checkpoint_cat, args.device, base_path=args.base_path)
    clef = carregar_codigo_clef(args.base_path)
    model.eval()
    dtype = next(model.head.parameters()).dtype

    heads = {"b2": model.head,
             "e1": _carregar_head(clef, args.checkpoint_dec, args.device, dtype)}

    # Adapter de urgência: backbone NÃO é PeftModel (frozen_base) — wrap limpo.
    from peft import PeftModel
    model.language_model = PeftModel.from_pretrained(
        model.language_model, args.adapter, is_trainable=False,
        local_files_only=True)
    base = Path(args.base_path)
    cfg = json.loads((base / "joint_head_config.json").read_text(encoding="utf-8"))
    head_release = clef.JointSchemaHead(**cfg)
    head_release.load_state_dict(load_file(str(base / "joint_head.safetensors")),
                                 strict=True)
    heads["release"] = head_release.to(device=args.device, dtype=dtype).eval()
    model.eval()

    MODEL.update({
        "model": model, "processor": processor, "clef": clef, "meta": meta,
        "heads": heads, "device": args.device, "modo": "multi_especialista",
        "especialistas": {
            "categoria": args.checkpoint_cat.rstrip("/").split("/")[-1],
            "decisao": args.checkpoint_dec.rstrip("/").split("/")[-1],
            "urgencia": args.adapter.rstrip("/").split("/")[-1] + " + head_release",
        },
    })
    print(f"✓ Veredicto MULTI na porta {args.porta} — "
          f"cat←{MODEL['especialistas']['categoria']} "
          f"dec←{MODEL['especialistas']['decisao']} "
          f"urg←{MODEL['especialistas']['urgencia']}", flush=True)
    ThreadingHTTPServer(("0.0.0.0", args.porta), Handler).serve_forever()


if __name__ == "__main__":
    main()

	#!/usr/bin/env python3
	"""servir_veredicto_multi.py — API MULTI-ESPECIALISTA do Veredicto ("Frankenstein").

	Roteamento por tarefa sobre UM backbone clef-flash (bf16, congelado):

	categoria ← cabeça do B2 (híbrido α=0,5) — melhor cat da régua: 0,7012
	decisão ← cabeça do E1 (híbrido α=margem) — melhor dec: 0,6892
	urgência ← adapter LoRA v4 + cabeça release — melhor urg: 0,7450

	B2 e E1 são frozen_base (só a cabeça muda); o único LoRA é o v4. Logo não é
	multi-LoRA de verdade: é 1 backbone + 3 cabeças + 1 adapter, 3 forwards/caso.

	Régua esperada (teste de mesa 22/out, preds compostas nos 753 casos seed 42):
	cat 0,7012 / dec 0,6892 / urg 0,7450 / geral 0,7118 — empate estatístico com
	o juiz de fronteira (0,7127). Este servidor deve reproduzir esses números.

	Mesmo contrato da produção (servir_ensemble.py): GET /saude,
	POST /decidir {"estado", "limiar"} → {"respostas", "confianca_roteamento", ...}.

	Uso (container treino, GPU 0, porta 8200):
	python3 scripts/servir_veredicto_multi.py \
	--checkpoint-cat models/veredicto_b2_step-2208 \
	--checkpoint-dec models/veredicto_e1_step-2576 \
	--base-path /workspace/.cache/huggingface/hub/models--Cloudflare--clef-flash/snapshots/8b2e5fd17c09fd49bc2880b805d5323ec7ff4ff3 \
	--adapter models/veredicto-lora-v4 \
	--porta 8200 --device cuda
	"""
	import argparse
	import json
	import sys
	import threading
	import time
	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 # noqa: E402

	OPCOES = {"categoria": list(CATEGORIAS), "decisao": list(DECISOES),
	"urgencia": list(URGENCIA)}
	INSTRUCOES = {
	"categoria": "Qual a categoria principal do chamado?",
	"decisao": "Qual decisão tomar para este caso?",
	"urgencia": "Qual a urgência? (0=baixa, 3=crítica)",
	}

	LOCK = threading.Lock() # troca de cabeça + toggle de adapter não são thread-safe
	MODEL = {}


	def _linha(tarefa: str, estado: str) -> dict:
	ops = OPCOES[tarefa]
	return {"estado": estado, "tarefa": tarefa,
	"pergunta": {"opcoes": ops, "instrucao": INSTRUCOES[tarefa]},
	"rotulo": 0} # rótulo dummy (índice): exigido pelo contrato, ignorado


	def _rotulos_codificados(qenc, tokenizer, canonicas: list[str]) -> list[str]:
	"""Posição codificada → texto da opção, NA ORDEM dos option_ids.

	A ordem canônica de negócio (DECISOES/CATEGORIAS) NÃO coincide com a
	ordem codificada (alfabética) — devolver list(canonicas) troca os
	rótulos (bug pego na paridade 22/out: probs certas, nomes trocados;
	urgência escapava porque lá as duas ordens coincidem). Fallback:
	ordem alfabética, a convenção da régua.
	"""
	try:
	decod = [tokenizer.decode([oid]).strip() for oid in qenc.option_ids]
	if sorted(decod) == sorted(o.strip() for o in canonicas):
	return decod
	except Exception:
	pass
	return sorted(o.strip() for o in canonicas)


	def _forward(head: str, usar_adapter: bool, batch):
	"""Um forward com a cabeça e o estado de adapter pedidos. Chamar sob LOCK.

	O retorno é CLONADO tensor a tensor: o clef reutiliza o buffer de saída
	entre forwards, então sem clone os logits do 1º forward seriam
	sobrescritos pelo 3º (bug pego na paridade 22/out: urgência 25/25 —
	era sempre a saída do ÚLTIMO forward — e cat/dec erradas).
	"""
	import torch
	m = MODEL
	model = m["model"]
	model.head = m["heads"][head]
	lm = model.language_model
	if not usar_adapter:
	with lm.disable_adapter(), torch.inference_mode():
	return [t.detach().clone() for t in model(batch)[0]]
	with torch.inference_mode():
	return [t.detach().clone() for t in model(batch)[0]]


	def decidir(estado: str, limiar: float) -> dict:
	import torch
	from veredicto_contrato import construir_caso, alinhar_perguntas
	m = MODEL
	clef, processor = m["clef"], m["processor"]
	tarefas = ["categoria", "decisao", "urgencia"]
	caso = construir_caso([_linha(t, estado) for t in tarefas],
	contexto=m["meta"]["contexto"])
	enc = clef.encode_record(processor.tokenizer, caso["record"],
	processor=processor, **caso["contexto"])
	batch = clef.collate_records([enc], processor.tokenizer.pad_token_id,
	torch.device(m["device"]))
	aligned = alinhar_perguntas(caso, enc.questions)

	t0 = time.time()
	with LOCK:
	logits_cat = _forward("b2", False, batch) # cat ← cabeça B2
	logits_dec = _forward("e1", False, batch) # dec ← cabeça E1
	logits_urg = _forward("release", True, batch) # urg ← v4 + release
	latencia_ms = round((time.time() - t0) * 1000)

	fonte = {"categoria": logits_cat, "decisao": logits_dec, "urgencia": logits_urg}
	respostas, confs = {}, []
	for k, (qenc, _, _) in enumerate(aligned):
	tarefa = qenc.question_id
	lg = fonte[tarefa][k]
	probs = lg.float().softmax(-1).cpu().tolist()
	rotulos = _rotulos_codificados(qenc, processor.tokenizer,
	caso["alvos"][tarefa]["opcoes_originais"])
	i = max(range(len(probs)), key=probs.__getitem__)
	respostas[tarefa] = {
	"resposta": rotulos[i],
	"confianca": round(probs[i], 4),
	"distribuicao": {r: round(p, 4) for r, p in zip(rotulos, probs)},
	}
	confs.append(probs[i])
	conf = min(confs)
	return {"respostas": respostas,
	"confianca_roteamento": round(conf, 4),
	"limiar": limiar,
	"destino": "automatico" if conf >= limiar else "system2",
	"latencia_ms": latencia_ms,
	"shadow": {"modo": m["modo"], "especialistas": m["especialistas"]}}


	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", "shadow": True,
	"modo": MODEL["modo"],
	"especialistas": MODEL["especialistas"],
	"calibrado": False})
	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
	try:
	self._json(decidir(estado, limiar))
	except Exception as e: # erro vira 500 isolado, nunca derruba o servidor
	self._json({"erro": f"falha de inferência no multi: {e}"}, 500)


	def _carregar_head(clef, ckpt_dir: str, device, dtype):
	"""Instancia uma JointSchemaHead de um checkpoint frozen_base e carrega os pesos."""
	import torch
	d = Path(ckpt_dir)
	cfg = json.loads((d / "joint_head_config.json").read_text(encoding="utf-8"))
	head = clef.JointSchemaHead(**cfg)
	head.load_state_dict(
	torch.load(d / "joint_head.pt", map_location=device, weights_only=True),
	strict=True)
	return head.to(device=device, dtype=dtype).eval()


	def main():
	ap = argparse.ArgumentParser(description=__doc__)
	ap.add_argument("--checkpoint-cat", required=True,
	help="checkpoint frozen_base cuja cabeça responde categoria (B2)")
	ap.add_argument("--checkpoint-dec", required=True,
	help="checkpoint frozen_base cuja cabeça responde decisão (E1)")
	ap.add_argument("--base-path", required=True)
	ap.add_argument("--adapter", required=True,
	help="adapter LoRA de urgência (v4) — usado com a cabeça de release")
	ap.add_argument("--device", default="cuda")
	ap.add_argument("--porta", type=int, default=8200)
	args = ap.parse_args()

	import torch
	from safetensors.torch import load_file
	from veredicto_contrato import carregar_checkpoint, carregar_codigo_clef

	# O checkpoint do B2 fornece backbone + processor + meta + cabeça de categoria.
	model, processor, meta = carregar_checkpoint(
	args.checkpoint_cat, args.device, base_path=args.base_path)
	clef = carregar_codigo_clef(args.base_path)
	model.eval()
	dtype = next(model.head.parameters()).dtype

	heads = {"b2": model.head,
	"e1": _carregar_head(clef, args.checkpoint_dec, args.device, dtype)}

	# Adapter de urgência: backbone NÃO é PeftModel (frozen_base) — wrap limpo.
	from peft import PeftModel
	model.language_model = PeftModel.from_pretrained(
	model.language_model, args.adapter, is_trainable=False,
	local_files_only=True)
	base = Path(args.base_path)
	cfg = json.loads((base / "joint_head_config.json").read_text(encoding="utf-8"))
	head_release = clef.JointSchemaHead(**cfg)
	head_release.load_state_dict(load_file(str(base / "joint_head.safetensors")),
	strict=True)
	heads["release"] = head_release.to(device=args.device, dtype=dtype).eval()
	model.eval()

	MODEL.update({
	"model": model, "processor": processor, "clef": clef, "meta": meta,
	"heads": heads, "device": args.device, "modo": "multi_especialista",
	"especialistas": {
	"categoria": args.checkpoint_cat.rstrip("/").split("/")[-1],
	"decisao": args.checkpoint_dec.rstrip("/").split("/")[-1],
	"urgencia": args.adapter.rstrip("/").split("/")[-1] + " + head_release",
	},
	})
	print(f"✓ Veredicto MULTI na porta {args.porta} — "
	f"cat←{MODEL['especialistas']['categoria']} "
	f"dec←{MODEL['especialistas']['decisao']} "
	f"urg←{MODEL['especialistas']['urgencia']}", flush=True)
	ThreadingHTTPServer(("0.0.0.0", args.porta), Handler).serve_forever()


	if __name__ == "__main__":
	main()