knowledge_graph_service.py

"""
Knowledge Graph Service for Neuro-Symbolic VQA
Uses ConceptNet API to provide common-sense reasoning capabilities
"""

import requests
import re
from typing import Dict, List, Optional
from functools import lru_cache
import time


class KnowledgeGraphService:
    """
    Lightweight ConceptNet API wrapper for common-sense reasoning.
    Enhances VQA answers with external knowledge about object properties,
    capabilities, uses, and relationships.
    """
    
    CONCEPTNET_API = "https://api.conceptnet.io"
    
    # Common-sense question patterns
    COMMONSENSE_PATTERNS = [
        # Capability questions
        (r'can .* (melt|freeze|fly|swim|float|sink|break|burn|explode)', 'CapableOf'),
        (r'is .* able to', 'CapableOf'),
        (r'does .* (float|sink)', 'CapableOf'),
        
        # Property questions
        (r'is .* (edible|poisonous|dangerous|safe|hot|cold|sweet|sour)', 'HasProperty'),
        (r'is this (food|drink|toy|tool|weapon)', 'HasProperty'),
        
        # Purpose questions
        (r'what .* (used for|for)', 'UsedFor'),
        (r'why .* (used|made)', 'UsedFor'),
        (r'how .* use', 'UsedFor'),
        
        # Composition questions
        (r'what .* made (of|from)', 'MadeOf'),
        (r'what .* (material|ingredient)', 'MadeOf'),
        
        # Location questions
        (r'where .* (found|located|kept|stored)', 'AtLocation'),
        (r'where (do|does) .* (live|grow)', 'AtLocation'),
    ]
    
    def __init__(self, cache_size=100, timeout=5):
        """
        Initialize Knowledge Graph service.
        
        Args:
            cache_size: Number of API responses to cache
            timeout: API request timeout in seconds
        """
        self.timeout = timeout
        self.cache_size = cache_size
        print("✅ Knowledge Graph service initialized (ConceptNet API)")
    
    @lru_cache(maxsize=100)
    def _query_conceptnet(self, concept: str, relation: str, limit: int = 10) -> Optional[Dict]:
        """
        Query ConceptNet API with caching.
        
        Args:
            concept: Concept to query (e.g., "ice_cream")
            relation: Relation type (e.g., "CapableOf", "HasProperty")
            limit: Maximum number of results
            
        Returns:
            API response dict or None if failed
        """
        try:
            # Normalize concept (replace spaces with underscores)
            concept = concept.lower().replace(' ', '_')
            
            # Build API URL
            url = f"{self.CONCEPTNET_API}/query"
            params = {
                'start': f'/c/en/{concept}',
                'rel': f'/r/{relation}',
                'limit': limit
            }
            
            # Make request
            response = requests.get(url, params=params, timeout=self.timeout)
            response.raise_for_status()
            
            return response.json()
            
        except requests.exceptions.Timeout:
            print(f"⚠️  ConceptNet API timeout for {concept}")
            return None
        except requests.exceptions.RequestException as e:
            print(f"⚠️  ConceptNet API error: {e}")
            return None
        except Exception as e:
            print(f"⚠️  Unexpected error querying ConceptNet: {e}")
            return None
    
    def get_concept_properties(self, concept: str) -> Dict[str, List[str]]:

        properties = {
            'CapableOf': [],
            'HasProperty': [],
            'UsedFor': [],
            'MadeOf': [],
            'AtLocation': []
        }
        
        # Query each relation type
        for relation in properties.keys():
            data = self._query_conceptnet(concept, relation)
            
            if data and 'edges' in data:
                for edge in data['edges']:
                    # Extract the end concept
                    if 'end' in edge and 'label' in edge['end']:
                        end_label = edge['end']['label']
                        properties[relation].append(end_label)
        
        return properties
    
    def is_commonsense_question(self, question: str) -> bool:
        """
        Detect if a question requires common-sense reasoning.
        
        Args:
            question: Question string
            
        Returns:
            True if question needs external knowledge
        """
        q_lower = question.lower()
        
        for pattern, _ in self.COMMONSENSE_PATTERNS:
            if re.search(pattern, q_lower):
                return True
        
        return False
    
    def _detect_question_type(self, question: str) -> Optional[str]:
        """
        Detect which ConceptNet relation the question is asking about.
        
        Args:
            question: Question string
            
        Returns:
            Relation type or None
        """
        q_lower = question.lower()
        
        for pattern, relation in self.COMMONSENSE_PATTERNS:
            if re.search(pattern, q_lower):
                return relation
        
        return None
    
    def answer_commonsense_question(self, object_name: str, question: str) -> Optional[str]:
        """
        Answer a common-sense question using Knowledge Graph.
        
        Args:
            object_name: Object detected by VQA (e.g., "ice cream")
            question: User's question
            
        Returns:
            Enhanced answer string or None
        """
        # Detect question type
        relation = self._detect_question_type(question)
        if not relation:
            return None
        
        # Query ConceptNet
        data = self._query_conceptnet(object_name, relation, limit=5)
        if not data or 'edges' not in data:
            return None
        
        # Extract relevant knowledge
        knowledge = []
        for edge in data['edges']:
            if 'end' in edge and 'label' in edge['end']:
                knowledge.append(edge['end']['label'])
        
        if not knowledge:
            return None
        
        # Generate natural language answer based on question type
        return self._synthesize_answer(object_name, question, relation, knowledge)
    
    def _synthesize_answer(self, object_name: str, question: str, 
                          relation: str, knowledge: List[str]) -> str:
        """
        Synthesize natural language answer from knowledge.
        
        Args:
            object_name: Detected object
            question: Original question
            relation: Relation type
            knowledge: List of related concepts from KG
            
        Returns:
            Natural language answer
        """
        q_lower = question.lower()
        
        # Capability questions (can X do Y?)
        if relation == 'CapableOf':
            # Check if specific capability is mentioned
            for capability in knowledge:
                if capability in q_lower:
                    return f"Yes, {object_name} can {capability}."
            
            # General capability answer
            if knowledge:
                caps = ', '.join(knowledge[:3])
                return f"{object_name.capitalize()} can {caps}."
        
        # Property questions (is X Y?)
        elif relation == 'HasProperty':
            # Check for specific property
            if 'edible' in q_lower:
                if 'edible' in knowledge:
                    return f"Yes, {object_name} is edible."
                else:
                    return f"No, {object_name} is not edible."
            
            if 'dangerous' in q_lower or 'safe' in q_lower:
                if any(prop in knowledge for prop in ['dangerous', 'harmful', 'poisonous']):
                    return f"Caution: {object_name} may be dangerous."
                else:
                    return f"{object_name.capitalize()} is generally safe."
            
            # General properties
            if knowledge:
                props = ', '.join(knowledge[:3])
                return f"{object_name.capitalize()} is {props}."
        
        # Purpose questions (what is X used for?)
        elif relation == 'UsedFor':
            if knowledge:
                uses = ', '.join(knowledge[:3])
                return f"{object_name.capitalize()} is used for {uses}."
        
        # Composition questions (what is X made of?)
        elif relation == 'MadeOf':
            if knowledge:
                materials = ', '.join(knowledge[:3])
                return f"{object_name.capitalize()} is made of {materials}."
        
        # Location questions (where is X found?)
        elif relation == 'AtLocation':
            if knowledge:
                locations = ', '.join(knowledge[:2])
                return f"{object_name.capitalize()} is typically found at {locations}."
        
        return None


# Test function
if __name__ == "__main__":
    print("=" * 80)
    print("🧪 Testing Knowledge Graph Service")
    print("=" * 80)
    
    kg = KnowledgeGraphService()
    
    # Test cases
    test_cases = [
        ("ice cream", "Can this melt?"),
        ("apple", "Is this edible?"),
        ("hammer", "What is this used for?"),
        ("knife", "Is this dangerous?"),
        ("bread", "What is this made of?"),
    ]
    
    for obj, question in test_cases:
        print(f"\n📝 Object: {obj}")
        print(f"❓ Question: {question}")
        
        # Check if common-sense question
        is_cs = kg.is_commonsense_question(question)
        print(f"🔍 Common-sense: {is_cs}")
        
        if is_cs:
            # Get answer
            answer = kg.answer_commonsense_question(obj, question)
            print(f"💬 Answer: {answer}")
        
        print("-" * 80)