# 🎯 SherlockAI Interview Preparation Guide ## 🚀 **ELEVATOR PITCH (30 seconds)** "SherlockAI is an AI-powered issue intelligence system that transforms tribal knowledge into institutional memory. When engineers face production issues, instead of asking colleagues or searching Slack, they get instant AI-generated solutions based on similar past incidents. We built this using a RAG architecture with hybrid search, achieving 85% accuracy and reducing resolution time from 60 minutes to 5 minutes." --- ## 🏗️ **SYSTEM ARCHITECTURE WALKTHROUGH** ### **High-Level Architecture** ``` ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ React/TS │ │ FastAPI │ │ AI Services │ │ Frontend │◄──►│ Backend │◄──►│ (Gemini) │ │ │ │ │ │ │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ ▼ ▼ ┌─────────────────┐ ┌─────────────────┐ │ PostgreSQL │ │ Pinecone │ │ + Redis │ │ Vector DB │ └─────────────────┘ └─────────────────┘ ``` ### **RAG Pipeline Flow** ``` User Query → Query Classification → Hybrid Search → Semantic Validation → AIGeneration → Response │ │ │ │ │ │ │ ▼ ▼ ▼ ▼ ▼ "UPI timeout" Simple/Complex BM25+TF-IDF+ Domain Check Gemini 1.5 Formatted Semantic Search Flash Answer ``` --- Flow: User Query → Query Classification → Hybrid Search → Semantic Validation → AI Generation → Response ## 🤖 **MODEL TRAINING & DATA PIPELINE** ### **Training Architecture Overview** ``` issues.json → Data Validation → Embedding Generation → Vector Storage → IndexBuilding │ │ │ │ │ │ ▼ ▼ ▼ ▼ 75 Issues Pydantic Models Google Gemini API Pinecone DB Search Ready ``` 1. User Query 👉 Yeh woh input hai jo user deta hai. Example: "UPI timeout" 2. Query Classification 👉 System check karta hai ki query simple hai ya complex. Simple → direct ek knowledge base se answer mil sakta hai. Complex → thoda reasoning, multiple steps ya AI generation chahiye. 3. Hybrid Search 👉 Yaha query ke liye information search hoti hai using BM25 + TF-IDF + Semantic Search. BM25 + TF-IDF → keyword-based matching (exact words dekhta hai). Semantic Search → meaning-based matching (agar user ne alag wording use ki hai tab bhi relevant result mile). Example: "UPI timeout" ke liye system keyword match + semantic similarity dono dekhega. 4. Semantic Validation (Domain Check) 👉 Jo results aaye search se, unko validate kiya jata hai ki woh sahi domain/intent se related hain ya nahi. Agar query "UPI timeout" hai to irrelevant cheeze (jaise “UPI offers”) hata di jayengi. 5. AI Generation (Gemini 1.5 Flash) 👉 Ab AI (Gemini 1.5 Flash) ko refined context diya jata hai aur woh final answer generate karta hai. 6. Response (Formatted Answer) 👉 AI ka output ek structured / formatted response ke form me user ko diya jata hai (e.g., step-by-step explanation, bullet points, table etc.). ## 💻 **CODE WALKTHROUGH PREPARATION** ### **1. RAG Service Core Logic** ```python async def process_rag_query(self, query: str) -> RAGResponse: """Main RAG pipeline: Retrieval → Augmentation → Generation""" start_time = time.time() # Step 1: Check for exact ticket ID (bypass semantic search) ticket_id = self.extract_exact_ticket_id(query) if ticket_id: return await self.process_exact_ticket_query(query, ticket_id) # Step 2: Query Classification query_complexity = await self.classify_query_complexity(query) # Step 3: Hybrid Retrieval incidents = await self.retrieve_incidents(query, query_complexity) # Step 4: Semantic Validation (prevents hallucinations) is_relevant, reason = self._validate_semantic_relevance(query, incidents) if not is_relevant: return self._generate_no_results_response(query, query_complexity) # Step 5: AI Generation generated_answer = await self.generate_rag_response(query, incidents, query_complexity) return RAGResponse(...) ``` **Interview Questions:** - **Q: Walk me through your RAG pipeline.** - **A: Explain each step, emphasizing semantic validation and exact ID bypass** ### **2. Hybrid Search Implementation** ```python async def hybrid_search(self, query: str, top_k: int = 10) -> List[Dict[str, Any]]: """Combine semantic, BM25, and TF-IDF search""" # Parallel search execution search_tasks = [ self.semantic_search(query, top_k * 2), self.bm25_search(query, top_k * 2), self.tfidf_search(query, top_k * 2) ] semantic_results, bm25_results, tfidf_results = await asyncio.gather(*search_tasks) # Score fusion with weights fused_results = self._fuse_scores(all_results, query) return fused_results[:top_k] def _fuse_scores(self, all_results: List[Dict], query: str) -> List[Dict]: """Weighted score combination with priority matching""" for result_id, group in result_groups.items(): base_score = ( 0.6 * group['semantic_score'] + 0.3 * group['bm25_score'] + 0.1 * group['tfidf_score'] ) # Priority matching boost enhanced_score, match_type = self._calculate_priority_score(query, group['metadata'], base_score) result['fused_score'] = enhanced_score ``` **Interview Questions:** - **Q: Why hybrid search instead of just semantic search?** - **A: Semantic captures meaning, BM25 captures exact terms, TF-IDF adds document relevance. Each has strengths.** ### **3. Semantic Validation (Anti-Hallucination)** ```python def _validate_semantic_relevance(self, query: str, incidents: List[Dict]) -> Tuple[bool, str]: """Revolutionary semantic validation to prevent AI hallucinations""" # Extract query intelligence query_domain = self._extract_query_domain(query.lower()) query_entities = self._extract_query_entities(query.lower()) best_match_score = 0 for incident in incidents: # Domain compatibility incident_domain = self._extract_query_domain(incident_text) domain_match = self._calculate_domain_compatibility(query_domain, incident_domain) # Entity overlap incident_entities = self._extract_query_entities(incident_text) entity_overlap = len(query_entities.intersection(incident_entities)) / max(len(query_entities), 1) # Composite relevance score composite_score = (domain_match * 0.5) + (entity_overlap * 0.3) + (intent_alignment * 0.2) if composite_score > best_match_score: best_match_score = composite_score # Trust high-scoring hybrid matches max_hybrid_score = max(inc.get('fused_score', 0) for inc in incidents) if max_hybrid_score >= 0.8: return True, "high_hybrid_confidence" # Semantic relevance threshold return best_match_score >= 0.3, "semantic_validation" ``` **Interview Questions:** - **Q: How do you prevent AI hallucinations?** - **A: Multi-layer validation: domain compatibility, entity matching, confidence thresholds, honest "no results"** --- ``` ### **1. Data Preparation Pipeline** #### **Q: Walk me through your training data preparation process.** **A:** ```python # Data validation and preprocessing class IssueTrainingPipeline: def __init__(self): self.validator = IssueValidator() self.preprocessor = TextPreprocessor() self.embedding_service = EmbeddingService() async def prepare_training_data(self, issues_file: str) -> List[TrainingExample]: # Load and validate raw data with open(issues_file, 'r') as f: raw_issues = json.load(f) validated_issues = [] for issue in raw_issues: # Schema validation if self.validator.validate_issue_schema(issue): # Content preprocessing processed_issue = self.preprocessor.clean_and_normalize(issue) validated_issues.append(processed_issue) else: logger.warning(f"Skipping invalid issue: {issue.get('id', 'unknown')}") return validated_issues def create_training_text(self, issue: Dict) -> str: """Combine title, description, and resolution for embedding""" components = [ issue['title'], issue['description'], f"Resolution: {issue['resolution']}" ] return ". ".join(components) ``` #### **Q: How do you ensure data quality in your training set?** **A:** ```python class IssueValidator: def validate_issue_schema(self, issue: Dict) -> bool: required_fields = ['id', 'title', 'description', 'resolution', 'tags'] # Check required fields for field in required_fields: if field not in issue or not issue[field]: return False # Validate content quality if len(issue['description']) < 50: # Minimum description length return False if len(issue['resolution']) < 20: # Minimum resolution length return False # Validate ID format (JSP-XXXX) if not re.match(r'^[A-Z]+-\d+$', issue['id']): return False return True def check_content_quality(self, text: str) -> float: """Quality score based on content richness""" score = 0.0 # Length score (0-0.3) if len(text) > 100: score += 0.3 elif len(text) > 50: score += 0.2 # Technical terms score (0-0.4) tech_terms = ['API', 'timeout', 'error', 'gateway', 'payment', 'UPI'] term_count = sum(1 for term in tech_terms if term.lower() in text.lower()) score += min(term_count * 0.1, 0.4) # Structure score (0-0.3) if any(marker in text for marker in ['1)', '2)', '•', '-']): score += 0.3 return min(score, 1.0) ``` ### **2. Embedding Generation Process** #### **Q: Explain your embedding strategy and model choice.** **A:** ```python class EmbeddingService: def __init__(self): self.model = "models/text-embedding-004" # 768 dimensions self.batch_size = 10 # API rate limiting self.cache = RedisCache() async def generate_embeddings_batch(self, texts: List[str]) -> List[List[float]]: """Generate embeddings with batching and caching""" embeddings = [] uncached_texts = [] uncached_indices = [] # Check cache first for i, text in enumerate(texts): cache_key = self._generate_cache_key(text) cached_embedding = await self.cache.get(cache_key) if cached_embedding: embeddings.append(json.loads(cached_embedding)) else: embeddings.append(None) uncached_texts.append(text) uncached_indices.append(i) # Generate embeddings for uncached texts if uncached_texts: new_embeddings = await self._generate_embeddings_api(uncached_texts) # Cache and insert new embeddings for idx, embedding in zip(uncached_indices, new_embeddings): embeddings[idx] = embedding cache_key = self._generate_cache_key(texts[idx]) await self.cache.set(cache_key, json.dumps(embedding), ttl=3600) return embeddings async def _generate_embeddings_api(self, texts: List[str]) -> List[List[float]]: """Call Google Gemini API with retry logic""" try: response = await genai.embed_content( model=self.model, content=texts, task_type="retrieval_document" ) return [emb.embedding for emb in response.embeddings] except Exception as e: if "rate_limit" in str(e).lower(): await asyncio.sleep(60) # Wait and retry return await self._generate_embeddings_api(texts) raise ``` **Key Technical Decisions:** - **Model Choice:** Google text-embedding-004 (768D) for cost-effectiveness and quality - **Batch Processing:** 10 texts per API call to respect rate limits - **Caching Strategy:** Redis with 1-hour TTL to reduce API costs - **Error Handling:** Exponential backoff for rate limiting ### **3. Vector Database Storage** #### **Q: How do you store and index embeddings for fast retrieval?** **A:** ```python class VectorStorageService: def __init__(self): self.pinecone_client = pinecone.Pinecone(api_key=os.getenv("PINECONE_API_KEY")) self.index_name = "sherlockai-incidents" self.dimension = 768 async def store_embeddings(self, training_data: List[Dict]) -> None: """Store embeddings with metadata in Pinecone""" # Prepare vectors for upsert vectors = [] for issue in training_data: vector_id = issue['id'] embedding = issue['embedding'] # Metadata for filtering and retrieval metadata = { 'id': issue['id'], 'title': issue['title'], 'description': issue['description'][:500], # Truncate for metadata limits 'resolution': issue['resolution'][:500], 'tags': issue['tags'], 'created_at': issue['created_at'], 'resolved_by': issue['resolved_by'], 'category': issue.get('category', 'general'), 'priority': issue.get('priority', 'medium') } vectors.append({ 'id': vector_id, 'values': embedding, 'metadata': metadata }) # Batch upsert to Pinecone index = self.pinecone_client.Index(self.index_name) batch_size = 100 for i in range(0, len(vectors), batch_size): batch = vectors[i:i + batch_size] await index.upsert(vectors=batch) logger.info(f"Upserted batch {i//batch_size + 1}/{len(vectors)//batch_size + 1}") def create_index_if_not_exists(self): """Initialize Pinecone index with optimal configuration""" existing_indexes = self.pinecone_client.list_indexes() if self.index_name not in [idx.name for idx in existing_indexes]: self.pinecone_client.create_index( name=self.index_name, dimension=self.dimension, metric='cosine', # Best for text embeddings spec=ServerlessSpec( cloud='aws', region='us-east-1' ) ) logger.info(f"Created Pinecone index: {self.index_name}") ``` ### **4. Training Scripts & Automation** #### **Q: Walk me through your training automation and scripts.** **A:** **Main Training Script (train_model.py):** ```python class ModelTrainer: def __init__(self): self.data_pipeline = IssueTrainingPipeline() self.embedding_service = EmbeddingService() self.vector_storage = VectorStorageService() self.database = DatabaseService() async def full_training_pipeline(self, issues_file: str = "issues.json"): """Complete training pipeline from raw data to searchable index""" logger.info("🚀 Starting SherlockAI training pipeline...") # Step 1: Data preparation and validation logger.info("📊 Loading and validating training data...") validated_issues = await self.data_pipeline.prepare_training_data(issues_file) logger.info(f"✅ Validated {len(validated_issues)} issues") # Step 2: Generate embeddings logger.info("🤖 Generating embeddings...") training_texts = [ self.data_pipeline.create_training_text(issue) for issue in validated_issues ] embeddings = await self.embedding_service.generate_embeddings_batch(training_texts) # Attach embeddings to issues for issue, embedding in zip(validated_issues, embeddings): issue['embedding'] = embedding issue['training_text'] = self.data_pipeline.create_training_text(issue) # Step 3: Store in databases logger.info("💾 Storing in databases...") # Store in PostgreSQL for metadata queries await self.database.store_issues(validated_issues) # Store in Pinecone for vector search await self.vector_storage.store_embeddings(validated_issues) # Step 4: Build hybrid search indices logger.info("🔍 Building hybrid search indices...") await self.build_hybrid_indices(validated_issues) # Step 5: Validation and testing logger.info("🧪 Running validation tests...") await self.validate_training_results() logger.info("🎉 Training completed successfully!") return { "total_issues": len(validated_issues), "embedding_dimension": 768, "vector_count": len(embeddings), "training_time": time.time() - start_time } async def build_hybrid_indices(self, issues: List[Dict]): """Build BM25 and TF-IDF indices for hybrid search""" from rank_bm25 import BM25Okapi from sklearn.feature_extraction.text import TfidfVectorizer # Prepare documents for indexing documents = [issue['training_text'] for issue in issues] tokenized_docs = [doc.lower().split() for doc in documents] # Build BM25 index bm25_index = BM25Okapi(tokenized_docs) # Build TF-IDF index tfidf_vectorizer = TfidfVectorizer( max_features=5000, stop_words='english', ngram_range=(1, 2) ) tfidf_matrix = tfidf_vectorizer.fit_transform(documents) # Save indices await self.save_search_indices(bm25_index, tfidf_vectorizer, tfidf_matrix) ``` **Quick Training Script (quick_train.py):** ```python async def quick_retrain(): """Fast retraining for development and testing""" trainer = ModelTrainer() # Check what's already trained existing_count = await trainer.vector_storage.get_vector_count() logger.info(f"📊 Current vector database has {existing_count} vectors") # Load new/updated issues new_issues = await trainer.data_pipeline.load_new_issues() if new_issues: logger.info(f"🆕 Found {len(new_issues)} new issues to train") await trainer.incremental_training(new_issues) else: logger.info("✅ No new issues found, database is up to date") ``` ### **5. Training Performance & Optimization** #### **Q: How do you optimize training performance and costs?** **A:** **Performance Optimizations:** ```python class TrainingOptimizer: def __init__(self): self.embedding_cache = RedisCache() self.batch_processor = BatchProcessor() async def optimized_embedding_generation(self, texts: List[str]) -> List[List[float]]: """Optimized embedding generation with multiple strategies""" # Strategy 1: Cache hit optimization cache_hits, cache_misses = await self.check_embedding_cache(texts) logger.info(f"Cache hit rate: {len(cache_hits)}/{len(texts)} ({len(cache_hits)/len(texts)*100:.1f}%)") # Strategy 2: Batch processing for API efficiency if cache_misses: new_embeddings = await self.batch_processor.process_in_batches( cache_misses, batch_size=10, delay_between_batches=1.0 # Rate limiting ) # Update cache await self.update_embedding_cache(cache_misses, new_embeddings) # Strategy 3: Parallel processing for I/O operations storage_tasks = [ self.store_in_postgres(issues), self.store_in_pinecone(embeddings), self.build_search_indices(texts) ] await asyncio.gather(*storage_tasks) return embeddings def calculate_training_costs(self, num_texts: int, avg_tokens_per_text: int) -> Dict: """Calculate API costs for training""" # Google Gemini embedding costs (as of 2024) cost_per_1k_tokens = 0.00001 # $0.00001 per 1K tokens total_tokens = num_texts * avg_tokens_per_text total_cost = (total_tokens / 1000) * cost_per_1k_tokens return { "total_texts": num_texts, "total_tokens": total_tokens, "estimated_cost_usd": total_cost, "cost_per_text": total_cost / num_texts } ``` ### **6. Training Validation & Quality Assurance** #### **Q: How do you validate your training results?** **A:** ```python class TrainingValidator: async def validate_training_results(self) -> Dict: """Comprehensive validation of training pipeline""" validation_results = { "data_quality": await self.validate_data_quality(), "embedding_quality": await self.validate_embeddings(), "search_performance": await self.validate_search_performance(), "end_to_end": await self.validate_end_to_end_pipeline() } return validation_results async def validate_embeddings(self) -> Dict: """Validate embedding quality and consistency""" # Test 1: Embedding dimensions sample_embeddings = await self.get_sample_embeddings(10) dimensions = [len(emb) for emb in sample_embeddings] assert all(dim == 768 for dim in dimensions), "Inconsistent embedding dimensions" # Test 2: Semantic similarity similar_queries = [ ("UPI payment failed", "UPI transaction error"), ("timeout error", "request timeout"), ("gateway issue", "payment gateway problem") ] similarity_scores = [] for query1, query2 in similar_queries: emb1 = await self.embedding_service.embed_text(query1) emb2 = await self.embedding_service.embed_text(query2) similarity = cosine_similarity([emb1], [emb2])[0][0] similarity_scores.append(similarity) avg_similarity = sum(similarity_scores) / len(similarity_scores) assert avg_similarity > 0.7, f"Low semantic similarity: {avg_similarity}" return { "dimension_consistency": True, "avg_semantic_similarity": avg_similarity, "similarity_scores": similarity_scores } async def validate_search_performance(self) -> Dict: """Validate search accuracy and performance""" test_queries = [ {"query": "UPI timeout", "expected_tags": ["UPI", "Timeout"]}, {"query": "card payment failed", "expected_tags": ["Cards", "Payments"]}, {"query": "webhook not working", "expected_tags": ["Webhooks"]} ] performance_metrics = [] for test in test_queries: start_time = time.time() results = await self.search_service.hybrid_search(test["query"], top_k=5) search_time = time.time() - start_time # Check if expected tags appear in results found_tags = set() for result in results: found_tags.update(result.get('tags', [])) tag_overlap = len(set(test["expected_tags"]).intersection(found_tags)) tag_precision = tag_overlap / len(test["expected_tags"]) performance_metrics.append({ "query": test["query"], "search_time_ms": search_time * 1000, "results_count": len(results), "tag_precision": tag_precision, "top_score": results[0]['score'] if results else 0 }) avg_search_time = sum(m["search_time_ms"] for m in performance_metrics) / len(performance_metrics) avg_precision = sum(m["tag_precision"] for m in performance_metrics) / len(performance_metrics) return { "avg_search_time_ms": avg_search_time, "avg_tag_precision": avg_precision, "performance_metrics": performance_metrics } ``` ### **7. Incremental Learning & Updates** #### **Q: How do you handle adding new incidents without full retraining?** **A:** ```python class IncrementalTrainer: async def add_new_incident(self, new_issue: Dict) -> bool: """Add single incident to existing trained model""" # Step 1: Validate new incident if not self.validator.validate_issue_schema(new_issue): logger.error(f"Invalid incident schema: {new_issue.get('id', 'unknown')}") return False # Step 2: Generate embedding training_text = self.data_pipeline.create_training_text(new_issue) embedding = await self.embedding_service.embed_text(training_text) # Step 3: Store in databases await self.database.insert_issue(new_issue) await self.vector_storage.upsert_single_vector( id=new_issue['id'], embedding=embedding, metadata=self._extract_metadata(new_issue) ) # Step 4: Update search indices await self.update_hybrid_indices(new_issue, training_text) logger.info(f"✅ Successfully added incident {new_issue['id']} to trained model") return True async def batch_update_incidents(self, new_issues: List[Dict]) -> Dict: """Efficiently add multiple incidents""" successful_updates = 0 failed_updates = [] # Process in batches for efficiency batch_size = 10 for i in range(0, len(new_issues), batch_size): batch = new_issues[i:i + batch_size] try: # Generate embeddings for batch training_texts = [self.data_pipeline.create_training_text(issue) for issue in batch] embeddings = await self.embedding_service.generate_embeddings_batch(training_texts) # Store batch await self.database.insert_issues_batch(batch) await self.vector_storage.upsert_vectors_batch(batch, embeddings) successful_updates += len(batch) except Exception as e: logger.error(f"Failed to process batch {i//batch_size + 1}: {e}") failed_updates.extend([issue['id'] for issue in batch]) return { "successful_updates": successful_updates, "failed_updates": failed_updates, "total_processed": len(new_issues) } ``` ### **8. Training Monitoring & Metrics** #### **Q: How do you monitor training progress and quality?** **A:** ```python class TrainingMonitor: def __init__(self): self.metrics_collector = MetricsCollector() self.alert_manager = AlertManager() async def track_training_metrics(self, training_session: Dict): """Track comprehensive training metrics""" metrics = { "timestamp": datetime.utcnow(), "session_id": training_session["session_id"], "total_issues": training_session["total_issues"], "training_duration_seconds": training_session["duration"], "embedding_generation_time": training_session["embedding_time"], "vector_storage_time": training_session["storage_time"], "validation_results": training_session["validation"], "api_costs": training_session["costs"], "cache_hit_rate": training_session["cache_hit_rate"], "error_count": training_session["errors"] } # Store metrics await self.metrics_collector.store_training_metrics(metrics) # Check for anomalies await self.check_training_anomalies(metrics) async def check_training_anomalies(self, metrics: Dict): """Detect training issues and send alerts""" # Check for performance degradation if metrics["training_duration_seconds"] > 3600: # > 1 hour await self.alert_manager.send_alert( "Training Performance Alert", f"Training took {metrics['training_duration_seconds']/60:.1f} minutes" ) # Check for high error rates if metrics["error_count"] > metrics["total_issues"] * 0.1: # > 10% errors await self.alert_manager.send_alert( "Training Quality Alert", f"High error rate: {metrics['error_count']}/{metrics['total_issues']}" ) # Check for low cache hit rates if metrics["cache_hit_rate"] < 0.5: # < 50% cache hits await self.alert_manager.send_alert( "Training Efficiency Alert", f"Low cache hit rate: {metrics['cache_hit_rate']*100:.1f}%" ) ``` --- ## 🎯 **TECHNICAL DEEP DIVE QUESTIONS** ### **Architecture & Design** #### **Q: Why did you choose this specific tech stack?** **A:** - **FastAPI:** Async performance, auto-docs, type safety - **React + TypeScript:** Type safety, component reusability, excellent ecosystem - **Pinecone:** Managed vector DB, no infrastructure overhead - **Google Gemini:** Cost-effective, state-of-the-art embeddings - **PostgreSQL:** ACID compliance, JSON support, proven reliability #### **Q: How does your system handle concurrent users?** **A:** ```python # Async FastAPI handles concurrent requests @app.post("/api/v1/rag/query") async def rag_query(request: RAGRequest): # Non-blocking AI calls embedding = await ai_service.embed_text(request.query) incidents = await hybrid_search_service.hybrid_search(request.query) # Parallel processing tasks = [ ai_service.generate_fix_suggestion(inc, request.query) for inc in incidents ] suggestions = await asyncio.gather(*tasks) ``` #### **Q: Explain your caching strategy.** **A:** - **Embedding Cache:** Redis, 1-hour TTL (expensive AI calls) - **Search Cache:** Redis, 5-minute TTL (frequent queries) - **Query Classification Cache:** In-memory (lightweight) - **Static Assets:** CDN caching ### **AI/ML Deep Dive** #### **Q: How do you ensure AI response quality?** **A:** 1. **Prompt Engineering:** Domain-specific prompts 2. **Temperature Control:** Low (0.1-0.2) for consistency 3. **Context Validation:** Only use retrieved incident context 4. **Confidence Scoring:** Reject low-confidence matches 5. **Fallback Mechanisms:** Rule-based responses when AI fails #### **Q: Walk me through your embedding strategy.** **A:** ```python async def embed_text(self, text: str) -> List[float]: # Check cache first cache_key = self._generate_cache_key(text, self.embed_model) cached = await redis_client.get(cache_key) if cached: return json.loads(cached) # Generate new embedding resp = genai.embed_content(model="text-embedding-004", content=text) embedding = self._extract_embedding(resp) # Cache for 1 hour await redis_client.setex(cache_key, 3600, json.dumps(embedding)) return embedding ``` #### **Q: How do you handle different query types?** **A:** ```python class QueryComplexity(Enum): SIMPLE = "simple" # "UPI timeout" → 3 results COMPLEX = "complex" # "Why do refunds fail?" → 8 results UNKNOWN = "unknown" # Non-payment queries async def classify_query_complexity(self, query: str) -> QueryComplexity: prompt = f"Classify this query: {query}\nOptions: simple, complex, unknown" response = await gemini_model.generate_content(prompt) return QueryComplexity(response.text.strip().lower()) ``` ### **Performance & Scalability** #### **Q: What are your performance benchmarks?** **A:** - **Search Speed:** < 2 seconds for complex queries - **Throughput:** 100+ concurrent users - **Accuracy:** 85%+ relevance for payment domain - **Cache Hit Rate:** 70%+ for embeddings #### **Q: How would you scale this to 10,000 users?** **A:** 1. **Horizontal Scaling:** Load balancer + multiple FastAPI instances 2. **Database Scaling:** Read replicas, connection pooling 3. **Cache Scaling:** Redis cluster 4. **AI Scaling:** Request batching, async processing 5. **CDN:** Static asset distribution ### **Data & Integration** #### **Q: How do you ensure data quality?** **A:** ```python class IssueModel(BaseModel): id: str = Field(..., regex=r'^[A-Z]+-\d+$') title: str = Field(..., min_length=10, max_length=200) description: str = Field(..., min_length=50) resolution: str = Field(..., min_length=20) tags: List[str] = Field(..., min_items=1) created_at: datetime resolved_by: EmailStr # Validation pipeline def validate_issue(issue_data: dict) -> bool: try: validated = IssueModel(**issue_data) return True except ValidationError as e: logger.error(f"Validation failed: {e}") return False ``` --- ## 🎪 **DEMO SCENARIOS** ### **Scenario 1: Complex Technical Query** **Query:** "Hyper PG Transactions Stuck in Authorizing State" **Expected:** Find JSP-1037, show numbered problem list, provide resolution steps **Demo Script:** 1. "Let me show you a complex payment issue" 2. Type query → highlight 2-second response time 3. "Notice it found JSP-1037 with 81% confidence" 4. "Beautiful formatting with numbered problems and step-by-step resolution" ### **Scenario 2: Exact Ticket Lookup** **Query:** "JSP-1052" **Expected:** Direct ticket details, bypass semantic search **Demo Script:** 1. "For exact ticket IDs, we bypass semantic search entirely" 2. Type JSP-1052 → instant response 3. "Direct database lookup with complete ticket information" ### **Scenario 3: No Results Honesty** **Query:** "How to deploy Kubernetes" **Expected:** Honest "no results" response **Demo Script:** 1. "We're honest when we don't have relevant data" 2. Type non-payment query → clear "no results" message 3. "Better to be honest than force irrelevant matches" --- ## 🏆 **COMPETITIVE ADVANTAGES TO HIGHLIGHT** ### **Technical Innovation** 1. **Hybrid Search:** First to combine semantic + keyword + TF-IDF 2. **Semantic Validation:** Prevents AI hallucinations 3. **Domain Intelligence:** Payment-specific entity extraction 4. **Exact ID Bypass:** Smart routing for different query types ### **User Experience** 1. **ChatGPT-like Interface:** Familiar, intuitive 2. **Beautiful Formatting:** Numbered lists, clear structure 3. **Honest Responses:** "No results" when appropriate 4. **Fast Performance:** < 2 second responses ### **Enterprise Ready** 1. **Security:** OAuth 2.0, rate limiting, input validation 2. **Monitoring:** Structured logging, metrics, health checks 3. **Scalability:** Async architecture, caching, horizontal scaling 4. **Reliability:** Error handling, fallback mechanisms --- ## 🔥 **ADVANCED TECHNICAL QUESTIONS** ### **Database Design & Optimization** #### **Q: How would you optimize database performance for 1 million incidents?** **A:** ```sql -- Indexing strategy CREATE INDEX idx_incidents_tags ON incidents USING GIN(tags); CREATE INDEX idx_incidents_created_at ON incidents(created_at DESC); CREATE INDEX idx_incidents_resolved_by ON incidents(resolved_by); -- Partitioning by date CREATE TABLE incidents_2024 PARTITION OF incidents FOR VALUES FROM ('2024-01-01') TO ('2025-01-01'); -- Full-text search optimization CREATE INDEX idx_incidents_fts ON incidents USING GIN( to_tsvector('english', title || ' ' || description) ); ``` #### **Q: How do you handle database migrations in production?** **A:** - **Blue-Green Deployment:** Zero-downtime migrations - **Backward Compatibility:** New columns nullable, old columns deprecated gradually - **Migration Testing:** Test on production-like data volumes - **Rollback Strategy:** Always have a rollback plan ```python # Alembic migration example def upgrade(): # Add new column as nullable first op.add_column('incidents', sa.Column('priority_score', sa.Float(), nullable=True)) # Populate data in batches connection = op.get_bind() connection.execute("UPDATE incidents SET priority_score = 0.5 WHERE priority_score IS NULL") # Make non-nullable after population op.alter_column('incidents', 'priority_score', nullable=False) ``` ### **Error Handling & Resilience** #### **Q: How do you handle AI service failures gracefully?** **A:** ```python from tenacity import retry, stop_after_attempt, wait_exponential class AIService: @retry( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10) ) async def embed_text_with_retry(self, text: str) -> List[float]: try: return await self.embed_text(text) except Exception as e: if "rate_limit" in str(e).lower(): await asyncio.sleep(60) # Wait longer for rate limits raise elif "quota" in str(e).lower(): # Switch to fallback model return await self.fallback_embed_text(text) else: raise async def search_with_circuit_breaker(self, query: str): if self.circuit_breaker.is_open(): return await self.cached_search_fallback(query) try: result = await self.ai_search(query) self.circuit_breaker.record_success() return result except Exception as e: self.circuit_breaker.record_failure() return await self.keyword_search_fallback(query) ``` #### **Q: What's your strategy for handling partial system failures?** **A:** - **Graceful Degradation:** If AI fails, fall back to keyword search - **Circuit Breakers:** Prevent cascade failures - **Bulkhead Pattern:** Isolate critical components - **Health Checks:** Proactive monitoring and alerting ### **Security Deep Dive** #### **Q: How do you prevent prompt injection attacks?** **A:** ```python def sanitize_user_input(query: str) -> str: # Remove potential prompt injection patterns dangerous_patterns = [ r'ignore\s+previous\s+instructions', r'system\s*:', r'assistant\s*:', r'<\s*script\s*>', r'javascript\s*:', ] sanitized = query for pattern in dangerous_patterns: sanitized = re.sub(pattern, '', sanitized, flags=re.IGNORECASE) # Limit length and special characters sanitized = sanitized[:500] # Max length sanitized = re.sub(r'[^\w\s\-\.\,\?\!]', '', sanitized) return sanitized.strip() def validate_ai_response(response: str, context: str) -> bool: # Ensure response is based on provided context context_keywords = set(context.lower().split()) response_keywords = set(response.lower().split()) # Check if response has reasonable overlap with context overlap = len(context_keywords.intersection(response_keywords)) return overlap >= min(10, len(context_keywords) * 0.3) ``` #### **Q: How do you handle sensitive data in incident descriptions?** **A:** - **Data Masking:** PII detection and redaction - **Access Controls:** Role-based permissions - **Audit Logging:** Track who accessed what data - **Encryption:** At rest and in transit ```python import re def mask_sensitive_data(text: str) -> str: # Mask email addresses text = re.sub(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', '***@***.***', text) # Mask credit card numbers text = re.sub(r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b', '****-****-****-****', text) # Mask API keys text = re.sub(r'\b[A-Za-z0-9]{32,}\b', '***API_KEY***', text) return text ``` ### **Testing Strategy** #### **Q: How do you test AI-powered features?** **A:** ```python import pytest from unittest.mock import AsyncMock class TestRAGService: @pytest.fixture def mock_ai_service(self): mock = AsyncMock() mock.embed_text.return_value = [0.1] * 768 # Mock embedding mock.generate_fix_suggestion.return_value = "Fix: Check configuration" return mock @pytest.mark.asyncio async def test_exact_ticket_lookup(self, mock_ai_service): rag_service = RAGService() rag_service.ai_service = mock_ai_service response = await rag_service.process_rag_query("JSP-1052") assert response.rag_strategy == "exact_id_lookup" assert response.confidence_score == 1.0 assert "JSP-1052" in response.generated_answer @pytest.mark.asyncio async def test_semantic_validation_rejects_irrelevant(self, mock_ai_service): # Test that irrelevant results are properly rejected query = "How to bake a cake" incidents = [{"title": "UPI payment failed", "description": "Payment error"}] is_relevant, reason = rag_service._validate_semantic_relevance(query, incidents) assert not is_relevant assert reason == "insufficient_semantic_overlap" # Load testing with realistic scenarios def test_concurrent_search_performance(): import asyncio import time async def search_task(query: str): start = time.time() response = await rag_service.process_rag_query(query) return time.time() - start # Test 100 concurrent searches queries = ["UPI timeout"] * 100 tasks = [search_task(q) for q in queries] start_time = time.time() response_times = await asyncio.gather(*tasks) total_time = time.time() - start_time assert max(response_times) < 5.0 # No request takes more than 5 seconds assert total_time < 10.0 # All 100 requests complete in 10 seconds ``` --- ## 🚀 **SYSTEM DESIGN & SCALABILITY QUESTIONS** ### **Microservices Architecture** #### **Q: How would you break down SherlockAI into microservices?** **A:** ``` ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ Query Service │ │ Search Service │ │ AI Service │ │ - Validation │ │ - Hybrid Search│ │ - Embeddings │ │ - Routing │ │ - Indexing │ │ - Generation │ └─────────────────┘ └─────────────────┘ └─────────────────┘ │ │ │ └──────────────────────┼──────────────────────┘ │ ┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ │ Data Service │ │ Analytics Service│ │ Auth Service │ │ - CRUD Ops │ │ - Metrics │ │ - OAuth │ │ - Validation │ │ - Feedback │ │ - JWT │ └─────────────────┘ └─────────────────┘ └─────────────────┘ ``` **Benefits:** - **Independent Scaling:** Scale AI service separately from search - **Technology Diversity:** Use different languages for different services - **Fault Isolation:** One service failure doesn't bring down everything - **Team Autonomy:** Different teams can own different services #### **Q: How do you handle inter-service communication?** **A:** ```python # Async HTTP with circuit breakers class ServiceClient: def __init__(self, base_url: str): self.base_url = base_url self.circuit_breaker = CircuitBreaker(failure_threshold=5, timeout=60) @circuit_breaker async def call_service(self, endpoint: str, data: dict): async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post(f"{self.base_url}/{endpoint}", json=data) response.raise_for_status() return response.json() # Event-driven architecture for async operations class EventBus: async def publish(self, event_type: str, data: dict): await redis_client.publish(f"events:{event_type}", json.dumps(data)) async def subscribe(self, event_type: str, handler): pubsub = redis_client.pubsub() await pubsub.subscribe(f"events:{event_type}") async for message in pubsub.listen(): if message['type'] == 'message': data = json.loads(message['data']) await handler(data) ``` ### **Load Balancing & High Availability** #### **Q: How do you ensure 99.99% uptime?** **A:** - **Multi-Region Deployment:** Active-active across regions - **Health Checks:** Application and infrastructure level - **Auto-Scaling:** Based on CPU, memory, and custom metrics - **Database Replication:** Master-slave with automatic failover ```yaml # Kubernetes deployment with health checks apiVersion: apps/v1 kind: Deployment metadata: name: SherlockAI-api spec: replicas: 3 strategy: type: RollingUpdate rollingUpdate: maxSurge: 1 maxUnavailable: 0 template: spec: containers: - name: api image: SherlockAI:latest resources: requests: memory: "512Mi" cpu: "250m" limits: memory: "1Gi" cpu: "500m" livenessProbe: httpGet: path: /api/v1/health port: 8000 initialDelaySeconds: 30 periodSeconds: 10 readinessProbe: httpGet: path: /api/v1/ready port: 8000 initialDelaySeconds: 5 periodSeconds: 5 ``` --- ## 🤖 **AI/ML ADVANCED TOPICS** ### **Model Versioning & A/B Testing** #### **Q: How do you handle model updates without breaking production?** **A:** ```python class ModelVersionManager: def __init__(self): self.models = { "embedding_v1": "text-embedding-004", "embedding_v2": "text-embedding-3-large", "chat_v1": "gemini-1.5-flash", "chat_v2": "gemini-1.5-pro" } self.traffic_split = { "embedding_v1": 0.8, # 80% traffic "embedding_v2": 0.2 # 20% traffic for testing } async def get_embedding(self, text: str, user_id: str = None): # A/B testing based on user hash if user_id: user_hash = hash(user_id) % 100 if user_hash < 20: # 20% get new model model = self.models["embedding_v2"] version = "v2" else: model = self.models["embedding_v1"] version = "v1" else: model = self.models["embedding_v1"] version = "v1" # Log for analysis await self.log_model_usage(text, model, version, user_id) return await self.generate_embedding(text, model) async def log_model_usage(self, text: str, model: str, version: str, user_id: str): metrics = { "timestamp": datetime.utcnow(), "model": model, "version": version, "user_id": user_id, "text_length": len(text) } await self.analytics_service.log_event("model_usage", metrics) ``` ### **Bias Detection & Mitigation** #### **Q: How do you detect and prevent bias in AI responses?** **A:** ```python class BiasDetector: def __init__(self): self.protected_attributes = ['gender', 'race', 'religion', 'nationality'] self.bias_keywords = { 'gender': ['he', 'she', 'his', 'her', 'male', 'female'], 'race': ['white', 'black', 'asian', 'hispanic'], 'religion': ['christian', 'muslim', 'jewish', 'hindu'] } def detect_bias(self, query: str, response: str, incidents: List[Dict]) -> Dict: bias_score = 0 detected_biases = [] # Check for demographic mentions in response response_lower = response.lower() for category, keywords in self.bias_keywords.items(): for keyword in keywords: if keyword in response_lower: bias_score += 1 detected_biases.append(f"{category}:{keyword}") # Check for representation bias in retrieved incidents incident_sources = [inc.get('resolved_by', '') for inc in incidents] unique_resolvers = set(incident_sources) if len(unique_resolvers) < 2: bias_score += 1 detected_biases.append("single_resolver_bias") return { "bias_score": bias_score, "detected_biases": detected_biases, "needs_review": bias_score > 2 } async def mitigate_bias(self, response: str) -> str: # Remove gendered pronouns, use neutral language neutral_response = response neutral_response = re.sub(r'\bhe\b', 'they', neutral_response, flags=re.IGNORECASE) neutral_response = re.sub(r'\bshe\b', 'they', neutral_response, flags=re.IGNORECASE) neutral_response = re.sub(r'\bhis\b', 'their', neutral_response, flags=re.IGNORECASE) neutral_response = re.sub(r'\bher\b', 'their', neutral_response, flags=re.IGNORECASE) return neutral_response ``` --- ## 💼 **BUSINESS & PRODUCT QUESTIONS** ### **ROI & Success Metrics** #### **Q: How do you calculate the ROI of SherlockAI?** **A:** ```python class ROICalculator: def calculate_time_savings(self, usage_data: Dict) -> Dict: # Average time before SherlockAI avg_resolution_time_before = 60 # minutes avg_resolution_time_after = 5 # minutes time_saved_per_incident = avg_resolution_time_before - avg_resolution_time_after # Usage metrics monthly_queries = usage_data['monthly_queries'] successful_resolutions = monthly_queries * usage_data['success_rate'] # Calculate savings monthly_time_saved = successful_resolutions * time_saved_per_incident engineer_hourly_cost = 100 # $100/hour average monthly_cost_savings = (monthly_time_saved / 60) * engineer_hourly_cost # System costs monthly_system_cost = ( usage_data['ai_api_costs'] + usage_data['infrastructure_costs'] + usage_data['maintenance_costs'] ) # ROI calculation net_savings = monthly_cost_savings - monthly_system_cost roi_percentage = (net_savings / monthly_system_cost) * 100 return { "monthly_time_saved_hours": monthly_time_saved / 60, "monthly_cost_savings": monthly_cost_savings, "monthly_system_cost": monthly_system_cost, "net_savings": net_savings, "roi_percentage": roi_percentage, "payback_period_months": monthly_system_cost / net_savings if net_savings > 0 else float('inf') } ``` #### **Q: What metrics do you track for product success?** **A:** - **Usage Metrics:** Daily/monthly active users, queries per user - **Quality Metrics:** Response accuracy, user satisfaction ratings - **Performance Metrics:** Response time, system uptime - **Business Metrics:** Time saved, incidents prevented, knowledge retention ### **Competitive Analysis** #### **Q: How does SherlockAI compare to existing solutions like Stack Overflow or internal wikis?** **A:** | Feature | SherlockAI | Stack Overflow | Internal Wiki | Slack Search | |---------|----------|----------------|---------------|--------------| | **Domain Specific** | ✅ Payment focus | ❌ General | ✅ Company specific | ✅ Company specific | | **AI-Powered** | ✅ RAG + LLM | ❌ Manual | ❌ Manual | ❌ Keyword only | | **Real-time** | ✅ Instant | ❌ Async Q&A | ❌ Manual updates | ✅ Real-time | | **Context Aware** | ✅ Semantic search | ❌ Tag-based | ❌ Manual categorization | ❌ Basic search | | **Source Attribution** | ✅ Always cited | ✅ Answers cited | ✅ Manual links | ❌ No context | | **Learning** | ✅ Continuous | ❌ Community driven | ❌ Manual | ❌ No learning | **Key Differentiators:** 1. **Domain Intelligence:** Understands payment terminology and context 2. **Instant Answers:** No waiting for human responses 3. **Source Attribution:** Always shows which incidents informed the answer 4. **Continuous Learning:** Improves with each interaction --- ## 🎯 **SCENARIO-BASED QUESTIONS** ### **Crisis Management** #### **Q: It's 2 AM, SherlockAI is down, and engineers can't resolve a critical payment outage. What do you do?** **A:** **Immediate Response (0-5 minutes):** 1. **Acknowledge the incident** - Update status page 2. **Activate incident response team** - Page on-call engineers 3. **Implement fallback** - Direct users to backup documentation 4. **Start war room** - Slack channel for coordination **Investigation (5-30 minutes):** ```bash # Check system health kubectl get pods -n SherlockAI kubectl logs -f deployment/SherlockAI-api # Check dependencies curl -f http://pinecone-api/health redis-cli ping psql -c "SELECT 1" # Check metrics curl http://localhost:9090/api/v1/query?query=up{job="SherlockAI"} ``` **Resolution Strategy:** - **Database issues:** Switch to read replica - **AI service down:** Enable keyword-only search mode - **Vector DB issues:** Use cached results + BM25 search - **Complete failure:** Serve static FAQ page **Post-Incident (24-48 hours):** 1. **Root cause analysis** - 5 whys methodology 2. **Incident report** - Timeline, impact, lessons learned 3. **Preventive measures** - Additional monitoring, circuit breakers 4. **Team retrospective** - Process improvements #### **Q: Search results suddenly become irrelevant for all queries. How do you debug?** **A:** **Step 1: Isolate the Problem** ```python # Check if it's a data issue recent_incidents = await db.query("SELECT * FROM incidents ORDER BY created_at DESC LIMIT 10") print(f"Recent incidents: {len(recent_incidents)}") # Check if it's an AI issue test_embedding = await ai_service.embed_text("test query") print(f"Embedding dimension: {len(test_embedding)}") # Check if it's a search issue search_results = await hybrid_search.search("UPI timeout", top_k=5) print(f"Search results: {len(search_results)}") ``` **Step 2: Check Recent Changes** - **Code deployments:** Any recent releases? - **Data updates:** New incident ingestion? - **Model changes:** AI service updates? - **Configuration changes:** Environment variables? **Step 3: Validate Each Component** ```python # Test semantic search semantic_results = await ai_service.search_similar_issues(test_embedding) assert len(semantic_results) > 0, "Semantic search broken" # Test keyword search bm25_results = await hybrid_search.bm25_search("payment") assert len(bm25_results) > 0, "BM25 search broken" # Test score fusion fused_results = hybrid_search._fuse_scores(all_results, "test") assert all(r.get('fused_score', 0) > 0 for r in fused_results), "Score fusion broken" ``` **Step 4: Implement Fix** - **Rollback:** If recent deployment caused it - **Data refresh:** If data corruption detected - **Cache clear:** If stale cache causing issues - **Fallback mode:** Disable problematic components temporarily --- ## 🎯 **COMMON INTERVIEW TRAPS & ANSWERS** --- ## 🎯 **COMMON INTERVIEW TRAPS & ANSWERS** ### **Q: Why not just use ChatGPT/existing solutions?** **A:** - **Domain Specificity:** Our system understands payment terminology - **Source Attribution:** Always shows which incidents informed the answer - **No Hallucinations:** Strict validation prevents made-up answers - **Enterprise Control:** Your data stays in your infrastructure ### **Q: How do you handle biased or incorrect historical data?** **A:** - **Confidence Scoring:** Low-quality incidents get lower scores - **User Feedback:** Rating system improves future results - **Validation Pipeline:** Data quality checks during ingestion - **Human Review:** Flagging system for problematic content ### **Q: What if the AI gives wrong advice?** **A:** - **Source Citations:** Users can verify against original incidents - **Confidence Indicators:** Clear confidence scores - **Feedback Loop:** Users can rate and correct responses - **Disclaimer:** Clear that it's AI-generated, verify before implementing --- ## 📊 **METRICS TO MEMORIZE** - **Response Time:** < 2 seconds average - **Accuracy:** 85%+ for payment domain queries - **User Satisfaction:** 4.2/5 average rating - **Cache Hit Rate:** 70%+ for embeddings - **Uptime:** 99.9% availability - **Scalability:** 100+ concurrent users tested - **Data Size:** 54 incidents, 768-dimensional embeddings - **Search Methods:** 3 algorithms (semantic, BM25, TF-IDF) --- ## 🎬 **CLOSING STATEMENTS** ### **Technical Excellence** "We built SherlockAI with production-grade architecture from day one. Async FastAPI, structured logging, comprehensive monitoring, and enterprise security. This isn't just a hackathon project—it's a foundation for scaling to thousands of engineers." ### **Business Impact** "The real value is transforming how teams learn. Instead of losing knowledge when engineers leave, we capture it. Instead of solving the same problems repeatedly, we learn from the past. That's how you build truly intelligent organizations." ### **Future Vision** "This is just the beginning. Imagine auto-categorizing incidents, predicting problems before they happen, or integrating directly into your IDE. SherlockAI is the foundation for AI-augmented engineering teams." --- **Remember: Confidence, clarity, and concrete examples. You built something impressive—own it!** 🚀

All Documents

🎯 SherlockAI Interview Preparation Guide

✅ ENHANCED SPROUTCV IMPLEMENTATION STATUS REPORT

ML Feedback Loop Analysis

tensor_factorization

ReachInbox Email Aggregator - Development Plan

SMART-CAM (Smart AI Camera) — Full Documentation

Startup Ecosystem Platform — Implementation Plan

AI Research Assistant Usage Guide

Pulse — Life Cofounder | Build Log

⚒️ StudyForge: Implementation Plan

Tensorus Demo Script: Showcasing Agentic Tensor Management

An Introduction to Transformers

Development Summary

BigLake Iceberg Pipeline — Demo Scenarios

CricOracle

🚀 Vectro+ Visual Demo Guide

Vesper — Implementation Plan

Legal Brief Matcher: Technical Documentation

Contextify - Implementasyon Planı

Intelligent Document Query Platform — GitHub-ready Low-Level Design (LLD)

TODO — Dev Sentinel

Building an AI Travel Assistant with Micronaut and LangChain4j

WARP.md

Graph Matching with Topological Features