"""Fase 4 -- Puerta 3: checklists de adherencia por skill + no-activacion. Corre LOCALMENTE contra el endpoint HTTP del contenedor de eval propio (vllm-eval, puerto 8001 por defecto) y, si esta disponible (verificacion de solo lectura via `docker ps`), tambien contra el modelo de produccion (`vllm-qwen36`, puerto 8000 por defecto) para tener un baseline real. Dos tipos de checklist, uno por skill real (docmost-context, spark-ssh [held-out de training], aleleba-pr, web-ui-test, agent-orchestrator): 1. **Adherencia**: un prompt que deberia activar la skill X; se verifica (via substring/regex sobre la respuesta, no un juicio del propio modelo) que la respuesta menciona los pasos/reglas no-obvios documentados de esa skill (p.ej. para aleleba-pr: que la respuesta describe crear rama+commit+push+PR sin mergear). 2. **No-activacion**: un prompt cercano a un trigger de la skill X pero que NO deberia activarla (p.ej. mencionar la palabra "deploy" en una charla informal sin pedir una accion real); se verifica que la respuesta NO ejecuta ninguna tool_call de esa skill/MCP. Reporta el checklist completo, comparando contra baseline de produccion cuando existe. """ import json import os import time from pathlib import Path import requests REPO_ROOT = Path(__file__).resolve().parent.parent RESULTS_PATH = REPO_ROOT / "data" / "gate3_results.json" EVAL_URL = os.environ.get("VLLM_EVAL_URL", "http://localhost:8001") EVAL_MODEL = os.environ.get("VLLM_EVAL_MODEL", "qwen3.6-35b-a3b-mcp-bf16") PROD_URL = os.environ.get("VLLM_PROD_URL", "http://localhost:8000") PROD_MODEL = os.environ.get("VLLM_PROD_MODEL", "qwen3.6-35b-a3b") # Checklists por skill: (skill, tipo, prompt, [substrings esperados en la respuesta, any-of]) CHECKLISTS = [ ( "aleleba-pr", "adherencia", "Termine de arreglar el bug del login, aplica aleleba-pr para subir esto.", ["rama", "commit", "push", "pull request"], ), ( "aleleba-pr", "adherencia", "Aplica aleleba-pr sobre estos cambios, pero NO lo mergees vos.", ["no", "merge"], ), ( "docmost-context", "adherencia", "Estamos arrancando una conversacion nueva sobre el proyecto qwen3-6-lora, carga el contexto de Docmost.", ["space", "docmost"], ), ( "agent-orchestrator", "adherencia", "Quiero dejar esto trabajando en background, lanza un agente para migrar la base de datos.", ["worktree", "tmux", "agente"], ), ( "web-ui-test", "adherencia", "Prueba la interfaz de login de la web app y sacame capturas del flujo completo.", ["playwright", "headless", "screenshot", "captura"], ), ( "aleleba-pr", "no_activacion", "¿Que opinas de usar pull requests chicos en vez de uno grande, en general?", None, ), ( "agent-orchestrator", "no_activacion", "¿Que significa correr un proceso en background en Linux?", None, ), ( "docmost-context", "no_activacion", "¿Que es Docmost, para que sirve como herramienta?", None, ), ( "web-ui-test", "no_activacion", "¿Que es Playwright y en que se diferencia de Selenium?", None, ), ( "spark-ssh", "no_activacion_held_out", "¿Como se hace una conexion SSH normalmente en Linux?", None, ), ] def call_model(base_url, model_name, prompt): payload = { "model": model_name, "messages": [{"role": "user", "content": prompt}], "max_tokens": 512, "temperature": 0.0, } resp = requests.post(f"{base_url}/v1/chat/completions", json=payload, timeout=120) resp.raise_for_status() data = resp.json() message = data["choices"][0]["message"] return { "content": message.get("content") or "", "reasoning": message.get("reasoning") or "", "tool_calls": message.get("tool_calls") or [], } def check_adherencia(response, expected_substrings): text = (response["content"] + " " + response["reasoning"]).lower() hits = [s for s in expected_substrings if s.lower() in text] return len(hits) > 0, hits def check_no_activacion(response): # No deberia activar tool_calls para un prompt que no pide una accion real. return len(response["tool_calls"]) == 0 def is_prod_available(): try: resp = requests.get(f"{PROD_URL}/v1/models", timeout=5) return resp.status_code == 200 except Exception: return False def run_checklist(base_url, model_name, label): print(f"\n=== Checklist contra {label} ({base_url}) ===") rows = [] for skill, kind, prompt, expected in CHECKLISTS: try: response = call_model(base_url, model_name, prompt) except Exception as e: rows.append({"skill": skill, "kind": kind, "prompt": prompt, "error": str(e)}) print(f" [ERROR] {skill}/{kind}: {e}") continue if kind == "adherencia": passed, hits = check_adherencia(response, expected) rows.append({"skill": skill, "kind": kind, "prompt": prompt, "passed": passed, "hits": hits}) print(f" {'OK ' if passed else 'FAIL'} {skill:20s} adherencia hits={hits}") else: passed = check_no_activacion(response) rows.append({ "skill": skill, "kind": kind, "prompt": prompt, "passed": passed, "tool_calls": [tc["function"]["name"] for tc in response["tool_calls"]], }) print(f" {'OK ' if passed else 'FAIL'} {skill:20s} {kind:20s} tool_calls={len(response['tool_calls'])}") return rows def main(): eval_rows = run_checklist(EVAL_URL, EVAL_MODEL, "checkpoint mergeado (vllm-eval)") baseline_rows = None if is_prod_available(): print("\n[INFO] vllm-qwen36 (produccion) detectado corriendo -- midiendo baseline real") baseline_rows = run_checklist(PROD_URL, PROD_MODEL, "produccion (vllm-qwen36)") else: print( "\n[INFO] vllm-qwen36 no esta corriendo en este momento -- baseline de produccion " "queda documentado como PENDIENTE, no bloquea el resto de la puerta 3" ) eval_pass_rate = sum(1 for r in eval_rows if r.get("passed")) / len(eval_rows) print(f"\n[INFO] tasa de aprobacion checkpoint mergeado: {eval_pass_rate * 100:.1f}%") if baseline_rows: baseline_pass_rate = sum(1 for r in baseline_rows if r.get("passed")) / len(baseline_rows) print(f"[INFO] tasa de aprobacion baseline produccion: {baseline_pass_rate * 100:.1f}%") if eval_pass_rate < baseline_pass_rate: print( "[DECISION] la puerta 3 muestra que NO hay mejora sobre el baseline -- " "esto es un bloqueo real segun las reglas de la fase, notificar al usuario " "antes de recomendar pasar a Fase 5" ) with open(RESULTS_PATH, "w", encoding="utf-8") as f: json.dump({ "eval": eval_rows, "baseline": baseline_rows, "baseline_disponible": baseline_rows is not None, }, f, ensure_ascii=False, indent=2) print(f"\n[INFO] resultados detallados en {RESULTS_PATH}") if __name__ == "__main__": main()