Create knowledge_engine.py

knowledge_engine.py

@@ -0,0 +1,565 @@
+"""
+Knowledge Engine - Layer 2: Knowledge management and retrieval
+"""
+
+import json
+import numpy as np
+from typing import Dict, List, Any, Optional, Tuple
+from datetime import datetime
+import hashlib
+from dataclasses import dataclass
+from enum import Enum
+
+
+class KnowledgeType(Enum):
+    """Types of knowledge in the system"""
+    CONCEPT = "concept"
+    FORMULA = "formula"
+    EXAMPLE = "example"
+    PRACTICE_PROBLEM = "practice_problem"
+    STUDY_GUIDE = "study_guide"
+    EXAM_TIP = "exam_tip"
+
+
+@dataclass
+class KnowledgeNode:
+    """Knowledge node in the knowledge graph"""
+    id: str
+    type: KnowledgeType
+    content: Dict[str, Any]
+    topics: List[str]
+    difficulty: str
+    prerequisites: List[str]
+    related_nodes: List[str]
+    metadata: Dict[str, Any]
+    created_at: datetime
+    last_accessed: datetime
+    access_count: int
+
+
+class KnowledgeEngine:
+    """Knowledge management and retrieval engine"""
+    
+    def __init__(self, knowledge_base_path: str = "data/cfa_knowledge_base.json"):
+        self.knowledge_graph: Dict[str, KnowledgeNode] = {}
+        self.topic_index: Dict[str, List[str]] = {}
+        self.concept_index: Dict[str, List[str]] = {}
+        self.load_knowledge_base(knowledge_base_path)
+        self.access_log = []
+        
+    def load_knowledge_base(self, path: str):
+        """Load and index knowledge base"""
+        try:
+            with open(path, 'r', encoding='utf-8') as f:
+                knowledge_data = json.load(f)
+                self._build_knowledge_graph(knowledge_data)
+                print(f"✅ Knowledge base loaded: {len(self.knowledge_graph)} nodes")
+        except FileNotFoundError:
+            print(f"⚠️ Knowledge base not found at {path}, creating minimal graph")
+            self._create_minimal_graph()
+    
+    def _build_knowledge_graph(self, knowledge_data: Dict[str, Any]):
+        """Build knowledge graph from loaded data"""
+        # Process topics
+        for topic_name, topic_data in knowledge_data.get('topics', {}).items():
+            # Create concept nodes
+            concepts = topic_data.get('key_concepts', [])
+            for concept in concepts:
+                node_id = self._generate_node_id(concept['name'], KnowledgeType.CONCEPT)
+                node = KnowledgeNode(
+                    id=node_id,
+                    type=KnowledgeType.CONCEPT,
+                    content={
+                        'name': concept['name'],
+                        'definition': concept.get('definition', ''),
+                        'formula': concept.get('formula', ''),
+                        'example': concept.get('example', '')
+                    },
+                    topics=[topic_name],
+                    difficulty=topic_data.get('difficulty', 'medium'),
+                    prerequisites=[],
+                    related_nodes=[],
+                    metadata={
+                        'cfa_level': topic_data.get('cfa_level', 'I'),
+                        'weight': topic_data.get('weight', ''),
+                        'applications': topic_data.get('common_applications', [])
+                    },
+                    created_at=datetime.now(),
+                    last_accessed=datetime.now(),
+                    access_count=0
+                )
+                self.knowledge_graph[node_id] = node
+                
+                # Update indices
+                self._add_to_index(node_id, topic_name, concept['name'])
+        
+        # Process formulas
+        for category, formulas in knowledge_data.get('formulas', {}).items():
+            for formula in formulas:
+                node_id = self._generate_node_id(formula['name'], KnowledgeType.FORMULA)
+                node = KnowledgeNode(
+                    id=node_id,
+                    type=KnowledgeType.FORMULA,
+                    content={
+                        'name': formula['name'],
+                        'formula': formula.get('formula', ''),
+                        'description': formula.get('description', ''),
+                        'variables': formula.get('variables', {})
+                    },
+                    topics=[category],
+                    difficulty='medium',
+                    prerequisites=[],
+                    related_nodes=[],
+                    metadata={'category': category},
+                    created_at=datetime.now(),
+                    last_accessed=datetime.now(),
+                    access_count=0
+                )
+                self.knowledge_graph[node_id] = node
+                self._add_to_index(node_id, category, formula['name'])
+        
+        # Build relationships
+        self._build_relationships()
+    
+    def _create_minimal_graph(self):
+        """Create minimal knowledge graph if no data file"""
+        # Core TVM concepts
+        tvm_concepts = [
+            {
+                'name': 'Present Value',
+                'definition': 'Current worth of future cash flows',
+                'formula': 'PV = FV / (1 + r)^n'
+            },
+            {
+                'name': 'Future Value',
+                'definition': 'Future worth of current investment',
+                'formula': 'FV = PV × (1 + r)^n'
+            }
+        ]
+        
+        for concept in tvm_concepts:
+            node_id = self._generate_node_id(concept['name'], KnowledgeType.CONCEPT)
+            node = KnowledgeNode(
+                id=node_id,
+                type=KnowledgeType.CONCEPT,
+                content=concept,
+                topics=['Time Value of Money'],
+                difficulty='medium',
+                prerequisites=[],
+                related_nodes=[],
+                metadata={'cfa_level': 'I', 'weight': '8-12%'},
+                created_at=datetime.now(),
+                last_accessed=datetime.now(),
+                access_count=0
+            )
+            self.knowledge_graph[node_id] = node
+            self._add_to_index(node_id, 'Time Value of Money', concept['name'])
+    
+    def _generate_node_id(self, name: str, ktype: KnowledgeType) -> str:
+        """Generate unique node ID"""
+        hash_input = f"{name}_{ktype.value}_{datetime.now().timestamp()}"
+        return hashlib.md5(hash_input.encode()).hexdigest()[:12]
+    
+    def _add_to_index(self, node_id: str, topic: str, concept: str):
+        """Add node to search indices"""
+        # Topic index
+        if topic not in self.topic_index:
+            self.topic_index[topic] = []
+        self.topic_index[topic].append(node_id)
+        
+        # Concept index (by keywords)
+        keywords = concept.lower().split()
+        for keyword in keywords:
+            if len(keyword) > 3:  # Ignore short words
+                if keyword not in self.concept_index:
+                    self.concept_index[keyword] = []
+                self.concept_index[keyword].append(node_id)
+    
+    def _build_relationships(self):
+        """Build relationships between knowledge nodes"""
+        # Simple relationship building - in production, use NLP
+        node_ids = list(self.knowledge_graph.keys())
+        
+        for i, node_id in enumerate(node_ids):
+            node = self.knowledge_graph[node_id]
+            related = []
+            
+            # Find related nodes by topic
+            for other_id in node_ids:
+                if other_id != node_id:
+                    other_node = self.knowledge_graph[other_id]
+                    # Same topic
+                    if any(topic in other_node.topics for topic in node.topics):
+                        related.append(other_id)
+                    # Related concepts (simple keyword matching)
+                    node_content = str(node.content).lower()
+                    other_content = str(other_node.content).lower()
+                    common_words = set(node_content.split()) & set(other_content.split())
+                    if len(common_words) >= 2:
+                        related.append(other_id)
+            
+            # Limit relationships
+            node.related_nodes = list(set(related))[:10]
+            self.knowledge_graph[node_id] = node
+    
+    def retrieve_knowledge(self, 
+                          query: str, 
+                          user_profile: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
+        """
+        Retrieve knowledge based on query
+        
+        Args:
+            query: Knowledge query
+            user_profile: User's learning profile
+        
+        Returns:
+            Retrieved knowledge
+        """
+        # Log access
+        self.access_log.append({
+            'query': query,
+            'timestamp': datetime.now().isoformat(),
+            'user_profile': user_profile
+        })
+        
+        # Search for relevant nodes
+        search_results = self._search_knowledge(query)
+        
+        # Rank results
+        ranked_results = self._rank_results(search_results, query, user_profile)
+        
+        # Prepare response
+        response = self._prepare_knowledge_response(ranked_results, query, user_profile)
+        
+        # Update access counts
+        for node_id, _ in ranked_results[:5]:
+            if node_id in self.knowledge_graph:
+                node = self.knowledge_graph[node_id]
+                node.last_accessed = datetime.now()
+                node.access_count += 1
+                self.knowledge_graph[node_id] = node
+        
+        return response
+    
+    def _search_knowledge(self, query: str) -> List[str]:
+        """Search knowledge graph for relevant nodes"""
+        query_lower = query.lower()
+        found_nodes = set()
+        
+        # Search by topic
+        for topic in self.topic_index.keys():
+            if topic.lower() in query_lower:
+                found_nodes.update(self.topic_index[topic])
+        
+        # Search by keywords
+        query_words = query_lower.split()
+        for word in query_words:
+            if len(word) > 3 and word in self.concept_index:
+                found_nodes.update(self.concept_index[word])
+        
+        # If no results, try broader search
+        if not found_nodes:
+            for node_id, node in self.knowledge_graph.items():
+                node_text = str(node.content).lower()
+                if any(word in node_text for word in query_words[:3]):
+                    found_nodes.add(node_id)
+        
+        return list(found_nodes)
+    
+    def _rank_results(self, 
+                     node_ids: List[str], 
+                     query: str,
+                     user_profile: Optional[Dict[str, Any]]) -> List[Tuple[str, float]]:
+        """Rank search results by relevance"""
+        ranked = []
+        
+        for node_id in node_ids:
+            node = self.knowledge_graph.get(node_id)
+            if not node:
+                continue
+            
+            relevance_score = self._calculate_relevance(node, query, user_profile)
+            ranked.append((node_id, relevance_score))
+        
+        # Sort by relevance
+        ranked.sort(key=lambda x: x[1], reverse=True)
+        return ranked
+    
+    def _calculate_relevance(self, 
+                           node: KnowledgeNode, 
+                           query: str,
+                           user_profile: Optional[Dict[str, Any]]) -> float:
+        """Calculate relevance score for a node"""
+        score = 0.0
+        
+        # Text similarity (simplified)
+        query_lower = query.lower()
+        node_text = str(node.content).lower()
+        
+        # Keyword matches
+        query_words = set(query_lower.split())
+        node_words = set(node_text.split())
+        keyword_overlap = len(query_words & node_words) / max(len(query_words), 1)
+        score += keyword_overlap * 0.4
+        
+        # Topic relevance
+        if user_profile:
+            study_focus = user_profile.get('study_focus', [])
+            topic_overlap = len(set(node.topics) & set(study_focus))
+            score += topic_overlap * 0.2
+        
+        # Popularity/recency
+        recency_factor = 1.0 - ((datetime.now() - node.last_accessed).days / 365)
+        popularity_factor = min(node.access_count / 100, 1.0)
+        score += (recency_factor * 0.2 + popularity_factor * 0.2)
+        
+        # Difficulty match
+        if user_profile:
+            experience = user_profile.get('experience_level', 'beginner')
+            difficulty_score = self._match_difficulty(node.difficulty, experience)
+            score += difficulty_score * 0.1
+        
+        return min(score, 1.0)
+    
+    def _match_difficulty(self, node_difficulty: str, user_experience: str) -> float:
+        """Match node difficulty to user experience"""
+        difficulty_map = {'easy': 0, 'medium': 1, 'hard': 2}
+        experience_map = {'beginner': 0, 'intermediate': 1, 'advanced': 2}
+        
+        node_level = difficulty_map.get(node_difficulty, 1)
+        user_level = experience_map.get(user_experience, 1)
+        
+        # Prefer nodes at or slightly above user level
+        diff = abs(node_level - user_level)
+        if diff == 0:
+            return 1.0
+        elif diff == 1:
+            return 0.7
+        else:
+            return 0.3
+    
+    def _prepare_knowledge_response(self, 
+                                   ranked_results: List[Tuple[str, float]], 
+                                   query: str,
+                                   user_profile: Optional[Dict[str, Any]]) -> Dict[str, Any]:
+        """Prepare structured knowledge response"""
+        response = {
+            'query': query,
+            'results': [],
+            'topics': set(),
+            'total_results': len(ranked_results),
+            'timestamp': datetime.now().isoformat()
+        }
+        
+        # Get top results
+        for node_id, relevance in ranked_results[:10]:  # Top 10 results
+            node = self.knowledge_graph.get(node_id)
+            if node:
+                node_data = {
+                    'id': node.id,
+                    'type': node.type.value,
+                    'content': node.content,
+                    'topics': node.topics,
+                    'difficulty': node.difficulty,
+                    'relevance': relevance,
+                    'metadata': node.metadata,
+                    'access_count': node.access_count
+                }
+                response['results'].append(node_data)
+                response['topics'].update(node.topics)
+        
+        response['topics'] = list(response['topics'])
+        
+        # Add learning path if user profile provided
+        if user_profile and response['results']:
+            response['learning_path'] = self._generate_learning_path(
+                response['results'], user_profile
+            )
+        
+        return response
+    
+    def _generate_learning_path(self, 
+                               results: List[Dict[str, Any]], 
+                               user_profile: Dict[str, Any]) -> Dict[str, Any]:
+        """Generate learning path based on results and user profile"""
+        # Group by topic
+        topics = {}
+        for result in results:
+            for topic in result['topics']:
+                if topic not in topics:
+                    topics[topic] = []
+                topics[topic].append(result)
+        
+        # Select primary topic
+        primary_topic = max(topics.items(), key=lambda x: len(x[1]))[0] if topics else None
+        
+        if not primary_topic:
+            return {'suggested_path': [], 'estimated_time': 0}
+        
+        # Generate learning path
+        learning_path = []
+        
+        # Start with concepts
+        concepts = [r for r in results if r['type'] == 'concept']
+        for concept in concepts[:3]:
+            learning_path.append({
+                'type': 'concept',
+                'title': concept['content'].get('name', 'Concept'),
+                'time': 15,  # minutes
+                'description': f"Learn: {concept['content'].get('definition', '')[:100]}..."
+            })
+        
+        # Add formulas
+        formulas = [r for r in results if r['type'] == 'formula']
+        for formula in formulas[:2]:
+            learning_path.append({
+                'type': 'formula',
+                'title': formula['content'].get('name', 'Formula'),
+                'time': 10,
+                'description': f"Master: {formula['content'].get('formula', '')}"
+            })
+        
+        # Add practice
+        learning_path.append({
+            'type': 'practice',
+            'title': f'Practice {primary_topic} Problems',
+            'time': 20,
+            'description': 'Apply concepts through practice problems'
+        })
+        
+        estimated_time = sum(step['time'] for step in learning_path)
+        
+        return {
+            'primary_topic': primary_topic,
+            'suggested_path': learning_path,
+            'estimated_time': estimated_time,
+            'suitable_for': user_profile.get('experience_level', 'beginner')
+        }
+    
+    def add_knowledge(self, 
+                     knowledge_type: KnowledgeType, 
+                     content: Dict[str, Any], 
+                     topics: List[str],
+                     difficulty: str = 'medium',
+                     metadata: Optional[Dict[str, Any]] = None) -> str:
+        """
+        Add new knowledge to the graph
+        
+        Args:
+            knowledge_type: Type of knowledge
+            content: Knowledge content
+            topics: Related topics
+            difficulty: Difficulty level
+            metadata: Additional metadata
+        
+        Returns:
+            Node ID
+        """
+        # Generate node
+        node_name = content.get('name', 'Unnamed Knowledge')
+        node_id = self._generate_node_id(node_name, knowledge_type)
+        
+        node = KnowledgeNode(
+            id=node_id,
+            type=knowledge_type,
+            content=content,
+            topics=topics,
+            difficulty=difficulty,
+            prerequisites=[],
+            related_nodes=[],
+            metadata=metadata or {},
+            created_at=datetime.now(),
+            last_accessed=datetime.now(),
+            access_count=0
+        )
+        
+        # Add to graph
+        self.knowledge_graph[node_id] = node
+        
+        # Update indices
+        for topic in topics:
+            if topic not in self.topic_index:
+                self.topic_index[topic] = []
+            self.topic_index[topic].append(node_id)
+        
+        # Index content keywords
+        content_str = str(content).lower()
+        for word in content_str.split():
+            if len(word) > 3:
+                if word not in self.concept_index:
+                    self.concept_index[word] = []
+                self.concept_index[word].append(node_id)
+        
+        print(f"✅ Added new knowledge: {node_name} ({node_id})")
+        return node_id
+    
+    def get_knowledge_stats(self) -> Dict[str, Any]:
+        """Get knowledge engine statistics"""
+        stats = {
+            'total_nodes': len(self.knowledge_graph),
+            'by_type': {},
+            'by_topic': {},
+            'by_difficulty': {},
+            'access_stats': {
+                'total_accesses': sum(node.access_count for node in self.knowledge_graph.values()),
+                'most_accessed': []
+            }
+        }
+        
+        # Count by type
+        for node in self.knowledge_graph.values():
+            ktype = node.type.value
+            stats['by_type'][ktype] = stats['by_type'].get(ktype, 0) + 1
+            
+            # Count by topic
+            for topic in node.topics:
+                stats['by_topic'][topic] = stats['by_topic'].get(topic, 0) + 1
+            
+            # Count by difficulty
+            stats['by_difficulty'][node.difficulty] = stats['by_difficulty'].get(node.difficulty, 0) + 1
+        
+        # Most accessed nodes
+        sorted_nodes = sorted(self.knowledge_graph.values(), 
+                            key=lambda x: x.access_count, 
+                            reverse=True)[:5]
+        stats['access_stats']['most_accessed'] = [
+            {
+                'id': node.id,
+                'name': node.content.get('name', 'Unnamed'),
+                'access_count': node.access_count,
+                'type': node.type.value
+            }
+            for node in sorted_nodes
+        ]
+        
+        return stats
+    
+    def export_knowledge_base(self, path: str):
+        """Export knowledge base to file"""
+        export_data = {
+            'metadata': {
+                'export_date': datetime.now().isoformat(),
+                'total_nodes': len(self.knowledge_graph),
+                'version': '1.0'
+            },
+            'nodes': []
+        }
+        
+        for node in self.knowledge_graph.values():
+            node_data = {
+                'id': node.id,
+                'type': node.type.value,
+                'content': node.content,
+                'topics': node.topics,
+                'difficulty': node.difficulty,
+                'metadata': node.metadata,
+                'access_count': node.access_count,
+                'created_at': node.created_at.isoformat(),
+                'last_accessed': node.last_accessed.isoformat()
+            }
+            export_data['nodes'].append(node_data)
+        
+        with open(path, 'w', encoding='utf-8') as f:
+            json.dump(export_data, f, indent=2, ensure_ascii=False)
+        
+        print(f"✅ Knowledge base exported to {path}")