| """ |
| Claude-powered heat alert explanation generator. |
| |
| When a heat risk trigger fires, enrolled workers need a plain-language |
| explanation of WHY, what it means for their payout, and what protective |
| actions to take. This module generates bilingual (English + Swahili) |
| explanations using Claude with RAG context, with a template-based |
| fallback if Claude is unavailable. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import logging |
| import os |
| from dataclasses import dataclass, field |
| from datetime import datetime, timezone |
| from typing import Any, Dict, Optional |
|
|
| from config import HEAT_THRESHOLDS, PAYOUT_PER_EVENT_USD, ZONE_MAP, UrbanZone |
| from src.explanation.knowledge_base import ( |
| INSURANCE_PRODUCT_INFO, |
| SWAHILI_TERMS, |
| get_emergency_contacts, |
| get_protective_actions, |
| get_zone_context, |
| ) |
|
|
| log = logging.getLogger(__name__) |
|
|
|
|
| |
|
|
| @dataclass |
| class TriggerEvent: |
| """A triggered insurance event to be explained.""" |
| zone_id: str |
| trigger_level: str |
| triggered_at: str |
| max_temp_c: float = 0.0 |
| max_wbgt_c: float = 0.0 |
| consecutive_days: int = 0 |
| heat_risk_score: float = 0.0 |
| contributing_factors: list[str] = field(default_factory=list) |
|
|
|
|
| @dataclass |
| class ExplanationResult: |
| """Generated explanation for a trigger event.""" |
| zone_id: str |
| trigger_level: str |
| english_text: str |
| swahili_text: str |
| payout_estimate: Dict[str, Any] |
| protective_actions: list[str] |
| emergency_contacts: Dict[str, str] |
| generated_at: str = field( |
| default_factory=lambda: datetime.now(timezone.utc).isoformat() |
| ) |
| provider: str = "template" |
| zone_name: str = "" |
| city: str = "" |
| zone_context: str = "" |
| tokens_used: int = 0 |
|
|
|
|
| |
|
|
| class TriggerExplainer: |
| """Generates bilingual heat alert explanations using Claude with RAG.""" |
|
|
| def __init__(self, api_key: Optional[str] = None, model: str = "claude-haiku-4-5-20251001"): |
| self.api_key = api_key or os.environ.get("ANTHROPIC_API_KEY", "") |
| self.model = model |
| self._client = None |
|
|
| |
| try: |
| from src.explanation.rag_provider import HybridRetriever |
| self._retriever = HybridRetriever() |
| log.info("Hybrid RAG retriever loaded successfully") |
| except Exception as exc: |
| log.info("RAG retriever not available, using full context injection: %s", exc) |
| self._retriever = None |
|
|
| def _get_client(self): |
| if self._client is None: |
| try: |
| import anthropic |
| self._client = anthropic.AsyncAnthropic(api_key=self.api_key) |
| except Exception as exc: |
| log.warning("Could not init Anthropic client: %s", exc) |
| return self._client |
|
|
| async def explain(self, event, zone=None, basis_risk=None) -> ExplanationResult: |
| """Generate a full explanation for a heat trigger event.""" |
| |
| zone_id = getattr(event, 'zone_id', '') |
| trigger_level = getattr(event, 'trigger_level', 'watch') |
|
|
| zone_obj = zone or ZONE_MAP.get(zone_id) |
| if zone_obj is None: |
| raise ValueError(f"Unknown zone: {zone_id}") |
|
|
| payout = _compute_payout(event, zone_obj) |
| actions = get_protective_actions(trigger_level) |
| contacts = get_emergency_contacts(zone_obj.city) |
|
|
| if self.api_key: |
| try: |
| english, swahili = await self._generate_claude(event, zone_obj, payout) |
| return ExplanationResult( |
| zone_id=zone_id, |
| trigger_level=trigger_level, |
| english_text=english, |
| swahili_text=swahili, |
| payout_estimate=payout, |
| protective_actions=actions, |
| emergency_contacts=contacts, |
| provider="claude", |
| zone_name=zone_obj.name, |
| city=zone_obj.city, |
| zone_context=get_zone_context(zone_id), |
| ) |
| except Exception as exc: |
| log.warning("Claude explanation failed, using template: %s", exc) |
|
|
| english = _template_english(event, zone_obj, payout) |
| swahili = _template_swahili(event, zone_obj, payout) |
| return ExplanationResult( |
| zone_id=zone_id, |
| trigger_level=trigger_level, |
| english_text=english, |
| swahili_text=swahili, |
| payout_estimate=payout, |
| protective_actions=actions, |
| emergency_contacts=contacts, |
| provider="template", |
| zone_name=zone_obj.name, |
| city=zone_obj.city, |
| zone_context=get_zone_context(zone_id), |
| ) |
|
|
| async def _generate_claude( |
| self, event, zone: UrbanZone, payout: Dict[str, Any] |
| ) -> tuple[str, str]: |
| |
| if self._retriever: |
| query = ( |
| f"{event.trigger_level} heat alert {zone.name} {zone.city} " |
| f"{zone.settlement_type} outdoor workers" |
| ) |
| context_docs = self._retriever.retrieve(query, zone_id=event.zone_id) |
| context = "\n---\n".join(context_docs) |
| else: |
| context = get_zone_context(event.zone_id) |
|
|
| system = ( |
| "You are a heat safety notification system for outdoor workers in East Africa. " |
| "Your audience is workers in markets, construction sites, and informal " |
| "settlements — many with limited formal education. Write clearly, simply, " |
| "and with empathy. Do not use jargon or technical abbreviations. " |
| "Explain what happened, what their payout is, and what they should do to " |
| "stay safe. Keep the explanation to 4-6 sentences." |
| ) |
|
|
| max_temp = getattr(event, 'max_temp_c', 0) |
| max_wbgt = getattr(event, 'max_wbgt_c', 0) |
| consec = getattr(event, 'consecutive_days', getattr(event, 'consecutive_days_above', 0)) |
| thresholds = HEAT_THRESHOLDS.get(event.trigger_level, {}) |
|
|
| user = ( |
| f"A {event.trigger_level.upper()} heat alert has been triggered for " |
| f"outdoor workers in {zone.name}, {zone.city}.\n\n" |
| f"What happened:\n" |
| f"- Maximum temperature: {max_temp:.1f}C " |
| f"(threshold: {thresholds.get('temp_c', 'N/A')}C)\n" |
| f"- Maximum WBGT (feels-like for workers): {max_wbgt:.1f}C " |
| f"(threshold: {thresholds.get('wbgt_c', 'N/A')}C)\n" |
| f"- Consecutive days above threshold: {consec} " |
| f"(required: {thresholds.get('consecutive_days', 'N/A')})\n" |
| f"- Settlement type: {zone.settlement_type}\n" |
| f"- Outdoor worker exposure: {zone.outdoor_exposure_pct:.0%}\n" |
| f"- Estimated workers affected: {zone.worker_population_est:,}\n\n" |
| f"Payout: {payout['currency_symbol']}{payout['amount']} per worker\n\n" |
| f"Knowledge base:\n{context}\n\n" |
| f"Write a 4-6 sentence explanation for the worker. Tell them what is " |
| f"happening with the heat, their payout amount, and the most important " |
| f"thing they should do right now to stay safe." |
| ) |
|
|
| client = self._get_client() |
| if client is None: |
| raise RuntimeError("Anthropic client not available") |
|
|
| msg = await client.messages.create( |
| model=self.model, |
| max_tokens=500, |
| system=system, |
| messages=[{"role": "user", "content": user}], |
| ) |
| english = msg.content[0].text.strip() |
| swahili = await self._translate_to_swahili(english, zone.name) |
| return english, swahili |
|
|
| async def _translate_to_swahili(self, english_text: str, zone_name: str) -> str: |
| system = ( |
| "You are a professional translator specializing in Swahili (Kiswahili). " |
| "Translate the given English heat safety notification to simple, clear " |
| "Swahili that a non-technical person can understand. Keep numbers, " |
| "currency amounts, and proper nouns unchanged. Return only the translated text." |
| ) |
| user = ( |
| f"Translate this heat alert notification for workers in {zone_name} " |
| f"to Swahili:\n\n{english_text}" |
| ) |
|
|
| client = self._get_client() |
| if client is None: |
| raise RuntimeError("Anthropic client not available") |
|
|
| msg = await client.messages.create( |
| model=self.model, |
| max_tokens=600, |
| system=system, |
| messages=[{"role": "user", "content": user}], |
| ) |
| return msg.content[0].text.strip() |
|
|
|
|
| |
|
|
| class TemplateExplainer: |
| """Template-based fallback when Claude is unavailable.""" |
|
|
| async def explain(self, event, zone=None, basis_risk=None) -> ExplanationResult: |
| zone_id = getattr(event, 'zone_id', '') |
| trigger_level = getattr(event, 'trigger_level', 'watch') |
| zone_obj = zone or ZONE_MAP.get(zone_id) |
| if zone_obj is None: |
| raise ValueError(f"Unknown zone: {zone_id}") |
|
|
| payout = _compute_payout(event, zone_obj) |
| actions = get_protective_actions(trigger_level) |
| contacts = get_emergency_contacts(zone_obj.city) |
|
|
| english = _template_english(event, zone_obj, payout) |
| swahili = _template_swahili(event, zone_obj, payout) |
|
|
| return ExplanationResult( |
| zone_id=zone_id, |
| trigger_level=trigger_level, |
| english_text=english, |
| swahili_text=swahili, |
| payout_estimate=payout, |
| protective_actions=actions, |
| emergency_contacts=contacts, |
| provider="template", |
| zone_name=zone_obj.name, |
| city=zone_obj.city, |
| zone_context=get_zone_context(zone_id), |
| ) |
|
|
|
|
| |
|
|
| def _compute_payout(event, zone: UrbanZone) -> Dict[str, Any]: |
| trigger_level = getattr(event, 'trigger_level', 'watch') |
| amount = PAYOUT_PER_EVENT_USD.get(trigger_level, 0) |
|
|
| return { |
| "amount": amount, |
| "currency": "USD", |
| "currency_symbol": "$", |
| "settlement_type": zone.settlement_type, |
| "trigger_level": trigger_level, |
| "delivery_method": "M-Pesa" if zone.settlement_type in ("informal", "mixed") else "Mobile money", |
| "expected_delivery": "Within 48 hours of trigger verification", |
| "is_payout": amount > 0, |
| "workers_covered": zone.worker_population_est, |
| "total_payout": amount * zone.worker_population_est, |
| } |
|
|
|
|
| |
|
|
| _LEVEL_DESCRIPTIONS = { |
| "critical": ( |
| "A CRITICAL heat alert has been triggered", |
| "Onyo la HATARI la joto kali limetolewa", |
| ), |
| "warning": ( |
| "A WARNING heat alert has been issued", |
| "Onyo la joto limetolewa", |
| ), |
| "watch": ( |
| "A WATCH heat advisory has been issued", |
| "Tahadhari ya joto imetolewa", |
| ), |
| } |
|
|
|
|
| def _template_english(event, zone: UrbanZone, payout: Dict[str, Any]) -> str: |
| trigger_level = getattr(event, 'trigger_level', 'watch') |
| level_en, _ = _LEVEL_DESCRIPTIONS.get(trigger_level, ("A heat alert has been issued", "")) |
|
|
| max_temp = getattr(event, 'max_temp_c', 0) |
| max_wbgt = getattr(event, 'max_wbgt_c', 0) |
| consec = getattr(event, 'consecutive_days', getattr(event, 'consecutive_days_above', 0)) |
|
|
| lines = [f"{level_en} for outdoor workers in {zone.name}, {zone.city}."] |
|
|
| thresholds = HEAT_THRESHOLDS.get(trigger_level, {}) |
| reasons: list[str] = [] |
| if max_temp >= thresholds.get("temp_c", 999): |
| reasons.append( |
| f"temperatures reached {max_temp:.0f}C, above the " |
| f"{thresholds['temp_c']}C safety threshold" |
| ) |
| if max_wbgt >= thresholds.get("wbgt_c", 999): |
| reasons.append( |
| f"the heat-humidity index (WBGT) reached {max_wbgt:.0f}C, " |
| f"above the {thresholds['wbgt_c']}C danger level" |
| ) |
| if consec >= thresholds.get("consecutive_days", 999): |
| reasons.append( |
| f"these dangerous conditions lasted {consec} consecutive days" |
| ) |
|
|
| if reasons: |
| lines.append("This was triggered because " + " and ".join(reasons) + ".") |
| else: |
| lines.append("Heat conditions have exceeded the safety threshold for your area.") |
|
|
| if payout["is_payout"]: |
| lines.append( |
| f"Your payout is {payout['currency_symbol']}{payout['amount']} per worker, " |
| f"which will be sent via {payout['delivery_method']} within 48 hours." |
| ) |
|
|
| actions = get_protective_actions(trigger_level) |
| if actions: |
| lines.append(f"Most important: {actions[0]}") |
|
|
| return " ".join(lines) |
|
|
|
|
| def _template_swahili(event, zone: UrbanZone, payout: Dict[str, Any]) -> str: |
| trigger_level = getattr(event, 'trigger_level', 'watch') |
| _, level_sw = _LEVEL_DESCRIPTIONS.get(trigger_level, ("", "Onyo la joto limetolewa")) |
|
|
| max_temp = getattr(event, 'max_temp_c', 0) |
| consec = getattr(event, 'consecutive_days', getattr(event, 'consecutive_days_above', 0)) |
|
|
| lines = [f"{level_sw} kwa wafanyakazi wa nje katika {zone.name}, {zone.city}."] |
|
|
| lines.append( |
| f"Joto limefika {max_temp:.0f}C kwa siku {consec} mfululizo." |
| ) |
|
|
| if payout["is_payout"]: |
| lines.append( |
| f"Malipo yako ni {payout['currency_symbol']}{payout['amount']} kwa kila mfanyakazi. " |
| f"Utapokea kupitia {payout['delivery_method']} ndani ya masaa 48." |
| ) |
|
|
| if trigger_level == "critical": |
| lines.append( |
| "Acha kazi za nje sasa hivi. Nenda kivulini au ndani ya nyumba. " |
| "Kunywa maji mengi." |
| ) |
| elif trigger_level == "warning": |
| lines.append( |
| "Punguza kazi za nje. Fanya kazi nzito asubuhi mapema au jioni. " |
| "Pumzika kivulini kila saa." |
| ) |
| else: |
| lines.append( |
| "Kuwa makini. Kunywa maji zaidi. Panga kazi nzito kwa masaa ya baridi." |
| ) |
|
|
| return " ".join(lines) |
|
|
