Spaces:
Sleeping
Sleeping
File size: 10,663 Bytes
3bb6958 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 |
# agent.py - AR Collection Agent implementation
from google import genai
from google.genai import types
from datetime import datetime
from typing import Dict, List
from config import GEMINI_API_KEY, SYSTEM_PROMPT, EMAIL_TEMPLATES
from database import query_database as execute_query, log_activity, store_mock_email
class ARCollectionAgent:
def __init__(self):
self.email_history = []
self.current_date = datetime.now()
self.chat_history = []
# Initialize Gemini client
self.client = genai.Client(api_key=GEMINI_API_KEY)
# Create tools for function calling
self.tools = self._create_tools()
# Set system instruction
self.system_instruction = SYSTEM_PROMPT.format(
current_date=self.current_date.strftime("%Y-%m-%d")
)
def _create_tools(self):
"""Create tool definitions for the new Gemini API."""
return [
self.query_database,
self.create_mock_email,
self.send_bulk_collection_emails,
self.get_current_datetime
]
def query_database(self, query: str) -> Dict:
"""Execute database query."""
return execute_query(query)
def create_mock_email(
self,
customer_email: str,
customer_name: str,
subject: str,
invoice_details: Dict,
tone: str = "friendly"
) -> Dict:
"""Generate mock collection email."""
# Get appropriate template
template = EMAIL_TEMPLATES.get(tone, EMAIL_TEMPLATES["friendly"])
# Format email body
body = template.format(
customer_name=customer_name,
invoice_id=invoice_details.get("invoice_id", "N/A"),
amount=float(invoice_details.get("amount", 0)),
days_overdue=invoice_details.get("days_overdue", 0),
due_date=invoice_details.get("due_date", "N/A")
)
# Create email record
email_record = {
"timestamp": datetime.now().isoformat(),
"recipient": customer_email,
"subject": subject,
"body": body,
"status": "MOCK - NOT SENT",
"tone": tone,
"invoice_id": invoice_details.get("invoice_id")
}
# Add to history
self.email_history.append(email_record)
# Store in dedicated mock_emails table
store_mock_email(email_record)
# Log activity in demo_activity_log
log_activity(
"mock_email_created",
customer_email.split('@')[0],
email_record
)
return email_record
def _select_email_tone(self, days_past_due: int, vip_flag: bool, num_late_12m: int, prior_promises_broken: int) -> str:
"""Select appropriate email tone based on customer risk profile."""
# Always gentle with VIP customers
if vip_flag:
if days_past_due > 45:
return "friendly" # Still gentle but noting urgency
return "friendly"
# Non-VIP customers - escalate based on behavior
if days_past_due > 60 or prior_promises_broken > 2:
return "final" # Final notice for seriously overdue or promise breakers
elif days_past_due > 30 or num_late_12m > 2:
return "firm" # Firm approach for repeat offenders
else:
return "friendly" # Standard friendly approach
def send_bulk_collection_emails(self, target_segments: str = "all") -> Dict:
"""Send collection emails to overdue customers based on target segments.
Args:
target_segments: "all", "vip", "high_risk", "nordic", or country names like "sweden"
"""
from database import query_database as db_query
try:
# Query for overdue customers
overdue_query_result = db_query("overdue invoices")
if not overdue_query_result.get("success", False):
return {
"success": False,
"error": "Failed to query overdue customers",
"emails_sent": 0
}
overdue_data = overdue_query_result.get("data", [])
if not overdue_data:
return {
"success": True,
"message": "No overdue customers found",
"emails_sent": 0
}
# Filter based on target segments
filtered_customers = []
target_lower = target_segments.lower()
for customer in overdue_data:
include = False
if target_lower == "all":
include = True
elif target_lower == "vip" and customer.get("vip_flag", False):
include = True
elif target_lower == "high_risk" and (customer.get("num_late_12m", 0) > 2 or customer.get("days_past_due", 0) >= 45):
include = True
elif target_lower in ["nordic", "sweden", "norway", "denmark"]:
customer_country = customer.get("country", "").lower()
if target_lower == "nordic" or target_lower in customer_country:
include = True
if include:
filtered_customers.append(customer)
# Generate emails for filtered customers
emails_generated = []
for customer in filtered_customers:
# Select appropriate tone
tone = self._select_email_tone(
customer.get("days_past_due", 0),
customer.get("vip_flag", False),
customer.get("num_late_12m", 0),
customer.get("prior_promises_broken", 0)
)
# Create subject based on tone and VIP status
if customer.get("vip_flag", False):
subject = f"Gentle Reminder: Invoice {customer.get('invoice_id')} - Valued Customer"
elif tone == "final":
subject = f"FINAL NOTICE: Invoice {customer.get('invoice_id')} - Immediate Action Required"
elif tone == "firm":
subject = f"Second Notice: Invoice {customer.get('invoice_id')} - Payment Due"
else:
subject = f"Payment Reminder: Invoice {customer.get('invoice_id')}"
# Prepare invoice details
invoice_details = {
"invoice_id": customer.get("invoice_id"),
"amount": customer.get("amount"),
"days_overdue": customer.get("days_past_due", 0),
"due_date": customer.get("due_date")
}
# Generate the email
email_record = self.create_mock_email(
customer_email=customer.get("customer_email", ""),
customer_name=customer.get("company_name", ""),
subject=subject,
invoice_details=invoice_details,
tone=tone
)
emails_generated.append(email_record)
return {
"success": True,
"message": f"Successfully generated {len(emails_generated)} collection emails for {target_segments} customers",
"emails_sent": len(emails_generated),
"target_segments": target_segments,
"email_details": [
{
"recipient": email["recipient"],
"subject": email["subject"],
"tone": email["tone"],
"invoice_id": email["invoice_id"]
}
for email in emails_generated[:5] # Return first 5 for summary
]
}
except Exception as e:
return {
"success": False,
"error": f"Error in bulk email generation: {str(e)}",
"emails_sent": 0
}
def get_current_datetime(self) -> Dict:
"""Return current datetime for calculations."""
return {
"current_date": datetime.now().strftime("%Y-%m-%d"),
"current_time": datetime.now().strftime("%H:%M:%S"),
"timestamp": datetime.now().isoformat()
}
async def process_message(self, message: str) -> str:
"""Process user message through Gemini."""
try:
# Add user message to chat history
self.chat_history.append({"role": "user", "content": message})
# Prepare contents for the API call
contents = []
for msg in self.chat_history:
if msg["role"] == "user":
contents.append(types.UserContent(parts=[types.Part.from_text(text=msg["content"])]))
elif msg["role"] == "assistant":
contents.append(types.ModelContent(parts=[types.Part.from_text(text=msg["content"])]))
# Generate content with tools
response = self.client.models.generate_content(
model='gemini-2.5-flash',
contents=contents,
config=types.GenerateContentConfig(
system_instruction=self.system_instruction,
tools=self.tools,
temperature=0.1,
max_output_tokens=4000
)
)
# Get the response text
response_text = response.text if response.text else "I apologize, but I couldn't generate a response."
# Add assistant response to chat history
self.chat_history.append({"role": "assistant", "content": response_text})
return response_text
except Exception as e:
error_msg = f"Error processing request: {str(e)}"
print(error_msg)
return error_msg
def get_email_history(self) -> List[Dict]:
"""Get email history for display."""
return self.email_history
def clear_history(self):
"""Clear chat and email history."""
self.email_history = []
self.chat_history = [] |