gateway/modules/workflows/processing/adaptive/intentAnalyzer.py

# intentAnalyzer.py
# Intent analysis for adaptive React mode - AI-based, language-agnostic

import json
import logging
from typing import Dict, Any, List

logger = logging.getLogger(__name__)

class IntentAnalyzer:
    """Analyzes user intent using AI - language-agnostic and generic"""
    
    def __init__(self, services=None):
        self.services = services
    
    async def analyzeUserIntent(self, userPrompt: str, context: Any) -> Dict[str, Any]:
        """Analyzes user intent from prompt and context using AI"""
        try:
            # Use AI to analyze intent
            aiAnalysis = await self._analyzeIntentWithAI(userPrompt, context)
            if aiAnalysis:
                return aiAnalysis
            
            # Fallback to basic analysis if AI fails
            return self._createBasicIntentAnalysis(userPrompt)
            
        except Exception as e:
            logger.error(f"Error analyzing user intent: {str(e)}")
            return self._createDefaultIntentAnalysis(userPrompt)
    
    async def _analyzeIntentWithAI(self, userPrompt: str, context: Any) -> Dict[str, Any]:
        """Uses AI to analyze user intent - language-agnostic"""
        try:
            if not self.services or not hasattr(self.services, 'ai'):
                return None
            
            # Create AI analysis prompt
            analysisPrompt = f"""
You are an intent analyzer. Analyze the user's request to understand what they want delivered.

USER REQUEST: {userPrompt}

CONTEXT: {getattr(context.task_step, 'objective', '') if hasattr(context, 'task_step') and context.task_step else ''}

Analyze the user's intent and determine:
1. What type of data/content they want (numbers, text, documents, analysis, code, etc.)
2. What format they expect (raw data, formatted, structured, visual, etc.)
3. What quality requirements they have (accuracy, completeness, format)
4. What specific success criteria define completion

CRITICAL: Respond with ONLY the JSON object below. Do not include any explanatory text, analysis, or other content before or after the JSON.

{{
    "primaryGoal": "The main objective the user wants to achieve",
    "dataType": "numbers|text|documents|analysis|code|unknown",
    "expectedFormat": "raw_data|formatted|structured|visual|unknown",
    "qualityRequirements": {{
        "accuracyThreshold": 0.0-1.0,
        "completenessThreshold": 0.0-1.0,
        "formatRequirement": "any|formatted|raw|structured"
    }},
    "successCriteria": ["specific criterion 1", "specific criterion 2"],
    "confidenceScore": 0.0-1.0
}}
"""
            
            # Call AI service for analysis
            from modules.datamodels.datamodelAi import AiCallOptions, OperationType
            request_options = AiCallOptions()
            request_options.operationType = OperationType.GENERAL
            
            response = await self.services.ai.callAi(
                prompt=analysisPrompt,
                documents=None,
                options=request_options
            )
            
            # If first attempt fails, try with more explicit prompt
            if response and not self._isValidJsonResponse(response):
                logger.debug("First AI intent analysis attempt failed, retrying with explicit JSON-only prompt")
                explicitPrompt = f"""
{analysisPrompt}

IMPORTANT: You must respond with ONLY valid JSON. No explanations, no analysis, no text before or after. Just the JSON object.
"""
                response = await self.services.ai.callAi(
                    prompt=explicitPrompt,
                    documents=None,
                    options=request_options
                )
            
            if not response or not response.strip():
                logger.warning("AI intent analysis returned empty response")
                return None
            
            # Clean and extract JSON from response
            result = response.strip()
            logger.debug(f"AI intent analysis response length: {len(result)}")
            
            # Try to find JSON in the response with multiple strategies
            import re
            
            # Strategy 1: Look for JSON in markdown code blocks
            json_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', result, re.DOTALL)
            if json_match:
                result = json_match.group(1)
                logger.debug(f"Extracted JSON from markdown code block: {result[:200]}...")
            else:
                # Strategy 2: Look for JSON object with proper structure
                json_match = re.search(r'\{[^{}]*"primaryGoal"[^{}]*\}', result, re.DOTALL)
                if not json_match:
                    # Strategy 3: Look for any JSON object
                    json_match = re.search(r'\{.*\}', result, re.DOTALL)
                
                if not json_match:
                    logger.warning(f"All AI intent analysis attempts failed - no JSON found in response: {result[:200]}...")
                    logger.debug(f"Full AI response: {result}")
                    return None
                
                result = json_match.group(0)
                logger.debug(f"Extracted JSON directly: {result[:200]}...")
            
            try:
                aiResult = json.loads(result)
                logger.info("AI intent analysis JSON parsed successfully")
                return aiResult
                
            except json.JSONDecodeError as json_error:
                logger.warning(f"All AI intent analysis attempts failed - invalid JSON: {str(json_error)}")
                logger.debug(f"JSON content: {result}")
                return None
            
            return None
            
        except Exception as e:
            logger.error(f"AI intent analysis failed: {str(e)}")
            return None
    
    def _createBasicIntentAnalysis(self, userPrompt: str) -> Dict[str, Any]:
        """Creates basic intent analysis without AI"""
        return {
            "primaryGoal": userPrompt.strip(),
            "dataType": "unknown",
            "expectedFormat": "unknown",
            "qualityRequirements": {
                "accuracyThreshold": 0.8,
                "completenessThreshold": 0.8,
                "formatRequirement": "any"
            },
            "successCriteria": ["Delivers what the user requested"],
            "confidenceScore": 0.5
        }
    
    def _createDefaultIntentAnalysis(self, userPrompt: str) -> Dict[str, Any]:
        """Creates a default intent analysis when analysis fails"""
        return {
            "primaryGoal": userPrompt,
            "dataType": "unknown",
            "expectedFormat": "unknown",
            "qualityRequirements": {
                "accuracyThreshold": 0.8,
                "completenessThreshold": 0.8,
                "formatRequirement": "any"
            },
            "successCriteria": ["Delivers what the user requested"],
            "confidenceScore": 0.1
        }
    
    def _isValidJsonResponse(self, response: str) -> bool:
        """Checks if response contains valid JSON structure"""
        try:
            import re
            # Look for JSON with expected structure
            json_match = re.search(r'\{[^{}]*"primaryGoal"[^{}]*\}', response, re.DOTALL)
            if json_match:
                json.loads(json_match.group(0))
                return True
            return False
        except:
            return False