va/agent
0
0
Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Format/Lint

Browse Source
This commit is contained in:
Anibal Angulo
2026-03-05 06:06:01 +00:00
parent 5941c41296
commit db879cee9f
6 changed files with 102 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
AGENTS.md
View File

@@ -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.

5
src/va_agent/agent.py
View File

@@ -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(

2
src/va_agent/config.py
View File

@@ -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

29
src/va_agent/dynamic_instruction.py
View File

@@ -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

88
src/va_agent/governance.py
View File

@@ -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,

32
src/va_agent/notifications.py
View File

@@ -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")