michael/AI-Profile-Router
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
0e4a9de5bac8230ce1ad7384dabcf03c48decddf
AI-Profile-Router/router/ai_profile_router.py
T

2151 lines
86 KiB
Python
Executable File
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python3
"""AI Profile Router – OpenAI-kompatibler Proxy vor llama.cpp.
Leitet OpenAI-kompatible Requests transparent an den lokalen llama.cpp-Server
weiter (Streaming, Tool Calls, JSON) und schaltet zwischen drei festen
Profilen um:
Profil Kontext
------ --------
fast 73728
medium 94208
long 131072
Virtuelle Modelle: qwen-fast, qwen-medium, qwen-long
Kommandos: POST /fast, /medium, /long (Profilwechsel)
GET /status (Zustand)
Bildgenerierung (FLUX.2 [klein] 4B Base):
POST /v1/images/generations (OpenAI-kompatibel)
GET /images (Liste)
GET /images/<datei> (PNG-Download)
Sprachausgabe (XTTS-v2, multilingual, CPU-only):
POST /v1/audio/speech (OpenAI-kompatibel)
GET /v1/audio/voices (verfügbare Stimmen)
Spracherkennung (whisper.cpp, deutsch, CPU-only):
POST /v1/audio/transcriptions (OpenAI-kompatibel)
GET /v1/audio/models (verfügbare Audio-Modelle)
Der TTS-Worker (mike-ai-xtts.service) und der STT-Worker
(mike-ai-whisper.service) laufen als separate, langlebige Prozesse.
Der Router leitet /v1/audio/speech und /v1/audio/transcriptions
per HTTP an die Worker weiter.
Der Router agiert als Modell-Orchestrator: vor der Generierung wird
llama.cpp gestoppt, der Bild-Worker lädt FLUX, generiert und entlädt
das Modell wieder; danach wird das vorherige Qwen-Profil wiederher-
gestellt und erst dann geantwortet (try/finally – Qwen wird auch bei
Fehlgeschlagener Generierung wiederhergestellt).
Vision-Orchestrierung (Q3 "Augen", temporär):
POST /v1/chat/completions mit Bild im letzten
User-Message → ein temporäres Q3-Vision-Modell
(llama-server + mmproj) wird geladen, analysiert
das Bild und wird wieder entladen; das Hauptprofil
wird immer wiederhergestellt (try/finally) und
erzeugt die Endantwort. Die Vision-Analyse bleibt
intern (kein sichtbarer Assistant-Turn).
Nur Python-Standardbibliothek. Logging nach stdout (journald).
"""
from __future__ import annotations
import base64
import email
import hashlib
import json
import logging
import os
import queue
import re
import subprocess
import sys
import threading
import time
import uuid
import http.client
from collections import OrderedDict
from email.parser import BytesParser
from email.policy import compat32
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
# ---------------------------------------------------------------------------
# Konfiguration (über Umgebungsvariablen, vgl. systemd-Unit)
# ---------------------------------------------------------------------------
HOST = os.environ.get("ROUTER_HOST", "0.0.0.0")
PORT = int(os.environ.get("ROUTER_PORT", "8081"))
UPSTREAM_URL = os.environ.get("UPSTREAM_URL", "http://127.0.0.1:8080").rstrip("/")
PROFILE_SCRIPT = os.environ.get("PROFILE_SCRIPT", "/usr/local/bin/llama-profile")
PROFILE_DIR = os.environ.get(
"PROFILE_DIR", "/etc/systemd/system/mike-ai-llama-ui.service.d")
SWITCH_TIMEOUT = float(os.environ.get("SWITCH_TIMEOUT", "600")) # s, Warten auf llama.cpp
REQUEST_TIMEOUT = float(os.environ.get("REQUEST_TIMEOUT", "600")) # s, Read-Timeout Upstream
CONNECT_TIMEOUT = float(os.environ.get("CONNECT_TIMEOUT", "10")) # s, Connect-Timeout
POLL_INTERVAL = float(os.environ.get("POLL_INTERVAL", "2")) # s, Polling-Intervall
# --- Bildgenerierung (FLUX.2 [klein] 4B Base) ---
LLAMA_SERVICE = os.environ.get("LLAMA_SERVICE", "mike-ai-llama-ui.service")
SYSTEMCTL_BIN = os.environ.get("SYSTEMCTL_BIN", "systemctl")
IMAGE_WORKER = os.environ.get(
"IMAGE_WORKER", "/opt/mike-ai/ai-profile-router/image_worker.py")
IMAGE_PYTHON = os.environ.get(
"IMAGE_PYTHON", "/opt/mike-ai/ai-profile-router/venv/bin/python")
IMAGE_DIR = os.environ.get(
"IMAGE_DIR", "/opt/mike-ai/ai-profile-router/images")
IMAGE_WORKER_LOG = os.environ.get(
"IMAGE_WORKER_LOG", "/opt/mike-ai/ai-profile-router/image_worker.log")
IMAGE_START_TIMEOUT = float(os.environ.get("IMAGE_START_TIMEOUT", "120")) # s, Worker-Start
IMAGE_GEN_TIMEOUT = float(os.environ.get("IMAGE_GEN_TIMEOUT", "1800")) # s, pro Bild
IMAGE_VRAM_FREE_TIMEOUT = float(os.environ.get("IMAGE_VRAM_FREE_TIMEOUT", "90")) # s, VRAM-Abgabe
# --- Vision-Orchestrierung (Q3 "Augen", temporär) ---
# Chat-Requests mit Bild im letzten User-Message lösen einen
# temporären Hotswap aus: Hauptprofil raus → Q3+mmproj rein →
# Bild analysieren → Q3 raus → Hauptprofil rein (try/finally).
LLAMA_SERVER_BIN = os.environ.get(
"LLAMA_SERVER_BIN", "/opt/mike-ai/llama.cpp/build/bin/llama-server")
VISION_MODEL = os.environ.get(
"VISION_MODEL", "/opt/mike-ai/models/qwen3.8-27b/Qwen3.8-27B-Q3_K_M.gguf")
VISION_MMPROJ = os.environ.get(
"VISION_MMPROJ", "/opt/mike-ai/models/qwen3.8-27b-nvfp4/mmproj-BF16.gguf")
VISION_CTX = int(os.environ.get("VISION_CTX", "32768"))
VISION_PORT = int(os.environ.get("VISION_PORT", "8086"))
VISION_ALIAS = "qwen38-27b-q3-vision"
VISION_LOAD_TIMEOUT = float(os.environ.get("VISION_LOAD_TIMEOUT", "300")) # s
VISION_INFER_TIMEOUT = float(os.environ.get("VISION_INFER_TIMEOUT", "300")) # s
VISION_UNLOAD_TIMEOUT = float(os.environ.get("VISION_UNLOAD_TIMEOUT", "120")) # s
VISION_MAX_TOKENS = int(os.environ.get("VISION_MAX_TOKENS", "4096"))
VISION_CACHE_MAX = int(os.environ.get("VISION_CACHE_MAX", "64"))
VISION_LOG = os.environ.get(
"VISION_LOG", "/opt/mike-ai/ai-profile-router/vision_server.log")
# Erlaubte Auflösungen (Breite x Höhe). FLUX.2 klein ist für 1 MP
# ausgelegt; 1920x1088 (≈2 MP) wird zusätzlich unterstützt.
IMAGE_SIZES = {
"1024x1024": (1024, 1024),
"1536x1024": (1536, 1024),
"1024x1536": (1024, 1536),
"1920x1088": (1920, 1088),
"1088x1920": (1088, 1920),
}
# Qualitätsstufen → Inference-Schritte (guidance bleibt offiziell 4.0).
# Auf der RTX 5080 gemessen: 30 vs. 50 Steps liefern praktisch dieselbe
# Qualität (1024x1024: 31,3 s vs. 45,3 s). Default ist daher "standard".
IMAGE_QUALITY = {"standard": 30, "high": 50}
IMAGE_DEFAULT_QUALITY = "standard"
IMAGE_MAX_N = 4
# --- Sprachausgabe (XTTS-v2, multilingual, CPU-only) ---
TTS_WORKER_URL = os.environ.get("TTS_WORKER_URL", "http://127.0.0.1:8085")
TTS_TIMEOUT = float(os.environ.get("TTS_TIMEOUT", "300")) # s, pro Synthese
TTS_CONNECT_TIMEOUT = float(os.environ.get("TTS_CONNECT_TIMEOUT", "5"))
TTS_MODEL = "xtts-v2" # virtuelles Modell für /v1/audio/speech
TTS_VOICES = ("claribel",)
TTS_DEFAULT_VOICE = "claribel"
TTS_FORMATS = ("mp3", "wav")
TTS_DEFAULT_FORMAT = "mp3"
# --- Spracherkennung (whisper.cpp, deutsch, CPU-only) ---
STT_WORKER_URL = os.environ.get("STT_WORKER_URL", "http://127.0.0.1:8084")
STT_TIMEOUT = float(os.environ.get("STT_TIMEOUT", "120")) # s, pro Transkription
STT_CONNECT_TIMEOUT = float(os.environ.get("STT_CONNECT_TIMEOUT", "5"))
STT_MODEL = "whisper-1" # virtuelles Modell für /v1/audio/transcriptions
# Maximale Upload-Größe (Bytes) – verhindert unbegrenzten RAM-Verbrauch.
# 50 MB ist für Audio-Dateien (WebM/Opus, WAV, MP3) mehr als ausreichend.
MAX_UPLOAD_SIZE = int(os.environ.get("MAX_UPLOAD_SIZE", 50 * 1024 * 1024))
# Chat-Waiting: Während eines Image-Jobs oder Profilwechsels ist Qwen
# down. Chat-Requests warten (statt 502) bis Qwen wieder bereit ist.
CHAT_WAIT_TIMEOUT = float(os.environ.get("CHAT_WAIT_TIMEOUT", "300")) # s, max. Warten
CHAT_DRAIN_TIMEOUT = float(os.environ.get("CHAT_DRAIN_TIMEOUT", "60")) # s, max. Warten auf aktive Chats
PROFILES = {"fast": 73728, "medium": 94208, "long": 131072}
VIRTUAL_MODELS = {f"qwen-{name}": name for name in PROFILES}
log = logging.getLogger("ai-profile-router")
# Hop-by-hop-Header, die nicht an Upstream/Client weitergereicht werden.
HOP_BY_HOP = {
"host", "connection", "keep-alive", "proxy-authenticate",
"proxy-authorization", "te", "trailer", "transfer-encoding",
"upgrade", "content-length",
}
def _parse_upstream(url: str) -> tuple[str, int]:
"""'http://127.0.0.1:8080' -> ('127.0.0.1', 8080)"""
hostport = url.split("://", 1)[-1]
host, _, port = hostport.partition(":")
return host, int(port) if port else 80
UPSTREAM_HOST, UPSTREAM_PORT = _parse_upstream(UPSTREAM_URL)
# ---------------------------------------------------------------------------
# Zustand
# ---------------------------------------------------------------------------
class _ImageState:
"""Zustand der Bildgenerierung (nur für Status-Reporting)."""
def __init__(self) -> None:
self.phase = "idle" # siehe PHASES unten
self.worker: "_Worker | None" = None
self.last_error: str | None = None
self.last_image: str | None = None
self.last_seconds: float | None = None
IMAGE_PHASES = (
"idle", "stopping-qwen", "loading-image", "generating",
"unloading-image", "restoring-qwen",
)
class _VisionState:
"""Zustand der Vision-Orchestrierung (Status-Reporting + Analyse-Cache)."""
def __init__(self) -> None:
self.phase = "idle" # siehe VISION_PHASES unten
self.last_error: str | None = None
self.last_turnaround: float | None = None # s, letzter kompletter Swap
self.last_profile: str | None = None # Profil, das gesichert wurde
# Cache: stabiler Bild-Hash → Vision-Analyse-Text. Folgefragen im
# selben Chat (Open WebUI schickt den multimodalen Verlauf erneut)
# senden das Bild NICHT erneut durch Q3, sondern verwenden die
# gecachte Analyse. LRU-begrenzt auf VISION_CACHE_MAX Einträge.
self.analysis_cache: "OrderedDict[str, str]" = OrderedDict()
self.cache_lock = threading.Lock()
VISION_PHASES = (
"idle", "stopping-main", "loading-vision", "analyzing",
"unloading-vision", "restoring-main",
)
class _State:
"""Gemeinsamer, thread-sicherer Zustand.
lock : zentraler GPU-/Model-Lock. Wird von Profilwechsel,
Image-Generation UND Vision-Swap gehalten →
gegenseitiger Ausschluss, kein Race zwischen beiden.
avail_lock : schützt qwen_unavailable + active_chats (Chat-Waiting).
"""
lock = threading.Lock() # GPU-/Model-Lock (Profilwechsel + Image + Vision)
switching: str | None = None # Profil, das gerade gewechselt wird
started = time.time()
image = _ImageState()
vision = _VisionState()
# Qwen-Verfügbarkeit für das Chat-Waiting:
qwen_unavailable = False # True, wenn Qwen down/neu geladen wird
active_chats = 0 # Anzahl laufender Chat-Requests
avail_lock = threading.Lock() # schützt die beiden Felder oben
STATE = _State()
def _wait_chats_drained(timeout: float | None = None) -> None:
"""Wartet, bis keine aktiven Chat-Requests mehr laufen.
Wird von Profilwechsel/Image-Job aufgerufen, BEVOR Qwen gestoppt wird.
Verhindert, dass ein laufender Chat auf ein gestopptes Qwen trifft (502).
"""
timeout = CHAT_DRAIN_TIMEOUT if timeout is None else timeout
deadline = time.monotonic() + timeout
while True:
with STATE.avail_lock:
if STATE.active_chats == 0:
return
n = STATE.active_chats
if time.monotonic() > deadline:
log.warning("Chat-Drain-Timeout nach %.0f s (%d aktive Chats) – "
"fahre trotzdem fort", timeout, n)
return
time.sleep(0.5)
def _set_qwen_unavailable(unavailable: bool) -> None:
with STATE.avail_lock:
STATE.qwen_unavailable = unavailable
# ---------------------------------------------------------------------------
# Upstream (llama.cpp)
# ---------------------------------------------------------------------------
def tts_status() -> dict:
"""Prüft den TTS-Worker: erreichbar? bereit? welche Stimmen?"""
hostport = TTS_WORKER_URL.split("://", 1)[-1]
host, _, port = hostport.partition(":")
try:
conn = http.client.HTTPConnection(host, int(port) if port else 80,
timeout=TTS_CONNECT_TIMEOUT)
conn.request("GET", "/status")
resp = conn.getresponse()
data = json.loads(resp.read())
conn.close()
return {"reachable": True, **data}
except (OSError, ValueError) as e:
return {"reachable": False, "error": str(e)}
def tts_synthesize(text: str, voice: str, speed: float,
fmt: str) -> tuple[bytes, str]:
"""Synthetisiert Audio über den TTS-Worker.
Liefert (audio_bytes, content_type). Wirft RuntimeError bei Fehler.
"""
hostport = TTS_WORKER_URL.split("://", 1)[-1]
host, _, port = hostport.partition(":")
payload = json.dumps({"text": text, "voice": voice,
"speed": speed, "format": fmt}).encode()
try:
conn = http.client.HTTPConnection(host, int(port) if port else 80,
timeout=TTS_CONNECT_TIMEOUT)
conn.request("POST", "/tts", body=payload,
headers={"Content-Type": "application/json"})
conn.sock.settimeout(TTS_TIMEOUT)
resp = conn.getresponse()
body = resp.read()
conn.close()
except (OSError, http.client.HTTPException) as e:
raise RuntimeError(f"TTS-Worker nicht erreichbar: {e}")
if resp.status != 200:
try:
err = json.loads(body)
msg = err.get("error", str(err))
except ValueError:
msg = body.decode(errors="replace")[:200]
raise RuntimeError(f"TTS-Fehler ({resp.status}): {msg}")
content_type = {"mp3": "audio/mpeg", "wav": "audio/wav",
"flac": "audio/flac",
"pcm": "application/octet-stream"}[fmt]
return body, content_type
def stt_status() -> dict:
"""Prüft den STT-Worker: erreichbar? bereit?"""
hostport = STT_WORKER_URL.split("://", 1)[-1]
host, _, port = hostport.partition(":")
try:
conn = http.client.HTTPConnection(host, int(port) if port else 80,
timeout=STT_CONNECT_TIMEOUT)
conn.request("GET", "/status")
resp = conn.getresponse()
data = json.loads(resp.read())
conn.close()
return {"reachable": True, **data}
except (OSError, ValueError) as e:
return {"reachable": False, "error": str(e)}
def stt_transcribe(file_data: bytes, filename: str,
language: str | None = None,
prompt: str | None = None,
temperature: float | None = None) -> dict:
"""Transkribiert Audio über den STT-Worker.
Liefert dict mit 'text'. Wirft RuntimeError bei Fehler.
"""
hostport = STT_WORKER_URL.split("://", 1)[-1]
host, _, port = hostport.partition(":")
# Multipart-Form-Data bauen
boundary = "----STTBoundary" + uuid.uuid4().hex[:16]
parts = []
parts.append(
f"--{boundary}\r\n"
f'Content-Disposition: form-data; name="file"; filename="{filename}"\r\n'
f"Content-Type: application/octet-stream\r\n\r\n".encode("utf-8")
)
parts.append(file_data)
parts.append(b"\r\n")
for key, value in [("language", language), ("prompt", prompt),
("temperature", temperature)]:
if value is not None:
parts.append(
f"--{boundary}\r\n"
f'Content-Disposition: form-data; name="{key}"\r\n\r\n'
f"{value}\r\n".encode("utf-8")
)
parts.append(f"--{boundary}--\r\n".encode("utf-8"))
body = b"".join(parts)
try:
conn = http.client.HTTPConnection(host, int(port) if port else 80,
timeout=STT_CONNECT_TIMEOUT)
conn.request("POST", "/transcribe", body=body,
headers={"Content-Type":
f"multipart/form-data; boundary={boundary}"})
conn.sock.settimeout(STT_TIMEOUT)
resp = conn.getresponse()
data = json.loads(resp.read())
conn.close()
except (OSError, http.client.HTTPException) as e:
raise RuntimeError(f"STT-Worker nicht erreichbar: {e}")
if resp.status != 200:
msg = data.get("error", str(data)) if isinstance(data, dict) else str(data)
raise RuntimeError(f"STT-Fehler ({resp.status}): {msg}")
return data
def upstream_status() -> dict:
"""Prüft llama.cpp: erreichbar? welches Modell? welcher Kontext?"""
try:
conn = http.client.HTTPConnection(UPSTREAM_HOST, UPSTREAM_PORT,
timeout=CONNECT_TIMEOUT)
conn.request("GET", "/v1/models")
resp = conn.getresponse()
data = json.loads(resp.read())
conn.close()
except (OSError, ValueError) as e:
return {"reachable": False, "error": str(e)}
models = data.get("data") or []
if not models:
return {"reachable": True, "model": None, "ctx": None}
m = models[0]
return {"reachable": True,
"model": m.get("id"),
"ctx": (m.get("meta") or {}).get("n_ctx")}
# ---------------------------------------------------------------------------
# Profile
# ---------------------------------------------------------------------------
def _read(path: str) -> str:
with open(path, encoding="utf-8") as f:
return f.read().strip()
def current_profile() -> str | None:
"""Aktives Profil, ermittelt durch Vergleich der override.conf."""
try:
override = _read(os.path.join(PROFILE_DIR, "override.conf"))
except OSError:
return None
for name in PROFILES:
try:
ref = _read(os.path.join(PROFILE_DIR, f"profile-{name}.conf.disabled"))
except OSError:
continue
if override == ref:
return name
return None
def _wait_ready(profile: str, deadline: float) -> None:
"""Wartet, bis llama.cpp das Profil geladen hat (Modell + ctx)."""
expected_ctx = PROFILES[profile]
while True:
status = upstream_status()
if (status["reachable"] and status.get("model")
and status.get("ctx") == expected_ctx):
log.info("llama.cpp bereit: Profil=%s Modell=%s ctx=%s",
profile, status.get("model"), status.get("ctx"))
return
if time.monotonic() > deadline:
raise RuntimeError(
f"llama.cpp nach {SWITCH_TIMEOUT:.0f} s nicht bereit "
f"(erwartet ctx {expected_ctx}, aktuell: {status.get('ctx')})")
time.sleep(POLL_INTERVAL)
def switch_profile(profile: str, implicit: bool = False) -> None:
"""Stellt sicher, dass das Profil aktiv ist, und wartet bis es geladen ist.
Wirft RuntimeError, wenn das Profil nicht aktiviert werden konnte.
implicit=True (ausgelöst durch ein virtuelles Modell in einem Chat-Request):
Wenn das Profil bereits aktiv ist, aber llama.cpp down ist, wird sofort
eine RuntimeError geworfen (kein stiller Neustart). Der Nutzer kann den
Neustart explizit über /<profil> anstoßen.
"""
if profile not in PROFILES:
raise ValueError(f"unbekanntes Profil: {profile!r} "
f"(erlaubt: {', '.join(PROFILES)})")
# Kein Fast-Fail: Wenn ein Image-Job läuft (hält den GPU-Lock), wartet
# der Profilwechsel auf den GPU-Lock (blockiert), bis der Image-Job
# fertig ist. So bekommen Chat-Requests kein 502, sondern warten.
with STATE.lock:
STATE.switching = profile
try:
cur = current_profile()
up = upstream_status()
ready = (up["reachable"] and up.get("model")
and up.get("ctx") == PROFILES[profile])
if cur == profile and ready:
log.info("Profil %s ist bereits aktiv", profile)
return
# Qwen wird neu geladen/gewechselt → für Chats nicht verfügbar.
_set_qwen_unavailable(True)
try:
_wait_chats_drained()
if cur == profile and up["reachable"] and not ready:
# Modell wird gerade geladen (z.B. nach einem Wechsel)
log.info("Warte, bis Profil %s geladen ist ...", profile)
_wait_ready(profile, time.monotonic() + SWITCH_TIMEOUT)
return
if cur == profile and not up["reachable"] and implicit:
raise RuntimeError(
f"llama.cpp nicht erreichbar (Profil {profile} ist "
f"bereits aktiv; Neustart über /{profile})")
log.info("Profilwechsel: %s -> %s", cur, profile)
try:
proc = subprocess.run(
[PROFILE_SCRIPT, profile],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
timeout=120,
)
out = proc.stdout.decode(errors="replace").strip()
if out:
log.info("llama-profile: %s", out[-500:])
if proc.returncode != 0:
# whiptail bricht das Skript ohne TTY ab – der Wechsel
# selbst (cp + systemctl restart) ist dann erledigt.
log.warning("llama-profile Exit-Code %d (ohne TTY "
"erwartet)", proc.returncode)
except subprocess.TimeoutExpired:
log.error("llama-profile hat 120 s überschritten")
if current_profile() != profile:
raise RuntimeError(
f"Profildatei wurde nicht gesetzt (erwartet: {profile})")
log.info("Warte, bis llama.cpp das Profil geladen hat ...")
_wait_ready(profile, time.monotonic() + SWITCH_TIMEOUT)
finally:
_set_qwen_unavailable(False)
finally:
STATE.switching = None
# ---------------------------------------------------------------------------
# Bildgenerierung (FLUX.2 [klein] 4B Base)
# ---------------------------------------------------------------------------
class _Worker:
"""Verwaltet den Bild-Worker-Prozess (stdin/stdout-JSON-Protokoll)."""
def __init__(self) -> None:
self.proc: subprocess.Popen | None = None
self.model_loaded = False
self._queue: queue.Queue[dict] = queue.Queue()
self._reader: threading.Thread | None = None
def alive(self) -> bool:
return self.proc is not None and self.proc.poll() is None
def start(self) -> None:
if self.alive():
return
log.info("starte Bild-Worker: %s %s", IMAGE_PYTHON, IMAGE_WORKER)
logf = open(IMAGE_WORKER_LOG, "ab")
self.proc = subprocess.Popen(
[IMAGE_PYTHON, IMAGE_WORKER],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=logf,
text=True,
bufsize=1,
)
self._reader = threading.Thread(target=self._read_loop, daemon=True)
self._reader.start()
try:
msg = self._queue.get(timeout=IMAGE_START_TIMEOUT)
except queue.Empty:
self.stop()
raise RuntimeError("Bild-Worker hat nicht gestartet")
if msg.get("status") != "ready":
self.stop()
raise RuntimeError(f"Bild-Worker-Startfehler: {msg}")
log.info("Bild-Worker bereit")
def _read_loop(self) -> None:
assert self.proc is not None and self.proc.stdout is not None
for line in self.proc.stdout:
line = line.strip()
if not line:
continue
try:
self._queue.put(json.loads(line))
except ValueError:
log.warning("Worker-Zeile (kein JSON): %s", line[:200])
def request(self, payload: dict, timeout: float) -> dict:
if not self.alive():
raise RuntimeError("Bild-Worker ist nicht aktiv")
assert self.proc is not None and self.proc.stdin is not None
self.proc.stdin.write(json.dumps(payload) + "\n")
self.proc.stdin.flush()
try:
return self._queue.get(timeout=timeout)
except queue.Empty:
raise RuntimeError(
f"Bild-Worker hat nach {timeout:.0f} s nicht geantwortet "
f"(cmd={payload.get('cmd')})")
def stop(self) -> None:
if self.proc is not None and self.proc.poll() is None:
self.proc.terminate()
try:
self.proc.wait(timeout=10)
except subprocess.TimeoutExpired:
self.proc.kill()
self.proc = None
self.model_loaded = False
def _worker() -> _Worker:
"""Worker-Instanz liefern (startet bei Bedarf)."""
img = STATE.image
if not img.worker or not img.worker.alive():
if img.worker:
img.worker.stop()
img.worker = _Worker()
img.worker.start()
return img.worker
def _wait_upstream_down(deadline: float) -> None:
"""Wartet, bis llama.cpp den Port freigegeben hat (VRAM frei)."""
while time.monotonic() < deadline:
if not upstream_status()["reachable"]:
return
time.sleep(1)
raise RuntimeError("llama.cpp gibt Port/VRAM nicht frei")
def _vram_used_mib() -> int | None:
"""Aktuelle VRAM-Belegung in MiB (via nvidia-smi), None bei Fehler."""
try:
out = subprocess.run(
["nvidia-smi", "--query-gpu=memory.used",
"--format=csv,noheader,nounits"],
stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, timeout=10,
).stdout.decode().strip()
return int(out.splitlines()[0].split()[0])
except (OSError, ValueError, IndexError):
return None
def _wait_vram_free(threshold_mib: int = 1000,
timeout: float | None = None) -> None:
"""Wartet, bis der VRAM unter threshold_mib fällt (FLUX entladen).
Wird nach dem Beenden des Bild-Workers aufgerufen, um sicherzustellen,
dass der VRAM (inkl. CUDA-Kontext) frei ist, bevor Qwen neu startet.
Wenn nvidia-smi nicht verfügbar ist (z.B. lokale Tests), wird der
Check übersprungen.
"""
timeout = IMAGE_VRAM_FREE_TIMEOUT if timeout is None else timeout
deadline = time.monotonic() + timeout
last = _vram_used_mib()
if last is None:
log.info("VRAM-Check übersprungen (nvidia-smi nicht verfügbar)")
return
while time.monotonic() < deadline:
if last <= threshold_mib:
log.info("VRAM frei: %d MiB", last)
return
time.sleep(1)
last = _vram_used_mib()
if last is None:
log.info("VRAM-Check übersprungen (nvidia-smi nicht verfügbar)")
return
raise RuntimeError(
f"VRAM nach {timeout:.0f} s nicht frei (letzte Messung: "
f"{last} MiB, erwartet <= {threshold_mib} MiB)")
def _restore_qwen(profile: str) -> None:
"""Startet llama.cpp mit dem gemerkten Profil und wartet auf Readiness."""
log.info("stelle Qwen-Profil %s wieder her ...", profile)
try:
subprocess.run([SYSTEMCTL_BIN, "start", LLAMA_SERVICE],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
timeout=120)
except subprocess.TimeoutExpired:
log.error("systemctl start hat 120 s überschritten")
_wait_ready(profile, time.monotonic() + SWITCH_TIMEOUT)
def generate_image(prompt: str, width: int, height: int, steps: int,
guidance: float, seed: int | None, n: int,
quality: str = "standard"
) -> tuple[list[str], str | None]:
"""Orchestriert die Bildgenerierung inkl. Qwen-Hotswap.
Hält den zentralen GPU-Lock (gegenseitiger Ausschluss mit Profilwechsel).
Ablauf: Qwen stoppen → Worker laden → generieren → Worker beenden
(VRAM + CUDA-Kontext frei) → Qwen wiederherstellen. Qwen wird auch bei
Fehlern wiederhergestellt (try/finally).
"""
img = STATE.image
with STATE.lock:
if img.phase != "idle":
raise RuntimeError(f"Bildgenerierung läuft ({img.phase})")
profile = current_profile()
if profile is None:
raise RuntimeError("kein aktives Qwen-Profil (override.conf?)")
os.makedirs(IMAGE_DIR, exist_ok=True)
results: list[str] = []
warning: str | None = None
# Qwen wird gestoppt → für Chats nicht verfügbar (die warten).
_set_qwen_unavailable(True)
try:
_wait_chats_drained()
# 1) Qwen stoppen (VRAM freigeben).
img.phase = "stopping-qwen"
subprocess.run([SYSTEMCTL_BIN, "stop", LLAMA_SERVICE],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
timeout=120)
_wait_upstream_down(time.monotonic() + 60)
# 2) Worker starten (Modell wird beim ersten generate geladen).
img.phase = "loading-image"
worker = _worker()
# 3) Generieren.
for i in range(n):
img.phase = "generating"
filename = time.strftime("%Y%m%d-%H%M%S") + \
f"-{os.urandom(2).hex()}.png"
output = os.path.join(IMAGE_DIR, filename)
resp = worker.request({
"cmd": "generate",
"prompt": prompt,
"width": width,
"height": height,
"steps": steps,
"guidance": guidance,
"seed": seed,
"output": output,
}, timeout=IMAGE_GEN_TIMEOUT)
if resp.get("status") != "ok":
raise RuntimeError(
resp.get("message", "Bildgenerierung fehlgeschlagen"))
worker.model_loaded = True
results.append(filename)
img.last_image = filename
img.last_seconds = resp.get("seconds")
# Metadaten speichern (Sidecar-JSON).
meta = {
"prompt": prompt,
"seed": seed,
"width": width,
"height": height,
"size": f"{width}x{height}",
"steps": steps,
"guidance": guidance,
"quality": quality,
"seconds": resp.get("seconds"),
"model": "FLUX.2-klein-base-4B",
"created": time.strftime("%Y-%m-%dT%H:%M:%S"),
}
meta_path = os.path.join(IMAGE_DIR, filename[:-4] + ".json")
try:
with open(meta_path, "w", encoding="utf-8") as f:
json.dump(meta, f, ensure_ascii=False, indent=2)
except OSError as e:
log.warning("Metadaten-Speicherung fehlgeschlagen: %s", e)
log.info("Bild %d/%d: %s (%.1f s)", i + 1, n, filename,
resp.get("seconds", 0))
# 4) Worker vollständig beenden (VRAM + CUDA-Kontext freigeben).
img.phase = "unloading-image"
worker.stop()
img.worker = None
try:
_wait_vram_free()
except RuntimeError as e:
log.warning("VRAM-Check: %s (fahre mit Qwen-Restore fort)", e)
except Exception as e:
img.last_error = str(e)
log.error("Bildgenerierung fehlgeschlagen: %s", e)
# Worker sicher beenden (falls noch aktiv), VRAM freigeben.
if img.worker is not None:
img.worker.stop()
img.worker = None
raise
finally:
# 5) Qwen immer wiederherstellen.
img.phase = "restoring-qwen"
try:
_restore_qwen(profile)
_set_qwen_unavailable(False)
except Exception as e:
warning = f"Qwen-Wiederherstellung fehlgeschlagen: {e}"
img.last_error = warning
log.error(warning)
# Qwen ist down → qwen_unavailable bleibt True.
img.phase = "idle"
return results, warning
def _image_filename_ok(name: str) -> bool:
return bool(re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]*\.png", name))
# ---------------------------------------------------------------------------
# Vision-Orchestrierung (Q3 "Augen", temporär)
# ---------------------------------------------------------------------------
VISION_ANALYST_PROMPT = (
"Du bist ein reiner Bild- und Screenshot-Analyst. Du beantwortest die "
"Benutzerfrage NICHT selbst. Du extrahierst aus dem Bild alle "
"Informationen, die für die Beantwortung relevant sein könnten.\n\n"
"Antworte NUR mit einer strukturierten Analyse in dieser Form:\n"
"1. SIEHTBARER TEXT: alle Texte wörtlich und vollständig, mit Anordnung\n"
"2. UI-ELEMENTE: Felder, Buttons, Menüs, Tabs, Dropdowns, Checkboxen – "
"mit Namen, Werten und Zustand (aktiv/inaktiv, gefüllt/leer, ausgewählt)\n"
"3. FEHLER- UND WARNMELDUNGEN: wörtlich, mit Farbe und Position\n"
"4. POSITIONEN UND BEZIEHUNGEN: räumliche Anordnung (oben/unten, "
"links/rechts, Reihenfolge)\n"
"5. ZUSTÄNDE: Statusanzeigen, Farben (rot/grün/gelb), Ladezustände\n"
"6. OBJEKTE: relevante Objekte und Beziehungen zwischen Elementen\n"
"7. WEITERES: alles Weitere, was für die Benutzerfrage relevant sein "
"könnte\n\n"
"Regeln: Bei Screenshots hat Text- und UI-Genauigkeit Vorrang vor "
"schöner Beschreibung. Keine Interpretation, keine Vermutungen – nur "
"was sichtbar ist. Unleserliches als [unleserlich] markieren."
)
IMAGE_PLACEHOLDER = "[Bild angehänggt – siehe Vision-Analyse]"
class _VisionServer:
"""Temporärer llama-server (Q3 + mmproj) für die Bildanalyse."""
def __init__(self) -> None:
self.proc: subprocess.Popen | None = None
self._logf = None
def alive(self) -> bool:
return self.proc is not None and self.proc.poll() is None
def start(self) -> None:
if self.alive():
return
cmd = [
LLAMA_SERVER_BIN,
"--model", VISION_MODEL,
"--mmproj", VISION_MMPROJ,
"--alias", VISION_ALIAS,
"--ctx-size", str(VISION_CTX),
"--flash-attn", "on",
"--cache-type-k", "q4_0",
"--cache-type-v", "q4_0",
"--threads", "6",
"--threads-batch", "6",
"--batch-size", "64",
"--ubatch-size", "32",
"--parallel", "1",
"--jinja",
"--host", "127.0.0.1",
"--port", str(VISION_PORT),
"--metrics",
"--fit", "off",
"--n-gpu-layers", "all",
"--mmproj-offload",
"--no-mmap",
"--temperature", "0.2",
"--top-p", "0.8",
"--top-k", "20",
"--device", "CUDA0",
"--split-mode", "none",
]
self._logf = open(VISION_LOG, "ab")
self.proc = subprocess.Popen(
cmd, stdin=subprocess.DEVNULL,
stdout=self._logf, stderr=subprocess.STDOUT)
log.info("Vision-Server gestartet (PID %d, Port %d, ctx %d)",
self.proc.pid, VISION_PORT, VISION_CTX)
def wait_ready(self, deadline: float) -> None:
"""Wartet, bis der Vision-Server das Modell mit erwartetem ctx meldet."""
while True:
try:
conn = http.client.HTTPConnection("127.0.0.1", VISION_PORT,
timeout=3)
conn.request("GET", "/v1/models")
resp = conn.getresponse()
data = json.loads(resp.read())
conn.close()
models = data.get("data") or []
if models and (models[0].get("meta") or {}).get("n_ctx") == VISION_CTX:
return
except (OSError, ValueError):
pass
if not self.alive():
raise RuntimeError(
"Vision-Server-Prozess beendet sich während des Ladens "
f"(Details: {VISION_LOG})")
if time.monotonic() > deadline:
raise RuntimeError(
f"Vision-Server nach {VISION_LOAD_TIMEOUT:.0f} s nicht "
f"bereit (Details: {VISION_LOG})")
time.sleep(2)
def stop(self) -> None:
if self.proc is not None and self.proc.poll() is None:
self.proc.terminate()
try:
self.proc.wait(timeout=30)
except subprocess.TimeoutExpired:
log.warning("Vision-Server reagiert nicht auf SIGTERM – SIGKILL")
self.proc.kill()
try:
self.proc.wait(timeout=10)
except subprocess.TimeoutExpired:
pass
if self._logf is not None:
try:
self._logf.close()
except OSError:
pass
self._logf = None
self.proc = None
def _extract_last_user_image(data: dict) -> tuple[str | None, str]:
"""Liefert (image_url, question) aus der letzten User-Message.
Nur Bilder in der LETZTEN User-Message lösen eine Vision-Analyse aus.
Bilder in früheren Nachrichten sind bereits durch die vorherige
Antwort abgedeckt (Chat-Historie) und werden nur durch einen
Platzhalter ersetzt.
"""
messages = data.get("messages")
if not isinstance(messages, list):
return None, ""
for msg in reversed(messages):
if not isinstance(msg, dict) or msg.get("role") != "user":
continue
content = msg.get("content")
if isinstance(content, str):
return None, content
if isinstance(content, list):
image_url: str | None = None
texts: list[str] = []
for part in content:
if not isinstance(part, dict):
continue
if part.get("type") == "image_url":
iu = part.get("image_url")
url = iu.get("url") if isinstance(iu, dict) else iu
if isinstance(url, str) and url:
image_url = url
elif (part.get("type") == "text"
and isinstance(part.get("text"), str)):
texts.append(part["text"])
question = " ".join(t.strip() for t in texts if t.strip())
return image_url, question
return None, ""
def _image_hash(image_url: str) -> str:
"""Stabiler Hash für ein Bild (Base64-Payload oder URL).
Dasselbe Bild → derselbe Hash (unabhängig von Chat/Turn). Dient als
Schlüssel für den Vision-Analyse-Cache.
"""
if image_url.startswith("data:"):
payload = image_url.split(",", 1)[1] if "," in image_url else ""
try:
raw = base64.b64decode(payload)
return "img:" + hashlib.sha256(raw).hexdigest()[:32]
except (ValueError, TypeError):
return "b64:" + hashlib.sha256(payload.encode()).hexdigest()[:32]
return "url:" + hashlib.sha256(image_url.encode()).hexdigest()[:32]
def _vision_cached_analysis(img_hash: str) -> str | None:
"""Liefert die gecachte Vision-Analyse für einen Bild-Hash (oder None)."""
with STATE.vision.cache_lock:
return STATE.vision.analysis_cache.get(img_hash)
def _vision_store_analysis(img_hash: str, analysis: str) -> None:
"""Speichert eine Vision-Analyse im Cache (LRU-begrenzt)."""
with STATE.vision.cache_lock:
STATE.vision.analysis_cache[img_hash] = analysis
STATE.vision.analysis_cache.move_to_end(img_hash)
while len(STATE.vision.analysis_cache) > VISION_CACHE_MAX:
STATE.vision.analysis_cache.popitem(last=False)
def _request_has_image(data: dict) -> bool:
"""True, wenn irgendwo im Request ein image_url-Part vorkommt."""
messages = data.get("messages")
if not isinstance(messages, list):
return False
for msg in messages:
if not isinstance(msg, dict):
continue
content = msg.get("content")
if isinstance(content, list):
for part in content:
if isinstance(part, dict) and part.get("type") == "image_url":
return True
return False
def _sanitize_for_main_model(data: dict) -> dict:
"""Erzeugt eine sanisierte Kopie des Requests für das Hauptmodell.
Das Hauptmodell (Fast/Medium/Long) hat KEIN Vision-Modell. Deshalb
werden alle Bild-Parts (image_url) durch den zu diesem Bild erzeugten
Vision-Analyse-Text (aus dem Cache) ersetzt:
- Bilddaten (Base64/URL) werden vollständig entfernt.
- Der ursprüngliche Text des Users bleibt erhalten.
- Die Analyse wird klar gekennzeichnet in denselben Turn eingesetzt.
Beispiel:
user: [image_url, "Was ist hier falsch?"]
→ user: "Was ist hier falsch?\n\n[Vision-Analyse des hochgeladenen
Bildes: ...]"
Wird bei JEDER Folgefrage erneut angewendet, weil Open WebUI den
ursprünglichen multimodalen Verlauf wieder mitsendet.
"""
out = json.loads(json.dumps(data))
messages = out.get("messages")
if not isinstance(messages, list):
return out
for msg in messages:
if not isinstance(msg, dict):
continue
content = msg.get("content")
if not isinstance(content, list):
continue
texts: list[str] = []
had_image = False
for part in content:
if isinstance(part, dict) and part.get("type") == "image_url":
had_image = True
iu = part.get("image_url")
url = iu.get("url") if isinstance(iu, dict) else iu
analysis = None
if isinstance(url, str) and url:
analysis = _vision_cached_analysis(_image_hash(url))
if analysis:
texts.append("[Vision-Analyse des hochgeladenen Bildes: "
+ analysis + "]")
else:
texts.append(IMAGE_PLACEHOLDER)
elif (isinstance(part, dict) and part.get("type") == "text"
and isinstance(part.get("text"), str)):
texts.append(part["text"])
# andere Part-Typen (z. B. audio) werden verworfen
if had_image:
msg["content"] = "\n\n".join(t for t in texts if t.strip())
return out
def _vision_analyze(image_url: str, question: str) -> str:
"""Sendet Bild + Frage an den Vision-Server, liefert den Analysen-Text."""
payload = {
"model": VISION_ALIAS,
"messages": [
{"role": "system", "content": VISION_ANALYST_PROMPT},
{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": image_url}},
{"type": "text",
"text": ("Benutzerfrage (nur zur Orientierung, NICHT "
"beantworten: " + (question or "(keine Frage)")
+ "\n\nErstelle jetzt die strukturierte Vision-Analyse.")},
]},
],
"max_tokens": VISION_MAX_TOKENS,
"temperature": 0.1,
"stream": False,
}
body = json.dumps(payload).encode()
# timeout=VISION_INFER_TIMEOUT gilt für Connect UND Read. (Nicht
# conn.sock.settimeout() – vor connect() ist conn.sock noch None.)
conn = http.client.HTTPConnection("127.0.0.1", VISION_PORT,
timeout=VISION_INFER_TIMEOUT)
conn.request("POST", "/v1/chat/completions", body=body,
headers={"Content-Type": "application/json"})
resp = conn.getresponse()
raw = resp.read()
conn.close()
try:
data = json.loads(raw)
except ValueError:
raise RuntimeError(
f"Vision-Inferenz: ungültige Antwort (HTTP {resp.status})")
if resp.status != 200:
msg = (data.get("error") or {}).get("message", str(data)) \
if isinstance(data, dict) else str(data)
raise RuntimeError(f"Vision-Inferenz fehlgeschlagen ({resp.status}): {msg}")
choices = data.get("choices") or []
if not choices:
raise RuntimeError("Vision-Inferenz: leere Antwort")
content = (choices[0].get("message") or {}).get("content")
if not isinstance(content, str) or not content.strip():
raise RuntimeError("Vision-Inferenz: leere Analyse")
return content.strip()
def _vision_swap(data: dict) -> str:
"""Kompletter Vision-Hotswap (Q3 "Augen").
Hält den zentralen GPU-Lock (gegenseitiger Ausschluss mit
Profilwechsel und FLUX). Normaler Ablauf (explizit, Schritt für
Schritt – die Wiederherstellung des Hauptprofils erfolgt erst NACH
abgeschlossener Vision-Inferenz):
1. Hauptprofil entladen (VRAM freigeben)
2. Q3-Vision-Server starten und auf Ready warten
3. Bild direkt an 127.0.0.1:VISION_PORT analysieren (Inferenz)
4. Q3-Vision-Server stoppen
5. Hauptprofil wiederherstellen
finally dient NUR als Fehler-/Cleanup-Sicherung (Q3 stoppen,
Hauptprofil retten, Phase zurücksetzen, Timing loggen) – nicht als
normaler Ablauf.
Liefert den Analysen-Text. Die Analyse wird zusätzlich im
Vision-Cache (keyed by Bild-Hash) gespeichert, damit Folgefragen das
Bild nicht erneut durch Q3 schicken (siehe _sanitize_for_main_model).
"""
vis = STATE.vision
profile = current_profile()
if profile is None:
raise RuntimeError("kein aktives Qwen-Profil (override.conf?)")
image_url, question = _extract_last_user_image(data)
if not image_url:
raise RuntimeError("kein Bild im Request")
img_hash = _image_hash(image_url)
vis.last_profile = profile
vis.last_error = None
t_total = time.monotonic()
timings: dict[str, float] = {}
with STATE.lock:
if vis.phase != "idle":
raise RuntimeError(f"Vision-Analyse läuft ({vis.phase})")
# Qwen wird gestoppt → für Chats nicht verfügbar (die warten).
_set_qwen_unavailable(True)
vision = _VisionServer()
analysis: str | None = None
main_restored = False
try:
_wait_chats_drained()
# 1) Hauptprofil entladen (VRAM freigeben).
vis.phase = "stopping-main"
t0 = time.monotonic()
subprocess.run([SYSTEMCTL_BIN, "stop", LLAMA_SERVICE],
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
timeout=120)
_wait_upstream_down(time.monotonic() + 60)
timings["main_unload"] = time.monotonic() - t0
log.info("Vision: Hauptprofil %s entladen (%.1f s)",
profile, timings["main_unload"])
# 2) Q3-Vision-Server starten und auf Ready warten.
vis.phase = "loading-vision"
t0 = time.monotonic()
vision.start()
vision.wait_ready(time.monotonic() + VISION_LOAD_TIMEOUT)
timings["vision_load"] = time.monotonic() - t0
log.info("Vision: Q3-Vision-Modell geladen (%.1f s)",
timings["vision_load"])
# 3) Bild direkt an den Vision-Server analysieren.
vis.phase = "analyzing"
t0 = time.monotonic()
log.info("Vision: Inferenz gestartet (127.0.0.1:%d)", VISION_PORT)
analysis = _vision_analyze(image_url, question)
timings["vision_infer"] = time.monotonic() - t0
log.info("Vision: Inferenz abgeschlossen (%.1f s, %d Zeichen)",
timings["vision_infer"], len(analysis))
_vision_store_analysis(img_hash, analysis)
log.info("Vision: Analyse im Cache gespeichert (Hash %s…)",
img_hash[:16])
# 4) Q3-Vision-Server stoppen.
vis.phase = "unloading-vision"
t0 = time.monotonic()
vision.stop()
try:
_wait_vram_free(timeout=VISION_UNLOAD_TIMEOUT)
except RuntimeError as e:
log.warning("Vision VRAM-Check: %s (fahre mit Restore fort)", e)
timings["vision_unload"] = time.monotonic() - t0
log.info("Vision: Q3 gestoppt (%.1f s)", timings["vision_unload"])
# 5) Hauptprofil wiederherstellen (erst NACH der Inferenz).
vis.phase = "restoring-main"
t0 = time.monotonic()
log.info("Vision: Hauptprofil %s wird wiederhergestellt", profile)
_restore_qwen(profile)
timings["main_restore"] = time.monotonic() - t0
log.info("Vision: Hauptprofil %s wiederhergestellt (%.1f s)",
profile, timings["main_restore"])
main_restored = True
_set_qwen_unavailable(False)
except Exception as e:
# Fehler-/Cleanup-Pfad: Q3 stoppen, Hauptprofil retten.
log.error("Vision-Fehler in Phase %s: %s", vis.phase, e)
vis.last_error = str(e)
if vision.alive():
try:
vision.stop()
log.info("Vision: Q3 gestoppt (Cleanup nach Fehler)")
except Exception:
log.exception("Vision: Q3-Cleanup fehlgeschlagen")
if not main_restored:
try:
_restore_qwen(profile)
_set_qwen_unavailable(False)
log.info("Vision: Hauptprofil %s wiederhergestellt "
"(Cleanup nach Fehler)", profile)
except Exception as e2:
vis.last_error = (f"Qwen-Wiederherstellung "
f"fehlgeschlagen: {e2}")
log.error("Vision: %s", vis.last_error)
# qwen_unavailable bleibt True (Qwen ist down).
raise
finally:
# NUR Cleanup: Phase zurücksetzen, Timing loggen.
vis.phase = "idle"
vis.last_turnaround = round(time.monotonic() - t_total, 1)
log.info("Vision-Timing: %s | Gesamt %.1f s",
" ".join(f"{k}={v:.1f}s" for k, v in timings.items()),
vis.last_turnaround)
if analysis is None:
# Defensive Absicherung (der except-Pfad wirft immer weiter).
raise RuntimeError(
f"Vision-Analyse fehlgeschlagen: "
f"{vis.last_error or 'unbekannter Fehler'}")
return analysis
# ---------------------------------------------------------------------------
# HTTP-Handler
# ---------------------------------------------------------------------------
class Handler(BaseHTTPRequestHandler):
server_version = "AIProfileRouter/1.0"
timeout = 60 # Socket-Timeout für Client-Requests (s)
# ---------- Routing ----------
def do_GET(self):
self._route()
def do_POST(self):
self._route()
def _route(self):
path = self.path.split("?", 1)[0]
started = time.monotonic()
try:
if path == "/v1/models" and self.command == "GET":
self._send_json(200, self._models_payload())
elif path == "/status":
self._send_json(200, self._status_payload())
elif path == "/v1/audio/models" and self.command == "GET":
self._send_json(200, self._audio_models_payload())
elif path == "/v1/audio/voices" and self.command == "GET":
self._send_json(200, self._audio_voices_payload())
elif path == "/v1/images/generations" and self.command == "POST":
self._image_generate()
elif path == "/v1/audio/speech" and self.command == "POST":
self._speech()
elif path == "/v1/audio/transcriptions" and self.command == "POST":
self._transcribe()
elif path == "/vision/test" and self.command == "POST":
self._vision_test()
elif path == "/images" and self.command == "GET":
self._images_list()
elif path.startswith("/images/") and self.command == "GET":
self._image_serve(path[len("/images/"):])
elif path in ("/fast", "/medium", "/long"):
self._switch(path[1:])
elif (self.command == "POST" and path.startswith("/")
and path.count("/") == 1):
# Kommandonamensraum: unbekanntes Profil
self._send_error(400, f"unbekanntes Profil: {path[1:]}",
"invalid_request_error", "invalid_profile")
else:
self._forward()
except BrokenPipeError:
log.warning("Client getrennt: %s %s", self.command, path)
except Exception:
log.exception("Fehler bei %s %s", self.command, path)
self._safe_error(500, "interner Router-Fehler")
finally:
log.info("%s %s -> %s in %.3f s", self.command, path,
getattr(self, "_last_code", "-"), time.monotonic() - started)
# ---------- Request-Body-Lesen (Content-Length + chunked) ----------
def _read_body(self) -> bytes:
"""Liest den HTTP-Request-Body (Content-Length oder chunked).
Liefert die Body-Bytes. Wirft ValueError bei:
- malformed chunked encoding
- Upload größer als MAX_UPLOAD_SIZE
- unvollständiger Body
"""
te = self.headers.get("Transfer-Encoding", "").lower()
if "chunked" in te:
return self._read_chunked_body()
length = int(self.headers.get("Content-Length") or 0)
if length > MAX_UPLOAD_SIZE:
raise ValueError(
f"Upload zu groß: {length} bytes (max {MAX_UPLOAD_SIZE})")
if length == 0:
return b""
data = self.rfile.read(length)
if len(data) != length:
raise ValueError(
f"Unvollständiger Body: {len(data)}/{length} bytes")
return data
def _read_chunked_body(self) -> bytes:
"""Liest und dekodiert einen HTTP/1.1 chunked-Transfer-Encoding Body.
RFC 7230 §4.1:
chunked-body = *chunk last-chunk trailer-part CRLF
chunk = chunk-size [chunk-ext] CRLF chunk-data CRLF
chunk-size = 1*HEXDIG
last-chunk = 0 [chunk-ext] CRLF
trailer-part = *( field-line CRLF )
- Chunk-Größen werden hexadezimal geparst.
- Chunk Extensions (nach ';') werden toleriert/ignoriert.
- 0-Chunk markiert das Ende.
- Trailer werden konsumiert und ignoriert.
- MAX_UPLOAD_SIZE wird durchgesetzt.
"""
chunks: list[bytes] = []
total_size = 0
while True:
# Chunk-Size-zeile lesen: "hex-size [chunk-ext] CRLF"
size_line = self.rfile.readline(65537)
if not size_line:
raise ValueError("Chunked Body: unerwartetes Ende")
# CRLF/LF entfernen
size_line = size_line.rstrip(b"\r\n")
# Chunk Extension entfernen (alles nach dem ersten ';')
if b";" in size_line:
size_line = size_line.split(b";", 1)[0]
# Hexadezimale Größe parsen
size_str = size_line.strip()
if not size_str:
raise ValueError("Chunked Body: leere Chunk-Size")
try:
chunk_size = int(size_str, 16)
except ValueError:
raise ValueError(
f"Malformed Chunk-Size: {size_str!r}")
# 0-Chunk = Ende des chunked-body
if chunk_size == 0:
break
# Uploadgrößenlimit prüfen
total_size += chunk_size
if total_size > MAX_UPLOAD_SIZE:
raise ValueError(
f"Upload zu groß: {total_size} bytes "
f"(max {MAX_UPLOAD_SIZE})")
# Chunk-Daten lesen
chunk_data = self.rfile.read(chunk_size)
if len(chunk_data) != chunk_size:
raise ValueError(
f"Unvollständiges Chunk: {len(chunk_data)}/{chunk_size} bytes")
chunks.append(chunk_data)
# CRLF nach Chunk-Daten lesen
crlf = self.rfile.read(2)
if crlf != b"\r\n":
raise ValueError(
f"Erwartet CRLF nach Chunk, erhalten: {crlf!r}")
# Trailer lesen und ignorieren
# trailer-part = *( field-line CRLF ), beendet durch leere Zeile
while True:
line = self.rfile.readline(65537)
if not line or line in (b"\r\n", b"\n"):
break
# Trailer-Header ignorieren
return b"".join(chunks)
# ---------- Router-eigene Endpunkte ----------
@staticmethod
def _models_payload() -> dict:
return {
"object": "list",
"data": [
{
"id": f"qwen-{name}",
"object": "model",
"created": 0,
"owned_by": "ai-profile-router",
"context_length": ctx,
"context_window": ctx,
}
for name, ctx in PROFILES.items()
],
}
def _status_payload(self) -> dict:
up = upstream_status()
img = STATE.image
with STATE.avail_lock:
qwen_unavailable = STATE.qwen_unavailable
active_chats = STATE.active_chats
return {
"router": "ai-profile-router",
"uptime_seconds": round(time.time() - STATE.started, 1),
"current_profile": current_profile(),
"switching": STATE.switching,
"profiles": PROFILES,
"upstream": {
"url": UPSTREAM_URL,
"reachable": up["reachable"],
"model": up.get("model"),
"ctx": up.get("ctx"),
},
"qwen": {
"available": not qwen_unavailable,
"active_chats": active_chats,
},
"image": {
"phase": img.phase,
"worker": "running" if (img.worker and img.worker.alive())
else "stopped",
"model_loaded": bool(img.worker and img.worker.model_loaded),
"last_image": img.last_image,
"last_seconds": img.last_seconds,
"last_error": img.last_error,
},
"vision": {
"phase": STATE.vision.phase,
"last_error": STATE.vision.last_error,
"last_profile": STATE.vision.last_profile,
"last_turnaround_seconds": STATE.vision.last_turnaround,
"analysis_cache_size": len(STATE.vision.analysis_cache),
},
"tts": tts_status(),
"stt": stt_status(),
}
# ---------- Bildgenerierung ----------
def _image_generate(self) -> None:
try:
body = self._read_body()
except ValueError as e:
self._send_error(400, str(e),
"invalid_request_error", "invalid_body")
return
try:
data = json.loads(body)
except ValueError:
self._send_error(400, "ungültiges JSON",
"invalid_request_error", "invalid_json")
return
if not isinstance(data, dict):
self._send_error(400, "Request muss ein JSON-Objekt sein",
"invalid_request_error", "invalid_request")
return
prompt = data.get("prompt")
if not isinstance(prompt, str) or not prompt.strip():
self._send_error(400, "'prompt' fehlt oder ist leer",
"invalid_request_error", "missing_prompt")
return
if len(prompt) > 8000:
self._send_error(400, "'prompt' zu lang (max 8000 Zeichen)",
"invalid_request_error", "prompt_too_long")
return
# Größe
size = data.get("size", "1024x1024")
if size not in IMAGE_SIZES:
self._send_error(
400, f"ungültige Größe: {size!r} "
f"(erlaubt: {', '.join(IMAGE_SIZES)})",
"invalid_request_error", "invalid_size")
return
width, height = IMAGE_SIZES[size]
# Anzahl
n = data.get("n", 1)
if not isinstance(n, int) or isinstance(n, bool) or not 1 <= n <= IMAGE_MAX_N:
self._send_error(400, f"'n' muss eine Ganzzahl 1..{IMAGE_MAX_N} sein",
"invalid_request_error", "invalid_n")
return
# Qualität / Schritte / Guidance
quality = data.get("quality", IMAGE_DEFAULT_QUALITY)
if quality not in IMAGE_QUALITY:
self._send_error(400, f"ungültige Qualität: {quality!r} "
f"(erlaubt: {', '.join(IMAGE_QUALITY)})",
"invalid_request_error", "invalid_quality")
return
steps = data.get("steps", IMAGE_QUALITY[quality])
if not isinstance(steps, int) or isinstance(steps, bool) or not 4 <= steps <= 150:
self._send_error(400, "'steps' muss eine Ganzzahl 4..150 sein",
"invalid_request_error", "invalid_steps")
return
guidance = data.get("guidance", 4.0)
try:
guidance = float(guidance)
except (TypeError, ValueError):
self._send_error(400, "'guidance' muss eine Zahl sein",
"invalid_request_error", "invalid_guidance")
return
if not 1.0 <= guidance <= 10.0:
self._send_error(400, "'guidance' muss zwischen 1.0 und 10.0 sein",
"invalid_request_error", "invalid_guidance")
return
seed = data.get("seed")
if seed is not None:
try:
seed = int(seed)
except (TypeError, ValueError):
self._send_error(400, "'seed' muss eine Ganzzahl sein",
"invalid_request_error", "invalid_seed")
return
if not 0 <= seed <= 2**32 - 1:
self._send_error(400, "'seed' muss zwischen 0 und 4294967295 sein",
"invalid_request_error", "invalid_seed")
return
response_format = data.get("response_format", "url")
if response_format not in ("url", "b64_json"):
self._send_error(400, "'response_format' muss 'url' oder 'b64_json' sein",
"invalid_request_error", "invalid_response_format")
return
# Generierung (blockt mehrere Minuten – eigener Thread-Timeout).
self.timeout = None
try:
results, warning = generate_image(
prompt.strip(), width, height, steps, guidance, seed, n,
quality)
except (ValueError, RuntimeError) as e:
self._send_error(503, str(e), "server_error", "image_generation_failed")
return
# Antwort bauen
host = self.headers.get("Host") or f"{HOST}:{PORT}"
if not host.startswith(("http://", "https://")):
host = f"http://{host}"
items = []
for filename in results:
path = os.path.join(IMAGE_DIR, filename)
item: dict = {"url": f"{host}/images/{filename}", "b64_json": None}
if response_format == "b64_json":
with open(path, "rb") as f:
item["b64_json"] = base64.b64encode(f.read()).decode()
item["url"] = None
items.append(item)
payload: dict = {"created": int(time.time()), "data": items}
if warning:
payload["router_warning"] = warning
self._send_json(200, payload)
def _images_list(self) -> None:
if not os.path.isdir(IMAGE_DIR):
self._send_json(200, {"images": []})
return
entries = []
for name in sorted(os.listdir(IMAGE_DIR), reverse=True):
if not _image_filename_ok(name):
continue
path = os.path.join(IMAGE_DIR, name)
try:
st = os.stat(path)
except OSError:
continue
entry = {
"name": name,
"url": f"/images/{name}",
"bytes": st.st_size,
"modified": int(st.st_mtime),
}
# Metadaten laden (Sidecar-JSON, falls vorhanden).
meta_path = os.path.join(IMAGE_DIR, name[:-4] + ".json")
if os.path.isfile(meta_path):
try:
with open(meta_path, encoding="utf-8") as f:
entry["meta"] = json.load(f)
except (OSError, ValueError):
pass
entries.append(entry)
self._send_json(200, {"images": entries[:200]})
def _image_serve(self, name: str) -> None:
if not _image_filename_ok(name):
self._send_error(400, "ungültiger Dateiname",
"invalid_request_error", "invalid_filename")
return
path = os.path.join(IMAGE_DIR, name)
if not os.path.isfile(path):
self._send_error(404, "Bild nicht gefunden",
"invalid_request_error", "not_found")
return
data = open(path, "rb").read()
self._last_code = 200
self.send_response(200)
self.send_header("Content-Type", "image/png")
self.send_header("Content-Length", str(len(data)))
self.send_header("Cache-Control", "public, max-age=86400")
self.send_header("Connection", "close")
self.end_headers()
self.wfile.write(data)
# ---------- Sprachausgabe (XTTS-v2) ----------
def _speech(self) -> None:
try:
body = self._read_body()
except ValueError as e:
self._send_error(400, str(e),
"invalid_request_error", "invalid_body")
return
try:
data = json.loads(body)
except ValueError:
self._send_error(400, "ungültiges JSON",
"invalid_request_error", "invalid_json")
return
if not isinstance(data, dict):
self._send_error(400, "Request muss ein JSON-Objekt sein",
"invalid_request_error", "invalid_request")
return
# input (OpenAI) – auch 'text' akzeptieren (bequemer für curl)
text = data.get("input", data.get("text"))
if not isinstance(text, str) or not text.strip():
self._send_error(400, "'input' fehlt oder ist leer",
"invalid_request_error", "missing_input")
return
if len(text) > 8000:
self._send_error(400, "'input' zu lang (max 8000 Zeichen)",
"invalid_request_error", "input_too_long")
return
voice = data.get("voice", TTS_DEFAULT_VOICE)
if voice not in TTS_VOICES:
self._send_error(
400, f"ungültige Stimme: {voice!r} "
f"(erlaubt: {', '.join(TTS_VOICES)})",
"invalid_request_error", "invalid_voice")
return
fmt = data.get("response_format", TTS_DEFAULT_FORMAT)
if fmt not in TTS_FORMATS:
self._send_error(
400, f"ungültiges response_format: {fmt!r} "
f"(erlaubt: {', '.join(TTS_FORMATS)})",
"invalid_request_error", "invalid_format")
return
speed = data.get("speed", 1.0)
try:
speed = float(speed)
except (TypeError, ValueError):
self._send_error(400, "'speed' muss eine Zahl sein",
"invalid_request_error", "invalid_speed")
return
if not 0.5 <= speed <= 2.0:
self._send_error(400, "'speed' muss zwischen 0.5 und 2.0 sein",
"invalid_request_error", "invalid_speed")
return
# Modell-Name optional; falls angegeben, muss es xtts-v2 sein.
model = data.get("model")
if model is not None and model != TTS_MODEL:
self._send_error(400, f"unbekanntes Modell: {model!r} "
f"(erwartet: {TTS_MODEL})",
"invalid_request_error", "unknown_model")
return
self.timeout = None # Synthese kann dauern
try:
audio, content_type = tts_synthesize(
text.strip(), voice, speed, fmt)
except RuntimeError as e:
self._send_error(503, str(e), "server_error", "tts_failed")
return
self._last_code = 200
self.send_response(200)
self.send_header("Content-Type", content_type)
self.send_header("Content-Length", str(len(audio)))
self.send_header("Connection", "close")
self.end_headers()
self.wfile.write(audio)
# ---------- Audio-Discovery ----------
def _audio_models_payload(self) -> dict:
"""Listet verfügbare Audio-Modelle (STT + TTS)."""
tts = tts_status()
stt = stt_status()
models = []
if stt.get("ready"):
models.append({
"id": STT_MODEL,
"object": "model",
"owned_by": "whisper.cpp",
"type": "transcription",
})
if tts.get("ready"):
models.append({
"id": TTS_MODEL,
"object": "model",
"owned_by": "coqui-xtts",
"type": "speech",
})
return {"object": "list", "data": models}
def _audio_voices_payload(self) -> dict:
"""Listet verfügbare TTS-Stimmen."""
tts = tts_status()
voices = []
for v in tts.get("voices", []):
voices.append({
"id": v,
"object": "voice",
"language": "de",
})
return {"object": "list", "data": voices}
# ---------- STT (Spracherkennung) ----------
def _parse_multipart(self, data: bytes, content_type: str
) -> tuple[bytes, str, dict]:
"""Parst multipart/form-data. Liefert (file_data, filename, fields).
Nutzt email.parser.BytesParser (Standardbibliothek) für robustes
MIME-Parsing. Handhabt quoted und unquoted Boundaries, beliebige
Feldreihenfolge, zusätzliche Header und binäre Payloads.
"""
# MIME-Message aus rohen Bytes + Content-Type-Header bauen
raw = (f"Content-Type: {content_type}\r\n\r\n"
).encode("utf-8") + data
msg = BytesParser(policy=compat32).parsebytes(raw)
if not msg.is_multipart():
raise ValueError("Kein multipart/form-data")
file_data = b""
filename = ""
fields = {}
for part in msg.get_payload():
disposition = part.get("Content-Disposition", "")
name = None
part_filename = None
for kv in disposition.split(";"):
kv = kv.strip()
if kv.startswith("name="):
name = kv[len("name="):].strip('"')
elif kv.startswith("filename="):
part_filename = kv[len("filename="):].strip('"')
if name is None:
continue
payload = part.get_payload(decode=True)
if payload is None:
payload = b""
if part_filename is not None:
# Dateifeld (binär, nicht dekodieren)
file_data = payload
filename = part_filename or ""
else:
# Textfeld
fields[name] = payload.decode("utf-8", errors="replace")
return file_data, filename, fields
def _transcribe(self) -> None:
"""POST /v1/audio/transcriptions – STT (OpenAI-kompatibel)."""
content_type = self.headers.get("Content-Type", "")
if "multipart/form-data" not in content_type:
self._send_error(400,
"Content-Type muss multipart/form-data sein",
"invalid_request_error", "invalid_content_type")
return
try:
data = self._read_body()
except ValueError as e:
self._send_error(400, str(e),
"invalid_request_error", "invalid_body")
return
try:
file_data, filename, fields = self._parse_multipart(
data, content_type)
except ValueError as e:
self._send_error(400, str(e),
"invalid_request_error", "invalid_multipart")
return
if not file_data:
self._send_error(400, "Keine Datei im Request",
"invalid_request_error", "missing_file")
return
# Modell-Validierung
model = fields.get("model", STT_MODEL)
if model not in (STT_MODEL, "whisper"):
self._send_error(400, f"unbekanntes Modell: {model!r} "
f"(erwartet: {STT_MODEL})",
"invalid_request_error", "unknown_model")
return
# Optionale Felder
language = fields.get("language")
prompt = fields.get("prompt")
temperature = None
if fields.get("temperature"):
try:
temperature = float(fields["temperature"])
except ValueError:
self._send_error(400, "'temperature' muss eine Zahl sein",
"invalid_request_error", "invalid_temperature")
return
response_format = fields.get("response_format", "json")
self.timeout = None # Transkription kann dauern
try:
result = stt_transcribe(
file_data, filename,
language=language, prompt=prompt,
temperature=temperature)
except RuntimeError as e:
self._send_error(503, str(e), "server_error", "stt_failed")
return
# OpenAI-kompatibles Antwort-Format
if response_format == "verbose_json":
resp = {
"text": result.get("text", ""),
"language": result.get("language", "de"),
"duration": result.get("audio_duration_ms", 0) / 1000.0,
}
else:
resp = {"text": result.get("text", "")}
self._send_json(200, resp)
def _switch(self, profile: str) -> None:
if profile not in PROFILES:
self._send_error(400, f"unbekanntes Profil: {profile}",
"invalid_request_error", "invalid_profile")
return
try:
switch_profile(profile)
except (ValueError, RuntimeError) as e:
self._send_error(503, str(e), "server_error", "profile_switch_failed")
return
up = upstream_status()
self._send_json(200, {
"status": "ok",
"profile": profile,
"context_length": PROFILES[profile],
"model": up.get("model"),
})
# ---------- Interner Vision-Test ----------
def _vision_test(self) -> None:
"""Interner Vision-Test: führt den kompletten Q3-Hotswap durch und
liefert die Analyse + Timing (ohne finale Hauptmodell-Inferenz).
Body: {"image_url": "data:image/png;base64,..." | "http://...",
"question": "optional"}
"""
try:
body = self._read_body()
except ValueError as e:
self._send_error(400, str(e),
"invalid_request_error", "invalid_body")
return
try:
req = json.loads(body) if body else {}
except ValueError:
self._send_error(400, "ungültiges JSON",
"invalid_request_error", "invalid_body")
return
if not isinstance(req, dict):
self._send_error(400, "Body muss ein JSON-Objekt sein",
"invalid_request_error", "invalid_body")
return
image_url = req.get("image_url")
if not isinstance(image_url, str) or not image_url:
self._send_error(400, "image_url fehlt (data-URL oder http-URL)",
"invalid_request_error", "missing_image")
return
question = req.get("question") or ""
# Request bauen, der den Vision-Pfad triggert.
data = {
"model": "qwen-medium",
"messages": [
{"role": "user", "content": [
{"type": "image_url", "image_url": {"url": image_url}},
{"type": "text",
"text": question or "Analysiere das Bild."},
]},
],
}
self.timeout = None # Vision-Swap kann Minuten dauern
try:
analysis = _vision_swap(data)
except (ValueError, RuntimeError) as e:
self._send_error(502, str(e), "server_error", "vision_failed")
return
vis = STATE.vision
self._send_json(200, {
"status": "ok",
"profile": vis.last_profile,
"turnaround_seconds": vis.last_turnaround,
"analysis_chars": len(analysis),
"analysis": analysis,
})
# ---------- Transparentes Forwarding ----------
def _forward(self) -> None:
path = self.path.split("?", 1)[0]
try:
body = self._read_body() or None
except ValueError as e:
self._send_error(400, str(e),
"invalid_request_error", "invalid_body")
return
# /v1/streams/lookup (Open-WebUI-Stream-Recovery): darf NIEMALS auf
# die Qwen-Wiederherstellung warten (während Vision-Hotswap,
# Profilwechsel oder Image-Job ist Qwen down). Wenn Qwen down ist,
# gibt es per Definition keine aktiven Streams → sofortige lokale
# Antwort []. Ansonsten normal an llama.cpp weiterleiten.
if path == "/v1/streams/lookup":
with STATE.avail_lock:
qwen_unavailable = STATE.qwen_unavailable
if qwen_unavailable:
self._send_json(200, [])
return
self._proxy(body)
return
data = None
# Virtuelles Modell? -> Profil sicherstellen, dann Modell ersetzen.
if body is not None and self.path.startswith("/v1/"):
try:
data = json.loads(body)
except ValueError:
data = None
model = data.get("model") if isinstance(data, dict) else None
if isinstance(model, str) and model in VIRTUAL_MODELS:
profile = VIRTUAL_MODELS[model]
try:
switch_profile(profile, implicit=True)
except (ValueError, RuntimeError) as e:
self._send_error(502, str(e), "server_error",
"upstream_unavailable")
return
up = upstream_status()
if not up["reachable"] or not up.get("model"):
self._send_error(502, "llama.cpp nicht erreichbar",
"server_error", "upstream_unavailable")
return
data["model"] = up["model"]
body = json.dumps(data).encode()
elif isinstance(model, str) and model.startswith("qwen-"):
# qwen-* ist der Namensraum des Routers
self._send_error(400, f"unbekanntes virtuelles Modell: {model}",
"invalid_request_error", "unknown_model")
return
# Vision: Bilder im Request → Q3-Vision-Analyse (nur für NEUE,
# noch nicht analysierte Bilder), danach erzeugt das
# (wiederhergestellte) Hauptmodell die Endantwort. Die gesamte
# History wird sanisiert: alle Bild-Parts werden durch ihre
# (gecachten) Vision-Analysen ersetzt, damit das Nicht-Vision-
# Hauptmodell (Fast/Medium/Long) keine Bilddaten bekommt. Das ist
# wichtig, weil Open WebUI bei Folgefragen den ursprünglichen
# multimodalen Verlauf erneut mitsendet.
if isinstance(data, dict) and path == "/v1/chat/completions":
image_url, _ = _extract_last_user_image(data)
if image_url:
if _vision_cached_analysis(_image_hash(image_url)) is None:
# Neues Bild → Vision-Hotswap (Q3 analysiert + cacht).
self.timeout = None # Vision-Swap kann Minuten dauern
try:
_vision_swap(data)
except (ValueError, RuntimeError) as e:
self._send_error(502, str(e), "server_error",
"vision_failed")
return
else:
log.info("Vision: Bild bereits analysiert "
"(Cache-Treffer) – kein Hotswap")
if _request_has_image(data):
data = _sanitize_for_main_model(data)
body = json.dumps(data).encode()
log.info("Vision: finale Hauptmodell-Inferenz gestartet")
# An llama.cpp weiterleiten (mit Chat-Waiting, Streaming bleibt erhalten).
self._proxy_with_wait(body)
def _proxy_with_wait(self, body: bytes | None) -> None:
"""Leitet an llama.cpp weiter, wartet aber erst, bis Qwen verfügbar ist.
Während eines Image-Jobs oder Profilwechsels ist Qwen down. Statt
502 zu liefern, wartet der Request (mit Timeout), bis Qwen wieder
bereit ist. Mehrere Chats können parallel laufen (active_chats).
Race-frei: Der Check auf qwen_unavailable und das Inkrement von
active_chats sind atomar (avail_lock). Ein Image-Job/Profilwechsel
setzt qwen_unavailable=True und wartet auf active_chats==0, BEVOR
er Qwen stoppt – ein laufender Chat wird daher nie unterbrochen.
"""
deadline = time.monotonic() + CHAT_WAIT_TIMEOUT
while True:
with STATE.avail_lock:
if not STATE.qwen_unavailable:
STATE.active_chats += 1
break
if time.monotonic() > deadline:
self._send_error(
503,
"Qwen wird neu geladen (Image-Job oder Profilwechsel), "
"bitte später erneut",
"server_error", "qwen_reloading")
return
time.sleep(0.5)
try:
self._proxy(body)
finally:
with STATE.avail_lock:
STATE.active_chats -= 1
def _proxy(self, body: bytes | None) -> None:
# An llama.cpp weiterleiten (Streaming bleibt erhalten).
try:
conn = http.client.HTTPConnection(UPSTREAM_HOST, UPSTREAM_PORT,
timeout=CONNECT_TIMEOUT)
conn.connect()
conn.sock.settimeout(REQUEST_TIMEOUT)
headers = {k: v for k, v in self.headers.items()
if k.lower() not in HOP_BY_HOP}
conn.request(self.command, self.path, body=body, headers=headers)
resp = conn.getresponse()
except (OSError, http.client.HTTPException) as e:
self._send_error(502, f"llama.cpp nicht erreichbar: {e}",
"server_error", "upstream_unavailable")
return
self._last_code = resp.status
self.send_response(resp.status)
for k, v in resp.getheaders():
if k.lower() not in HOP_BY_HOP:
self.send_header(k, v)
self.send_header("Connection", "close")
self.end_headers()
try:
while True:
chunk = resp.read(16384)
if not chunk:
break
self.wfile.write(chunk)
self.wfile.flush()
except (OSError, http.client.HTTPException) as e:
log.warning("Upstream-Stream abgebrochen: %s", e)
finally:
conn.close()
# ---------- Antworten ----------
def _send_json(self, code: int, payload: dict) -> None:
body = json.dumps(payload).encode()
self._last_code = code
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.send_header("Connection", "close")
self.end_headers()
self.wfile.write(body)
def _send_error(self, code: int, message: str, etype: str, ecode: str) -> None:
# OpenAI-kompatibles Fehlerformat
self._send_json(code, {"error": {"message": message,
"type": etype,
"code": ecode}})
def _safe_error(self, code: int, message: str) -> None:
try:
self._send_error(code, message, "server_error", "internal_error")
except Exception:
pass
# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------
class _FlushHandler(logging.StreamHandler):
"""StreamHandler, der nach jedem Record flusht (journald)."""
def emit(self, record):
super().emit(record)
self.flush()
def main() -> None:
handler = _FlushHandler(sys.stdout)
handler.setFormatter(logging.Formatter(
"%(asctime)s %(levelname)s %(message)s"))
logging.basicConfig(level=os.environ.get("LOG_LEVEL", "INFO"),
handlers=[handler])
log.info("AI Profile Router startet: %s:%s -> %s (Profile: %s)",
HOST, PORT, UPSTREAM_URL, ", ".join(PROFILES))
server = ThreadingHTTPServer((HOST, PORT), Handler)
server.daemon_threads = True
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
server.server_close()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink