OneTeam/NaliiaBot
16
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: PEP8 Style

Browse Source
This commit is contained in:
aserrador
2026-03-04 16:20:56 -05:00
parent aab046b2ef
commit 6fc6338413
17 changed files with 1086 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/naliiabot/bot/agent/agent.py
View File

@@ -6,24 +6,22 @@ from langchain_core.runnables import RunnableConfig
from typing import Literal, Callable, Any, Union
from .schemas import MessagesState
from dataclasses import dataclass
import logging
@dataclass
class AgentConfig:
"""Configuración inmutable del agente para fácil testing."""
system_prompt: str = (
"helpful assistant that can call tools when needed. Always respond with a message. "
"helpful assistant that can call tools when needed."
" Always respond with a message."
)
max_iterations: int = 10
max_iterations: int = 200
timeout_seconds: float = 30.0
class Agent:
"""
Agente conversacional basado en LangGraph.
Soporta tools pasadas como lista de instancias.
"""
@@ -36,7 +34,7 @@ class Agent:
):
"""
Inicializa el agente.
Args:
model: Modelo LLM a usar
config: Configuración del agente (AgentConfig)
@@ -49,7 +47,7 @@ class Agent:
self._config = config or AgentConfig()
self._tools = tools or []
self._checkpointer = checkpointer or InMemorySaver()
self._tools_by_name: dict[str, Any] = {
tool.name: tool for tool in self._tools
}
@@ -73,7 +71,9 @@ class Agent:
"""Expone el grafo para inspección en tests."""
return self._build_agent()
async def ainvoke(self, state: dict, config: RunnableConfig | None = None) -> dict:
async def ainvoke(
self, state: dict, config: RunnableConfig | None = None
) -> dict:
"""
Versión asíncrona de invoke, necesaria para checkpointers de DB.
"""
@@ -86,7 +86,9 @@ class Agent:
# Llamamos al método ainvoke del grafo compilado
return await self._compiled_agent.ainvoke(state, config=config)
def invoke(self, state: dict, config: RunnableConfig | None = None) -> dict:
def invoke(
self, state: dict, config: RunnableConfig | None = None
) -> dict:
"""
Ejecuta el agente con el estado inicial.
@@ -98,7 +100,7 @@ class Agent:
"""
if not config:
config : RunnableConfig = {"configurable": {"thread_id": "1"}}
config: RunnableConfig = {"configurable": {"thread_id": "1"}}
if "messages" not in state:
raise ValueError("State must contain 'messages' key")

22
src/naliiabot/bot/prompts/prompts/prompts_template.yaml
View File

@@ -1,7 +1,7 @@
prompts:
NALIIA_PROMPT: |
"""
Eres Naliia, asistente virtual de un centro de belleza.
Eres Naliia, asistente virtual.
Responde usando formato Markdown:
- Usa **negrita** para énfasis.
@@ -9,19 +9,11 @@ prompts:
- Usa saltos de línea entre secciones.
- Destaca opciones importantes con `código inline`.
Ejemplo de formato:
¡Hola! Bienvenido/a, soy **Naliia**, tu asistente virtual.
*Lo que NUNCA debes hacer:*
Puedo ayudarte con:
**Agendar una cita**
Ver horarios disponibles y reservar tu servicio
**Información sobre servicios y productos**
Conocer nuestros tratamientos y productos
**Cancelar una cita**
Si necesitas cancelar alguna reserva previa
¿En qué puedo ayudarte hoy?
❌ Dar diagnósticos médicos
❌ Recomendar tratamientos clínicos sin evaluación
❌ Minimizar riesgos ("es super seguro, no pasa nada")
❌ Prometer resultados específicos
❌ Agendar procedimientos invasivos sin mencionar evaluación previa
"""

99
src/naliiabot/bot/tools/naliia_tools.py
View File

@@ -1,67 +1,70 @@
import asyncio
from datetime import datetime
from fastmcp import Client
from langchain.tools import tool
from ..logger import logger
from .schemes import ScheduleSchema
MCP_SERVER_URL = "http://localhost:8000/mcp"
_MCP_CLIENT = Client(MCP_SERVER_URL)
class NaliiaTools:
def __init__(self):
self._session_active = False
def get_tools(self):
return [
self.verificar_usuario_registrado,
self.registrar_usuario_nuevo,
self.consultar_sedes,
self.consultar_agenda_disponible,
self.agendar_cita
self.get_current_datetime,
self.schedule_appointment
]
@tool
def verificar_usuario_registrado(phone: str) -> bool:
def get_current_datetime() -> datetime:
"""
Verifica si el usuario esta registrado.
Args:
phone: Numero de contacto del cliente ejemplo 30123334
Consulta la fecha actual, no se aceptan fechas
en el pasado con respecto a esta fecha.
"""
logger.info("Llamando a Verificar usuario.")
return datetime.now()
return False
@tool
def registrar_usuario_nuevo(full_name: str, phone: str) -> bool:
@tool(args_schema=ScheduleSchema)
def schedule_appointment(
schedule_date, schedule_time,
service_center, customer,
professional, description
) -> bool:
"""
En caso de que no se haya podido verificar el usuario, sera necesario
registrarlo como un usuario nuevo.
Args:
full_name: Nombre completo del cliente.
phone: Numero de contacto del cliente.
"""
logger.info("Llamando a Registrar Nuevo Usuario.")
return True
@tool
def consultar_sedes() -> list:
"""
Brinda informacion al usuario acerca de las sedes disponibles y sus horarios.
"""
logger.info("Llamando a consultar sedes.")
return []
@tool
def consultar_agenda_disponible() -> list:
"""
Ayuda a consultar al cliente los horarios de atencion disponibles para agendar su cita.
"""
logger.info("Llamando a consultar agenda disponible.")
return []
@tool
def agendar_cita() -> bool:
"""
Ayuda al cliente a confirmar su cita.
Permite al Cliente al Agendar una Cita
"""
logger.info("Llamando a agendar cita.")
logger.info([
schedule_date,
schedule_time,
service_center,
customer,
professional,
description
])
example_schedule = {
'professional': 6,
'description': "Bien Bonito Todo!",
'customer': 4,
'date': '2026-03-25 14:00',
'service_center': 11
}
async def call_tool():
async with _MCP_CLIENT:
result = await _MCP_CLIENT.call_tool(
"create_schedule", example_schedule)
logger.info(result)
asyncio.run(call_tool())
return True

37
src/naliiabot/bot/tools/schemes.py Normal file
View File

@@ -0,0 +1,37 @@
from datetime import date, time
from typing import Optional
from pydantic import BaseModel, Field, PositiveInt
class ScheduleSchema(BaseModel):
"""
Scheme for Schedule Appointment
"""
schedule_date: date = Field(
description="Date in format DD-MM-YYYY")
schedule_time: time = Field(
description="Time in format 12h (HH:MM)")
service_center: Optional[PositiveInt] = Field(
None, description="ID Service center Example: 1")
customer: Optional[PositiveInt] = Field(
None, description="ID Customer Example: 1")
professional: Optional[PositiveInt] = Field(
None, description="ID Professional Example: 1")
description: str = Field(
default="General Schedule",
max_length=255,
description="Max 255 characters."
)
class Config:
json_schema_extra = {
"example": {
"schedule_date": "15-03-2026",
"schedule_time": "02:00",
"service_center": 1,
"professional": 12,
"customer": 999,
"description": "Review"
}
}