Spaces:
Sleeping
Sleeping
File size: 2,928 Bytes
5a87a66 83e1e78 5a87a66 06e7858 5a87a66 06e7858 5a87a66 06e7858 5a87a66 13b9fde 5a87a66 83e1e78 5a87a66 8dfd504 5a87a66 a95acc0 5a87a66 4a8f804 5a87a66 | 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 | """
Thin HTTP client for the ALAS Streamlit frontend.
All backend calls go through here. Swap BASE_URL to point at a deployed server
without touching app.py.
"""
import os
import requests
from dotenv import load_dotenv
load_dotenv()
BASE_URL = os.getenv("BACKEND_URL", "http://localhost:8000")
def get_scenarios() -> dict:
"""GET /scenarios -> {scenarios: [{id, title, skill_area, ai_name, ai_role, learner_role, learning_goal, has_autoplay}, ...]}"""
r = requests.get(f"{BASE_URL}/scenarios", timeout=10)
r.raise_for_status()
return r.json()
def create_chat(scenario_id: str = "onboarding_report") -> dict:
"""POST /chats -> {chat_id, scenario_id, messages}"""
r = requests.post(
f"{BASE_URL}/chats", json={"scenario_id": scenario_id}, timeout=10
)
r.raise_for_status()
return r.json()
def get_chat(chat_id: str) -> dict:
"""GET /chats/{chat_id} -> {chat_id, messages}"""
r = requests.get(f"{BASE_URL}/chats/{chat_id}", timeout=10)
r.raise_for_status()
return r.json()
def get_skills() -> dict:
"""GET /skills -> {skills: [{slug, name, order}, ...]}"""
r = requests.get(f"{BASE_URL}/skills", timeout=10)
r.raise_for_status()
return r.json()
def get_notes(chat_id: str) -> dict:
"""GET /chats/{chat_id}/notes -> {chat_id, notes: [{skill, note, turn}, ...]}"""
r = requests.get(f"{BASE_URL}/chats/{chat_id}/notes", timeout=10)
r.raise_for_status()
return r.json()
def send_message(chat_id: str, message: str) -> dict:
"""POST /chats/{chat_id}/message -> {chat_id, reply, turn, messages}"""
r = requests.post(
f"{BASE_URL}/chats/{chat_id}/message",
json={"message": message},
timeout=60,
)
r.raise_for_status()
return r.json()
def stream_message(chat_id: str, message: str):
"""
POST /chats/{chat_id}/stream
Yields decoded token strings. Stops on [DONE] sentinel.
"""
with requests.post(
f"{BASE_URL}/chats/{chat_id}/stream",
json={"message": message},
stream=True,
timeout=(10, 120),
) as r:
r.raise_for_status()
for raw_line in r.iter_lines():
if not raw_line:
continue
line = raw_line.decode("utf-8") if isinstance(raw_line, bytes) else raw_line
if not line.startswith("data: "):
continue
token = line[len("data: ") :]
if token == "[DONE]":
break
yield token
def get_assessment(chat_id: str) -> dict:
"""POST /chats/{chat_id}/assessment -> Assessment (cached server-side per session)"""
r = requests.post(f"{BASE_URL}/chats/{chat_id}/assessment", timeout=60)
r.raise_for_status()
return r.json()
def delete_chat(chat_id: str) -> None:
"""DELETE /chats/{chat_id}"""
r = requests.delete(f"{BASE_URL}/chats/{chat_id}", timeout=10)
r.raise_for_status()
|