diff --git a/AGENTS.md b/AGENTS.md index 3434537..a4792cf 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -1,3 +1,4 @@ Use `uv` for project management. -Use `uv run ruff check` for linting, and `uv run ty check` for type checking +Use `uv run ruff check` for linting +Use `uv run ty check` for type checking Use `uv run pytest` for testing. diff --git a/src/va_agent/agent.py b/src/va_agent/agent.py index 72085d6..11b458b 100644 --- a/src/va_agent/agent.py +++ b/src/va_agent/agent.py @@ -1,7 +1,5 @@ """ADK agent with vector search RAG tool.""" -from functools import partial - from google import genai from google.adk.agents.llm_agent import Agent from google.adk.runners import Runner @@ -12,10 +10,9 @@ from google.genai.types import Content, Part from va_agent.auth import auth_headers_provider from va_agent.config import settings -from va_agent.dynamic_instruction import provide_dynamic_instruction +from va_agent.governance import GovernancePlugin from va_agent.notifications import NotificationService from va_agent.session import FirestoreSessionService -from va_agent.governance import GovernancePlugin # MCP Toolset for RAG knowledge search toolset = McpToolset( diff --git a/src/va_agent/config.py b/src/va_agent/config.py index f3502b5..ae33d2d 100644 --- a/src/va_agent/config.py +++ b/src/va_agent/config.py @@ -39,7 +39,7 @@ class AgentSettings(BaseSettings): model_config = SettingsConfigDict( yaml_file=CONFIG_FILE_PATH, extra="ignore", # Ignore extra fields from config.yaml - env_file=".env" + env_file=".env", ) @classmethod diff --git a/src/va_agent/dynamic_instruction.py b/src/va_agent/dynamic_instruction.py index 86f190d..cc64fb9 100644 --- a/src/va_agent/dynamic_instruction.py +++ b/src/va_agent/dynamic_instruction.py @@ -34,17 +34,19 @@ async def provide_dynamic_instruction( """ # Only check notifications on the first message - if not ctx or not ctx._invocation_context: + if not ctx: logger.debug("No context available for dynamic instruction") return "" - session = ctx._invocation_context.session + session = ctx.session if not session: logger.debug("No session available for dynamic instruction") return "" - # FOR TESTING: Always check for notifications (comment out to enable first-message-only) - # Only check on first message (when events list is empty or has only 1-2 events) + # FOR TESTING: Always check for notifications + # (comment out to enable first-message-only) + # Only check on first message (when events list is empty + # or has only 1-2 events) # Events include both user and agent messages, so < 2 means first interaction # event_count = len(session.events) if session.events else 0 # @@ -74,7 +76,11 @@ async def provide_dynamic_instruction( return "" # Build dynamic instruction with notification details - notification_ids = [n.get("id_notificacion") for n in pending_notifications] + notification_ids = [ + nid + for n in pending_notifications + if (nid := n.get("id_notificacion")) is not None + ] count = len(pending_notifications) # Format notification details for the agent @@ -97,9 +103,11 @@ INSTRUCCIONES: - Menciona estas notificaciones de forma natural en tu respuesta inicial - No necesitas leerlas todas literalmente, solo hazle saber que las tiene - Sé breve y directo según tu personalidad (directo y cálido) -- Si el usuario pregunta algo específico, prioriza responder eso primero y luego menciona las notificaciones +- Si el usuario pregunta algo específico, prioriza responder eso primero\ + y luego menciona las notificaciones -Ejemplo: "¡Hola! 👋 Antes de empezar, veo que tienes {count} notificación(es) pendiente(s) en tu cuenta. ¿Te gustaría revisarlas o prefieres que te ayude con algo más?" +Ejemplo: "¡Hola! 👋 Tienes {count} notificación(es)\ + pendiente(s). ¿Te gustaría revisarlas?" """ # Mark notifications as notified in Firestore @@ -111,10 +119,11 @@ Ejemplo: "¡Hola! 👋 Antes de empezar, veo que tienes {count} notificación(es phone_number, ) - return instruction - except Exception: logger.exception( - "Error building dynamic instruction for user %s", phone_number + "Error building dynamic instruction for user %s", + phone_number, ) return "" + else: + return instruction diff --git a/src/va_agent/governance.py b/src/va_agent/governance.py index 936c668..a65d5a3 100644 --- a/src/va_agent/governance.py +++ b/src/va_agent/governance.py @@ -1,4 +1,5 @@ """GovernancePlugin: Guardrails for VAia, the virtual assistant for VA.""" + import logging import re @@ -9,10 +10,57 @@ logger = logging.getLogger(__name__) FORBIDDEN_EMOJIS = [ - "🥵","🔪","🎰","🎲","🃏","😤","🤬","😡","😠","🩸","🧨","🪓","☠️","💀", - "💣","🔫","👗","💦","🍑","🍆","👄","👅","🫦","💩","⚖️","⚔️","✝️","🕍", - "🕌","⛪","🍻","🍸","🥃","🍷","🍺","🚬","👹","👺","👿","😈","🤡","🧙", - "🧙‍♀️", "🧙‍♂️", "🧛", "🧛‍♀️", "🧛‍♂️", "🔞","🧿","💊", "💏" + "🥵", + "🔪", + "🎰", + "🎲", + "🃏", + "😤", + "🤬", + "😡", + "😠", + "🩸", + "🧨", + "🪓", + "☠️", + "💀", + "💣", + "🔫", + "👗", + "💦", + "🍑", + "🍆", + "👄", + "👅", + "🫦", + "💩", + "⚖️", + "⚔️", + "✝️", + "🕍", + "🕌", + "⛪", + "🍻", + "🍸", + "🥃", + "🍷", + "🍺", + "🚬", + "👹", + "👺", + "👿", + "😈", + "🤡", + "🧙", + "🧙‍♀️", + "🧙‍♂️", + "🧛", + "🧛‍♀️", + "🧛‍♂️", + "🔞", + "🧿", + "💊", + "💏", ] @@ -20,29 +68,31 @@ class GovernancePlugin: """Guardrail executor for VAia requests as a Agent engine callbacks.""" def __init__(self) -> None: - """Initialize guardrail model (structured output), prompt and emojis patterns.""" + """Initialize guardrail model, prompt and emojis patterns.""" self._combined_pattern = self._get_combined_pattern() - def _get_combined_pattern(self): - person_pattern = r"(?:🧑|👩|👨)" - tone_pattern = r"[\U0001F3FB-\U0001F3FF]?" - - # Unique pattern that combines all forbidden emojis, including complex ones with skin tones - combined_pattern = re.compile( - rf"{person_pattern}{tone_pattern}\u200d❤️?\u200d💋\u200d{person_pattern}{tone_pattern}" # kiss - rf"|{person_pattern}{tone_pattern}\u200d❤️?\u200d{person_pattern}{tone_pattern}" # lovers - rf"|🖕{tone_pattern}" # middle finger with all skin tone variations - rf"|{'|'.join(map(re.escape, sorted(FORBIDDEN_EMOJIS, key=len, reverse=True)))}" # simple emojis - rf"|\u200d|\uFE0F" # residual ZWJ and variation selectors + def _get_combined_pattern(self) -> re.Pattern[str]: + person = r"(?:🧑|👩|👨)" + tone = r"[\U0001F3FB-\U0001F3FF]?" + simple = "|".join( + map(re.escape, sorted(FORBIDDEN_EMOJIS, key=len, reverse=True)) ) - return combined_pattern - + + # Combines all forbidden emojis, including complex + # ones with skin tones + return re.compile( + rf"{person}{tone}\u200d❤️?\u200d💋\u200d{person}{tone}" + rf"|{person}{tone}\u200d❤️?\u200d{person}{tone}" + rf"|🖕{tone}" + rf"|{simple}" + rf"|\u200d|\uFE0F" + ) + def _remove_emojis(self, text: str) -> tuple[str, list[str]]: removed = self._combined_pattern.findall(text) text = self._combined_pattern.sub("", text) return text.strip(), removed - def after_model_callback( self, callback_context: CallbackContext | None = None, diff --git a/src/va_agent/notifications.py b/src/va_agent/notifications.py index e7cc57a..8536fb2 100644 --- a/src/va_agent/notifications.py +++ b/src/va_agent/notifications.py @@ -58,9 +58,7 @@ class NotificationService: """ try: # Query Firestore document by phone number - doc_ref = self._db.collection(self._collection_path).document( - phone_number - ) + doc_ref = self._db.collection(self._collection_path).document(phone_number) doc = await doc_ref.get() if not doc.exists: @@ -78,9 +76,7 @@ class NotificationService: # Filter notifications that have NOT been notified by the agent pending = [ - n - for n in all_notifications - if not n.get("notified_by_agent", False) + n for n in all_notifications if not n.get("notified_by_agent", False) ] if not pending: @@ -90,9 +86,7 @@ class NotificationService: return [] # Sort by timestamp_creacion (most recent first) - pending.sort( - key=lambda n: n.get("timestamp_creacion", 0), reverse=True - ) + pending.sort(key=lambda n: n.get("timestamp_creacion", 0), reverse=True) # Return top N most recent result = pending[: self._max_to_notify] @@ -104,13 +98,13 @@ class NotificationService: len(result), ) - return result - except Exception: logger.exception( "Failed to fetch notifications for phone: %s", phone_number ) return [] + else: + return result async def mark_as_notified( self, phone_number: str, notification_ids: list[str] @@ -133,9 +127,7 @@ class NotificationService: return True try: - doc_ref = self._db.collection(self._collection_path).document( - phone_number - ) + doc_ref = self._db.collection(self._collection_path).document(phone_number) doc = await doc_ref.get() if not doc.exists: @@ -184,18 +176,16 @@ class NotificationService: phone_number, ) - return True - except Exception: logger.exception( "Failed to mark notifications as notified for phone: %s", phone_number, ) return False + else: + return True - def format_notification_summary( - self, notifications: list[dict[str, Any]] - ) -> str: + def format_notification_summary(self, notifications: list[dict[str, Any]]) -> str: """Format notifications into a human-readable summary. Args: @@ -209,9 +199,7 @@ class NotificationService: return "" count = len(notifications) - summary_lines = [ - f"El usuario tiene {count} notificación(es) pendiente(s):" - ] + summary_lines = [f"El usuario tiene {count} notificación(es) pendiente(s):"] for i, notif in enumerate(notifications, 1): texto = notif.get("texto", "Sin texto")