732982e66c
- /api/loaded backend endpoint proxies Ollama /api/ps and returns
{models: [{name, expires_at}]}
- Status line under the controls shows "Hot: gemma2:9b — unloads in 4m"
with a glowing orange dot, or "No model loaded" in muted style
- Loaded models in the dropdown get a "● " prefix
- Refreshes on page load, after every translation, and 500ms after the
user pauses typing in the input box
109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
import os
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
from fastapi import FastAPI, HTTPException
|
|
from fastapi.responses import FileResponse, StreamingResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
from pydantic import BaseModel
|
|
|
|
OLLAMA_URL = os.getenv("OLLAMA_URL", "http://radagast.jpnadas.xyz:11434")
|
|
# Comma-separated substrings; a model name matches if it contains any of them.
|
|
# Set to empty string to show every installed model.
|
|
MODEL_FILTER = os.getenv("MODEL_FILTER", "gemma,aya").lower()
|
|
|
|
STATIC_DIR = Path(__file__).parent / "static"
|
|
|
|
app = FastAPI(title="finrod")
|
|
|
|
|
|
class TranslateRequest(BaseModel):
|
|
text: str
|
|
source: str
|
|
target: str
|
|
model: str
|
|
|
|
|
|
@app.get("/")
|
|
async def index() -> FileResponse:
|
|
return FileResponse(STATIC_DIR / "index.html")
|
|
|
|
|
|
@app.get("/healthz")
|
|
async def healthz() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
|
|
@app.get("/api/models")
|
|
async def list_models() -> list[str]:
|
|
async with httpx.AsyncClient(timeout=10) as client:
|
|
try:
|
|
r = await client.get(f"{OLLAMA_URL}/api/tags")
|
|
r.raise_for_status()
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(502, f"Ollama unreachable: {e}") from e
|
|
names = [m["name"] for m in r.json().get("models", [])]
|
|
filters = [f.strip() for f in MODEL_FILTER.split(",") if f.strip()]
|
|
if filters:
|
|
names = [n for n in names if any(f in n.lower() for f in filters)]
|
|
return sorted(names)
|
|
|
|
|
|
@app.get("/api/loaded")
|
|
async def loaded_models() -> dict[str, list[dict[str, str | None]]]:
|
|
"""Models currently resident in Ollama's memory (from /api/ps)."""
|
|
async with httpx.AsyncClient(timeout=5) as client:
|
|
try:
|
|
r = await client.get(f"{OLLAMA_URL}/api/ps")
|
|
r.raise_for_status()
|
|
except httpx.HTTPError as e:
|
|
raise HTTPException(502, f"Ollama unreachable: {e}") from e
|
|
return {
|
|
"models": [
|
|
{"name": m["name"], "expires_at": m.get("expires_at")}
|
|
for m in r.json().get("models", [])
|
|
]
|
|
}
|
|
|
|
|
|
@app.post("/api/translate")
|
|
async def translate(req: TranslateRequest) -> StreamingResponse:
|
|
text = req.text.strip()
|
|
if not text:
|
|
raise HTTPException(400, "text is empty")
|
|
|
|
if req.source.lower() == "auto":
|
|
source_clause = "Detect the source language."
|
|
else:
|
|
source_clause = f"The source language is {req.source}."
|
|
|
|
prompt = (
|
|
f"You are a translation engine. {source_clause} "
|
|
f"Translate the text below into {req.target}. "
|
|
"Output ONLY the translation: no quotes, no explanations, no preamble, "
|
|
"no language labels. Preserve line breaks and formatting.\n\n"
|
|
f"Text:\n{text}"
|
|
)
|
|
|
|
async def relay():
|
|
async with httpx.AsyncClient(timeout=300) as client:
|
|
async with client.stream(
|
|
"POST",
|
|
f"{OLLAMA_URL}/api/generate",
|
|
json={
|
|
"model": req.model,
|
|
"prompt": prompt,
|
|
"stream": True,
|
|
"options": {"temperature": 0.2},
|
|
},
|
|
) as r:
|
|
r.raise_for_status()
|
|
async for line in r.aiter_lines():
|
|
if line:
|
|
yield line + "\n"
|
|
|
|
return StreamingResponse(relay(), media_type="application/x-ndjson")
|
|
|
|
|
|
app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
|