Financial Services & Banking LLM Implementation

Implementation Example

# Banking Customer Service LLM Implementation
import asyncio
from typing import Dict, List, Optional, Tuple, Any
from dataclasses import dataclass
from datetime import datetime, timedelta
import re
from enum import Enum

class ConversationIntent(Enum):
    """Banking conversation intents"""
    ACCOUNT_BALANCE = "account_balance"
    TRANSACTION_HISTORY = "transaction_history"
    TRANSFER_MONEY = "transfer_money"
    CARD_SERVICES = "card_services"
    LOAN_INQUIRY = "loan_inquiry"
    FRAUD_REPORT = "fraud_report"
    GENERAL_INQUIRY = "general_inquiry"

@dataclass
class BankingContext:
    """Customer banking context"""
    customer_id: str
    authenticated: bool
    account_numbers: List[str]
    session_id: str
    risk_score: float
    conversation_history: List[Dict]

class BankingLLMChatbot:
    """SOX-compliant banking chatbot with LLM integration"""
    
    def __init__(self, llm_provider: str = "gpt-4"):
        self.llm_provider = llm_provider
        self.compliance_logger = ComplianceAuditLogger()
        self.fraud_detector = FraudDetectionModule()
        self.knowledge_base = BankingKnowledgeBase()
        self.prompt_templates = self._load_prompt_templates()
        
    async def handle_customer_query(
        self,
        query: str,
        context: BankingContext,
        language: str = "en"
    ) -> Dict[str, Any]:
        """Handle customer banking query with compliance"""
        
        # Step 1: Classify intent and extract entities
        intent, entities = await self._classify_intent(query, context)
        
        # Step 2: Check authentication requirements
        if not self._check_authentication(intent, context):
            return {
                "response": "Please authenticate first to access this service.",
                "action_required": "authentication",
                "intent": intent.value
            }
        
        # Step 3: Risk assessment
        risk_assessment = await self.fraud_detector.assess_request(
            query, context, intent
        )
        
        if risk_assessment["block"]:
            await self._handle_high_risk_request(risk_assessment, context)
            return {
                "response": "For security reasons, please contact our support team.",
                "action_required": "manual_review",
                "risk_level": "high"
            }
        
        # Step 4: Generate appropriate response
        response = await self._generate_llm_response(
            query, intent, entities, context
        )
        
        # Step 5: Compliance logging
        await self.compliance_logger.log_interaction(
            customer_id=context.customer_id,
            intent=intent,
            query=self._sanitize_for_logging(query),
            response=self._sanitize_for_logging(response["text"]),
            risk_score=risk_assessment["score"]
        )
        
        return response
    
    async def _classify_intent(
        self,
        query: str,
        context: BankingContext
    ) -> Tuple[ConversationIntent, Dict]:
        """Classify query intent using LLM"""
        
        classification_prompt = f"""
Classify the following banking query into one of these intents:
- ACCOUNT_BALANCE: Checking account balance or status
- TRANSACTION_HISTORY: Viewing transactions or statements
- TRANSFER_MONEY: Sending money or making payments
- CARD_SERVICES: Card activation, blocking, or issues
- LOAN_INQUIRY: Questions about loans or credit
- FRAUD_REPORT: Reporting suspicious activity
- GENERAL_INQUIRY: Other banking questions

Also extract any relevant entities (account numbers, amounts, dates).

Query: "{query}"
Recent context: {self._get_recent_context(context)}

Respond in JSON format:
{{
    "intent": "INTENT_NAME",
    "entities": {{
        "accounts": [],
        "amounts": [],
        "dates": [],
        "merchant_names": []
    }},
    "confidence": 0.95
}}
"""
        
        # Call LLM for classification
        llm_response = await self._call_llm(
            classification_prompt,
            temperature=0.1  # Low temperature for consistency
        )
        
        # Parse response
        import json
        parsed = json.loads(llm_response)
        
        intent = ConversationIntent[parsed["intent"]]
        entities = parsed["entities"]
        
        return intent, entities
    
    async def _generate_llm_response(
        self,
        query: str,
        intent: ConversationIntent,
        entities: Dict,
        context: BankingContext
    ) -> Dict[str, Any]:
        """Generate appropriate response based on intent"""
        
        # Get appropriate prompt template
        template = self.prompt_templates[intent]
        
        # Retrieve relevant data
        if intent == ConversationIntent.ACCOUNT_BALANCE:
            account_data = await self._get_account_balances(
                context.account_numbers
            )
            prompt = template.format(
                query=query,
                accounts=account_data,
                customer_name=await self._get_customer_name(context.customer_id)
            )
            
        elif intent == ConversationIntent.TRANSACTION_HISTORY:
            transactions = await self._get_recent_transactions(
                context.account_numbers,
                entities.get("dates", {})
            )
            prompt = template.format(
                query=query,
                transactions=self._format_transactions(transactions),
                period=entities.get("dates", {}).get("period", "last 30 days")
            )
            
        elif intent == ConversationIntent.TRANSFER_MONEY:
            # For transfers, provide guidance but don't execute
            prompt = template.format(
                query=query,
                transfer_limits=await self._get_transfer_limits(context.customer_id),
                authentication_status=context.authenticated
            )
            
        elif intent == ConversationIntent.FRAUD_REPORT:
            # Handle fraud reports with high priority
            prompt = template.format(
                query=query,
                fraud_hotline="1-800-XXX-XXXX",
                case_number=await self._create_fraud_case(context, query)
            )
            
        else:
            # General inquiry
            knowledge = await self.knowledge_base.search(query)
            prompt = template.format(
                query=query,
                knowledge=knowledge,
                contact_info=self._get_contact_info()
            )
        
        # Add compliance and safety instructions
        safety_prompt = """
IMPORTANT COMPLIANCE RULES:
1. Never disclose full account numbers or sensitive data
2. Always verify authentication before providing account-specific information
3. For transactions, guide the customer but don't execute directly
4. Include relevant disclosures and disclaimers
5. Be helpful but maintain security
6. If unsure, direct to human support
"""
        
        full_prompt = safety_prompt + "\n\n" + prompt
        
        # Generate response
        llm_response = await self._call_llm(
            full_prompt,
            temperature=0.3,
            max_tokens=500
        )
        
        # Post-process for compliance
        sanitized_response = self._sanitize_response(llm_response)
        
        return {
            "text": sanitized_response,
            "intent": intent.value,
            "requires_action": self._determine_required_actions(intent),
            "suggested_actions": self._get_suggested_actions(intent, entities),
            "compliance_notes": self._get_compliance_notes(intent)
        }
    
    def _load_prompt_templates(self) -> Dict[ConversationIntent, str]:
        """Load banking-specific prompt templates"""
        return {
            ConversationIntent.ACCOUNT_BALANCE: """
Customer Query: {query}
Customer Name: {customer_name}

Available Account Information:
{accounts}

Provide a friendly response about their account balance(s). 
- Show balances in a clear format
- Mention any pending transactions if relevant
- Don't show full account numbers (only last 4 digits)
- Include the current date/time of the balance
""",
            
            ConversationIntent.TRANSACTION_HISTORY: """
Customer Query: {query}
Period: {period}

Recent Transactions:
{transactions}

Provide a helpful summary of their transaction history:
- Highlight any unusual patterns if asked
- Group by category if helpful
- Mention the total spending/income for the period
- Flag any pending transactions
""",
            
            ConversationIntent.TRANSFER_MONEY: """
Customer Query: {query}
Authentication Status: {authentication_status}
Daily Transfer Limits: {transfer_limits}

Guide the customer on how to make a transfer:
- Explain the steps in our mobile app or online banking
- Mention security requirements (2FA, etc.)
- Note applicable limits and fees
- DO NOT execute the transfer directly
- Remind about fraud prevention tips
""",
            
            ConversationIntent.FRAUD_REPORT: """
URGENT - Potential Fraud Report
Customer Query: {query}
Fraud Hotline: {fraud_hotline}
Case Number: {case_number}

Respond with empathy and urgency:
- Acknowledge their concern
- Provide the case number for reference
- Give clear next steps
- Mention we're reviewing their account
- Provide the fraud hotline for immediate assistance
- Assure them their funds are protected
""",
            
            ConversationIntent.GENERAL_INQUIRY: """
Customer Query: {query}

Relevant Information:
{knowledge}

Contact Information:
{contact_info}

Provide a helpful response:
- Answer their question using the knowledge base
- If you can't fully answer, provide relevant contact info
- Be professional and friendly
- Include any relevant disclaimers
"""
        }
    
    async def _call_llm(
        self,
        prompt: str,
        temperature: float = 0.3,
        max_tokens: int = 500
    ) -> str:
        """Call LLM with retry logic and error handling"""
        # Implementation would call actual LLM API
        # This is a placeholder
        
        # Add conversation context
        full_prompt = f"""
You are a helpful banking assistant for SecureBank. 
You must follow all compliance rules and prioritize customer security.

{prompt}
"""
        
        try:
            # Actual LLM API call would go here
            response = "LLM response placeholder"
            return response
        except Exception as e:
            # Log error and return safe fallback
            await self.compliance_logger.log_error(
                "LLM_CALL_FAILED",
                str(e)
            )
            return "I apologize, but I'm having trouble processing your request. Please try again or contact our support team."
    
    def _sanitize_response(self, response: str) -> str:
        """Sanitize LLM response for compliance"""
        # Remove any accidentally included sensitive data
        
        # Mask account numbers (keep only last 4 digits)
        response = re.sub(
            r'd{8,16}',
            lambda m: '*' * (len(m.group()) - 4) + m.group()[-4:],
            response
        )
        
        # Remove SSNs if accidentally included
        response = re.sub(
            r'd{3}-d{2}-d{4}',
            '[REDACTED]',
            response
        )
        
        # Add required disclaimers
        if any(term in response.lower() for term in ['invest', 'loan', 'credit']):
            response += "\n\nThis information is for general purposes only. Please consult with a financial advisor for personalized advice."
        
        return response
    
    def _check_authentication(
        self,
        intent: ConversationIntent,
        context: BankingContext
    ) -> bool:
        """Check if intent requires authentication"""
        public_intents = [
            ConversationIntent.GENERAL_INQUIRY,
            ConversationIntent.LOAN_INQUIRY  # General info only
        ]
        
        if intent in public_intents:
            return True
            
        return context.authenticated
    
    async def _get_account_balances(
        self,
        account_numbers: List[str]
    ) -> str:
        """Get formatted account balances"""
        # This would connect to core banking system
        # Placeholder implementation
        balances = []
        for acc in account_numbers:
            balances.append(f"- Checking ****{acc[-4:]}: $5,432.10")
            balances.append(f"- Savings ****{acc[-4:]}: $25,678.90")
        
        return "\n".join(balances)
    
    def _format_transactions(
        self,
        transactions: List[Dict]
    ) -> str:
        """Format transactions for display"""
        # Format transaction list
        formatted = []
        for tx in transactions[:10]:  # Limit to recent 10
            formatted.append(
                f"- {tx['date']}: {tx['description']} "
                f"{'+'if tx['type']=='credit' else '-'}${tx['amount']}"
            )
        
        return "\n".join(formatted)

