import os import streamlit as st import pandas as pd from datetime import datetime, timedelta import time from huggingface_hub import InferenceClient import dotenv # Load environment variables dotenv.load_dotenv() # Set page configuration st.set_page_config( page_title="Debt Collection AI", page_icon="💰", layout="wide" ) # Custom CSS for styling st.markdown(""" """, unsafe_allow_html=True) # Configuration MODEL_NAME = "meta-llama/Meta-Llama-3-8B-Instruct" HF_API_KEY = os.getenv("HF_API_KEY") COMPLIANCE_RULES = {"prohibited_words": ["sue", "arrest", "threaten"], "max_installment_ratio": 0.3} # Initialize Hugging Face client @st.cache_resource def get_client(): return InferenceClient( model=MODEL_NAME, token=HF_API_KEY ) # Sample data (Mock Database) DEBTORS = { 1000: { "debtor_id": 1000, "name": "Rami", "age": 35, "preferred_language": "English", "preferred_channel": "Email", "income": 60000, "outstanding_balance": 1500, "installments": 15, "last_payment_date": "2023-11-01", "dispute_history": ["Billing error on 2023-10-01"] }, 1001: { "debtor_id": 1001, "name": "Jane Doe", "age": 35, "preferred_language": "English", "preferred_channel": "Email", "income": 45000, "outstanding_balance": 2500, "installments": 12, "last_payment_date": "2023-10-15", "dispute_history": ["Billing error on 2023-09-01"] } } # Business Logic Functions def optimize_channel(debtor_data: dict) -> dict: age = debtor_data["age"] return { "channel": "SMS" if age < 35 else "Email", "best_time": "18:00-20:00" if age < 35 else "09:00-12:00" } def generate_message(debtor_data: dict) -> str: try: client = get_client() prompt = f"""<|begin_of_text|> <|start_header_id|>system<|end_header_id|> Generate a friendly debt collection message in {debtor_data['preferred_language']}. Avoid these words: {COMPLIANCE_RULES['prohibited_words']}.<|eot_id|> <|start_header_id|>user<|end_header_id|> Debtor: {debtor_data['name']} Last Payment: {debtor_data['last_payment_date']} Balance: ${debtor_data['outstanding_balance']}<|eot_id|> <|start_header_id|>assistant<|end_header_id|> """ response = client.text_generation( prompt, max_new_tokens=200, temperature=0.7 ) print(response) return response.strip() except Exception as e: st.error(f"Error generating message: {str(e)}") return "Hello, this is a friendly reminder about your outstanding balance. Please contact us to discuss payment options." def calculate_payment_plan(debtor_data: dict) -> dict: max_payment = debtor_data["income"] * COMPLIANCE_RULES["max_installment_ratio"] installments = debtor_data["installments"] return { "monthly_payment": round(debtor_data["outstanding_balance"] / installments, 2), "duration_months": installments, "due_date": (datetime.now() + timedelta(days=15)).strftime("%Y-%m-%d") } def generate_negotiation_strategy(debtor_data: dict) -> str: try: client = get_client() prompt = f"""<|begin_of_text|> <|start_header_id|>system<|end_header_id|> Generate a negotiation strategy for a debt collection agent. Include personalized payment options and dispute handling recommendations.<|eot_id|> <|start_header_id|>user<|end_header_id|> Debtor: {debtor_data['name']} Income: ${debtor_data['income']} Balance: ${debtor_data['outstanding_balance']} Installments: {debtor_data['installments']} Dispute History: {debtor_data['dispute_history'] if debtor_data['dispute_history'] else "None"}<|eot_id|> <|start_header_id|>assistant<|end_header_id|> """ response = client.text_generation( prompt, max_new_tokens=300, temperature=0.7 ) print(response) return response.strip() except Exception as e: st.error(f"Error generating negotiation strategy: {str(e)}") return "Recommend standard payment plan with flexible options. Address any disputes promptly and professionally." def get_message_response(debtor_id): debtor = DEBTORS.get(debtor_id) if not debtor: return {"error": "Debtor not found"} channel_info = optimize_channel(debtor) return { "debtor_id": debtor_id, "message": generate_message(debtor), "channel": channel_info["channel"], "send_time": f"{datetime.now().date()}T{channel_info['best_time']}" } def get_payment_plan_response(debtor_id): debtor = DEBTORS.get(debtor_id) if not debtor: return {"error": "Debtor not found"} return { "debtor_id": debtor_id, "payment_plan": calculate_payment_plan(debtor), "negotiation_strategy": generate_negotiation_strategy(debtor) } # Initialize session state if 'selected_debtor_id' not in st.session_state: st.session_state.selected_debtor_id = None if 'message_response' not in st.session_state: st.session_state.message_response = None if 'payment_plan' not in st.session_state: st.session_state.payment_plan = None # Calculate total outstanding balance total_outstanding = sum(debtor["outstanding_balance"] for debtor in DEBTORS.values()) # Check for API key if not HF_API_KEY: st.error("Hugging Face API key not found. Please set the HF_API_KEY environment variable.") st.stop() # Sidebar - Debtor selection with st.sidebar: st.markdown("
© 2025 Debt Collection AI System
", unsafe_allow_html=True)