255 lines
8.2 KiB
Python
255 lines
8.2 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
FastAPI web frontend for the text-based LLM interaction system.
|
|
Serves a simple web UI and exposes an API to interact with the LLM.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Optional, Tuple
|
|
from uuid import uuid4
|
|
from threading import Lock
|
|
|
|
from fastapi import FastAPI, Request, Response, Cookie, HTTPException
|
|
from fastapi.responses import FileResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
|
|
from config import Config
|
|
from llm_client import LLMClient
|
|
from conversation import ConversationManager
|
|
from game_config import load_game_config
|
|
from game_state import GameState
|
|
|
|
|
|
# Application setup
|
|
app = FastAPI(title="Text Adventure - Web UI")
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
STATIC_DIR = BASE_DIR / "static"
|
|
STATIC_DIR.mkdir(parents=True, exist_ok=True)
|
|
SESSIONS_DIR = (BASE_DIR.parent / "data" / "sessions")
|
|
SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Mount /static for assets (requires 'aiofiles' in env for async file serving)
|
|
app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")
|
|
|
|
|
|
# Models
|
|
class ChatRequest(BaseModel):
|
|
message: str
|
|
|
|
|
|
# Globals
|
|
CONFIG = Config()
|
|
CLIENT = LLMClient(CONFIG)
|
|
GAMECFG = load_game_config()
|
|
_SESSIONS: dict[str, ConversationManager] = {}
|
|
_GAME_STATES: dict[str, GameState] = {}
|
|
_SESSIONS_LOCK = Lock()
|
|
|
|
SESSION_COOKIE_NAME = "session_id"
|
|
|
|
|
|
def _get_or_create_session(session_id: Optional[str]) -> Tuple[str, ConversationManager, bool]:
|
|
"""
|
|
Return an existing conversation manager by session_id,
|
|
or create a new session if not present.
|
|
Returns (session_id, manager, created_flag).
|
|
"""
|
|
created = False
|
|
with _SESSIONS_LOCK:
|
|
if not session_id or session_id not in _SESSIONS:
|
|
session_id = uuid4().hex
|
|
cm = ConversationManager()
|
|
# Seed with system prompt if configured
|
|
if getattr(GAMECFG, "system_prompt", None):
|
|
cm.add_system_message(GAMECFG.system_prompt)
|
|
_SESSIONS[session_id] = cm
|
|
created = True
|
|
manager = _SESSIONS[session_id]
|
|
return session_id, manager, created
|
|
|
|
|
|
def _ensure_state(session_id: str) -> GameState:
|
|
"""
|
|
Ensure a GameState exists for a given session and return it.
|
|
Loads from disk if present, otherwise creates a new state and persists it.
|
|
"""
|
|
with _SESSIONS_LOCK:
|
|
if session_id not in _GAME_STATES:
|
|
path = SESSIONS_DIR / f"{session_id}.json"
|
|
gs = None
|
|
if path.exists():
|
|
try:
|
|
with path.open("r", encoding="utf-8") as f:
|
|
data = json.load(f) or {}
|
|
gs = GameState()
|
|
for field in [
|
|
"room_description",
|
|
"door_description",
|
|
"door_locked",
|
|
"door_open",
|
|
"door_id",
|
|
"lock_id",
|
|
"lock_key_id",
|
|
"key_description",
|
|
"key_hidden",
|
|
"key_revealed",
|
|
"key_taken",
|
|
"key_id",
|
|
"container_id",
|
|
"inventory",
|
|
"exits",
|
|
"completed",
|
|
]:
|
|
if field in data:
|
|
setattr(gs, field, data[field])
|
|
except Exception:
|
|
gs = None
|
|
if gs is None:
|
|
gs = GameState.from_files("state")
|
|
try:
|
|
with path.open("w", encoding="utf-8") as f:
|
|
json.dump(gs.__dict__, f)
|
|
except Exception:
|
|
pass
|
|
_GAME_STATES[session_id] = gs
|
|
return _GAME_STATES[session_id]
|
|
|
|
|
|
# Routes
|
|
@app.get("/", response_class=FileResponse)
|
|
def index() -> FileResponse:
|
|
"""
|
|
Serve the main web UI.
|
|
"""
|
|
index_path = STATIC_DIR / "index.html"
|
|
if not index_path.exists():
|
|
# Provide a minimal fallback page if index.html is missing
|
|
# This should not happen in normal usage.
|
|
fallback = """<!doctype html>
|
|
<html>
|
|
<head><meta charset="utf-8"><title>Text Adventure</title></head>
|
|
<body><h1>Text Adventure Web UI</h1><p>index.html is missing.</p></body>
|
|
</html>
|
|
"""
|
|
tmp_path = STATIC_DIR / "_fallback_index.html"
|
|
tmp_path.write_text(fallback, encoding="utf-8")
|
|
return FileResponse(str(tmp_path))
|
|
return FileResponse(str(index_path))
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health() -> dict:
|
|
"""
|
|
Health check endpoint.
|
|
"""
|
|
return {"status": "ok"}
|
|
|
|
@app.get("/api/session")
|
|
def session_info(
|
|
response: Response,
|
|
session_id: Optional[str] = Cookie(default=None, alias=SESSION_COOKIE_NAME),
|
|
) -> JSONResponse:
|
|
"""
|
|
Ensure a session exists, set cookie if needed, and return scenario metadata and start message.
|
|
Also returns the current public game state snapshot.
|
|
"""
|
|
sid, conv, created = _get_or_create_session(session_id)
|
|
if created:
|
|
response.set_cookie(
|
|
key=SESSION_COOKIE_NAME,
|
|
value=sid,
|
|
httponly=True,
|
|
samesite="lax",
|
|
max_age=7 * 24 * 3600,
|
|
path="/",
|
|
)
|
|
gs = _ensure_state(sid)
|
|
payload = {
|
|
"session_id": sid,
|
|
"created": created,
|
|
"start_message": getattr(GAMECFG, "start_message", "") or "",
|
|
"scenario": getattr(GAMECFG, "scenario", {}) or {},
|
|
"rules": getattr(GAMECFG, "rules", []) or [],
|
|
"state": gs.to_public_dict(),
|
|
}
|
|
return JSONResponse(payload)
|
|
|
|
|
|
@app.post("/api/chat")
|
|
def chat(
|
|
req: ChatRequest,
|
|
response: Response,
|
|
session_id: Optional[str] = Cookie(default=None, alias=SESSION_COOKIE_NAME),
|
|
) -> JSONResponse:
|
|
"""
|
|
Accept a user message, apply deterministic game logic to update state,
|
|
then ask the LLM to narrate the outcome. Maintains a server-side session.
|
|
"""
|
|
message = req.message.strip()
|
|
if not message:
|
|
raise HTTPException(status_code=400, detail="Message cannot be empty")
|
|
|
|
sid, conv, created = _get_or_create_session(session_id)
|
|
|
|
# Set session cookie if new
|
|
if created:
|
|
response.set_cookie(
|
|
key=SESSION_COOKIE_NAME,
|
|
value=sid,
|
|
httponly=True,
|
|
samesite="lax",
|
|
max_age=7 * 24 * 3600, # 7 days
|
|
path="/",
|
|
)
|
|
|
|
# Determine outcome via the game engine, then request narration
|
|
try:
|
|
gs = _ensure_state(sid)
|
|
conv.add_user_message(message)
|
|
|
|
engine_outcome = gs.apply_action(message) # {"events": [...]}
|
|
|
|
# Build a transient system message with canonical facts for the narrator
|
|
narrator_directive = {
|
|
"ENGINE_OUTCOME": {
|
|
"events": engine_outcome.get("events", []),
|
|
"state": gs.to_public_dict(),
|
|
},
|
|
"NARRATION_RULES": [
|
|
"Narrate strictly according to ENGINE_OUTCOME. Do not invent state.",
|
|
"Do not add items, unlock objects, or change inventory; the engine already did that.",
|
|
"Use 2-5 sentences, present tense, second person. Be concise and vivid.",
|
|
"If the action was impossible, explain why using the facts provided.",
|
|
],
|
|
}
|
|
transient_system = {
|
|
"role": "system",
|
|
"content": "ENGINE CONTEXT (JSON): " + json.dumps(narrator_directive),
|
|
}
|
|
|
|
messages = list(conv.get_history()) + [transient_system]
|
|
reply = CLIENT.get_response(messages)
|
|
|
|
conv.add_assistant_message(reply)
|
|
|
|
# Persist updated state
|
|
try:
|
|
with (SESSIONS_DIR / f"{sid}.json").open("w", encoding="utf-8") as f:
|
|
json.dump(gs.__dict__, f)
|
|
except Exception:
|
|
pass
|
|
|
|
return JSONResponse({
|
|
"reply": reply,
|
|
"completed": gs.completed,
|
|
"events": engine_outcome.get("events", []),
|
|
"state": gs.to_public_dict(),
|
|
})
|
|
except Exception as e:
|
|
# Do not leak internal details in production; log as needed.
|
|
raise HTTPException(status_code=502, detail=f"LLM backend error: {e}") from e |