class ComplianceAuditLogger:
    """SOX-compliant audit logging for banking LLMs"""
    
    async def log_interaction(
        self,
        customer_id: str,
        intent: ConversationIntent,
        query: str,
        response: str,
        risk_score: float
    ):
        """Log all interactions for compliance"""
        log_entry = {
            "timestamp": datetime.utcnow().isoformat(),
            "customer_id_hash": self._hash_customer_id(customer_id),
            "session_id": self._generate_session_id(),
            "intent": intent.value,
            "query_length": len(query),
            "response_length": len(response),
            "risk_score": risk_score,
            "llm_provider": "gpt-4",
            "compliance_version": "2.1",
            "retention_days": 2555  # 7 years for SOX
        }
        
        # Write to immutable audit log
        await self._write_to_audit_log(log_entry)
    
    def _hash_customer_id(self, customer_id: str) -> str:
        """Hash customer ID for privacy"""
        import hashlib
        return hashlib.sha256(
            f"{customer_id}:{datetime.utcnow().date()}".encode()
        ).hexdigest()

class FraudDetectionModule:
    """Real-time fraud detection for LLM interactions"""
    
    async def assess_request(
        self,
        query: str,
        context: BankingContext,
        intent: ConversationIntent
    ) -> Dict[str, Any]:
        """Assess fraud risk of request"""
        risk_score = 0.0
        risk_factors = []
        
        # Check for high-risk patterns
        high_risk_patterns = [
            (r"transfer.*all.*money", 0.8, "Full balance transfer request"),
            (r"urgent|immediately|asap", 0.3, "Urgency indicator"),
            (r"gift.*card|bitcoin|crypto", 0.7, "Suspicious payment method"),
            (r"irs|tax.*agent|police", 0.6, "Impersonation risk"),
        ]
        
        query_lower = query.lower()
        for pattern, score, description in high_risk_patterns:
            if re.search(pattern, query_lower):
                risk_score += score
                risk_factors.append(description)
        
        # Check velocity (rapid requests)
        if self._check_velocity_risk(context):
            risk_score += 0.4
            risk_factors.append("High request velocity")
        
        # Check unusual access patterns
        if self._check_access_pattern_risk(context):
            risk_score += 0.3
            risk_factors.append("Unusual access pattern")
        
        # Normalize risk score
        risk_score = min(risk_score, 1.0)
        
        return {
            "score": risk_score,
            "block": risk_score > 0.8,
            "factors": risk_factors,
            "recommendation": self._get_risk_recommendation(risk_score)
        }
    
    def _check_velocity_risk(self, context: BankingContext) -> bool:
        """Check for rapid-fire requests"""
        # Check if too many requests in short time
        recent_requests = [
            req for req in context.conversation_history
            if (datetime.now() - req["timestamp"]).seconds < 60
        ]
        return len(recent_requests) > 10
    
    def _get_risk_recommendation(self, score: float) -> str:
        """Get recommendation based on risk score"""
        if score > 0.8:
            return "Block and escalate to fraud team"
        elif score > 0.5:
            return "Additional authentication required"
        elif score > 0.3:
            return "Monitor closely"
        else:
            return "Standard processing"

