""" Advanced LLM Reasoning Engine for Query Analysis and Response Generation Handles complex reasoning, clause referencing, and structured response generation """ import os import json import logging import re from datetime import datetime from typing import List, Dict, Any, Optional, Tuple from dataclasses import dataclass from pathlib import Path # LLM and AI libraries from llama_cpp import Llama from transformers import pipeline import torch import numpy as np # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class ReasoningResult: """Represents the result of LLM reasoning""" decision: str # approved, denied, pending, unclear confidence_score: float justification: str relevant_clauses: List[str] amount: Optional[float] = None waiting_period: Optional[str] = None conditions: List[str] = None exclusions: List[str] = None required_documents: List[str] = None processing_time: Optional[str] = None reasoning_steps: List[str] = None source_references: List[Dict[str, Any]] = None @dataclass class ClauseReference: """Represents a reference to a specific policy clause""" clause_id: str clause_text: str relevance_score: float page_number: Optional[int] = None section_type: Optional[str] = None class AdvancedLLMReasoning: """Advanced LLM reasoning engine with clause referencing and structured analysis""" def __init__(self, model_path: str = None, use_gpu: bool = True, max_tokens: int = 2048): self.model_path = model_path self.use_gpu = use_gpu self.max_tokens = max_tokens # Initialize LLM self._initialize_llm() # Initialize reasoning patterns self._initialize_reasoning_patterns() # Initialize clause extraction self._initialize_clause_extraction() logger.info("Advanced LLM Reasoning Engine initialized") def _initialize_llm(self): """Initialize the LLM model""" try: # Set default model path if none provided if self.model_path is None: self.model_path = "./mistral-7b-instruct-v0.1.Q4_K_M.gguf" # Check if model file exists if not os.path.exists(self.model_path): logger.warning(f"Model file not found: {self.model_path}") logger.info("LLM reasoning will use fallback mode without local model") self.llm = None return # Initialize Llama model with more robust configuration try: self.llm = Llama( model_path=self.model_path, n_ctx=4096, n_gpu_layers=50 if self.use_gpu else 0, verbose=False, use_mmap=True, use_mlock=False, seed=42 ) logger.info(f"LLM model loaded successfully with GPU: {self.use_gpu}") except Exception as gpu_error: logger.warning(f"GPU loading failed: {gpu_error}") # Try CPU-only configuration try: self.llm = Llama( model_path=self.model_path, n_ctx=2048, n_gpu_layers=0, # CPU only verbose=False, use_mmap=True, use_mlock=False, seed=42 ) logger.info("LLM model loaded successfully with CPU configuration") except Exception as cpu_error: logger.error(f"CPU loading also failed: {cpu_error}") self.llm = None logger.info(f"LLM model loaded: {self.model_path}") except Exception as e: logger.error(f"Error initializing LLM: {e}") logger.info("LLM reasoning will use fallback mode") self.llm = None def _initialize_reasoning_patterns(self): """Initialize reasoning patterns and templates""" try: # Reasoning templates for different query types and domains self.reasoning_templates = { 'coverage_check': """ You are an expert document analyst. Analyze the provided document information carefully to answer the user's query. USER QUERY: {query} DOCUMENT SECTIONS: {context} INSTRUCTIONS: 1. Read and understand the document sections provided 2. Look for specific clauses, conditions, limitations, and exclusions 3. Pay attention to time periods, limits, and restrictions 4. Consider both what is allowed/permitted AND what is explicitly prohibited/excluded 5. Base your decision ONLY on the information provided in the document ANALYSIS REQUIREMENTS: - Decision: APPROVED, DENIED, CONDITIONAL, or PENDING (be precise based on document text) - Confidence Score: 0.0 to 1.0 (higher if document clearly states the answer) - Justification: Quote specific document text and explain your reasoning - Relevant Clauses: List the exact document sections that support your decision - Conditions: Any specific conditions, time limits, or restrictions mentioned - Exclusions: What is explicitly excluded or not permitted - Required Documents: Documents mentioned as required for this type of request IMPORTANT: If the document explicitly states something is NOT permitted or has limitations, you must reflect that in your decision. Do not assume approval unless the document clearly states it. Respond in valid JSON format. """, 'legal_compliance': """ You are an expert legal compliance analyst. Analyze the provided legal documents to determine compliance status. USER QUERY: {query} LEGAL DOCUMENT SECTIONS: {context} INSTRUCTIONS: 1. Read and understand the legal document sections provided 2. Look for specific regulations, requirements, and compliance criteria 3. Pay attention to deadlines, obligations, and legal requirements 4. Consider both what is required AND what is explicitly prohibited 5. Base your decision ONLY on the information provided in the legal documents ANALYSIS REQUIREMENTS: - Decision: COMPLIANT, NON_COMPLIANT, CONDITIONAL, or NEEDS_REVIEW - Confidence Score: 0.0 to 1.0 (higher if document clearly states the answer) - Justification: Quote specific legal text and explain your reasoning - Relevant Regulations: List the exact legal sections that apply - Requirements: Any specific legal requirements or obligations mentioned - Violations: What would constitute non-compliance - Required Actions: Steps needed to achieve or maintain compliance Respond in valid JSON format. """, 'hr_policy': """ You are an expert HR policy analyst. Analyze the provided HR documents to answer employee-related queries. USER QUERY: {query} HR DOCUMENT SECTIONS: {context} INSTRUCTIONS: 1. Read and understand the HR document sections provided 2. Look for specific policies, procedures, and employee rights 3. Pay attention to eligibility criteria, time limits, and benefits 4. Consider both what is permitted AND what is explicitly prohibited 5. Base your decision ONLY on the information provided in the HR documents ANALYSIS REQUIREMENTS: - Decision: APPROVED, DENIED, CONDITIONAL, or PENDING_REVIEW - Confidence Score: 0.0 to 1.0 (higher if document clearly states the answer) - Justification: Quote specific policy text and explain your reasoning - Relevant Policies: List the exact policy sections that apply - Eligibility: Any specific eligibility criteria or conditions - Benefits: What benefits or entitlements are available - Required Documentation: Documents needed to support the request Respond in valid JSON format. """, 'contract_analysis': """ You are an expert contract analyst. Analyze the provided contract documents to answer contract-related queries. USER QUERY: {query} CONTRACT DOCUMENT SECTIONS: {context} INSTRUCTIONS: 1. Read and understand the contract document sections provided 2. Look for specific terms, conditions, and contractual obligations 3. Pay attention to deadlines, deliverables, and performance requirements 4. Consider both what is required AND what is explicitly prohibited 5. Base your decision ONLY on the information provided in the contract documents ANALYSIS REQUIREMENTS: - Decision: PERMITTED, PROHIBITED, CONDITIONAL, or NEEDS_CLARIFICATION - Confidence Score: 0.0 to 1.0 (higher if contract clearly states the answer) - Justification: Quote specific contract text and explain your reasoning - Relevant Clauses: List the exact contract sections that apply - Obligations: Any specific contractual obligations or requirements - Restrictions: What is explicitly prohibited or limited - Remedies: Available remedies or consequences for non-compliance Respond in valid JSON format. """, 'claim_processing': """ Analyze the claim processing requirements: Query: {query} Policy Information: {context} Please provide: 1. Decision: APPROVED, DENIED, or PENDING 2. Confidence Score: 0.0 to 1.0 3. Justification: Detailed explanation 4. Required Documents: List of needed documents 5. Processing Time: Expected processing duration 6. Steps: Claim processing steps 7. Relevant Clauses: Policy clauses for claims Respond in JSON format. """, 'policy_review': """ Review the policy terms and conditions: Query: {query} Policy Content: {context} Please provide: 1. Decision: CLEAR, UNCLEAR, or NEEDS_CLARIFICATION 2. Confidence Score: 0.0 to 1.0 3. Justification: Detailed explanation 4. Relevant Clauses: Specific policy sections 5. Key Points: Important policy points 6. Recommendations: Suggested actions Respond in JSON format. """ } # Decision mapping for different domains self.decision_mapping = { # Insurance domain 'COVERED': 'approved', 'NOT_COVERED': 'denied', 'CONDITIONAL': 'pending', 'APPROVED': 'approved', 'REJECTED': 'denied', 'DENIED': 'denied', 'PENDING': 'pending', 'PENDING_REVIEW': 'pending', # Legal compliance domain 'COMPLIANT': 'approved', 'NON_COMPLIANT': 'denied', 'NEEDS_REVIEW': 'pending', # Contract domain 'PERMITTED': 'approved', 'PROHIBITED': 'denied', 'NEEDS_CLARIFICATION': 'pending', # Policy review 'CLEAR': 'approved', 'UNCLEAR': 'pending' } logger.info("Reasoning patterns initialized") except Exception as e: logger.error(f"Error initializing reasoning patterns: {e}") def _initialize_clause_extraction(self): """Initialize clause extraction patterns""" try: # Patterns for extracting policy clauses self.clause_patterns = { 'coverage_clause': [ r'coverage.*?shall.*?include', r'covered.*?expenses.*?include', r'benefits.*?shall.*?cover', r'policy.*?covers.*?following' ], 'exclusion_clause': [ r'exclusions.*?include', r'not.*?covered.*?following', r'excluded.*?from.*?coverage', r'coverage.*?does.*?not.*?include' ], 'condition_clause': [ r'conditions.*?precedent', r'requirements.*?for.*?coverage', r'must.*?meet.*?following', r'coverage.*?subject.*?to' ], 'amount_clause': [ r'maximum.*?benefit.*?\$[\d,]+', r'coverage.*?limit.*?\$[\d,]+', r'benefit.*?amount.*?\$[\d,]+', r'up.*?to.*?\$[\d,]+' ], 'waiting_period': [ r'waiting.*?period.*?\d+.*?(days?|weeks?|months?)', r'coverage.*?begins.*?after.*?\d+', r'benefits.*?available.*?after.*?\d+' ] } logger.info("Clause extraction patterns initialized") except Exception as e: logger.error(f"Error initializing clause extraction: {e}") def analyze_query(self, query: str, context: List[Dict[str, Any]], query_type: str = 'coverage_check') -> ReasoningResult: """Main method to analyze a query using LLM reasoning with detailed analysis""" try: # Prepare context from relevant chunks formatted_context = self._format_context_for_llm(context) # Create comprehensive reasoning prompt prompt = f""" You are an expert insurance policy analyzer. Analyze the following query against the provided policy documents. User Query: {query} Relevant Policy Clauses: {formatted_context} Please provide a structured analysis in the following JSON format: {{ "decision": "approved/rejected/conditional", "amount": , "justification": "", "relevant_clauses": [""], "confidence_score": <0.0 to 1.0>, "conditions": [""], "exclusions": [""], "waiting_period": "", "required_documents": [""], "reasoning_steps": [""] }} Base your decision on: 1. Policy coverage and exclusions 2. Eligibility criteria 3. Waiting periods 4. Pre-existing conditions 5. Specific terms and conditions 6. Time limitations and restrictions 7. Required documentation IMPORTANT: - Quote specific policy text in your justification - Reference exact clause IDs from the provided context - If the policy explicitly states limitations, reflect them accurately - Provide detailed reasoning, not just yes/no answers - Consider both what is covered AND what is excluded - Look for specific amounts, time periods, and conditions JSON Response: """ # Generate response using LLM response = self._generate_llm_response(prompt) # Parse the response parsed_result = self._parse_llm_response(response, query_type) # Extract clause references clause_references = self._extract_clause_references(context, parsed_result) # Build final result with comprehensive analysis result = ReasoningResult( decision=parsed_result.get('decision', 'pending'), confidence_score=parsed_result.get('confidence_score', 0.5), justification=parsed_result.get('justification', 'Unable to determine'), relevant_clauses=parsed_result.get('relevant_clauses', []), amount=parsed_result.get('amount'), waiting_period=parsed_result.get('waiting_period'), conditions=parsed_result.get('conditions', []), exclusions=parsed_result.get('exclusions', []), required_documents=parsed_result.get('required_documents', []), processing_time=parsed_result.get('processing_time'), reasoning_steps=parsed_result.get('reasoning_steps', []), source_references=clause_references ) logger.info(f"Query analysis completed: {result.decision} ({result.confidence_score:.2f})") return result except Exception as e: logger.error(f"Error analyzing query: {e}") return self._create_fallback_result(query) def _format_context_for_llm(self, context: List[Dict[str, Any]]) -> str: """Format context for LLM consumption""" try: formatted_parts = [] for i, item in enumerate(context, 1): content = item.get('content', '') source = item.get('source_file', 'Unknown') similarity = item.get('similarity_score', 0.0) formatted_parts.append(f"Section {i} (Source: {source}, Relevance: {similarity:.2f}):\n{content}\n") return "\n".join(formatted_parts) except Exception as e: logger.error(f"Error formatting context: {e}") return str(context) def _generate_llm_response(self, prompt: str) -> str: """Generate response using the LLM""" try: # Check if LLM is available if self.llm is None: logger.warning("LLM not available, using fallback response") return self._generate_fallback_response(prompt) # Create system prompt system_prompt = """You are an expert insurance policy analyzer. Your job is to: 1. Carefully read and understand the policy document sections provided 2. Answer the user's query based ONLY on the information in the policy document 3. Look for specific clauses, conditions, limitations, and exclusions 4. Pay attention to time periods, coverage limits, and restrictions 5. If the policy explicitly states something is NOT covered, you must say it's REJECTED 6. If there are specific conditions or limitations, you must mention them 7. Always respond in valid JSON format with accurate information 8. Quote specific policy text in your justification 9. Reference exact clause IDs from the provided context 10. Provide detailed reasoning, not just yes/no answers 11. Consider both what is covered AND what is excluded 12. Look for specific amounts, time periods, and conditions 13. Do not make assumptions - base your decision only on what the policy document states""" # Generate response response = self.llm.create_completion( prompt=f"{system_prompt}\n\n{prompt}", max_tokens=self.max_tokens, temperature=0.1, stop=["```", "Human:", "Assistant:"] ) return response['choices'][0]['text'].strip() except Exception as e: logger.error(f"Error generating LLM response: {e}") return self._generate_fallback_response(prompt) def _parse_llm_response(self, response: str, query_type: str) -> Dict[str, Any]: """Parse the LLM response into structured data""" try: # Try to extract JSON from response json_match = re.search(r'\{.*\}', response, re.DOTALL) if json_match: json_str = json_match.group(0) parsed = json.loads(json_str) else: # Fallback parsing parsed = self._fallback_parse_response(response) # Map decision to standard format if 'decision' in parsed: parsed['decision'] = self.decision_mapping.get( parsed['decision'].upper(), 'pending' ) # Ensure confidence score is float if 'confidence_score' in parsed: try: parsed['confidence_score'] = float(parsed['confidence_score']) except: parsed['confidence_score'] = 0.5 return parsed except Exception as e: logger.error(f"Error parsing LLM response: {e}") return { 'decision': 'pending', 'confidence_score': 0.5, 'justification': 'Unable to parse response', 'relevant_clauses': [] } def _fallback_parse_response(self, response: str) -> Dict[str, Any]: """Fallback parsing when JSON extraction fails""" try: result = { 'decision': 'pending', 'confidence_score': 0.5, 'justification': response[:500], 'relevant_clauses': [] } # Try to extract decision if 'covered' in response.lower(): result['decision'] = 'approved' elif 'not covered' in response.lower() or 'excluded' in response.lower(): result['decision'] = 'denied' # Try to extract confidence confidence_match = re.search(r'confidence.*?(\d+\.?\d*)', response, re.IGNORECASE) if confidence_match: try: result['confidence_score'] = float(confidence_match.group(1)) except: pass return result except Exception as e: logger.error(f"Error in fallback parsing: {e}") return { 'decision': 'pending', 'confidence_score': 0.5, 'justification': 'Analysis failed', 'relevant_clauses': [] } def _extract_clause_references(self, context: List[Dict[str, Any]], parsed_result: Dict[str, Any]) -> List[Dict[str, Any]]: """Extract specific clause references from context""" try: references = [] for item in context: content = item.get('content', '') source = item.get('source_file', 'Unknown') # Extract clauses using patterns for clause_type, patterns in self.clause_patterns.items(): for pattern in patterns: matches = re.findall(pattern, content, re.IGNORECASE) for match in matches: references.append({ 'clause_type': clause_type, 'clause_text': match, 'source_file': source, 'relevance_score': item.get('similarity_score', 0.0) }) return references except Exception as e: logger.error(f"Error extracting clause references: {e}") return [] def _generate_fallback_response(self, prompt: str) -> str: """Generate a fallback response when LLM is not available""" try: # Extract query and context from prompt query_match = re.search(r'USER QUERY:\s*(.+?)(?=\n\n|$)', prompt, re.DOTALL | re.IGNORECASE) context_match = re.search(r'POLICY DOCUMENT SECTIONS:\s*(.+?)(?=\n\n|$)', prompt, re.DOTALL | re.IGNORECASE) query = query_match.group(1).strip() if query_match else "Unknown query" context = context_match.group(1).strip() if context_match else "No policy context provided" # Analyze the context for key information context_lower = context.lower() query_lower = query.lower() # Look for specific policy terms and conditions decision = "PENDING" confidence = 0.5 justification = "Unable to determine coverage without proper policy analysis" relevant_clauses = [] conditions = [] exclusions = [] # Check for coverage limitations and restrictions if any(term in context_lower for term in ['until first discharge', 'discharge from hospital', 'hospitalization period', 'limited to']): if any(term in query_lower for term in ['after discharge', 'post discharge', 'discharge', 'beyond']): decision = "REJECTED" confidence = 0.9 justification = "Policy explicitly states coverage is limited to hospitalization period until first discharge. Post-discharge care is not covered under this policy." relevant_clauses = ["newborn coverage", "discharge limitation", "hospitalization period"] conditions = ["Coverage only during hospitalization", "Until first discharge"] exclusions = ["Post-discharge care", "Outpatient newborn care"] waiting_period = "Until first discharge" required_documents = ["Hospital discharge summary", "Birth certificate"] # Check for waiting periods elif any(term in context_lower for term in ['waiting period', 'waiting periods', 'time requirement']): decision = "CONDITIONAL" confidence = 0.7 justification = "Coverage subject to waiting period requirements as specified in the policy" relevant_clauses = ["waiting periods", "time requirements"] conditions = ["Waiting period must be satisfied"] waiting_period = "As specified in policy" # Check for exclusions elif any(term in context_lower for term in ['not covered', 'excluded', 'exclusions', 'prohibited', 'not permitted']): decision = "REJECTED" confidence = 0.8 justification = "Policy explicitly excludes or prohibits this type of coverage" relevant_clauses = ["exclusions", "prohibitions"] exclusions = ["Excluded per policy terms"] # Check for covered items elif any(term in context_lower for term in ['covered', 'coverage', 'benefits', 'permitted', 'allowed']): decision = "APPROVED" confidence = 0.7 justification = "Policy indicates this type of coverage is permitted" relevant_clauses = ["coverage", "benefits"] conditions = ["Subject to policy terms"] # Default case - analyze based on context content else: # Look for positive indicators in context if any(term in context_lower for term in ['covered', 'coverage', 'benefits', 'permitted']): decision = "APPROVED" confidence = 0.6 justification = "Policy appears to provide coverage for this type of request" relevant_clauses = ["general coverage"] conditions = ["Subject to policy terms"] elif any(term in context_lower for term in ['excluded', 'not covered', 'prohibited']): decision = "REJECTED" confidence = 0.6 justification = "Policy appears to exclude this type of coverage" relevant_clauses = ["exclusions"] exclusions = ["Excluded per policy terms"] else: # Try to extract any relevant information from the context if len(context) > 0: decision = "CONDITIONAL" confidence = 0.5 justification = f"Based on the available policy information, this request requires further review. Found {len(context)} relevant policy sections." relevant_clauses = ["policy sections found"] conditions = ["Policy review required", "Additional documentation may be needed"] else: decision = "PENDING" confidence = 0.3 justification = "No relevant policy information found. Please ensure the policy document has been properly uploaded and processed." relevant_clauses = ["no policy data"] conditions = ["Policy document upload required"] # Check for specific amounts in context amount_match = re.search(r'(\d+(?:,\d+)*(?:\.\d+)?)\s*(?:rs?|rupees?|inr|\$)', context_lower) if amount_match: amount = float(amount_match.group(1).replace(',', '')) else: amount = None return json.dumps({ "decision": decision, "confidence_score": confidence, "justification": justification, "relevant_clauses": relevant_clauses, "amount": amount, "waiting_period": waiting_period, "conditions": conditions, "exclusions": exclusions, "required_documents": ["Policy document", "Claim form"] }) except Exception as e: logger.error(f"Error generating fallback response: {e}") return json.dumps({ "decision": "PENDING", "confidence_score": 0.5, "justification": "Unable to analyze policy information", "relevant_clauses": [], "conditions": [], "exclusions": [] }) def _create_fallback_result(self, query: str) -> ReasoningResult: """Create a fallback result when analysis fails""" return ReasoningResult( decision='pending', confidence_score=0.0, justification='Unable to analyze query due to technical issues', relevant_clauses=[], reasoning_steps=['Analysis failed'], source_references=[] ) def explain_decision(self, result: ReasoningResult) -> str: """Generate a human-readable explanation of the decision""" try: explanation_parts = [] # Main decision explanation_parts.append(f"Decision: {result.decision.upper()}") explanation_parts.append(f"Confidence: {result.confidence_score:.1%}") # Justification if result.justification: explanation_parts.append(f"\nJustification:\n{result.justification}") # Relevant clauses if result.relevant_clauses: explanation_parts.append(f"\nRelevant Policy Clauses:") for clause in result.relevant_clauses: explanation_parts.append(f"- {clause}") # Amount information if result.amount: explanation_parts.append(f"\nCoverage Amount: ${result.amount:,.2f}") # Waiting period if result.waiting_period: explanation_parts.append(f"\nWaiting Period: {result.waiting_period}") # Conditions if result.conditions: explanation_parts.append(f"\nConditions:") for condition in result.conditions: explanation_parts.append(f"- {condition}") # Exclusions if result.exclusions: explanation_parts.append(f"\nExclusions:") for exclusion in result.exclusions: explanation_parts.append(f"- {exclusion}") # Required documents if result.required_documents: explanation_parts.append(f"\nRequired Documents:") for doc in result.required_documents: explanation_parts.append(f"- {doc}") return "\n".join(explanation_parts) except Exception as e: logger.error(f"Error explaining decision: {e}") return f"Decision: {result.decision.upper()}\nConfidence: {result.confidence_score:.1%}\nJustification: {result.justification}" def validate_decision(self, result: ReasoningResult) -> bool: """Validate the reasoning result for consistency""" try: # Check if confidence score is valid if not (0.0 <= result.confidence_score <= 1.0): return False # Check if decision is valid valid_decisions = ['approved', 'denied', 'pending'] if result.decision not in valid_decisions: return False # Check if justification is provided if not result.justification or len(result.justification.strip()) < 10: return False return True except Exception as e: logger.error(f"Error validating decision: {e}") return False # Example usage if __name__ == "__main__": # Initialize reasoning engine reasoning_engine = AdvancedLLMReasoning() # Test query test_query = "Is heart surgery covered under this policy?" test_context = [ { 'content': 'This policy covers medical procedures including heart surgery up to $50,000.', 'source_file': 'policy.pdf', 'similarity_score': 0.9 } ] # Analyze query result = reasoning_engine.analyze_query(test_query, test_context, 'coverage_check') print(f"Decision: {result.decision}") print(f"Confidence: {result.confidence_score:.2f}") print(f"Justification: {result.justification}") # Explain decision explanation = reasoning_engine.explain_decision(result) print(f"\nExplanation:\n{explanation}")