from __future__ import annotations import json from typing import Any, Dict, List, Optional, Tuple import pandas as pd from audio_client import openai_client from guidance_resolver import resolve_guidance from nightscout_client import ( calc_delta_last_minutes, direction_to_trend, traffic_light, ) from nlu import is_target_range_question def _norm_str(x: Any, default: str = "") -> str: if x is None or (isinstance(x, float) and pd.isna(x)): return default s = str(x).strip() return s if s else default def build_general_tips_block() -> str: return ( "\n\n---\n" "### Allgemeine Hinweise (außerhalb eurer Leitplanken)\n" "_Ich finde dazu gerade keine passende Regel in euren Guidances. " "Die folgenden Punkte sind allgemeine Tipps und **keine** guidance-basierte Empfehlung._\n\n" "- Wenn Werte sich schnell veraendern: Verlauf engmaschiger beobachten.\n" "- Bei Symptomen, Unsicherheit oder unklarer Lage: lieber frueh Kontaktperson einbinden.\n" "- Wenn Essen/Sport ansteht: Kontext hilft (was, wieviel, wann), dann kann ich besser einordnen.\n" ) def deterministic_guidance_text( action: str, why_line: str, bullets: List[str], add_general_tips: bool = False, ) -> str: parts: List[str] = [] parts.append(f"**Kurz gesagt:** {action}") parts.append( f"**Warum:** {why_line}" if why_line else "**Warum:** (keine Zusatzinfo)" ) if bullets: parts.append( "**Was jetzt sinnvoll ist:**\n- " + "\n- ".join(bullets[:2]) ) if add_general_tips: parts.append(build_general_tips_block()) return "\n\n".join(parts) def llm_coach_only( user_text: str, why_line: str, bullets: List[str], *, openai_api_key: str, model_reply: str = "gpt-4.1-mini", max_bullets: int = 2, ) -> Tuple[str, List[str]]: """ LLM darf nur WHY + Bullets sprachlich verbessern. Keine neuen Handlungen, keine Insulin-Dosis, keine Vorhersagezahlen. """ client = openai_client(openai_api_key) if client is None: return why_line, bullets[:max_bullets] payload = { "user_text": user_text, "why_line": why_line, "bullets": bullets[:max_bullets], "constraints": { "no_new_actions": True, "no_insulin_dosing": True, "no_numeric_predictions": True, "tone": "ruhig, alltagsnah, fuer Sprachausgabe", }, } schema = """ Gib NUR JSON zurueck: { "why": "string", "bullets": ["string", "string"] } Regeln: - Verbessere nur Formulierung (why + bullets). - Erfinde keine neuen Aktionen. - Keine Insulin-Dosis. - Keine Vorhersagezahlen. - max 2 bullets. """ try: resp = client.chat.completions.create( model=model_reply, messages=[ { "role": "system", "content": "Du formulierst kurz, klar und sicher. Antworte ausschliesslich als JSON.", }, { "role": "system", "content": schema, }, { "role": "user", "content": json.dumps(payload, ensure_ascii=False), }, ], temperature=0.3, ) data = json.loads((resp.choices[0].message.content or "").strip()) out_why = str(data.get("why") or "").strip() or why_line out_bullets = data.get("bullets") or [] if not isinstance(out_bullets, list): return why_line, bullets[:max_bullets] cleaned: List[str] = [] for b in out_bullets[:max_bullets]: s = str(b).strip() if s: cleaned.append(s) return out_why, (cleaned if cleaned else bullets[:max_bullets]) except Exception: return why_line, bullets[:max_bullets] def _target_range_status(sgv: Optional[float]) -> Optional[str]: if sgv is None: return None try: v = float(sgv) except Exception: return None if 80 <= v <= 180: return "Ja – aktuell im Zielbereich." if v < 80: return "Aktuell unter dem Zielbereich." return "Aktuell ueber dem Zielbereich." def llm_smart_reply_situation( user_text: str, latest: dict, delta_30m: Optional[float], matched_rule: dict, *, openai_api_key: str, model_reply: str = "gpt-4.1-mini", ) -> str: sgv = latest.get("sgv") if latest.get("ok") else None direction = latest.get("direction", "") if latest.get("ok") else "" age = latest.get("age_min") if latest.get("ok") else None trend_simple = direction_to_trend(direction) trend_word = { "falling": "fallend", "double_falling": "schnell fallend", "rising": "steigend", "stable": "stabil", }.get(trend_simple, "unklar") if is_target_range_question(user_text): status = _target_range_status(sgv) if status is None: return ( "**Kurz gesagt:** Ich kann den aktuellen Wert gerade nicht sicher lesen.\n\n" "**Was jetzt sinnvoll ist:**\n" "- Nightscout-Verbindung/Sensor kurz pruefen." ) return ( f"**Kurz gesagt:** {status}\n\n" f"**Aktuell:** {sgv} mg/dl, Trend wirkt {trend_word}." ) guidance_id = (matched_rule.get("guidance_id") or "").upper() no_rule = guidance_id in ( "NO_MATCH", "NO_GUIDANCES", "NO_GLUCOSE", "DEFAULT_NO_MATCH", ) action = ( _norm_str(matched_rule.get("action"), "") .strip() or "Bitte beobachten." ) if sgv is not None and age is not None: why_line = f"Aktuell {sgv} mg/dl, Trend wirkt {trend_word}." if delta_30m is not None: if delta_30m <= -30: why_line += " In den letzten ~30 min ging es deutlich nach unten." elif delta_30m >= 30: why_line += " In den letzten ~30 min ging es deutlich nach oben." else: why_line = "Nightscout-Daten sind gerade nicht sicher verfuegbar." bullets: List[str] = [] fu = _norm_str(matched_rule.get("follow_up"), "").strip() cg = _norm_str(matched_rule.get("carbs_g"), "").strip() if fu: bullets.append(fu) if cg and cg != "0": bullets.append(f"Richtwert: {cg} g KH (laut eurer Regel)") if no_rule: return deterministic_guidance_text( action=action, why_line=why_line, bullets=bullets, add_general_tips=True, ) why2, bullets2 = llm_coach_only( user_text=user_text, why_line=why_line, bullets=bullets, openai_api_key=openai_api_key, model_reply=model_reply, max_bullets=2, ) return deterministic_guidance_text( action=action, why_line=why2, bullets=bullets2, add_general_tips=False, ) def llm_smart_reply_what_if( user_text: str, what_if_glucose: float, what_if_trend: str, matched_rule: dict, *, openai_api_key: str, model_reply: str = "gpt-4.1-mini", ) -> str: guidance_id = (matched_rule.get("guidance_id") or "").upper() no_rule = guidance_id in ( "NO_MATCH", "NO_GUIDANCES", "NO_GLUCOSE", "DEFAULT_NO_MATCH", ) action = ( _norm_str(matched_rule.get("action"), "") .strip() or "Bitte beobachten." ) cn = _norm_str(matched_rule.get("condition_note"), "").strip() why_line = ( cn if cn else f"Szenario: {int(what_if_glucose)} mg/dl ({what_if_trend})." ) bullets: List[str] = [] cg = _norm_str(matched_rule.get("carbs_g"), "").strip() fe = _norm_str(matched_rule.get("food_examples"), "").strip() fu = _norm_str(matched_rule.get("follow_up"), "").strip() if cg and cg != "0": bullets.append(f"Richtwert: {cg} g KH") if fe: bullets.append(f"Beispiele: {fe}") if fu: bullets.append(f"Dann: {fu}") if no_rule: header = ( f"**Kurz gesagt:** Fuer das Szenario " f"({int(what_if_glucose)} mg/dl, {what_if_trend}) " f"finde ich keine passende Regel." ) return header + build_general_tips_block() why2, bullets2 = llm_coach_only( user_text=user_text, why_line=why_line, bullets=bullets, openai_api_key=openai_api_key, model_reply=model_reply, max_bullets=2, ) parts: List[str] = [] parts.append( f"**Kurz gesagt:** Wenn er bei **{int(what_if_glucose)} mg/dl** " f"liegt ({what_if_trend}), wuerdet ihr: **{action}**." ) parts.append(f"**Warum:** {why2}") if bullets2: parts.append( "**Was jetzt sinnvoll ist:**\n- " + "\n- ".join(bullets2[:2]) ) return "\n\n".join(parts) def run_agent_situation( user_text: str, latest: dict, hist: dict, guidances_df: pd.DataFrame, *, openai_api_key: str, model_reply: str = "gpt-4.1-mini", ) -> str: glucose = latest.get("sgv") if latest.get("ok") else None trend = direction_to_trend(latest.get("direction", "")) if latest.get("ok") else "any" age_min = latest.get("age_min") if latest.get("ok") else None match = resolve_guidance( guidances_df, glucose, trend, user_text=user_text, data_age_min=age_min, ) rule = match.rule or {} delta_30m = None if hist.get("ok") and hist.get("df") is not None: delta_30m = calc_delta_last_minutes(hist["df"], minutes=30) return llm_smart_reply_situation( user_text, latest, delta_30m, rule, openai_api_key=openai_api_key, model_reply=model_reply, ) def run_agent_what_if( user_text: str, guidances_df: pd.DataFrame, what_if_glucose: Optional[float], what_if_trend: Optional[str], *, openai_api_key: str, model_reply: str = "gpt-4.1-mini", ) -> str: if what_if_glucose is None: return ( "**Kurz gesagt:** Klar – ich kann das als Szenario durchspielen.\n\n" "**Sag mir bitte:** einen Beispielwert " "(z.B. 70 oder 'siebzig') und optional ob er stabil/fallend/steigend ist." ) tr = what_if_trend if what_if_trend else "any" match = resolve_guidance( guidances_df, float(what_if_glucose), tr, user_text=user_text, data_age_min=None, ) return llm_smart_reply_what_if( user_text, float(what_if_glucose), tr, match.rule or {}, openai_api_key=openai_api_key, model_reply=model_reply, ) def make_start_message( latest: dict, hist: dict, guidances_df: pd.DataFrame, ) -> str: sgv = latest.get("sgv") if latest.get("ok") else None direction = latest.get("direction", "") if latest.get("ok") else "" trend = direction_to_trend(direction) if latest.get("ok") else "any" age = latest.get("age_min") if latest.get("ok") else None delta_30m = None if hist.get("ok") and hist.get("df") is not None: delta_30m = calc_delta_last_minutes(hist["df"], minutes=30) _, emoji = traffic_light(sgv, trend, delta_30m) match = resolve_guidance( guidances_df, sgv, trend, user_text="start", data_age_min=age, ) action = ( _norm_str((match.rule or {}).get("action"), "") .strip() or "Beobachten." ) range_label = ( "im Zielbereich" if isinstance(sgv, (int, float)) and 80 <= float(sgv) <= 180 else "außerhalb des Zielbereichs" ) return ( f"Hey, ich bin **Gluco** 👋\n\n" f"**Aktueller Wert:** {sgv if sgv is not None else '-'} mg/dl, " f"Trend: {direction or '-'}.\n\n" f"**Status:** {emoji} {range_label}.\n\n" f"**Leitplanken-Empfehlung:** {action}\n\n" f"Womit kann ich dir helfen? " f"(z.B. *„Sind wir im Zielbereich?“*, " f"*„Was waere wenn er auf neunzig faellt?“*, " f"*„Wieviele KH sind in 30g Apfel?“*)" )