Spaces:
Sleeping
Sleeping
File size: 1,687 Bytes
72e2b6e 7fa2eda 72e2b6e 7fa2eda | 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 | import os
import requests
import streamlit as st
API_BASE_URL = os.getenv("API_BASE_URL", "http://localhost:8000")
def api_predict(text: str, model_type: str | None = None) -> dict:
payload = {"text": text}
if model_type:
payload["model_type"] = model_type
response = requests.post(f"{API_BASE_URL}/predict", json=payload, timeout=30)
response.raise_for_status()
return response.json()
def api_health() -> dict:
response = requests.get(f"{API_BASE_URL}/health", timeout=10)
response.raise_for_status()
return response.json()
def api_models() -> dict:
response = requests.get(f"{API_BASE_URL}/models", timeout=10)
response.raise_for_status()
return response.json()
def api_ab_stats() -> dict:
response = requests.get(f"{API_BASE_URL}/monitoring/ab-stats", timeout=10)
response.raise_for_status()
return response.json()
def api_drift() -> dict:
response = requests.get(f"{API_BASE_URL}/monitoring/drift", timeout=30)
response.raise_for_status()
return response.json()
def api_reset_monitoring() -> dict:
response = requests.post(f"{API_BASE_URL}/monitoring/reset", timeout=10)
response.raise_for_status()
return response.json()
def check_api_connection() -> bool:
try:
api_health()
return True
except requests.exceptions.RequestException:
return False
def require_api():
if not check_api_connection():
st.error(
f"Cannot connect to API at {API_BASE_URL}. "
"Start it with: `uv run uvicorn api.main:app --reload --port 8000`"
)
st.stop() |