368 lines
12 KiB
Python
368 lines
12 KiB
Python
"""
|
|
BAN YARO — KI-Abstraktions-Layer
|
|
|
|
Drei Modi:
|
|
- "off" → kein KI, Feature deaktiviert (Free-User ohne lokales Modell)
|
|
- "local" → LM Studio auf DS1621 (OpenAI-kompatibler Endpunkt, kostenlos)
|
|
- "cloud" → Claude API (nur für Premium-User, kostet Geld)
|
|
|
|
Wird über KI_MODE Umgebungsvariable gesteuert:
|
|
KI_MODE=local → Entwicklung + Free-User auf DS
|
|
KI_MODE=cloud → Production + Premium-User
|
|
KI_MODE=off → kein KI verfügbar
|
|
|
|
Wichtig: cloud-Aufrufe IMMER mit requires_premium=True schützen.
|
|
Kein API-Geld ohne zahlenden User.
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
from typing import Optional
|
|
from functools import wraps
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
KI_MODE = os.getenv("KI_MODE", "local") # off | local | cloud
|
|
LOCAL_BASE_URL = os.getenv("KI_LOCAL_URL", "http://10.47.11.70:11435/v1")
|
|
LOCAL_MODEL = os.getenv("KI_LOCAL_MODEL", "gemma-4-31b-it")
|
|
CLOUD_MODEL = os.getenv("KI_CLOUD_MODEL", "claude-opus-4-6")
|
|
ANTHROPIC_KEY = os.getenv("ANTHROPIC_API_KEY", "")
|
|
|
|
# Lazy Imports — nur laden wenn wirklich benötigt
|
|
_openai_client = None
|
|
_anthropic_client = None
|
|
|
|
|
|
def _get_local_client():
|
|
global _openai_client
|
|
if _openai_client is None:
|
|
from openai import OpenAI
|
|
_openai_client = OpenAI(base_url=LOCAL_BASE_URL, api_key="lm-studio")
|
|
return _openai_client
|
|
|
|
|
|
def _get_cloud_client():
|
|
global _anthropic_client
|
|
if _anthropic_client is None:
|
|
import anthropic
|
|
_anthropic_client = anthropic.Anthropic(api_key=ANTHROPIC_KEY)
|
|
return _anthropic_client
|
|
|
|
|
|
class KIUnavailableError(Exception):
|
|
"""KI-Feature nicht verfügbar (Off-Modus oder kein Premium)."""
|
|
pass
|
|
|
|
|
|
class KIPremiumRequired(Exception):
|
|
"""Dieses Feature erfordert Ban Yaro Premium."""
|
|
pass
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Haupt-Aufruf-Funktion — zentraler Eingang für alle KI-Requests
|
|
# ------------------------------------------------------------------
|
|
async def complete(
|
|
prompt: str,
|
|
system: str = "Du bist ein hilfreicher Assistent für Hundebesitzer.",
|
|
max_tokens: int = 512,
|
|
requires_premium: bool = False,
|
|
user_is_premium: bool = False,
|
|
json_mode: bool = False,
|
|
) -> str:
|
|
"""
|
|
KI-Completion. Wählt automatisch den richtigen Backend.
|
|
|
|
Args:
|
|
prompt: User-Nachricht
|
|
system: System-Prompt
|
|
max_tokens: Maximale Antwortlänge
|
|
requires_premium: True = nur für Premium-User (nutzt Cloud)
|
|
user_is_premium: Ob der anfragende User Premium hat
|
|
json_mode: Antwort als JSON anfordern
|
|
|
|
Returns:
|
|
KI-Antwort als String
|
|
|
|
Raises:
|
|
KIPremiumRequired: Cloud-Feature ohne Premium
|
|
KIUnavailableError: KI komplett deaktiviert
|
|
"""
|
|
if KI_MODE == "off":
|
|
raise KIUnavailableError("KI ist deaktiviert.")
|
|
|
|
# Premium-Check vor Cloud-Aufrufen
|
|
if requires_premium and not user_is_premium:
|
|
raise KIPremiumRequired(
|
|
"Dieses Feature ist Teil von Ban Yaro Premium."
|
|
)
|
|
|
|
# Cloud-Aufruf: nur wenn Premium UND cloud-Modus
|
|
if requires_premium and user_is_premium and KI_MODE == "cloud":
|
|
return await _cloud_complete(prompt, system, max_tokens, json_mode)
|
|
|
|
# Lokaler Aufruf: Entwicklung + Free-User
|
|
if KI_MODE in ("local", "cloud"):
|
|
try:
|
|
return await _local_complete(prompt, system, max_tokens, json_mode)
|
|
except Exception as e:
|
|
logger.warning(f"Lokales KI-Modell nicht erreichbar: {e}")
|
|
# Cloud-Fallback: im cloud-Modus immer, sonst nur für Premium-User
|
|
if ANTHROPIC_KEY and (KI_MODE == "cloud" or (requires_premium and user_is_premium)):
|
|
logger.info("Fallback auf Cloud-KI.")
|
|
return await _cloud_complete(prompt, system, max_tokens, json_mode)
|
|
raise KIUnavailableError(
|
|
"KI-Modell momentan nicht erreichbar. Bitte später erneut versuchen."
|
|
) from e
|
|
|
|
raise KIUnavailableError("Unbekannter KI-Modus.")
|
|
|
|
|
|
async def _local_complete(
|
|
prompt: str, system: str, max_tokens: int, json_mode: bool
|
|
) -> str:
|
|
"""LM Studio lokale Completion (synchron in Thread-Pool)."""
|
|
import asyncio
|
|
|
|
def _sync():
|
|
client = _get_local_client()
|
|
kwargs = dict(
|
|
model=LOCAL_MODEL,
|
|
messages=[
|
|
{"role": "system", "content": system},
|
|
{"role": "user", "content": prompt},
|
|
],
|
|
max_tokens=max_tokens,
|
|
temperature=0.3,
|
|
)
|
|
if json_mode:
|
|
kwargs["response_format"] = {"type": "json_object"}
|
|
|
|
response = client.chat.completions.create(**kwargs)
|
|
return response.choices[0].message.content.strip()
|
|
|
|
return await asyncio.get_event_loop().run_in_executor(None, _sync)
|
|
|
|
|
|
async def _cloud_complete(
|
|
prompt: str, system: str, max_tokens: int, json_mode: bool
|
|
) -> str:
|
|
"""Claude API Completion mit Prompt Caching."""
|
|
import asyncio
|
|
|
|
def _sync():
|
|
client = _get_cloud_client()
|
|
system_content = [
|
|
{
|
|
"type": "text",
|
|
"text": system,
|
|
"cache_control": {"type": "ephemeral"}, # Prompt Caching
|
|
}
|
|
]
|
|
response = client.messages.create(
|
|
model=CLOUD_MODEL,
|
|
max_tokens=max_tokens,
|
|
system=system_content,
|
|
messages=[{"role": "user", "content": prompt}],
|
|
)
|
|
return response.content[0].text.strip()
|
|
|
|
return await asyncio.get_event_loop().run_in_executor(None, _sync)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Spezialisierte KI-Funktionen
|
|
# (hier liegen die konkreten Prompts — zentral, nicht in den Routes)
|
|
# ------------------------------------------------------------------
|
|
|
|
async def symptom_check(symptoms: str, dog_info: dict,
|
|
user_is_premium: bool = False) -> dict:
|
|
"""
|
|
Symptom-Triage für Hunde.
|
|
Lokal für alle, Premium bekommt detailliertere Analyse.
|
|
"""
|
|
rasse = dog_info.get("rasse", "unbekannt")
|
|
alter = dog_info.get("alter_jahre", "unbekannt")
|
|
system = (
|
|
"Du bist ein erfahrener Veterinär-Assistent. "
|
|
"Deine Aufgabe ist eine erste Einschätzung von Symptomen beim Hund. "
|
|
"Antworte IMMER auf Deutsch und IMMER im angegebenen JSON-Format. "
|
|
"Überschreite nie deine Kompetenz — weise bei Unsicherheit zum Tierarzt."
|
|
)
|
|
prompt = f"""
|
|
Hund: {rasse}, {alter} Jahre alt.
|
|
Symptome: {symptoms}
|
|
|
|
Antworte NUR als JSON:
|
|
{{
|
|
"dringlichkeit": "beobachten" | "tierarzt_heute" | "notfall",
|
|
"einschaetzung": "Kurze Einschätzung in 1-2 Sätzen",
|
|
"hinweise": ["Hinweis 1", "Hinweis 2"],
|
|
"zum_tierarzt_wenn": "Wann unbedingt zum Tierarzt"
|
|
}}
|
|
"""
|
|
result = await complete(
|
|
prompt, system,
|
|
max_tokens=400,
|
|
requires_premium=False, # Basis-Triage ist kostenlos
|
|
user_is_premium=user_is_premium,
|
|
json_mode=True,
|
|
)
|
|
import json
|
|
try:
|
|
return json.loads(result)
|
|
except json.JSONDecodeError:
|
|
return {
|
|
"dringlichkeit": "tierarzt_heute",
|
|
"einschaetzung": result,
|
|
"hinweise": [],
|
|
"zum_tierarzt_wenn": "Bei Verschlechterung sofort.",
|
|
}
|
|
|
|
|
|
async def training_plan(problem: str, dog_info: dict,
|
|
user_is_premium: bool = False) -> str:
|
|
"""
|
|
Trainingsplan generieren — Premium-Feature.
|
|
"""
|
|
system = (
|
|
"Du bist ein zertifizierter Hundetrainer. "
|
|
"Erstelle konkrete, positive, gewaltfreie Trainingspläne auf Deutsch."
|
|
)
|
|
prompt = f"""
|
|
Hund: {dog_info.get('rasse')}, {dog_info.get('alter_jahre')} Jahre.
|
|
Problem: {problem}
|
|
|
|
Erstelle einen 2-Wochen-Trainingsplan mit täglichen Übungen (max. 15 Min/Tag).
|
|
Strukturiert, konkret, motivierend.
|
|
"""
|
|
return await complete(
|
|
prompt, system,
|
|
max_tokens=800,
|
|
requires_premium=True, # Cloud-Feature — kostet Geld
|
|
user_is_premium=user_is_premium,
|
|
)
|
|
|
|
|
|
async def diary_tags(entry_text: str) -> list[str]:
|
|
"""
|
|
Automatische Tags für Tagebucheinträge — läuft lokal, kostenlos.
|
|
"""
|
|
if KI_MODE == "off":
|
|
return []
|
|
prompt = f"""
|
|
Tagebucheintrag: "{entry_text}"
|
|
|
|
Vergib 2-4 passende Tags aus dieser Liste:
|
|
training, spaziergang, gesundheit, freunde, spielen, futter, tierarzt,
|
|
reise, meilenstein, lustig, traurig, wetter, sonstige
|
|
|
|
Antworte NUR mit den Tags, kommagetrennt, keine Erklärung.
|
|
"""
|
|
try:
|
|
result = await complete(prompt, max_tokens=50, requires_premium=False)
|
|
return [t.strip().lower() for t in result.split(",") if t.strip()]
|
|
except Exception:
|
|
return []
|
|
|
|
|
|
async def poison_description_check(description: str) -> dict:
|
|
"""
|
|
Giftköder-Beschreibung auf Plausibilität und Typ prüfen — lokal, kostenlos.
|
|
"""
|
|
if KI_MODE == "off":
|
|
return {"valid": True, "typ": "unbekannt"}
|
|
prompt = f"""
|
|
Giftköder-Meldung: "{description}"
|
|
|
|
Bewerte als JSON:
|
|
{{
|
|
"plausibel": true/false,
|
|
"typ": "fleisch" | "koeder" | "objekt" | "unbekannt",
|
|
"hinweis": "Kurzer Hinweis für andere Hundebesitzer (max 1 Satz)"
|
|
}}
|
|
"""
|
|
try:
|
|
import json
|
|
result = await complete(prompt, max_tokens=150, json_mode=True)
|
|
return json.loads(result)
|
|
except Exception:
|
|
return {"plausibel": True, "typ": "unbekannt", "hinweis": description}
|
|
|
|
|
|
async def health_summary(health_data: list, dog_info: dict,
|
|
user_is_premium: bool = False) -> str:
|
|
"""
|
|
Tierärztliche Zusammenfassung aller Gesundheitsdaten — lokal für alle.
|
|
Fasst Impfungen, Besuche, Medikamente etc. in einem lesbaren Bericht zusammen.
|
|
"""
|
|
system = (
|
|
"Du bist ein erfahrener Veterinär-Assistent. "
|
|
"Erstelle präzise, strukturierte Gesundheitsberichte für Hunde auf Deutsch. "
|
|
"Weise auf fällige Impfungen und wichtige Termine hin."
|
|
)
|
|
|
|
# Daten nach Typ gruppieren für den Prompt
|
|
def _fmt(entries, typ):
|
|
subset = [e for e in entries if e.get("typ") == typ]
|
|
if not subset:
|
|
return " (keine Einträge)"
|
|
lines = []
|
|
for e in subset[:10]: # maximal 10 pro Typ
|
|
line = f" - {e.get('datum', '?')}: {e.get('bezeichnung', '?')}"
|
|
if e.get("naechstes"):
|
|
line += f" (nächste Fälligkeit: {e['naechstes']})"
|
|
if e.get("notiz"):
|
|
line += f" — {e['notiz']}"
|
|
lines.append(line)
|
|
return "\n".join(lines)
|
|
|
|
heute = __import__("datetime").date.today().isoformat()
|
|
prompt = f"""
|
|
Hund: {dog_info.get('name')}, {dog_info.get('rasse', 'unbekannt')},
|
|
Geburtstag: {dog_info.get('geburtstag', 'unbekannt')}, Gewicht: {dog_info.get('gewicht_kg', '?')} kg
|
|
|
|
Heutiges Datum: {heute}
|
|
|
|
=== IMPFUNGEN ===
|
|
{_fmt(health_data, 'impfung')}
|
|
|
|
=== ENTWURMUNG ===
|
|
{_fmt(health_data, 'entwurmung')}
|
|
|
|
=== TIERARZTBESUCHE ===
|
|
{_fmt(health_data, 'tierarzt')}
|
|
|
|
=== MEDIKAMENTE ===
|
|
{_fmt(health_data, 'medikament')}
|
|
|
|
=== ALLERGIEN ===
|
|
{_fmt(health_data, 'allergie')}
|
|
|
|
Erstelle einen strukturierten Gesundheitsbericht mit:
|
|
1. Aktueller Status (2-3 Sätze)
|
|
2. Fällige / überfällige Impfungen und Termine
|
|
3. Wichtige Hinweise für den nächsten Tierarztbesuch
|
|
4. Allgemeine Empfehlungen
|
|
|
|
Schreibe klar und verständlich für den Hundebesitzer.
|
|
"""
|
|
return await complete(
|
|
prompt, system,
|
|
max_tokens=700,
|
|
requires_premium=False, # Kostenlos für alle
|
|
user_is_premium=user_is_premium,
|
|
)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Status-Endpoint (für Admin/Debug)
|
|
# ------------------------------------------------------------------
|
|
def status() -> dict:
|
|
return {
|
|
"mode": KI_MODE,
|
|
"local_url": LOCAL_BASE_URL if KI_MODE != "off" else None,
|
|
"local_model": LOCAL_MODEL if KI_MODE != "off" else None,
|
|
"cloud_model": CLOUD_MODEL,
|
|
"cloud_key_set": bool(ANTHROPIC_KEY),
|
|
}
|