llm_robust.py

"""

Robust LLM wrapper with aggressive timeout protection and lightweight fallbacks

Prevents node.js/model server crashes during summarization

"""

import os
import signal
import time
from contextlib import contextmanager
from typing import Tuple, Dict, Optional

class TimeoutException(Exception):
    pass

@contextmanager
def timeout(seconds):
    """Context manager for enforcing hard timeouts"""
    def signal_handler(signum, frame):
        raise TimeoutException(f"Operation timed out after {seconds} seconds")

    # Set the signal handler
    old_handler = signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)

    try:
        yield
    finally:
        signal.alarm(0)
        signal.signal(signal.SIGALRM, old_handler)

def query_llm_with_timeout(

    prompt: str,

    user_context: str,

    interviewee_type: str,

    extract_structured: bool = True,

    is_summary: bool = False,

    max_timeout: int = 60  # Reduced from 120 to 60 seconds

) -> Tuple[str, Dict]:
    """

    Query LLM with aggressive timeout protection

    Falls back to lightweight processing if heavy models fail

    """

    print(f"[LLM] Starting {'summary' if is_summary else 'analysis'} generation...")
    print(f"[LLM] Timeout limit: {max_timeout}s")

    # Import here to avoid circular dependencies
    from llm import query_llm

    try:
        # Try with timeout protection
        with timeout(max_timeout):
            result = query_llm(
                prompt,
                user_context,
                interviewee_type,
                extract_structured=extract_structured,
                is_summary=is_summary
            )
            print(f"[LLM] ✓ Completed successfully")
            return result

    except TimeoutException as e:
        print(f"[LLM] ✗ Timeout after {max_timeout}s")
        print(f"[LLM] Generating lightweight fallback...")

        # Generate lightweight fallback
        if is_summary:
            return generate_lightweight_summary(prompt, interviewee_type)
        else:
            return generate_lightweight_analysis(prompt, interviewee_type)

    except Exception as e:
        print(f"[LLM] ✗ Error: {type(e).__name__}: {str(e)}")
        print(f"[LLM] Generating emergency fallback...")

        # Emergency fallback
        if is_summary:
            return generate_emergency_summary(interviewee_type)
        else:
            return generate_emergency_analysis(interviewee_type)

def generate_lightweight_summary(prompt: str, interviewee_type: str) -> Tuple[str, Dict]:
    """

    Generate a lightweight summary without heavy LLM processing

    Extracts key points from the prompt itself

    """

    print("[Fallback] Creating lightweight summary from prompt data...")

    # Extract numbers from prompt
    import re

    # Find participant counts
    participant_matches = re.findall(r'(\d+)\s+(?:participants|transcripts|interviews)', prompt, re.IGNORECASE)
    num_participants = int(participant_matches[0]) if participant_matches else 0

    # Find percentages
    percentages = re.findall(r'(\d+)%', prompt)

    # Find mentions of conditions/themes
    lines = prompt.split('\n')
    themes = []
    for line in lines:
        if ':' in line and not line.strip().startswith(('#', '-', '*', '=')):
            parts = line.split(':', 1)
            if len(parts) == 2:
                theme = parts[0].strip()
                if len(theme) < 50:  # Reasonable theme length
                    themes.append(theme)

    summary = f"""LIGHTWEIGHT SUMMARY REPORT

(Generated due to LLM timeout - data extracted from available information)



SAMPLE OVERVIEW:

Total {interviewee_type} interviews analyzed: {num_participants}



KEY OBSERVATIONS:

This analysis is based on structured data extraction rather than full LLM synthesis.

For detailed narrative analysis, please:

1. Reduce the number of transcripts being analyzed simultaneously

2. Check LLM server (LMStudio/HuggingFace) connectivity

3. Consider using a lighter model



DATA EXTRACTED:

"""

    if themes:
        summary += f"\nIdentified themes ({len(themes)} total):\n"
        for i, theme in enumerate(themes[:10], 1):
            summary += f"{i}. {theme}\n"

    if percentages:
        summary += f"\nPercentages mentioned: {', '.join(set(percentages))}%\n"

    summary += f"""



RECOMMENDATIONS:

1. Review the CSV output file for structured data

2. Individual transcript analyses contain detailed information

3. For full narrative synthesis, retry with:

   - Fewer transcripts per batch

   - Increased timeout limits

   - Verified LLM server connectivity



This lightweight summary preserves data integrity while avoiding server crashes.

For production use, ensure LLM backend is properly configured and responsive.

"""

    return summary, {}

