""" LIA LLM Layer ============= Optional LLM-powered reply generator that sits behind the URE. Uses OpenAI to generate personalized, brand-compliant WhatsApp messages based on the template "spine" provided by URE. The URE still handles: - Tier classification (T1-T6) - Stage selection (18/12/9/6/3) - Template key and text selection - State machine updates The LLM adds: - Natural language variation - Brand voice consistency - Personalization - tone_tag and intent_tag for analytics Contract: - The external response shape is UNCHANGED - tone_tag and intent_tag are added INSIDE updated_state only """ import os import json import logging from typing import Dict, Any, Optional, Tuple from datetime import datetime, timezone import anyio try: from openai import OpenAI OPENAI_AVAILABLE = True OpenAIClient = OpenAI except ImportError: OPENAI_AVAILABLE = False OpenAIClient = None # type: ignore from app.prompts.lia_llm_system_prompt import ( LIA_SYSTEM_PROMPT, VALID_TONE_TAGS, VALID_INTENT_TAGS, ) logger = logging.getLogger(__name__) LLM_ENABLED = os.environ.get("LIA_LLM_ENABLED", "false").lower() == "true" OPENAI_MODEL = os.environ.get("LIA_LLM_MODEL", "gpt-4o-mini") def _get_openai_client() -> Optional[Any]: """Get OpenAI client if available and configured.""" if not OPENAI_AVAILABLE or OpenAIClient is None: logger.warning("OpenAI library not installed") return None api_key = os.environ.get("OPENAI_API_KEY") if not api_key: logger.warning("OPENAI_API_KEY not set") return None return OpenAIClient(api_key=api_key) def _build_llm_context( customer_state: Dict[str, Any], inbound_text: str, tier: str, template_key: str, template_text: str, recent_messages: Optional[list] = None, ) -> Dict[str, Any]: """ Build the structured context for the LLM as specified in the system prompt. This matches the INPUT FORMAT specified in lia_llm_system_prompt.py. """ customer_name = customer_state.get("customer_name") if customer_name == "Unknown" or not customer_name: customer_name = None context = { "customer_name": customer_name, "vehicle_model": customer_state.get("vehicle_model"), "target_model": customer_state.get("target_model"), "tier": tier, "stage": customer_state.get("stage"), "campaign_type": customer_state.get("campaign_type", "EOT"), "template_key": template_key, "template_text": template_text, "recent_messages": recent_messages or [], "soft_no_count": customer_state.get("soft_no_count", 0), "confusion_count": customer_state.get("confusion_count", 0), "appointment_status": customer_state.get("appointment_status", "NONE"), "status": "dnc" if customer_state.get("opted_out") else "active", "out_of_hours": _is_out_of_hours(), "opening_hours_summary": "Monday–Friday: 8:30am–6:00pm, Saturday: 9:00am–5:00pm, Sunday: closed", "today_date": datetime.now(timezone.utc).strftime("%Y-%m-%d"), "dealer_name": "Lincoln Audi", "escalation_reason": customer_state.get("escalation_reason"), "customer_message": inbound_text, } return context def _is_out_of_hours() -> bool: """ Check if current time is outside showroom opening hours. Opening hours: - Monday–Friday: 8:30am–6:00pm - Saturday: 9:00am–5:00pm - Sunday: closed """ now = datetime.now(timezone.utc) weekday = now.weekday() hour = now.hour minute = now.minute current_time = hour + minute / 60.0 if weekday == 6: return True elif weekday == 5: return current_time < 9.0 or current_time >= 17.0 else: return current_time < 8.5 or current_time >= 18.0 def _parse_llm_response(response_text: str) -> Dict[str, Any]: """ Parse the LLM's JSON response. Expected format: { "reply": "...", "tone_tag": "...", "intent_tag": "...", "use_template_key": "..." } """ try: start_idx = response_text.find("{") end_idx = response_text.rfind("}") + 1 if start_idx >= 0 and end_idx > start_idx: json_str = response_text[start_idx:end_idx] parsed = json.loads(json_str) if "reply" not in parsed: raise ValueError("Missing 'reply' field in LLM response") if parsed.get("tone_tag") not in VALID_TONE_TAGS: parsed["tone_tag"] = "general_engagement" if parsed.get("intent_tag") not in VALID_INTENT_TAGS: parsed["intent_tag"] = "provide_information" return parsed else: raise ValueError("No JSON object found in response") except (json.JSONDecodeError, ValueError) as e: logger.error(f"Failed to parse LLM response: {e}") logger.debug(f"Raw response: {response_text}") return { "reply": response_text.strip(), "tone_tag": "general_engagement", "intent_tag": "provide_information", "use_template_key": "FALLBACK", } def generate_llm_reply( customer_state: Dict[str, Any], inbound_text: str, tier: str, template_key: str, template_text: str, recent_messages: Optional[list] = None, ) -> Tuple[str, Dict[str, Any]]: """ Generate an LLM-powered reply using the template as the "spine". This function is called by the URE when LLM mode is enabled. It preserves the existing return shape but adds tone_tag and intent_tag to updated_state. Args: customer_state: Current customer state from URE inbound_text: Customer's WhatsApp message tier: Classified tier (T1_GENERAL, T2_SOFT_NO, etc.) template_key: Template key selected by URE template_text: Template text with placeholders resolved recent_messages: Optional list of recent conversation messages Returns: Tuple of (reply_text, updated_state) updated_state includes: - All existing fields (soft_no_count, confusion_count, etc.) - NEW: tone_tag (string) - NEW: intent_tag (string) - last_reply_type: "llm" """ new_state = customer_state.copy() new_state["last_message"] = inbound_text if not LLM_ENABLED: new_state["last_reply_type"] = "template" new_state["tone_tag"] = None new_state["intent_tag"] = None return (template_text, new_state) client = _get_openai_client() if not client: logger.warning("OpenAI client not available, falling back to template") new_state["last_reply_type"] = "template_fallback" new_state["tone_tag"] = None new_state["intent_tag"] = None return (template_text, new_state) context = _build_llm_context( customer_state=customer_state, inbound_text=inbound_text, tier=tier, template_key=template_key, template_text=template_text, recent_messages=recent_messages, ) try: response = client.chat.completions.create( model=OPENAI_MODEL, messages=[ {"role": "system", "content": LIA_SYSTEM_PROMPT}, {"role": "user", "content": json.dumps(context, indent=2)}, ], temperature=0.7, max_tokens=500, ) raw_response = response.choices[0].message.content parsed = _parse_llm_response(raw_response) reply_text = parsed.get("reply", template_text) tone_tag = parsed.get("tone_tag", "general_engagement") intent_tag = parsed.get("intent_tag", "provide_information") new_state["last_reply_type"] = "llm" new_state["tone_tag"] = tone_tag new_state["intent_tag"] = intent_tag logger.info(f"LLM reply generated: tier={tier}, template={template_key}, tone={tone_tag}, intent={intent_tag}") return (reply_text, new_state) except Exception as e: logger.error(f"LLM generation failed: {e}") new_state["last_reply_type"] = "template_fallback" new_state["tone_tag"] = None new_state["intent_tag"] = None return (template_text, new_state) async def generate_llm_reply_async( customer_state: Dict[str, Any], inbound_text: str, tier: str, template_key: str, template_text: str, recent_messages: Optional[list] = None, ) -> Tuple[str, Dict[str, Any]]: """ Async version of generate_llm_reply for use in async contexts. Uses anyio.to_thread.run_sync to run the synchronous OpenAI call in a thread pool. """ def _sync_call(): return generate_llm_reply( customer_state=customer_state, inbound_text=inbound_text, tier=tier, template_key=template_key, template_text=template_text, recent_messages=recent_messages, ) return await anyio.to_thread.run_sync(_sync_call) def build_ure_context( tier: str, stage: str, asset_type: str = "new", template_key: Optional[str] = None, template_text: Optional[str] = None, appointment_status: str = "NONE", soft_no_count: int = 0, confusion_count: int = 0, customer_name: Optional[str] = None, customer_message: Optional[str] = None, needs_human_review: bool = False, escalation_reason: Optional[str] = None, ) -> str: """ Build the URE context string to pass to the LLM (legacy format). This provides all the structured information the System Prompt needs. Note: This is the legacy string format. For new code, use _build_llm_context() which returns a structured dict. """ context_parts = [ f"tier: {tier}", f"stage: {stage}", f"asset_type: {asset_type}", f"template_key: {template_key or 'none'}", f"appointment_status: {appointment_status}", f"soft_no_count: {soft_no_count}", f"confusion_count: {confusion_count}", f"customer_name: {customer_name or 'there'}", f"needs_human_review: {needs_human_review}", ] if escalation_reason: context_parts.append(f"escalation_reason: {escalation_reason}") if template_text: context_parts.append(f"\ntemplate_text:\n{template_text}") if customer_message: context_parts.append(f"\ncustomer_message:\n{customer_message}") return "\n".join(context_parts) async def generate_lia_reply_llm( ure_context: str, model: str = "gpt-4o-mini", ) -> Optional[str]: """ Generate a LIA reply using OpenAI with System Prompt. Args: ure_context: The structured context string from build_ure_context() model: OpenAI model to use (default: gpt-4o-mini) Returns: The LLM-generated reply text, or None if OpenAI is unavailable Note: If OpenAI is unavailable, this returns None and the caller should fall back to the template-based URE reply. """ client = _get_openai_client() if not client: logger.warning("[LIA] OpenAI not available - returning None for LLM reply") return None def _call_openai(): return client.chat.completions.create( model=model, messages=[ {"role": "system", "content": LIA_SYSTEM_PROMPT}, {"role": "user", "content": ure_context}, ], max_tokens=500, temperature=0.7, ) try: completion = await anyio.to_thread.run_sync(_call_openai) reply = completion.choices[0].message.content.strip() logger.debug("[LIA] LLM generated reply: %s chars", len(reply)) return reply except Exception as e: logger.error("[LIA] OpenAI call failed: %s", e) return None async def generate_lia_reply_with_fallback( ure_context: str, template_fallback: str, model: str = "gpt-4o-mini", ) -> str: """ Generate a LIA reply using LLM, with fallback to template. This is the recommended function to use - it tries LLM first and falls back to the provided template if LLM fails. Args: ure_context: The structured context string from build_ure_context() template_fallback: The template-based reply to use if LLM fails model: OpenAI model to use Returns: The reply text (LLM-generated or template fallback) """ llm_reply = await generate_lia_reply_llm(ure_context, model) if llm_reply: return llm_reply return template_fallback def is_llm_enabled() -> bool: """Check if LLM mode is enabled.""" return LLM_ENABLED and OPENAI_AVAILABLE and bool(os.environ.get("OPENAI_API_KEY")) if __name__ == "__main__": print("=" * 70) print("LIA LLM LAYER - TEST") print("=" * 70) print(f"LLM Enabled: {LLM_ENABLED}") print(f"OpenAI Available: {OPENAI_AVAILABLE}") print(f"API Key Set: {bool(os.environ.get('OPENAI_API_KEY'))}") print(f"Model: {OPENAI_MODEL}") test_state = { "customer_id": "test-123", "customer_name": "Sarah Johnson", "vehicle_model": "A3 Sportback", "target_model": "Q4 e-tron", "stage": "12", "soft_no_count": 0, "confusion_count": 0, "appointment_status": "NONE", "opted_out": False, "campaign_type": "EOT", } test_template = ( "Hi [customer_name], it's Lia at Lincoln Audi.\n\n" "Now you're into the final year on your [vehicle_model], it's a good time to look at your options. " "Your Audi holds strong value and you've still got your existing-customer savings available.\n\n" "Would a weekday or Saturday work better for a quick visit?" ) reply, updated_state = generate_llm_reply( customer_state=test_state, inbound_text="Hi, I got your message. What are my options?", tier="T1_GENERAL", template_key="EOT_12_INTRO_NEW", template_text=test_template, recent_messages=[], ) print(f"\n{'=' * 70}") print("TEST RESULT:") print(f"{'=' * 70}") print(f"Reply:\n{reply}") print(f"\nUpdated State:") print(f" last_reply_type: {updated_state.get('last_reply_type')}") print(f" tone_tag: {updated_state.get('tone_tag')}") print(f" intent_tag: {updated_state.get('intent_tag')}")