generado bot de telegram con capacidades de kanboard

agentes/base.py

@@ -0,0 +1,118 @@
+"""Clases base para definir agentes del proyecto."""
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Callable, Dict, List, Optional
+
+from agno.db.sqlite import SqliteDb
+
+logger = logging.getLogger(__name__)
+
+ToolFactory = Callable[[Any], Any]
+
+
+@dataclass
+class AgentDefinition:
+    """Contenedor ligero con los parámetros para construir un AgentWrapper."""
+
+    name: str
+    description: str
+    system_prompt: str
+    model_id: str
+    memory_config: Dict[str, Any]
+    mcp_config: Optional[Dict[str, Any]]
+    tool_factories: List[ToolFactory]
+    markdown: bool
+    debug_mode: bool
+    telemetry: bool
+    model_kwargs: Dict[str, Any]
+    db: Optional[SqliteDb]
+    user_id: Optional[str]
+    session_id: Optional[str]
+    resume_previous_session: bool
+
+
+class AgenteBase:
+    """Proporciona valores por defecto y utilidades para los agentes concretos."""
+
+    key: str = "base"
+    name: str = "Agente base"
+    description: str = "Agente genérico."
+    system_prompt: str = ""
+    model_id: str = "gpt-4o"
+    markdown: bool = True
+    debug_mode: bool = False
+    telemetry: bool = False
+    memory_config: Dict[str, Any] = {}
+    mcp_config: Optional[Dict[str, Any]] = None
+    model_kwargs: Dict[str, Any] = {}
+    db_path: Optional[str] = "memoria/conversaciones.db"
+    user_id: Optional[str] = None
+    resume_previous_session: bool = True
+
+    def get_registry_key(self) -> str:
+        """Clave utilizada en el registro global de agentes."""
+        return self.key.lower()
+
+    def get_memory_config(self) -> Dict[str, Any]:
+        return dict(self.memory_config) if self.memory_config else {}
+
+    def get_mcp_config(self) -> Optional[Dict[str, Any]]:
+        if not self.mcp_config:
+            return None
+        return {**self.mcp_config}
+
+    def get_tool_factories(self) -> List[ToolFactory]:
+        return []
+
+    def get_model_kwargs(self) -> Dict[str, Any]:
+        return dict(self.model_kwargs) if self.model_kwargs else {}
+
+    def _resolve_db_path(self) -> Optional[Path]:
+        if not self.db_path:
+            return None
+        db_path = Path(self.db_path)
+        if not db_path.is_absolute():
+            project_root = Path(__file__).resolve().parent.parent
+            db_path = project_root / db_path
+        try:
+            db_path.parent.mkdir(parents=True, exist_ok=True)
+        except OSError as error:
+            logger.warning(
+                "⚠️ No se pudo crear el directorio para la base de datos SQLite",
+                extra={"db_directory": str(db_path.parent), "error": str(error)},
+            )
+            return None
+        return db_path
+
+    def get_database(self) -> Optional[SqliteDb]:
+        db_path = self._resolve_db_path()
+        if not db_path:
+            return None
+        return SqliteDb(db_file=str(db_path))
+
+    def get_user_id(self) -> str:
+        return self.user_id or self.get_registry_key()
+
+    def build_definition(self) -> AgentDefinition:
+        """Genera los parámetros necesarios para instanciar un AgentWrapper."""
+        db = self.get_database()
+        return AgentDefinition(
+            name=self.name,
+            description=self.description,
+            system_prompt=self.system_prompt,
+            model_id=self.model_id,
+            memory_config=self.get_memory_config(),
+            mcp_config=self.get_mcp_config(),
+            tool_factories=list(self.get_tool_factories()),
+            markdown=self.markdown,
+            debug_mode=self.debug_mode,
+            telemetry=self.telemetry,
+            model_kwargs=self.get_model_kwargs(),
+            db=db,
+            user_id=self.get_user_id(),
+            session_id=None,
+            resume_previous_session=self.resume_previous_session,
+        )