Project 5: Conversational AI for Real-time Fraud Analytics

Intent Classification and Entity Extraction

The system uses advanced NLU components to understand analyst queries:

# RASA-like NLU components for intent classification and entity extraction
class IntentClassifier:
    def __init__(self, model_path: str = None):
        # In a real implementation, this would load a trained model
        self.intents = {
            "greet": ["hello", "hi", "hey", "good morning", "good afternoon", "good evening"],
            "goodbye": ["bye", "goodbye", "see you", "talk to you later", "have a nice day"],
            "search_transaction": ["find transaction", "search transaction", "look up transaction", 
                                  "show me transaction", "transaction details"],
            "transaction_summary": ["summarize transactions", "transaction summary", "overview of transactions",
                                   "recent transactions", "show me recent activity"],
            "fraud_risk": ["fraud risk", "risk assessment", "risk score", "fraud probability", 
                          "likelihood of fraud", "is this fraudulent"],
            "similar_patterns": ["similar patterns", "similar cases", "pattern matching", 
                               "similar fraud", "matching cases", "fraud patterns"],
            "explain_decision": ["explain decision", "why flagged", "reason for alert", 
                               "explain alert", "why is this suspicious", "explain risk score"],
            "recommend_action": ["what should I do", "recommended action", "next steps", 
                               "how to proceed", "action plan", "investigation steps"]
        }
        
    def predict(self, text: str) -> Tuple[str, float]:
        """Predict the intent of a message"""
        best_intent = "unknown"
        best_score = 0.0
        
        # Simple keyword matching for demonstration
        text_lower = text.lower()
        for intent, keywords in self.intents.items():
            for keyword in keywords:
                if keyword in text_lower:
                    # Calculate a simple score based on keyword match
                    match_length = len(keyword)
                    text_length = len(text_lower)
                    score = match_length / text_length * 1.5  # Boost the score a bit
                    
                    if score > best_score:
                        best_score = min(score, 0.95)  # Cap at 0.95
                        best_intent = intent
        
        # If no good match, default to a low confidence unknown intent
        if best_score < 0.3:
            best_intent = "unknown"
            best_score = 0.1
            
        return best_intent, best_score
                    

The entity extraction component identifies key information in analyst queries:

class EntityExtractor:
    def __init__(self, model_path: str = None):
        # In a real implementation, this would load a trained model
        self.entity_patterns = {
            "transaction_id": [r"tx[_-]?\d+", r"transaction[_-]?\d+", r"#\d+"],
            "amount": [r"\$\d+\.?\d*", r"\d+\.?\d*\s?dollars", r"\d+\.?\d*\s?usd"],
            "date": [r"\d{1,2}/\d{1,2}/\d{2,4}", r"\d{1,2}-\d{1,2}-\d{2,4}", 
                    r"yesterday", r"today", r"last week", r"this month"],
            "merchant": [r"at\s+([A-Za-z0-9\s]+)", r"from\s+([A-Za-z0-9\s]+)", r"to\s+([A-Za-z0-9\s]+)"],
            "card_number": [r"\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}", r"card ending in \d{4}"]
        }
                    

RAG System for Fraud Knowledge

The system includes a Retrieval Augmented Generation component for accessing fraud knowledge:

