Compare commits

...

2 Commits

Author SHA1 Message Date
Anibal Angulo
014ceb9da6 formatting
All checks were successful
CI / ci (pull_request) Successful in 19s
2026-03-10 23:49:24 +00:00
Anibal Angulo
7926d9881c Add notification model
Some checks failed
CI / ci (pull_request) Failing after 12s
2026-03-10 23:47:11 +00:00
3 changed files with 92 additions and 49 deletions

View File

@@ -1,5 +1,6 @@
"""Configuration helper for ADK agent.""" """Configuration helper for ADK agent."""
import logging
import os import os
from pydantic_settings import ( from pydantic_settings import (
@@ -37,6 +38,9 @@ class AgentSettings(BaseSettings):
mcp_audience: str mcp_audience: str
mcp_remote_url: str mcp_remote_url: str
# Logging
log_level: str = "INFO"
model_config = SettingsConfigDict( model_config = SettingsConfigDict(
yaml_file=CONFIG_FILE_PATH, yaml_file=CONFIG_FILE_PATH,
extra="ignore", # Ignore extra fields from config.yaml extra="ignore", # Ignore extra fields from config.yaml
@@ -60,3 +64,6 @@ class AgentSettings(BaseSettings):
settings = AgentSettings.model_validate({}) settings = AgentSettings.model_validate({})
logging.basicConfig()
logging.getLogger("va_agent").setLevel(settings.log_level.upper())

View File

@@ -84,23 +84,16 @@ async def provide_dynamic_instruction(
return "" return ""
# Build dynamic instruction with notification details # Build dynamic instruction with notification details
notification_ids = [ notification_ids = [n.id_notificacion for n in recent_notifications]
nid
for n in recent_notifications
if (nid := n.get("id_notificacion")) is not None
]
count = len(recent_notifications) count = len(recent_notifications)
# Format notification details for the agent (most recent first) # Format notification details for the agent (most recent first)
now = time.time() now = time.time()
notification_details = [] notification_details = []
for i, notif in enumerate(recent_notifications, 1): for i, notif in enumerate(recent_notifications, 1):
evento = notif.get("nombre_evento_dialogflow", "notificacion") ago = _format_time_ago(now, notif.timestamp_creacion)
texto = notif.get("texto", "Sin texto")
ts = notif.get("timestamp_creacion", notif.get("timestampCreacion", 0))
ago = _format_time_ago(now, ts)
notification_details.append( notification_details.append(
f" {i}. [{ago}] Evento: {evento} | Texto: {texto}" f" {i}. [{ago}] Evento: {notif.nombre_evento} | Texto: {notif.texto}"
) )
details_text = "\n".join(notification_details) details_text = "\n".join(notification_details)
@@ -123,6 +116,7 @@ async def provide_dynamic_instruction(
count, count,
phone_number, phone_number,
) )
logger.debug("Dynamic instruction content:\n%s", instruction)
except Exception: except Exception:
logger.exception( logger.exception(

View File

@@ -7,32 +7,78 @@ import time
from datetime import datetime from datetime import datetime
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
from pydantic import AliasChoices, BaseModel, Field, field_validator
if TYPE_CHECKING: if TYPE_CHECKING:
from google.cloud.firestore_v1.async_client import AsyncClient from google.cloud.firestore_v1.async_client import AsyncClient
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
def _extract_ts(n: dict[str, Any]) -> float: class Notification(BaseModel):
"""Return the creation timestamp of a notification as epoch seconds.""" """A single notification, normalised from either schema.
raw = n.get("timestamp_creacion", n.get("timestampCreacion", 0))
if isinstance(raw, (int, float)): Handles snake_case (``id_notificacion``), camelCase
return float(raw) (``idNotificacion``), and English short names (``notificationId``)
if isinstance(raw, datetime): transparently via ``AliasChoices``.
return raw.timestamp() """
if isinstance(raw, str):
id_notificacion: str = Field(
validation_alias=AliasChoices(
"id_notificacion", "idNotificacion", "notificationId"
),
)
texto: str = Field(
default="Sin texto",
validation_alias=AliasChoices("texto", "text"),
)
nombre_evento: str = Field(
default="notificacion",
validation_alias=AliasChoices(
"nombre_evento_dialogflow", "nombreEventoDialogflow", "event"
),
)
timestamp_creacion: float = Field(
default=0.0,
validation_alias=AliasChoices("timestamp_creacion", "timestampCreacion"),
)
status: str = "active"
parametros: dict[str, Any] = Field(
default_factory=dict,
validation_alias=AliasChoices("parametros", "parameters"),
)
@field_validator("timestamp_creacion", mode="before")
@classmethod
def _coerce_timestamp(cls, v: Any) -> float:
"""Normalise Firestore timestamps (float, str, datetime) to float."""
if isinstance(v, (int, float)):
return float(v)
if isinstance(v, datetime):
return v.timestamp()
if isinstance(v, str):
try: try:
return float(raw) return float(v)
except ValueError: except ValueError:
return 0.0 return 0.0
return 0.0 return 0.0
class NotificationDocument(BaseModel):
"""Top-level Firestore / Redis document that wraps a list of notifications.
Mirrors the schema used by ``utils/check_notifications.py``
(``NotificationSession``) but keeps only what the agent needs.
"""
notificaciones: list[Notification] = Field(default_factory=list)
@runtime_checkable @runtime_checkable
class NotificationBackend(Protocol): class NotificationBackend(Protocol):
"""Backend-agnostic interface for notification storage.""" """Backend-agnostic interface for notification storage."""
async def get_recent_notifications(self, phone_number: str) -> list[dict[str, Any]]: async def get_recent_notifications(self, phone_number: str) -> list[Notification]:
"""Return recent notifications for *phone_number*.""" """Return recent notifications for *phone_number*."""
... ...
@@ -65,7 +111,7 @@ class FirestoreNotificationBackend:
self._max_to_notify = max_to_notify self._max_to_notify = max_to_notify
self._window_hours = window_hours self._window_hours = window_hours
async def get_recent_notifications(self, phone_number: str) -> list[dict[str, Any]]: async def get_recent_notifications(self, phone_number: str) -> list[Notification]:
"""Get recent notifications for a user. """Get recent notifications for a user.
Retrieves notifications created within the configured time window, Retrieves notifications created within the configured time window,
@@ -75,14 +121,7 @@ class FirestoreNotificationBackend:
phone_number: User's phone number (used as document ID) phone_number: User's phone number (used as document ID)
Returns: Returns:
List of notification dictionaries with structure: List of validated :class:`Notification` instances.
{
"id_notificacion": str,
"texto": str,
"status": str,
"timestamp_creacion": timestamp,
"parametros": {...}
}
""" """
try: try:
@@ -96,17 +135,19 @@ class FirestoreNotificationBackend:
return [] return []
data = doc.to_dict() or {} data = doc.to_dict() or {}
all_notifications = data.get("notificaciones", []) document = NotificationDocument.model_validate(data)
if not all_notifications: if not document.notificaciones:
logger.info("No notifications in array for phone: %s", phone_number) logger.info("No notifications in array for phone: %s", phone_number)
return [] return []
cutoff = time.time() - (self._window_hours * 3600) cutoff = time.time() - (self._window_hours * 3600)
recent = [n for n in all_notifications if _extract_ts(n) >= cutoff] parsed = [
n for n in document.notificaciones if n.timestamp_creacion >= cutoff
]
if not recent: if not parsed:
logger.info( logger.info(
"No notifications within the last %.0fh for phone: %s", "No notifications within the last %.0fh for phone: %s",
self._window_hours, self._window_hours,
@@ -114,13 +155,13 @@ class FirestoreNotificationBackend:
) )
return [] return []
recent.sort(key=_extract_ts, reverse=True) parsed.sort(key=lambda n: n.timestamp_creacion, reverse=True)
result = recent[: self._max_to_notify] result = parsed[: self._max_to_notify]
logger.info( logger.info(
"Found %d recent notifications for phone: %s (returning top %d)", "Found %d recent notifications for phone: %s (returning top %d)",
len(recent), len(parsed),
phone_number, phone_number,
len(result), len(result),
) )
@@ -165,7 +206,7 @@ class RedisNotificationBackend:
self._max_to_notify = max_to_notify self._max_to_notify = max_to_notify
self._window_hours = window_hours self._window_hours = window_hours
async def get_recent_notifications(self, phone_number: str) -> list[dict[str, Any]]: async def get_recent_notifications(self, phone_number: str) -> list[Notification]:
"""Get recent notifications for a user from Redis. """Get recent notifications for a user from Redis.
Reads from the ``notification:{phone}`` key, parses the JSON Reads from the ``notification:{phone}`` key, parses the JSON
@@ -185,10 +226,9 @@ class RedisNotificationBackend:
) )
return [] return []
data = json.loads(raw) document = NotificationDocument.model_validate(json.loads(raw))
all_notifications: list[dict[str, Any]] = data.get("notificaciones", [])
if not all_notifications: if not document.notificaciones:
logger.info( logger.info(
"No notifications in array for phone: %s", "No notifications in array for phone: %s",
phone_number, phone_number,
@@ -197,9 +237,11 @@ class RedisNotificationBackend:
cutoff = time.time() - (self._window_hours * 3600) cutoff = time.time() - (self._window_hours * 3600)
recent = [n for n in all_notifications if _extract_ts(n) >= cutoff] parsed = [
n for n in document.notificaciones if n.timestamp_creacion >= cutoff
]
if not recent: if not parsed:
logger.info( logger.info(
"No notifications within the last %.0fh for phone: %s", "No notifications within the last %.0fh for phone: %s",
self._window_hours, self._window_hours,
@@ -207,13 +249,13 @@ class RedisNotificationBackend:
) )
return [] return []
recent.sort(key=_extract_ts, reverse=True) parsed.sort(key=lambda n: n.timestamp_creacion, reverse=True)
result = recent[: self._max_to_notify] result = parsed[: self._max_to_notify]
logger.info( logger.info(
"Found %d recent notifications for phone: %s (returning top %d)", "Found %d recent notifications for phone: %s (returning top %d)",
len(recent), len(parsed),
phone_number, phone_number,
len(result), len(result),
) )