""" PromptLibrary - Carga y renderiza prompts desde archivos YAML. Este módulo maneja: - Carga de templates de prompts desde YAML - Renderizado con variables dinámicas - Herencia de personalidades - Cache para evitar lecturas repetidas de disco """ import os from pathlib import Path from typing import Dict, Optional, Any, List from functools import lru_cache import yaml from app.core.config import settings class PromptLibrary: """ Biblioteca de prompts cargados desde YAML. Estructura de directorios esperada: app/prompts/ ├── personalities/ # Personalidades de marca ├── templates/ # Templates de contenido ├── examples/ # Ejemplos para few-shot └── platforms/ # Configuración por plataforma """ def __init__(self, prompts_dir: Optional[str] = None): """ Inicializar la biblioteca de prompts. Args: prompts_dir: Directorio raíz de prompts. Si no se especifica, usa app/prompts/ relativo al proyecto. """ if prompts_dir: self.prompts_dir = Path(prompts_dir) else: # Detectar directorio del proyecto base_dir = Path(__file__).parent.parent.parent # app/services/ai -> app self.prompts_dir = base_dir / "prompts" self._cache: Dict[str, Any] = {} def _load_yaml(self, file_path: Path) -> Dict: """ Cargar archivo YAML con cache. Args: file_path: Ruta al archivo YAML Returns: Diccionario con contenido del YAML """ cache_key = str(file_path) if cache_key not in self._cache: if not file_path.exists(): raise FileNotFoundError(f"Archivo de prompt no encontrado: {file_path}") with open(file_path, "r", encoding="utf-8") as f: self._cache[cache_key] = yaml.safe_load(f) return self._cache[cache_key] def clear_cache(self) -> None: """Limpiar cache de YAMLs cargados.""" self._cache.clear() # === Personalidades === def get_personality(self, name: str = "default") -> Dict: """ Obtener una personalidad de marca. Args: name: Nombre de la personalidad (default, educational, promotional) Returns: Dict con configuración de personalidad """ file_path = self.prompts_dir / "personalities" / f"{name}.yaml" personality = self._load_yaml(file_path) # Si hereda de otra personalidad, mergear if "extends" in personality: base_name = personality["extends"] base = self.get_personality(base_name) personality = self._merge_personalities(base, personality) return personality def _merge_personalities(self, base: Dict, child: Dict) -> Dict: """ Mergear personalidad hija con base. La hija sobrescribe valores de la base. """ merged = base.copy() for key, value in child.items(): if key == "extends": continue if isinstance(value, dict) and key in merged and isinstance(merged[key], dict): merged[key] = {**merged[key], **value} else: merged[key] = value return merged def get_system_prompt(self, personality: str = "default") -> str: """ Obtener system prompt para una personalidad. Args: personality: Nombre de la personalidad Returns: System prompt renderizado con variables de negocio """ pers = self.get_personality(personality) system_prompt = pers.get("system_prompt", "") # Si tiene system_prompt_addition (de herencia), agregarlo if "system_prompt_addition" in pers: system_prompt += pers["system_prompt_addition"] # Renderizar variables de negocio system_prompt = system_prompt.format( brand_name=settings.BUSINESS_NAME, location=settings.BUSINESS_LOCATION, website=settings.BUSINESS_WEBSITE, tone=settings.CONTENT_TONE, ) return system_prompt # === Templates === def get_template(self, name: str) -> Dict: """ Obtener un template de contenido. Args: name: Nombre del template (tip_tech, product_post, etc.) Returns: Dict con configuración del template """ file_path = self.prompts_dir / "templates" / f"{name}.yaml" return self._load_yaml(file_path) def render_template( self, template_name: str, variables: Dict[str, Any], personality: str = "default" ) -> Dict[str, Any]: """ Renderizar un template con variables. Args: template_name: Nombre del template variables: Variables para sustituir en el template personality: Personalidad a usar Returns: Dict con: - system_prompt: Prompt del sistema - user_prompt: Prompt del usuario renderizado - parameters: Parámetros para la API (temperature, max_tokens) """ template = self.get_template(template_name) # Obtener system prompt de la personalidad especificada en template o parámetro pers_name = template.get("personality", personality) system_prompt = self.get_system_prompt(pers_name) # Agregar contexto del template al system prompt si existe if "system_context" in template: system_prompt += "\n\n" + template["system_context"] # Obtener y renderizar el user prompt user_template = template.get("template", "") # Aplicar defaults a variables faltantes template_vars = {} for var_def in template.get("variables", []): var_name = var_def["name"] if var_name in variables: template_vars[var_name] = variables[var_name] elif "default" in var_def: template_vars[var_name] = var_def["default"] elif var_def.get("required", False): raise ValueError(f"Variable requerida no proporcionada: {var_name}") # Renderizar template try: user_prompt = user_template.format(**template_vars) except KeyError as e: raise ValueError(f"Variable faltante en template: {e}") # Obtener parámetros parameters = template.get("parameters", {}) return { "system_prompt": system_prompt, "user_prompt": user_prompt, "parameters": parameters, "template_name": template_name, "personality": pers_name, } # === Plataformas === def get_platform_config(self, platform: str) -> Dict: """ Obtener configuración de una plataforma. Args: platform: Nombre de la plataforma (x, threads, instagram, facebook) Returns: Dict con configuración de la plataforma """ file_path = self.prompts_dir / "platforms" / f"{platform}.yaml" return self._load_yaml(file_path) def get_platform_limits(self, platform: str) -> Dict: """ Obtener límites de una plataforma. Args: platform: Nombre de la plataforma Returns: Dict con límites (max_characters, max_hashtags, etc.) """ config = self.get_platform_config(platform) return config.get("limits", {}) def get_platform_adaptation_rules(self, platform: str) -> str: """ Obtener reglas de adaptación para una plataforma. Args: platform: Nombre de la plataforma Returns: String con reglas de adaptación """ config = self.get_platform_config(platform) return config.get("adaptation_rules", "") # === Ejemplos === def get_examples(self, category: Optional[str] = None) -> Dict: """ Obtener ejemplos para few-shot learning. Args: category: Categoría específica (ia, productividad, seguridad) Si es None, retorna best_posts.yaml Returns: Dict con ejemplos """ if category: file_path = self.prompts_dir / "examples" / "by_category" / f"{category}.yaml" else: file_path = self.prompts_dir / "examples" / "best_posts.yaml" return self._load_yaml(file_path) def get_few_shot_examples( self, template_type: str, category: Optional[str] = None, max_examples: int = 3 ) -> List[str]: """ Obtener ejemplos formateados para few-shot prompting. Args: template_type: Tipo de template (tip_tech, product_post, etc.) category: Categoría de contenido (opcional) max_examples: Máximo de ejemplos a retornar Returns: Lista de strings con ejemplos formateados """ examples = [] # Primero intentar ejemplos de categoría específica if category: try: cat_examples = self.get_examples(category) if "examples" in cat_examples: for example_type, example_list in cat_examples["examples"].items(): for ex in example_list[:max_examples]: if "content" in ex: examples.append(ex["content"]) except FileNotFoundError: pass # Luego best_posts generales if len(examples) < max_examples: try: best = self.get_examples() if "examples" in best and template_type in best["examples"]: for ex in best["examples"][template_type]: if len(examples) >= max_examples: break if "content" in ex: examples.append(ex["content"]) elif "posts" in ex: # Para threads examples.append("\n\n".join(ex["posts"])) except FileNotFoundError: pass return examples[:max_examples] # === Utilidades === def list_templates(self) -> List[str]: """Listar todos los templates disponibles.""" templates_dir = self.prompts_dir / "templates" if not templates_dir.exists(): return [] return [f.stem for f in templates_dir.glob("*.yaml")] def list_personalities(self) -> List[str]: """Listar todas las personalidades disponibles.""" pers_dir = self.prompts_dir / "personalities" if not pers_dir.exists(): return [] return [f.stem for f in pers_dir.glob("*.yaml")] def list_platforms(self) -> List[str]: """Listar todas las plataformas configuradas.""" platforms_dir = self.prompts_dir / "platforms" if not platforms_dir.exists(): return [] return [f.stem for f in platforms_dir.glob("*.yaml")] # Instancia global prompt_library = PromptLibrary()