# RAG system for fraud knowledge retrieval
class FraudKnowledgeRAG:
    def __init__(self):
        self.knowledge_base = self._initialize_knowledge_base()
        
    def _initialize_knowledge_base(self) -> List[Dict[str, Any]]:
        """Initialize the knowledge base with fraud patterns and investigation procedures"""
        return [
            {
                "id": "pattern_001",
                "type": "fraud_pattern",
                "title": "Card Testing",
                "description": "Multiple small transactions in quick succession to test if a stolen card works",
                "indicators": [
                    "Multiple transactions under $10",
                    "Transactions at different merchants within minutes",
                    "Digital goods or services that don't require shipping",
                    "First-time merchants for the cardholder"
                ],
                "risk_level": "high",
                "investigation_steps": [
                    "Check for other small transactions in the last 24 hours",
                    "Look for common IP addresses across transactions",
                    "Verify if customer has reported card lost or stolen",
                    "Check for previous history of this pattern on the account"
                ],
                "embedding": np.random.rand(128)  # In a real system, this would be a pre-computed embedding
            },
            # Additional patterns...
        ]
        
    def search(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]:
        """Search the knowledge base for relevant information"""
        # In a real system, this would encode the query into a vector using a language model
        # Here we simulate by creating a random vector with some bias based on keywords
        
        query_vector = np.random.rand(128)
        
        # Calculate similarity scores
        results = []
        for item in self.knowledge_base:
            item_vector = item["embedding"]
            similarity = np.dot(query_vector, item_vector) / (np.linalg.norm(item_vector) * np.linalg.norm(query_vector))
            results.append((item, float(similarity)))
        
        # Sort by similarity (highest first) and return top_k
        results.sort(key=lambda x: x[1], reverse=True)
        return [{"item": item, "similarity": similarity} for item, similarity in results[:top_k]]
                    

Conversational AI Assistant

The main assistant class integrates all components to provide a seamless conversational experience:

# Conversational AI for fraud investigation
class FraudInvestigationAssistant:
    def __init__(self):
        self.intent_classifier = IntentClassifier()
        self.entity_extractor = EntityExtractor()
        self.knowledge_rag = FraudKnowledgeRAG()
        self.transaction_db = TransactionDatabase()
        self.risk_model = FraudRiskModel()
        self.conversation_history = []
        
    def process_message(self, message: str, session_id: str = "default") -> Dict[str, Any]:
        """Process a message from a fraud analyst"""
        # Add message to conversation history
        self.conversation_history.append({"role": "user", "content": message})
        
        # Analyze message
        intent, confidence = self.intent_classifier.predict(message)
        entities = self.entity_extractor.extract(message)
        
        # Prepare response
        response = {
            "text": "",
            "intent": intent,
            "confidence": confidence,
            "entities": entities,
            "data": {}
        }
        
        # Handle different intents
        if intent == "search_transaction":
            # Extract transaction identifiers from entities
            transaction_id = None
            search_params = {}
            
            for entity in entities:
                if entity["entity"] == "transaction_id":
                    transaction_id = entity["value"]
                # Additional entity processing...
            
            # Search for transactions
            if transaction_id:
                transaction = self.transaction_db.get_transaction(transaction_id)
                if transaction:
                    response["text"] = f"I found transaction {transaction_id}:"
                    response["data"] = {"transaction": transaction}
                else:
                    response["text"] = f"I couldn't find transaction {transaction_id}."
            else:
                transactions = self.transaction_db.search(search_params)
                if transactions:
                    response["text"] = f"I found {len(transactions)} transactions matching your criteria:"
                    response["data"] = {"transactions": transactions}
                else:
                    response["text"] = "I couldn't find any transactions matching your criteria."
        
        # Additional intent handlers...
                
        # Add response to conversation history
        self.conversation_history.append({"role": "assistant", "content": response["text"]})
        
        return response
                    

NLU Pipeline

The Natural Language Understanding pipeline consists of several components:

1. Intent Classification:
   • BERT-based model fine-tuned on fraud investigation conversations
   • Support for 8 core intents with extensible architecture
   • Confidence scoring for ambiguous queries
   • Fallback handling for unknown intents
2. Entity Extraction:
   • Named Entity Recognition for transaction-specific entities
   • Regular expression patterns for structured entities like transaction IDs
   • Date and time normalization
   • Numeric value extraction with unit handling
3. Context Management:
   • Session-based conversation tracking
   • Entity memory across conversation turns
   • Intent-based context switching
   • Slot filling for incomplete queries