def generate_emergency_summary(interviewee_type: str) -> Tuple[str, Dict]:
    """Emergency fallback when even lightweight processing fails"""

    summary = f"""EMERGENCY FALLBACK REPORT



LLM PROCESSING UNAVAILABLE



The system encountered critical errors during summary generation.

All structured data has been preserved in the CSV output file.



IMMEDIATE ACTIONS REQUIRED:

1. Check LLM server status (LMStudio/HuggingFace API)

2. Verify network connectivity

3. Review console logs for specific error messages

4. Check available system memory



DATA PRESERVATION:

✓ Individual transcript analyses completed

✓ Structured data extracted to CSV

✓ Quality scores calculated

✗ Cross-transcript narrative synthesis failed



NEXT STEPS:

1. Review the CSV file: Contains all extracted structured data

2. Check individual transcript results below this summary

3. Resolve LLM connectivity issues

4. Re-run summary generation once service is restored



This emergency report ensures no data loss while protecting system stability.

"""

    return summary, {}

def generate_lightweight_analysis(prompt: str, interviewee_type: str) -> Tuple[str, Dict]:
    """Lightweight analysis without heavy LLM"""

    # Extract basic structured data from prompt
    import re

    structured_data = {}

    if interviewee_type == "HCP":
        # Extract medical terms
        medical_pattern = r'\b(diagnos\w+|prescri\w+|treatment|medication|therapy)\b'
        terms = re.findall(medical_pattern, prompt, re.IGNORECASE)
        structured_data = {
            "diagnoses": list(set([t for t in terms if 'diagnos' in t.lower()])),
            "prescriptions": list(set([t for t in terms if 'prescri' in t.lower()])),
            "treatment_rationale": [],
            "key_insights": [f"Lightweight extraction: {len(terms)} medical terms identified"]
        }

    elif interviewee_type == "Patient":
        # Extract patient terms
        patient_pattern = r'\b(symptom|pain|concern|treatment|medication|side effect)\b'
        terms = re.findall(patient_pattern, prompt, re.IGNORECASE)
        structured_data = {
            "symptoms": list(set([t for t in terms if 'symptom' in t.lower() or 'pain' in t.lower()])),
            "concerns": [],
            "treatment_response": [],
            "key_insights": [f"Lightweight extraction: {len(terms)} patient-related terms identified"]
        }
    else:
        structured_data = {
            "key_insights": ["Lightweight analysis - full LLM processing unavailable"]
        }

    analysis = f"""[LIGHTWEIGHT ANALYSIS]

Due to LLM timeout, basic pattern extraction was used.

Structured data contains {sum(len(v) for v in structured_data.values() if isinstance(v, list))} items.



For full analysis, ensure LLM server is responsive.

"""

    return analysis, structured_data

def generate_emergency_analysis(interviewee_type: str) -> Tuple[str, Dict]:
    """Emergency fallback for individual transcript analysis"""

    structured_data = {
        "key_insights": ["Emergency fallback - LLM processing failed"],
        "processing_status": "FALLBACK_MODE"
    }

    analysis = "[EMERGENCY FALLBACK] LLM processing unavailable. Minimal data extraction performed."

    return analysis, structured_data

# Utility function to test LLM connectivity before processing
def test_llm_connection(timeout_seconds: int = 10) -> bool:
    """Test if LLM backend is responsive"""

    print("[LLM] Testing backend connectivity...")

    test_prompt = "Test"

    try:
        with timeout(timeout_seconds):
            from llm import query_llm
            result = query_llm(
                test_prompt,
                "",
                "Other",
                extract_structured=False,
                is_summary=False
            )
            print("[LLM] ✓ Backend responsive")
            return True
    except Exception as e:
        print(f"[LLM] ✗ Backend not responsive: {e}")
        return False