File size: 2,390 Bytes
2c35d7f | 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 | """
SafeChat API Client — Talks to the ML Service.
"""
import requests
import os
API_URL = os.getenv("SAFECHAT_API_URL", "http://localhost:8001")
def check_health() -> dict | None:
"""Check if ML Service is reachable. Returns health payload or None."""
try:
r = requests.get(f"{API_URL}/api/v1/health", timeout=3)
if r.status_code == 200:
return r.json()
return None
except requests.exceptions.RequestException:
return None
def moderate(text: str, context: list[str] | None = None) -> dict | None:
"""POST /api/v1/moderate — returns full moderation result."""
payload = {"text": text}
if context:
payload["context"] = context
try:
r = requests.post(f"{API_URL}/api/v1/moderate", json=payload, timeout=15)
r.raise_for_status()
return r.json()
except requests.exceptions.RequestException:
return None
def detoxify(text: str, context: list[str] | None = None) -> dict | None:
"""POST /api/v1/detoxify — returns detoxified text."""
payload = {"text": text}
if context:
payload["context"] = context
try:
r = requests.post(f"{API_URL}/api/v1/detoxify", json=payload, timeout=20)
r.raise_for_status()
return r.json()
except requests.exceptions.RequestException:
return None
def get_feedback_stats() -> dict | None:
"""GET /api/v1/feedback/stats — continuous learning metrics."""
try:
r = requests.get(f"{API_URL}/api/v1/feedback/stats", timeout=5)
r.raise_for_status()
return r.json()
except requests.exceptions.RequestException:
return None
def submit_feedback(
message_id: str,
moderator_id: str,
was_correct: bool,
correct_label: str | None = None,
notes: str | None = None,
) -> dict | None:
"""POST /api/v1/feedback — submit moderator correction."""
payload = {
"message_id": message_id,
"moderator_id": moderator_id,
"model_prediction_was_correct": was_correct,
}
if correct_label:
payload["correct_label"] = correct_label
if notes:
payload["notes"] = notes
try:
r = requests.post(f"{API_URL}/api/v1/feedback", json=payload, timeout=5)
r.raise_for_status()
return r.json()
except requests.exceptions.RequestException:
return None
|