Spaces:
Runtime error
Runtime error
| """ | |
| Knowledge Agent - Retrieves and explains CFA concepts using local embeddings | |
| Free version using sentence-transformers and ChromaDB | |
| """ | |
| import json | |
| from typing import Dict, List, Any, Optional, Tuple | |
| from datetime import datetime | |
| import re | |
| import numpy as np | |
| from sentence_transformers import SentenceTransformer | |
| import chromadb | |
| from chromadb.config import Settings | |
| import os | |
| class KnowledgeAgent: | |
| """Retrieves and explains CFA knowledge concepts using local embeddings""" | |
| def __init__(self, knowledge_base_path: str = "data/cfa_knowledge_base.json"): | |
| self.knowledge_base_path = knowledge_base_path | |
| self.knowledge_base = self._load_knowledge_base() | |
| self.search_history = [] | |
| # Initialize embedding model (free, runs locally) | |
| try: | |
| self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2') | |
| print("โ Sentence transformer model loaded") | |
| except Exception as e: | |
| print(f"โ ๏ธ Could not load sentence transformer: {e}") | |
| self.embedding_model = None | |
| # Initialize vector database | |
| self.chroma_client = None | |
| self.collection = None | |
| self._initialize_vector_db() | |
| def _load_knowledge_base(self) -> Dict[str, Any]: | |
| """Load CFA knowledge base from JSON file""" | |
| try: | |
| # Try multiple possible paths | |
| possible_paths = [ | |
| self.knowledge_base_path, | |
| os.path.join(os.path.dirname(__file__), '..', self.knowledge_base_path), | |
| os.path.join(os.getcwd(), self.knowledge_base_path) | |
| ] | |
| for path in possible_paths: | |
| if os.path.exists(path): | |
| with open(path, 'r', encoding='utf-8') as f: | |
| data = json.load(f) | |
| print(f"โ Knowledge base loaded from {path}") | |
| return data | |
| print("โ ๏ธ Knowledge base file not found, creating default") | |
| return self._create_default_knowledge_base() | |
| except Exception as e: | |
| print(f"โ Error loading knowledge base: {e}") | |
| return self._create_default_knowledge_base() | |
| def _create_default_knowledge_base(self) -> Dict[str, Any]: | |
| """Create a default knowledge base if file not found""" | |
| return { | |
| "topics": { | |
| "Time Value of Money": { | |
| "description": "Core concept of finance dealing with the value of money over time", | |
| "key_concepts": [ | |
| { | |
| "name": "Present Value", | |
| "definition": "Current worth of a future sum of money given a specified rate of return", | |
| "formula": "PV = FV / (1 + r)^n", | |
| "example": "The present value of $110 received one year from now at 10% discount rate is $100" | |
| }, | |
| { | |
| "name": "Future Value", | |
| "definition": "Value of a current asset at a specified date in the future", | |
| "formula": "FV = PV ร (1 + r)^n", | |
| "example": "$100 invested today at 10% annual interest will grow to $110 in one year" | |
| } | |
| ], | |
| "common_applications": [ | |
| "Valuing bonds and stocks", | |
| "Capital budgeting decisions", | |
| "Retirement planning", | |
| "Loan amortization" | |
| ], | |
| "cfa_level": "I", | |
| "weight": "8-12%", | |
| "difficulty": "Medium" | |
| }, | |
| "Statistics": { | |
| "description": "Collection, analysis, interpretation, and presentation of data", | |
| "key_concepts": [ | |
| { | |
| "name": "Mean", | |
| "definition": "Average of a set of numbers", | |
| "formula": "ฮผ = ฮฃ x_i / n", | |
| "example": "Mean of [2, 4, 6, 8] is (2+4+6+8)/4 = 5" | |
| }, | |
| { | |
| "name": "Standard Deviation", | |
| "definition": "Measure of the amount of variation or dispersion", | |
| "formula": "ฯ = โ[ฮฃ(x_i - ฮผ)ยฒ / (n-1)]", | |
| "example": "Standard deviation measures how spread out numbers are" | |
| } | |
| ], | |
| "common_applications": [ | |
| "Portfolio analysis", | |
| "Risk assessment", | |
| "Performance measurement" | |
| ], | |
| "cfa_level": "I", | |
| "weight": "5-8%", | |
| "difficulty": "Medium" | |
| } | |
| }, | |
| "formulas": { | |
| "TVM": [ | |
| {"name": "Future Value", "formula": "FV = PV(1+r)^n"}, | |
| {"name": "Present Value", "formula": "PV = FV/(1+r)^n"} | |
| ] | |
| }, | |
| "glossary": { | |
| "NPV": "Net Present Value - difference between present value of cash inflows and outflows", | |
| "IRR": "Internal Rate of Return - discount rate that makes NPV zero", | |
| "Beta": "Measure of a stock's volatility relative to the market", | |
| "Alpha": "Measure of performance on a risk-adjusted basis" | |
| } | |
| } | |
| def _initialize_vector_db(self): | |
| """Initialize ChromaDB vector database""" | |
| try: | |
| # Create persistent directory for ChromaDB | |
| chroma_dir = os.path.join(os.path.dirname(__file__), '..', 'chroma_db') | |
| os.makedirs(chroma_dir, exist_ok=True) | |
| self.chroma_client = chromadb.PersistentClient( | |
| path=chroma_dir, | |
| settings=Settings(anonymized_telemetry=False) | |
| ) | |
| # Create or get collection | |
| self.collection = self.chroma_client.get_or_create_collection( | |
| name="cfa_knowledge", | |
| metadata={"description": "CFA Knowledge Base Embeddings"} | |
| ) | |
| print("โ ChromaDB initialized") | |
| # Index knowledge if collection is empty | |
| if self.collection.count() == 0: | |
| self._index_knowledge_base() | |
| except Exception as e: | |
| print(f"โ ๏ธ Could not initialize ChromaDB: {e}") | |
| self.chroma_client = None | |
| self.collection = None | |
| def _index_knowledge_base(self): | |
| """Index knowledge base in vector database""" | |
| if not self.embedding_model or not self.collection: | |
| return | |
| print("๐ Indexing knowledge base...") | |
| documents = [] | |
| metadatas = [] | |
| ids = [] | |
| # Index topics | |
| for topic_name, topic_data in self.knowledge_base.get('topics', {}).items(): | |
| # Topic description | |
| doc_text = f"Topic: {topic_name}. Description: {topic_data.get('description', '')}" | |
| documents.append(doc_text) | |
| metadatas.append({ | |
| "type": "topic", | |
| "topic": topic_name, | |
| "cfa_level": topic_data.get('cfa_level', 'I'), | |
| "difficulty": topic_data.get('difficulty', 'Medium') | |
| }) | |
| ids.append(f"topic_{topic_name}") | |
| # Key concepts | |
| for concept in topic_data.get('key_concepts', []): | |
| doc_text = f"Concept: {concept['name']}. Definition: {concept.get('definition', '')}" | |
| if 'formula' in concept: | |
| doc_text += f" Formula: {concept['formula']}" | |
| if 'example' in concept: | |
| doc_text += f" Example: {concept['example']}" | |
| documents.append(doc_text) | |
| metadatas.append({ | |
| "type": "concept", | |
| "topic": topic_name, | |
| "concept": concept['name'], | |
| "has_formula": 'formula' in concept | |
| }) | |
| ids.append(f"concept_{topic_name}_{concept['name']}") | |
| # Index formulas | |
| for category, formulas in self.knowledge_base.get('formulas', {}).items(): | |
| for formula in formulas: | |
| doc_text = f"Formula: {formula['name']}. Formula: {formula.get('formula', '')}" | |
| if 'description' in formula: | |
| doc_text += f" Description: {formula['description']}" | |
| documents.append(doc_text) | |
| metadatas.append({ | |
| "type": "formula", | |
| "category": category, | |
| "formula_name": formula['name'] | |
| }) | |
| ids.append(f"formula_{category}_{formula['name']}") | |
| # Index glossary terms | |
| for term, definition in self.knowledge_base.get('glossary', {}).items(): | |
| doc_text = f"Term: {term}. Definition: {definition}" | |
| documents.append(doc_text) | |
| metadatas.append({ | |
| "type": "glossary", | |
| "term": term | |
| }) | |
| ids.append(f"glossary_{term}") | |
| # Generate embeddings and add to collection | |
| if documents: | |
| try: | |
| embeddings = self.embedding_model.encode(documents).tolist() | |
| self.collection.add( | |
| documents=documents, | |
| metadatas=metadatas, | |
| ids=ids, | |
| embeddings=embeddings | |
| ) | |
| print(f"โ Indexed {len(documents)} knowledge items") | |
| except Exception as e: | |
| print(f"โ Error indexing documents: {e}") | |
| def process(self, | |
| query: str, | |
| user_profile: Dict[str, Any], | |
| conversation_history: List[Dict] = None) -> Dict[str, Any]: | |
| """ | |
| Retrieve and explain CFA knowledge concepts | |
| Args: | |
| query: Knowledge query | |
| user_profile: User's learning profile | |
| conversation_history: Previous conversation | |
| Returns: | |
| Knowledge explanation | |
| """ | |
| # Analyze query to determine knowledge needs | |
| knowledge_type = self._analyze_knowledge_query(query) | |
| # Search knowledge base | |
| if self.collection and self.embedding_model: | |
| # Use vector search | |
| search_results = self._vector_search(query, knowledge_type, user_profile) | |
| else: | |
| # Fallback to keyword search | |
| search_results = self._keyword_search(query, knowledge_type) | |
| # Format response | |
| response = self._format_knowledge_response(query, search_results, knowledge_type, user_profile) | |
| # Log search | |
| self._log_search(query, knowledge_type, search_results) | |
| return { | |
| 'response': response, | |
| 'knowledge_type': knowledge_type, | |
| 'search_results': search_results, | |
| 'confidence': self._calculate_confidence(search_results), | |
| 'metadata': { | |
| 'search_method': 'vector' if self.collection else 'keyword', | |
| 'result_count': len(search_results.get('items', [])) | |
| } | |
| } | |
| def _analyze_knowledge_query(self, query: str) -> str: | |
| """Analyze query to determine type of knowledge needed""" | |
| query_lower = query.lower() | |
| if any(term in query_lower for term in ['what is', 'define', 'definition of', 'meaning of']): | |
| return 'definition' | |
| elif any(term in query_lower for term in ['explain', 'how does', 'how to', 'understand']): | |
| return 'explanation' | |
| elif any(term in query_lower for term in ['formula', 'calculate', 'equation', 'solve']): | |
| return 'formula' | |
| elif any(term in query_lower for term in ['example', 'case study', 'application', 'real world']): | |
| return 'example' | |
| elif any(term in query_lower for term in ['compare', 'difference between', 'vs', 'versus']): | |
| return 'comparison' | |
| elif any(term in query_lower for term in ['list', 'what are the', 'types of', 'categories']): | |
| return 'list' | |
| else: | |
| return 'general' | |
| def _vector_search(self, query: str, knowledge_type: str, user_profile: Dict[str, Any]) -> Dict[str, Any]: | |
| """Search using vector embeddings""" | |
| results = { | |
| 'items': [], | |
| 'topics': set(), | |
| 'method': 'vector' | |
| } | |
| try: | |
| # Generate query embedding | |
| query_embedding = self.embedding_model.encode(query).tolist() | |
| # Search collection | |
| search_results = self.collection.query( | |
| query_embeddings=[query_embedding], | |
| n_results=10, | |
| include=["documents", "metadatas", "distances"] | |
| ) | |
| # Process results | |
| if search_results['documents']: | |
| for i, (doc, metadata, distance) in enumerate(zip( | |
| search_results['documents'][0], | |
| search_results['metadatas'][0], | |
| search_results['distances'][0] | |
| )): | |
| # Convert distance to similarity score (0-1) | |
| similarity = 1 - (distance / 2) # Rough approximation | |
| item = { | |
| 'content': doc, | |
| 'metadata': metadata, | |
| 'similarity': similarity, | |
| 'type': metadata.get('type', 'unknown') | |
| } | |
| results['items'].append(item) | |
| # Extract topics | |
| if 'topic' in metadata: | |
| results['topics'].add(metadata['topic']) | |
| # Filter by knowledge type if needed | |
| if knowledge_type != 'general': | |
| filtered_items = [] | |
| for item in results['items']: | |
| if self._matches_knowledge_type(item, knowledge_type): | |
| filtered_items.append(item) | |
| results['items'] = filtered_items[:5] # Limit to 5 | |
| except Exception as e: | |
| print(f"โ Vector search error: {e}") | |
| # Fall back to keyword search | |
| return self._keyword_search(query, knowledge_type) | |
| return results | |
| def _keyword_search(self, query: str, knowledge_type: str) -> Dict[str, Any]: | |
| """Fallback keyword search""" | |
| results = { | |
| 'items': [], | |
| 'topics': set(), | |
| 'method': 'keyword' | |
| } | |
| query_lower = query.lower() | |
| query_terms = set(query_lower.split()) | |
| # Search topics | |
| for topic_name, topic_data in self.knowledge_base.get('topics', {}).items(): | |
| topic_lower = topic_name.lower() | |
| # Check topic name | |
| if any(term in topic_lower for term in query_terms): | |
| results['topics'].add(topic_name) | |
| results['items'].append({ | |
| 'content': f"Topic: {topic_name}. {topic_data.get('description', '')}", | |
| 'metadata': {'type': 'topic', 'topic': topic_name}, | |
| 'similarity': 0.8, | |
| 'type': 'topic' | |
| }) | |
| # Search concepts within topic | |
| for concept in topic_data.get('key_concepts', []): | |
| concept_lower = concept['name'].lower() | |
| if any(term in concept_lower for term in query_terms): | |
| results['topics'].add(topic_name) | |
| content = f"Concept: {concept['name']}. {concept.get('definition', '')}" | |
| if 'formula' in concept: | |
| content += f" Formula: {concept['formula']}" | |
| results['items'].append({ | |
| 'content': content, | |
| 'metadata': {'type': 'concept', 'topic': topic_name, 'concept': concept['name']}, | |
| 'similarity': 0.7, | |
| 'type': 'concept' | |
| }) | |
| # Search glossary | |
| for term, definition in self.knowledge_base.get('glossary', {}).items(): | |
| term_lower = term.lower() | |
| if term_lower in query_lower or any(term_word in query_terms for term_word in term_lower.split()): | |
| results['items'].append({ | |
| 'content': f"Term: {term}. Definition: {definition}", | |
| 'metadata': {'type': 'glossary', 'term': term}, | |
| 'similarity': 0.9, | |
| 'type': 'glossary' | |
| }) | |
| # Sort by similarity (keyword match quality) | |
| results['items'].sort(key=lambda x: x['similarity'], reverse=True) | |
| results['items'] = results['items'][:5] # Limit to 5 | |
| return results | |
| def _matches_knowledge_type(self, item: Dict[str, Any], knowledge_type: str) -> bool: | |
| """Check if item matches the requested knowledge type""" | |
| item_type = item.get('type') | |
| type_mapping = { | |
| 'definition': ['glossary', 'concept'], | |
| 'explanation': ['topic', 'concept'], | |
| 'formula': ['concept', 'formula'], | |
| 'example': ['concept'], | |
| 'comparison': ['topic', 'concept'], | |
| 'list': ['topic'] | |
| } | |
| if knowledge_type in type_mapping: | |
| return item_type in type_mapping[knowledge_type] | |
| return True | |
| def _format_knowledge_response(self, | |
| query: str, | |
| results: Dict[str, Any], | |
| knowledge_type: str, | |
| user_profile: Dict[str, Any]) -> str: | |
| """Format knowledge response based on query type""" | |
| response_parts = [] | |
| if not results.get('items'): | |
| response_parts.append(f"I couldn't find specific information about '{query}'.") | |
| response_parts.append("\n**Try asking about:**") | |
| response_parts.append("- Time Value of Money (TVM)") | |
| response_parts.append("- Statistics and probability") | |
| response_parts.append("- Portfolio management") | |
| response_parts.append("- Financial ratios") | |
| response_parts.append("- CFA ethics") | |
| return "\n".join(response_parts) | |
| # Add header based on knowledge type | |
| headers = { | |
| 'definition': "๐ Definitions", | |
| 'explanation': "๐ Explanations", | |
| 'formula': "๐งฎ Formulas", | |
| 'example': "๐ก Examples", | |
| 'comparison': "โ๏ธ Comparisons", | |
| 'list': "๐ Lists", | |
| 'general': "๐ Knowledge Results" | |
| } | |
| response_parts.append(f"## {headers.get(knowledge_type, '๐ Knowledge')}") | |
| # Add results | |
| for i, item in enumerate(results['items'][:3], 1): # Limit to 3 items | |
| response_parts.append(f"\n### Result {i}") | |
| content = item['content'] | |
| metadata = item.get('metadata', {}) | |
| # Format based on type | |
| if metadata.get('type') == 'topic': | |
| response_parts.append(f"**Topic:** {metadata.get('topic', 'Unknown')}") | |
| response_parts.append(f"{content}") | |
| elif metadata.get('type') == 'concept': | |
| response_parts.append(f"**Concept:** {metadata.get('concept', 'Unknown')}") | |
| response_parts.append(f"**Topic:** {metadata.get('topic', 'General')}") | |
| response_parts.append(f"{content}") | |
| elif metadata.get('type') == 'glossary': | |
| response_parts.append(f"**Term:** {metadata.get('term', 'Unknown')}") | |
| response_parts.append(f"{content}") | |
| elif metadata.get('type') == 'formula': | |
| response_parts.append(f"**Formula:** {metadata.get('formula_name', 'Unknown')}") | |
| response_parts.append(f"{content}") | |
| else: | |
| response_parts.append(f"{content}") | |
| # Add study tips if relevant | |
| if results.get('topics'): | |
| topics = list(results['topics'])[:2] # Limit to 2 topics | |
| if topics: | |
| response_parts.append("\n---\n") | |
| response_parts.append("**๐ Study Tips:**") | |
| for topic in topics: | |
| tips = self._get_study_tips(topic, user_profile) | |
| if tips: | |
| response_parts.append(f"\n**For {topic}:**") | |
| for tip in tips[:2]: | |
| response_parts.append(f"- {tip}") | |
| return "\n".join(response_parts) | |
| def _get_study_tips(self, topic: str, user_profile: Dict[str, Any]) -> List[str]: | |
| """Get study tips for a topic""" | |
| tips_db = { | |
| 'Time Value of Money': [ | |
| "Practice with different compounding periods (annual, semi-annual, monthly)", | |
| "Master the TVM calculator keys: N, I/Y, PV, PMT, FV", | |
| "Understand the relationship between discount rate and present value" | |
| ], | |
| 'Statistics': [ | |
| "Focus on interpreting results, not just calculations", | |
| "Understand when to use population vs sample statistics", | |
| "Practice hypothesis testing steps systematically" | |
| ], | |
| 'Portfolio Management': [ | |
| "Memorize the CAPM formula: E(R) = Rf + ฮฒ(Rm - Rf)", | |
| "Understand the efficient frontier and capital market line", | |
| "Practice calculating portfolio variance with different correlations" | |
| ] | |
| } | |
| tips = tips_db.get(topic, [ | |
| "Break down complex concepts into smaller parts", | |
| "Create flashcards for key terms and formulas", | |
| "Practice with real-world examples" | |
| ]) | |
| # Adjust based on user profile | |
| experience = user_profile.get('experience_level', 'beginner') | |
| if experience == 'beginner': | |
| tips.insert(0, "Start with foundational concepts before advancing") | |
| elif experience == 'advanced': | |
| tips.append("Focus on application and analysis questions") | |
| return tips[:3] # Return top 3 tips | |
| def _calculate_confidence(self, results: Dict[str, Any]) -> float: | |
| """Calculate confidence in search results""" | |
| if not results.get('items'): | |
| return 0.2 | |
| # Calculate average similarity | |
| similarities = [item.get('similarity', 0.5) for item in results['items']] | |
| avg_similarity = sum(similarities) / len(similarities) | |
| # Adjust based on result count | |
| count_factor = min(len(results['items']) / 5, 1.0) * 0.2 | |
| confidence = avg_similarity + count_factor | |
| return min(confidence, 0.95) | |
| def _log_search(self, query: str, knowledge_type: str, results: Dict[str, Any]): | |
| """Log knowledge searches""" | |
| log_entry = { | |
| 'timestamp': datetime.now().isoformat(), | |
| 'query': query, | |
| 'knowledge_type': knowledge_type, | |
| 'result_count': len(results.get('items', [])), | |
| 'method': results.get('method', 'unknown'), | |
| 'topics': list(results.get('topics', set())) | |
| } | |
| self.search_history.append(log_entry) | |
| # Keep only last 100 searches | |
| if len(self.search_history) > 100: | |
| self.search_history = self.search_history[-100:] | |
| def get_search_history(self, limit: int = 10) -> List[Dict]: | |
| """Get recent search history""" | |
| return self.search_history[-limit:] | |
| def get_metrics(self) -> Dict[str, Any]: | |
| """Get knowledge agent metrics""" | |
| if not self.search_history: | |
| return { | |
| 'total_searches': 0, | |
| 'vector_db_ready': self.collection is not None, | |
| 'embedding_model_ready': self.embedding_model is not None, | |
| 'knowledge_items': self.collection.count() if self.collection else 0 | |
| } | |
| # Calculate success rate (searches with results) | |
| successful = sum(1 for entry in self.search_history if entry['result_count'] > 0) | |
| success_rate = successful / len(self.search_history) if self.search_history else 0 | |
| # Get most searched topics | |
| topic_counts = {} | |
| for entry in self.search_history: | |
| for topic in entry.get('topics', []): | |
| topic_counts[topic] = topic_counts.get(topic, 0) + 1 | |
| common_topics = sorted(topic_counts.items(), key=lambda x: x[1], reverse=True)[:3] | |
| return { | |
| 'total_searches': len(self.search_history), | |
| 'success_rate': success_rate, | |
| 'common_topics': common_topics, | |
| 'vector_db_ready': self.collection is not None, | |
| 'embedding_model_ready': self.embedding_model is not None, | |
| 'knowledge_items': self.collection.count() if self.collection else 0, | |
| 'recent_queries': [entry['query'][:50] for entry in self.search_history[-5:]] | |
| } | |