# Document Analysis Implementation
class DocumentAnalysisLLM:
    """Contract and document analysis for banking"""
    
    def __init__(self, llm_provider: str = "gpt-4"):
        self.llm_provider = llm_provider
        self.compliance_checker = ComplianceChecker()
        
    async def analyze_contract(
        self,
        document_text: str,
        document_type: str = "loan_agreement",
        analysis_requirements: List[str] = None
    ) -> Dict[str, Any]:
        """Analyze banking contract with LLM"""
        
        if not analysis_requirements:
            analysis_requirements = [
                "key_terms",
                "obligations",
                "risks",
                "unusual_clauses",
                "compliance_issues"
            ]
        
        # Create specialized prompts for each requirement
        analysis_results = {}
        
        for requirement in analysis_requirements:
            prompt = self._create_analysis_prompt(
                document_text,
                document_type,
                requirement
            )
            
            result = await self._call_llm(prompt, temperature=0.1)
            analysis_results[requirement] = self._parse_analysis_result(
                result,
                requirement
            )
        
        # Compliance check
        compliance_results = await self.compliance_checker.check_document(
            document_text,
            document_type
        )
        
        return {
            "document_type": document_type,
            "analysis": analysis_results,
            "compliance": compliance_results,
            "summary": self._generate_executive_summary(analysis_results),
            "risk_rating": self._calculate_risk_rating(analysis_results),
            "processing_time": datetime.utcnow().isoformat()
        }
    
    def _create_analysis_prompt(
        self,
        document_text: str,
        document_type: str,
        requirement: str
    ) -> str:
        """Create specific prompt for each analysis requirement"""
        
        prompts = {
            "key_terms": f"""
Analyze this {document_type} and extract all key terms:
- Interest rates and fees
- Payment schedules
- Important dates and deadlines
- Parties involved
- Principal amounts

Document:
{document_text[:3000]}...

Format as structured JSON with clear categorization.
""",
            
            "obligations": f"""
List all obligations and requirements in this {document_type}:
- Borrower obligations
- Lender obligations  
- Conditions precedent
- Ongoing covenants
- Reporting requirements

Document:
{document_text[:3000]}...

Categorize by party and timing.
""",
            
            "risks": f"""
Identify all risks and potential issues in this {document_type}:
- Financial risks
- Legal risks
- Operational risks
- Market risks
- Counterparty risks

Document:
{document_text[:3000]}...

Rate each risk as High/Medium/Low with justification.
""",
            
            "unusual_clauses": f"""
Flag any unusual or non-standard clauses in this {document_type}:
- Clauses that deviate from market standard
- Potentially problematic terms
- Missing standard protections
- Ambiguous language

Document:
{document_text[:3000]}...

Explain why each clause is unusual and potential implications.
""",
            
            "compliance_issues": f"""
Review for regulatory compliance issues in this {document_type}:
- TILA (Truth in Lending) compliance
- Fair lending requirements
- Usury law compliance
- Required disclosures
- Prohibited terms

Document:
{document_text[:3000]}...

Cite specific regulations where relevant.
"""
        }
        
        return prompts.get(requirement, f"Analyze {requirement} in the document")

# Example usage
async def main():
    # Initialize banking chatbot
    chatbot = BankingLLMChatbot()
    
    # Example customer context
    context = BankingContext(
        customer_id="CUST-12345",
        authenticated=True,
        account_numbers=["1234567890", "0987654321"],
        session_id="SESSION-001",
        risk_score=0.2,
        conversation_history=[]
    )
    
    # Example queries
    queries = [
        "What's my checking account balance?",
        "Show me transactions from last week",
        "I want to report a fraudulent charge on my card",
        "How do I apply for a mortgage?"
    ]
    
    for query in queries:
        print(f"\nCustomer: {query}")
        response = await chatbot.handle_customer_query(query, context)
        print(f"Bot: {response['text']}")
        print(f"Intent: {response.get('intent', 'unknown')}")

if __name__ == "__main__":
    asyncio.run(main())