Spaces:
Runtime error
Runtime error
| """ | |
| 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" | |
| 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}") |