Spaces:
Paused
Paused
File size: 8,139 Bytes
75c718c 4b918ea 75c718c bd0e40f 75c718c bd0e40f 75c718c 4a27cdc 75c718c 4a27cdc 75c718c 4b918ea 4a27cdc 7605858 4b918ea 75c718c 4b918ea 75c718c | 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 | """
caller_memory.py — Persistent Caller Memory for Maya (Neon/Postgres Version)
"""
import asyncio
from dataclasses import dataclass, field
from typing import Optional, List
from datetime import datetime
import pytz
from src.database import _is_valid_uuid
IST = pytz.timezone("Asia/Kolkata")
@dataclass
class CallerProfile:
id: Optional[str] = None # UUID
phone_number: str = ""
name: Optional[str] = None
tenant_id: str = ""
call_count: int = 0
first_call_at: Optional[datetime] = None
last_call_at: Optional[datetime] = None
last_service: Optional[str] = None
is_vip: bool = False
notes: Optional[str] = None
recent_calls: list = field(default_factory=list)
@property
def is_return_caller(self) -> bool:
return self.call_count > 0
def get_context_for_llm(self) -> str:
# If we have a name, provide it regardless of call count (could be imported)
# If we have no name AND no history, return empty
if not self.name and not self.is_return_caller:
return ""
lines = []
if self.name:
lines.append(f"CALLER NAME: {self.name}")
lines.append(f" -> Address them as '{self.name}' naturally.")
if self.is_return_caller:
lines.append(f"CALL HISTORY: This is their call #{self.call_count + 1}")
if self.last_call_at:
# Simple "days ago" logic
delta = datetime.now(pytz.utc) - self.last_call_at.astimezone(pytz.utc)
days = delta.days
lines.append(f"LAST CALLED: {days} days ago" if days > 0 else "LAST CALLED: Today")
if self.last_service:
lines.append(f"LAST SERVICE: {self.last_service}")
if self.is_vip:
lines.append("VIP CALLER: Yes. Give them priority treatment.")
if self.recent_calls:
last = self.recent_calls[0]
if last.get("summary"):
lines.append(f"LAST CALL SUMMARY: {last['summary']}")
return "\n\n[CALLER MEMORY]\n" + "\n".join(lines) + "\n[END CALLER MEMORY]"
def get_personalized_greeting(self, language: str = "gujarati") -> Optional[str]:
if not self.name:
return None
# Standardizing on Devanagari for Hindi for better TTS
if language == "hindi":
if self.is_return_caller:
return f"नमस्ते {self.name} जी! स्माइलकेयर डेंटल क्लिनिक में आपका फिर से स्वागत है। आप कैसे हैं? मैं आपकी क्या मदद कर सकती हूँ?"
else:
return f"नमस्ते {self.name} जी! स्माइलकेयर डेंटल क्लिनिक में आपका स्वागत है। मैं माया हूँ, आपकी क्या मदद कर सकती हूँ?"
if language == "gujarati":
if self.is_return_caller:
return f"નમસ્તે {self.name} ભાઈ/બહેન! સ્માઇલકેર ડેન્ટલ ક્લિનિકમાં તમારું ફરી સ્વાગત છે. તમે કેમ છો? હું તમારી શું મદદ કરી શકું?"
else:
return f"નમસ્તે {self.name} ભાઈ/બહેન! સ્માઇલકેર ડેન્ટલ ક્લિનિકમાં તમારું સ્વાગત છે. હું માયા છું, હું તમારી શું મદદ કરી શકું?"
# English Fallback
if self.is_return_caller:
return f"Hello {self.name}! Welcome back to SmileCare Dental. How are you doing today? How can I help you?"
else:
return f"Hello {self.name}! Welcome to SmileCare Dental. I'm Maya, how can I assist you today?"
class CallerMemoryService:
def __init__(self, db_pool):
self.pool = db_pool
async def lookup_caller(self, phone_number: str, tenant_id: str) -> CallerProfile:
# Return blank profile if tenant_id is not a valid UUID (demo/test tenants)
if not _is_valid_uuid(tenant_id):
if tenant_id == "demo-tenant-123" and phone_number == "client:demo-caller":
# Mock return caller for testing Phase E4
return CallerProfile(
phone_number=phone_number, tenant_id=tenant_id,
name="Rudra", call_count=5, is_vip=True,
last_service="Dental Cleaning",
last_call_at=datetime.now(IST)
)
return CallerProfile(phone_number=phone_number, tenant_id=tenant_id)
async with self.pool.acquire() as conn:
# 1. Fetch Profile
row = await conn.fetchrow(
"SELECT * FROM caller_profiles WHERE tenant_id = $1 AND phone_number = $2",
tenant_id, phone_number
)
if not row:
return CallerProfile(phone_number=phone_number, tenant_id=tenant_id)
profile = CallerProfile(
id=str(row["id"]),
phone_number=row["phone_number"],
tenant_id=str(row["tenant_id"]),
name=row["name"],
call_count=row["call_count"],
first_call_at=row["first_call_at"],
last_call_at=row["last_call_at"],
last_service=row["last_service"],
is_vip=row["is_vip"],
notes=row["notes"]
)
# 2. Fetch Recent History
history_rows = await conn.fetch(
"SELECT called_at, summary FROM caller_history WHERE caller_id = $1 ORDER BY called_at DESC LIMIT 3",
row["id"]
)
profile.recent_calls = [dict(r) for r in history_rows]
return profile
async def update_caller_after_call(self, profile: CallerProfile, session_data: dict):
# Skip if tenant_id is not a valid UUID (demo/test tenants)
if not _is_valid_uuid(profile.tenant_id):
print(f"[Memory] Skipping DB update for non-UUID tenant: {profile.tenant_id}")
return
async with self.pool.acquire() as conn:
# 1. Upsert Profile
if profile.id:
await conn.execute(
"""
UPDATE caller_profiles
SET call_count = call_count + 1, last_call_at = NOW(), name = COALESCE($1, name), last_service = COALESCE($2, last_service)
WHERE id = $3
""",
session_data.get("name"), session_data.get("service"), profile.id
)
caller_id = profile.id
else:
caller_id = await conn.fetchval(
"""
INSERT INTO caller_profiles (tenant_id, phone_number, name, call_count, last_service)
VALUES ($1, $2, $3, 1, $4)
RETURNING id
""",
profile.tenant_id, profile.phone_number, session_data.get("name"), session_data.get("service")
)
# 2. Add History
await conn.execute(
"""
INSERT INTO caller_history (caller_id, tenant_id, call_sid, summary, emotion, appointment_booked)
VALUES ($1, $2, $3, $4, $5, $6)
""",
caller_id, profile.tenant_id, session_data.get("call_sid"), session_data.get("summary"),
session_data.get("emotion"), session_data.get("appointment_booked", False)
)
print(f"[Memory] Updated profile for {profile.phone_number}")
|