File size: 3,676 Bytes
7c8bb54 006d3e8 7c8bb54 006d3e8 7c8bb54 006d3e8 7c8bb54 0ba1a0c 7c8bb54 896ab23 7c8bb54 006d3e8 7c8bb54 006d3e8 054a72a 0ba1a0c 054a72a 006d3e8 054a72a 7c8bb54 006d3e8 7c8bb54 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | # type: ignore
"""
modules/external_gpu.py
Backend GPU externo (Kaggle T4 4-bit / Lightning / RunPod) — fallback
quando o ZeroGPU esgota a quota.
- A Space chama EXTERNAL_GPU_URL/generate (ver kaggle_backend.py).
- Sem quota ZeroGPU não há CUDA local: 4-bit SÓ faz sentido numa GPU
externa real (T4 16GB aguenta Lexi 8B em 4-bit ≈ 5.5GB).
"""
import os
from typing import Optional, List, Dict
try:
from loguru import logger # type: ignore
except Exception:
class _Dummy:
def info(self, *a, **k): pass
def success(self, *a, **k): pass
def warning(self, *a, **k): pass
def error(self, *a, **k): pass
logger = _Dummy() # type: ignore
_last_status: str = "nunca tentada"
def get_external_gpu_url() -> str:
return (os.getenv("EXTERNAL_GPU_URL") or "").strip().rstrip("/")
def is_external_gpu_configured() -> bool:
return bool(get_external_gpu_url())
def get_last_status() -> str:
"""Último resultado (para mostrar no chat/status)."""
return _last_status
def _set_status(msg: str) -> None:
global _last_status
_last_status = msg
def generate_external_gpu(
prompt: str,
system_prompt: Optional[str] = None,
context_history: Optional[List[Dict]] = None,
max_tokens: int = 512,
timeout: int = 120,
) -> Optional[str]:
"""POST {url}/generate → texto ou None (qualquer falha = None, sem exceção)."""
url = get_external_gpu_url()
if not url:
_set_status("off (sem EXTERNAL_GPU_URL nos Secrets)")
return None
try:
import requests # type: ignore
# 1) probe rápido: túnel morto (Kaggle parado/URL rodada) falha aqui
# em vez de esperar o timeout cheio do /generate.
try:
h = requests.get(f"{url}/health",
headers={"ngrok-skip-browser-warning": "1"},
timeout=15)
if h.status_code != 200:
_set_status("túnel morto? (reinicia o Kaggle e atualiza EXTERNAL_GPU_URL)")
except Exception:
pass
r = requests.post(
f"{url}/generate",
json={
"prompt": prompt,
"system_prompt": system_prompt or "",
"history": context_history or [],
"max_tokens": max_tokens,
},
headers={"ngrok-skip-browser-warning": "1"},
timeout=timeout,
)
if r.status_code == 200:
text = (r.json().get("text") or "").strip()
if text:
logger.success("[EXT-GPU] Resposta via backend externo")
_set_status("OK (última chamada com sucesso)")
return text
_set_status("falhou (200 vazio)")
logger.warning(f"[EXT-GPU] 200 vazio: {r.text[:200]}")
return None
ctype = (r.headers.get("content-type") or "")
if "html" in ctype or r.text.lstrip().lower().startswith(("<!doctype", "<html")):
_set_status("túnel morto (reinicia o Kaggle e atualiza EXTERNAL_GPU_URL)")
logger.warning("[EXT-GPU] túnel ngrok morto (HTML em vez de JSON)")
return None
try:
detail = r.json().get("error", "") or r.json().get("trace", "")
except Exception:
detail = r.text
_set_status(f"falhou (HTTP {r.status_code}: {str(detail)[:120]})")
logger.warning(f"[EXT-GPU] HTTP {r.status_code}: {str(detail)[:300]}")
return None
except Exception as e:
_set_status(f"falhou (inalcançável: {str(e)[:120]})")
logger.warning(f"[EXT-GPU] indisponível ({e})")
return None
|