Spaces:
Sleeping
Sleeping
| 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?“*)" | |
| ) |