from __future__ import annotations import json from typing import Any from .agent_router import AgentKind, agent_label, get_agent_system_prompt, route_to_agent from .chatbot_bridge import ChatbotBridge from .config import get_settings from . import langfuse_client as lf from .providers import BaseProvider, extract_json_document from .rag import rag_store from .ssh_exec import SSHCommandRunner from .storage import SupervisorStore OPS_KEYWORDS: dict[str, tuple[list[str], str, str, str]] = { "git_status": (["estado git", "git status", "despliegue actual", "estado producción"], "Comprobar estado del despliegue", "Inspeccionar la raíz del despliegue y, si existe repo Git, mostrar su estado antes de tocar nada.", "low"), "git_pull": (["deploy", "desplegar", "publica", "publicar", "actualiza producción", "git pull"], "Actualizar código en producción", "Traer la rama principal al servidor objetivo mediante pull seguro.", "high"), "composer_lint": (["lint php", "composer lint", "lint del proyecto"], "Ejecutar lint PHP", "Lanzar el lint PHP oficial del repositorio.", "low"), "phpstan": (["phpstan", "analiza phpstan"], "Ejecutar PHPStan", "Revisar estática del backend PHP con el comando oficial.", "low"), "psalm": (["psalm", "analiza psalm"], "Ejecutar Psalm", "Pasar Psalm sobre el proyecto para buscar regresiones de tipos.", "low"), "npm_minify": (["minify", "minifica", "minificar assets"], "Minificar frontend custom", "Ejecutar la minificación custom antes de desplegar cambios front.", "medium"), "npm_build_grid": (["grid-core", "build grid", "compila grid"], "Compilar grid-core", "Reconstruir el bundle grid-core consumido por la web.", "medium"), "tail_php_errorlog": (["logs", "php_errorlog", "errores php", "mira el log"], "Leer php_errorlog", "Inspeccionar las últimas líneas del log PHP del servidor.", "low"), "cache_flush": (["opcache", "flush cache", "vaciar cache", "reinicia cache"], "Vaciar OPcache", "Forzar limpieza de OPcache tras un despliegue o fix crítico.", "medium"), } GROWTH_WORDS = [ "captación", "captacion", "anunciantes", "crm", "ventas", "leads", "campaña", "campana", "reactivar", "outreach", "email", "whatsapp", "telegram", ] RESEARCH_WORDS = [ "estudia", "estudio", "analiza", "audita", "auditoría", "auditoria", "seo", "performance", "seguridad", "ga4", "gsc", "competencia", "roadmap", "plan", ] AUTOMATION_RULE_BLUEPRINTS: list[dict[str, Any]] = [ { "rule_key": "ops_daily", "title": "Revisión operativa diaria", "description": "Genera un estudio diario sobre despliegue, logs y chequeos técnicos antes de tocar producción.", "prompt_text": ( "Actúa como OpsAgent especialista en el proyecto {project_name} ({project_domain}).\n" "Genera un estudio de revisión operativa diaria con los siguientes puntos:\n" "1. Estado del despliegue: ¿hay cambios pendientes de publicar o commits recientes sin revisar?\n" "2. Errores PHP recientes: ¿hay errores graves en el log que requieran acción urgente?\n" "3. Salud del servidor: OPcache, permisos, espacio en disco estimado.\n" "4. ¿Conviene ejecutar lint PHP o PHPStan hoy antes de cambios?\n" "5. Propuesta de acción concreta para hoy (máx. 3 puntos priorizados).\n" "Sé conciso. Marca cada propuesta como low / medium / high riesgo." ), "plan_kind": "command", "schedule_hint": "diario", "enabled": False, "requires_approval": True, }, { "rule_key": "marketing_daily", "title": "Revisión comercial diaria", "description": "Prepara un estudio diario de campañas, métricas y oportunidades comerciales.", "prompt_text": ( "Actúa como MarketingAgent especialista en {project_name} ({project_domain}), un portal de clasificados.\n" "Analiza el estado comercial de hoy:\n" "1. ¿Cuántas campañas activas hay y cuál es su rendimiento estimado?\n" "2. ¿Hay anunciantes que llevan más de 15 días sin publicar y podrían reactivarse?\n" "3. ¿Qué canal tiene mejor conversión esta semana (WA, email, directo)?\n" "4. Propón 2-3 acciones de marketing concretas para hoy (mensajes, ofertas, canales).\n" "Sé directo y enfocado en acciones ejecutables en las próximas 24 h. Marca riesgo de cada propuesta." ), "plan_kind": "marketing", "schedule_hint": "diario", "enabled": False, "requires_approval": True, }, { "rule_key": "seo_weekly", "title": "Auditoría SEO y UX semanal", "description": "Genera una revisión semanal sobre SEO, UX, rendimiento y embudos del portal web.", "prompt_text": ( "Actúa como SeoAgent especialista en {project_domain}.\n" "Genera el informe SEO y UX semanal:\n" "1. Core Web Vitals estimados y puntos críticos de rendimiento del portal.\n" "2. Principales oportunidades de posicionamiento orgánico en España para el sector.\n" "3. Problemas de UX detectables: embudos de alta, registro de anunciantes, flujo de compra.\n" "4. Acciones de SEO técnico pendientes (canonical, schema, sitemap, robots.txt).\n" "5. Top 3 mejoras priorizadas para implementar esta semana.\n" "Sé específico y orientado a ROI. Marca riesgo bajo/medio para cada mejora." ), "plan_kind": "command", "schedule_hint": "semanal", "enabled": False, "requires_approval": True, }, { "rule_key": "stack_health", "title": "Chequeo de salud del stack IA", "description": "Revisa el estado completo del supervisor, Telegram, bridge del chatbot y guardarraíles.", "prompt_text": ( "Actúa como OpsAgent encargado de la salud del stack IA de {project_name}.\n" "Genera un informe de salud del sistema:\n" "1. Supervisor multiagente: ¿responde correctamente? ¿Hay errores en el scheduler?\n" "2. Telegram bot: ¿en polling activo? ¿Algún comando falló recientemente?\n" "3. Bridge del chatbot: ¿conectado? ¿Últimas campañas sin errores?\n" "4. Aprendizaje: ¿hay propuestas aprobadas pendientes de memorizar?\n" "5. Guardarraíles: ¿alguna regla SSH fue rechazada recientemente? ¿Todos los agentes responden?\n" "6. Acción recomendada si detectas algún problema (máx. 2 propuestas).\n" "Marca riesgo bajo si todo está OK, medio o alto si hay algo urgente." ), "plan_kind": "command", "schedule_hint": "diario", "enabled": False, "requires_approval": True, }, { "rule_key": "reactivacion_anunciantes", "title": "Plan de reactivación de anunciantes", "description": "Identifica anunciantes inactivos y propone campañas de reactivación personalizadas por segmento.", "prompt_text": ( "Actúa como MarketingAgent especialista en CRM para {project_name} ({project_domain}).\n" "Diseña el plan de reactivación semanal de anunciantes:\n" "1. Perfil del anunciante inactivo típico (días sin publicar, segmento, historial).\n" "2. Mensaje de reactivación personalizado por canal (WhatsApp, email, Telegram).\n" "3. Oferta concreta: ¿qué incentivo cierra más (descuento destaque, créditos extra, visibilidad garantizada)?\n" "4. Secuencia de seguimiento: cuándo enviar 2.º y 3.er contacto si no responden.\n" "5. KPI para medir éxito: tasa de reactivación esperada y ROI mínimo aceptable.\n" "Propón 3 campañas concretas listas para lanzar, una por segmento (nuevo, dormido, premium caducado)." ), "plan_kind": "marketing", "schedule_hint": "semanal", "enabled": False, "requires_approval": True, }, { "rule_key": "seguridad_semanal", "title": "Auditoría de seguridad semanal", "description": "Revisa guardarraíles operativos, accesos SSH, permisos y vectores de riesgo del portal.", "prompt_text": ( "Actúa como SecurityAgent para {project_name} ({project_domain}).\n" "Genera el informe de seguridad semanal:\n" "1. Accesos SSH recientes: ¿hay intentos fallidos o IPs no reconocidas?\n" "2. Permisos de archivos críticos en producción (wp-config, .env, uploads, logs).\n" "3. Versiones de componentes: PHP, CMS y plugins con actualizaciones pendientes.\n" "4. Headers HTTP de seguridad: CSP, HSTS, X-Frame-Options, Referrer-Policy.\n" "5. Top 3 riesgos OWASP aplicables al portal y propuesta de mitigación concreta.\n" "Marca riesgo alto si hay algo urgente. Propón máx. 3 acciones priorizadas." ), "plan_kind": "command", "schedule_hint": "semanal", "enabled": False, "requires_approval": True, }, { "rule_key": "git_code_analysis", "title": "Análisis CodeAgent: estado git + propuesta de acción", "description": "Lee el estado git remoto, últimos commits y diff, y propone si hay que actualizar producción o corregir algo.", "prompt_text": ( "Actúa como CodeAgent experto en despliegue para {project_name} ({project_domain}).\n" "Tu objetivo es analizar el estado actual del repositorio en producción y proponer acciones.\n" "Pasos:\n" "1. Propón ejecutar 'git_fetch' (bajo riesgo) para ver qué commits hay en remoto sin aplicar.\n" "2. Propón ejecutar 'git_status' para ver el estado actual del working tree.\n" "3. Propón ejecutar 'git_log' para ver los últimos 10 commits y detectar patrones.\n" "4. Si hay commits pendientes de aplicar: propón 'git_pull_current' (alto riesgo, requiere aprobación).\n" "5. Si hay errores PHP: propón 'tail_php_errorlog' (bajo riesgo) para inspeccionarlos.\n" "Para cada propuesta, indica: acción concreta, riesgo (low/medium/high), justificación en 1 línea.\n" "Ordena por prioridad. No hagas nada sin aprobación." ), "plan_kind": "command", "schedule_hint": "diario", "enabled": False, "requires_approval": True, }, { "rule_key": "whatsapp_reactivacion", "title": "Plan de campaña WhatsApp: reactivación de anunciantes", "description": "Genera textos y segmentación para una campaña WhatsApp de reactivación. Requiere aprobación antes de cualquier envío.", "prompt_text": ( "Actúa como MarketingAgent experto en WhatsApp Business para {project_name} ({project_domain}).\n" "Genera un plan de campaña WhatsApp de reactivación de anunciantes:\n" "1. SEGMENTOS: define 3 segmentos (inactivos <30 días, inactivos 30-90 días, nunca publicaron).\n" "2. MENSAJE POR SEGMENTO: redacta el texto exacto para cada plantilla (máx 160 car. por plantilla).\n" " Segmento A (caducados recientes): urgencia + descuento 20%.\n" " Segmento B (dormidos): novedad del portal + facilidad de reactivación.\n" " Segmento C (sin publicar): guía paso a paso para publicar el primer anuncio.\n" "3. NOMBRE DE PLANTILLA sugerido para cada segmento (formato: letras_minúsculas_sin_espacios).\n" "4. CTA: qué acción concreta debe hacer el receptor (botón de respuesta rápida sugerido).\n" "5. SECUENCIA: si no responden en 48h, qué mensaje de seguimiento enviar.\n" "Formato de salida: JSON estructurado con segments[], message, template_name, cta, followup.\n" "Recuerda: solo se envía a contactos con consent_accepted=1. Incluye enlace de baja en notas." ), "plan_kind": "marketing", "schedule_hint": "semanal", "enabled": False, "requires_approval": True, }, ] class MultiAgentOrchestrator: def __init__(self, store: SupervisorStore, provider: BaseProvider, runner: SSHCommandRunner, chatbot_bridge: ChatbotBridge | None = None) -> None: self.store = store self.provider = provider self.runner = runner self.chatbot_bridge = chatbot_bridge self.store.seed_automation_rules(AUTOMATION_RULE_BLUEPRINTS) async def create_command_plan(self, text: str, actor: str, channel: str) -> dict[str, Any]: trace_id = lf.start_trace( "create_command_plan", input_data={"text": text[:500], "actor": actor, "channel": channel}, tags=["ops", "plan"], ) lf.set_current_trace_id(trace_id) try: heuristic_plan = self._heuristic_plan(text) final_plan = await self._maybe_enrich_with_llm(text, heuristic_plan) result = self._persist_plan(text, actor, channel, final_plan) lf.finish_trace(trace_id, output={"proposals": len(result.get("proposals", []))}) return result except Exception: lf.finish_trace(trace_id, status="ERROR") raise finally: lf.set_current_trace_id(None) async def create_marketing_plan(self, brief: str, actor: str, channel: str) -> dict[str, Any]: context: dict[str, Any] = {} bridge_error: str | None = None if self.chatbot_bridge and self.chatbot_bridge.enabled: try: overview = await self.chatbot_bridge.get_overview(limit=6) campaigns = await self.chatbot_bridge.list_campaigns(limit=8) context = { "overview": overview, "campaigns": campaigns.get("campaigns", []), } except Exception as exc: bridge_error = str(exc) fallback = self._heuristic_marketing_plan(brief, context, bridge_error) trace_id = lf.start_trace( "create_marketing_plan", input_data={"brief": brief[:500], "actor": actor, "channel": channel}, tags=["marketing", "plan"], ) lf.set_current_trace_id(trace_id) try: final_plan = await self._maybe_enrich_marketing_with_llm(brief, fallback, context, bridge_error) result = self._persist_plan(brief, actor, channel, final_plan) lf.finish_trace(trace_id, output={"proposals": len(result.get("proposals", []))}) return result except Exception: lf.finish_trace(trace_id, status="ERROR") raise finally: lf.set_current_trace_id(None) async def get_chatbot_overview(self, limit: int = 6) -> dict[str, Any]: if not self.chatbot_bridge or not self.chatbot_bridge.enabled: raise RuntimeError("El bridge del chatbot no está configurado en el supervisor") return await self.chatbot_bridge.get_overview(limit=limit) async def list_chatbot_campaigns(self, limit: int = 8) -> dict[str, Any]: if not self.chatbot_bridge or not self.chatbot_bridge.enabled: raise RuntimeError("El bridge del chatbot no está configurado en el supervisor") return await self.chatbot_bridge.list_campaigns(limit=limit) async def get_chatbot_campaign(self, campaign_id: int, source: str | None = None) -> dict[str, Any]: if not self.chatbot_bridge or not self.chatbot_bridge.enabled: raise RuntimeError("El bridge del chatbot no está configurado en el supervisor") return await self.chatbot_bridge.get_campaign(campaign_id, source) async def list_chatbot_consents( self, limit: int = 50, status: str | None = None, source: str | None = None, phone: str | None = None, advertiser_id: int | None = None, ) -> dict[str, Any]: if not self.chatbot_bridge or not self.chatbot_bridge.enabled: raise RuntimeError("El bridge del chatbot no está configurado en el supervisor") return await self.chatbot_bridge.list_consents(limit=limit, status=status, source=source, phone=phone, advertiser_id=advertiser_id) async def create_chatbot_consent( self, actor: str, phone: str, source: str, name: str | None = None, proof: str | None = None, advertiser_id: int | None = None, ) -> dict[str, Any]: if not self.chatbot_bridge or not self.chatbot_bridge.enabled: raise RuntimeError("El bridge del chatbot no está configurado en el supervisor") result = await self.chatbot_bridge.create_consent( phone=phone, source=source, name=name, proof=proof, advertiser_id=advertiser_id, ) self.store.add_audit( "chatbot.consent.created", { "actor": actor, "phone": phone, "source": source, "advertiser_id": advertiser_id, "created": result.get("created", False), }, ) return result async def create_chatbot_campaign( self, actor: str, advertiser_id: int, name: str, message_template: str, targets: list[dict[str, Any]], description: str | None = None, scheduled_at: str | None = None, use_ai_personalization: bool = False, ) -> dict[str, Any]: if not self.chatbot_bridge or not self.chatbot_bridge.enabled: raise RuntimeError("El bridge del chatbot no está configurado en el supervisor") result = await self.chatbot_bridge.create_campaign( advertiser_id=advertiser_id, name=name, message_template=message_template, targets=targets, description=description, scheduled_at=scheduled_at, use_ai_personalization=use_ai_personalization, ) campaign = result.get("campaign") or {} self.store.add_audit( "chatbot.campaign.created", { "actor": actor, "advertiser_id": advertiser_id, "campaign_id": campaign.get("id"), "campaign_ref": campaign.get("campaign_ref"), "targets": result.get("created_targets", 0), }, ) return result async def send_chatbot_campaign( self, actor: str, campaign_id: int, immediate: bool = True, source: str | None = None, ) -> dict[str, Any]: if not self.chatbot_bridge or not self.chatbot_bridge.enabled: raise RuntimeError("El bridge del chatbot no está configurado en el supervisor") if source and source.strip().lower() not in {"marketing", ""}: raise RuntimeError("El envío real desde el supervisor solo está habilitado para campañas del motor marketing.") result = await self.chatbot_bridge.send_campaign(campaign_id=campaign_id, immediate=immediate) campaign = result.get("campaign") or {} self.store.add_audit( "chatbot.campaign.armed", { "actor": actor, "campaign_id": campaign_id, "campaign_ref": campaign.get("campaign_ref"), "mode": result.get("mode", "immediate"), "pending_targets": result.get("pending_targets", 0), }, ) return result def get_learning_mode(self) -> str: mode = str(self.store.get_runtime_state("learning.mode", "semi") or "semi").strip().lower() if mode not in {"manual", "semi", "auto"}: return "semi" return mode def list_automation_rules(self) -> list[dict[str, Any]]: return self.store.list_automation_rules(limit=20) def get_autonomy_state(self) -> dict[str, Any]: rules = self.list_automation_rules() enabled = sum(1 for rule in rules if rule.get("enabled")) return { "rules": rules, "enabled": enabled, "total": len(rules), "recent_runs": self.store.list_automation_runs(limit=6), } def set_automation_rule_enabled(self, rule_key: str, enabled: bool, actor: str, note: str | None = None) -> dict[str, Any]: updated = self.store.set_automation_rule_enabled(rule_key, enabled) if updated is None: raise KeyError("Automatización no encontrada") self.store.add_audit( "automation.rule.updated", { "actor": actor, "rule_key": rule_key, "enabled": enabled, "note": note or "", }, ) return updated async def run_automation_rule(self, rule_key: str, actor: str, note: str | None = None) -> dict[str, Any]: rule = self.store.get_automation_rule(rule_key) if rule is None: raise KeyError("Automatización no encontrada") # Sustituir placeholders portables en el prompt s = get_settings() raw_prompt = str(rule["prompt_text"]) prompt = raw_prompt.format( project_name=s.project_name, project_domain=s.project_domain, ) try: if rule.get("plan_kind") == "marketing": result = await self.create_marketing_plan(prompt, actor, f"automation:{rule_key}") else: result = await self.create_command_plan(prompt, actor, f"automation:{rule_key}") except Exception as exc: self.store.record_automation_rule_run(rule_key, "failed") self.store.add_automation_run(rule_key, actor, "failed", note or str(exc)) self.store.add_audit( "automation.rule.failed", {"actor": actor, "rule_key": rule_key, "error": str(exc)}, ) raise study = result.get("study") or {} summary = str(study.get("summary") or rule.get("title") or "Automatización ejecutada") self.store.record_automation_rule_run(rule_key, "planned") self.store.add_automation_run(rule_key, actor, "planned", summary, study_id=study.get("id")) self.store.add_audit( "automation.rule.executed", { "actor": actor, "rule_key": rule_key, "study_id": study.get("id"), "note": note or "", }, ) return {"rule": rule, "result": result} async def agent_chat(self, text: str, agent_kind_str: str, actor: str) -> dict[str, Any]: """Chat directo con un agente específico. Devuelve respuesta de texto libre.""" try: kind = AgentKind(agent_kind_str.lower()) except ValueError: kind = AgentKind.GENERAL system_prompt = get_agent_system_prompt(kind) label = agent_label(kind) if self.provider.name == "heuristic": return { "agent": label, "kind": kind.value, "response": ( f"El proveedor LLM es heurístico; no puede responder en modo chat libre. " f"Configura OLLAMA o ANTHROPIC en el .env para habilitar este agente." ), "actor": actor, } self.store.add_audit("agent.chat", {"actor": actor, "agent": kind.value, "text": text[:500]}) trace_id = lf.start_trace( "agent_chat", input_data={"agent": kind.value, "text": text[:500], "actor": actor}, tags=["chat", kind.value], ) lf.set_current_trace_id(trace_id) try: raw = await self.provider.complete(system_prompt, text) lf.finish_trace(trace_id, output=raw[:500]) # Push SSE event para dashboard en tiempo real self.store.push_sse_event("agent.chat.response", { "agent": label, "kind": kind.value, "actor": actor, "text_preview": text[:120], }) except Exception as exc: lf.finish_trace(trace_id, status="ERROR") lf.set_current_trace_id(None) return { "agent": label, "kind": kind.value, "response": f"Error comunicando con el proveedor LLM: {exc}", "actor": actor, } finally: lf.set_current_trace_id(None) return { "agent": label, "kind": kind.value, "response": raw, "actor": actor, } def get_learning_state(self) -> dict[str, Any]: mode = self.get_learning_mode() summary = self.store.feedback_summary() return { "mode": mode, "summary": summary, "recent_feedback": self.store.list_feedback(limit=8), "recent_memories": self.store.list_memories(limit=6), "automation": { "proposal_execution_requires_approval": True, "study_memory_autopromote": mode == "auto", "semi_mode_requires_feedback_review": mode == "semi", }, } def set_learning_mode(self, mode: str, actor: str) -> dict[str, Any]: normalized = str(mode or "").strip().lower() if normalized not in {"manual", "semi", "auto"}: raise RuntimeError("Modo de aprendizaje inválido. Usa manual, semi o auto.") self.store.set_runtime_state("learning.mode", normalized) self.store.add_audit("learning.mode.updated", {"actor": actor, "mode": normalized}) return self.get_learning_state() def record_learning_feedback( self, actor: str, score: int, note: str = "", proposal_id: str | None = None, study_id: str | None = None, source: str = "telegram", ) -> dict[str, Any]: safe_score = max(1, min(int(score), 5)) proposal = self.store.get_proposal(proposal_id) if proposal_id else None resolved_study_id = study_id or (proposal.get("study_id") if proposal else None) label = self._feedback_label(safe_score) feedback = self.store.create_feedback( actor=actor, score=safe_score, label=label, note=note or f"Feedback {label}", source=source, proposal_id=proposal_id, study_id=resolved_study_id, ) self.store.add_audit( "learning.feedback.created", { "actor": actor, "proposal_id": proposal_id, "study_id": resolved_study_id, "score": safe_score, "label": label, "source": source, }, ) created_memory = None mode = self.get_learning_mode() if mode == "auto" and resolved_study_id and safe_score >= 4 and not self.store.has_memory_for_study(resolved_study_id): created_memory = self.store.create_memory_from_study( resolved_study_id, actor, note=(note or "") + "\n\nAutoaprendizaje: feedback positivo desde Telegram.", ) if created_memory is not None: self.store.add_audit( "learning.memory.autopromoted", {"actor": actor, "study_id": resolved_study_id, "feedback_id": feedback["id"]}, ) return { "mode": mode, "feedback": feedback, "memory": created_memory, "study_id": resolved_study_id, "proposal_id": proposal_id, } def _persist_plan(self, text: str, actor: str, channel: str, final_plan: dict[str, Any]) -> dict[str, Any]: study = self.store.create_study( request_text=text, summary=final_plan["summary"], analysis=final_plan["analysis"], recommended_agents=final_plan["recommended_agents"], actor=actor, channel=channel, ) proposals = [] for proposal in final_plan["proposals"]: created = self.store.create_proposal( study_id=study["id"], agent_name=proposal["agent_name"], title=proposal["title"], summary=proposal["summary"], action_type=proposal["action_type"], payload=proposal.get("payload", {}), risk_level=proposal.get("risk_level", "medium"), ) proposals.append(created) self.store.add_audit( "command.created", { "actor": actor, "channel": channel, "study_id": study["id"], "recommended_agents": study["recommended_agents"], }, ) return {"study": study, "proposals": proposals} def _heuristic_marketing_plan(self, brief: str, context: dict[str, Any], bridge_error: str | None = None) -> dict[str, Any]: overview = context.get("overview") if isinstance(context.get("overview"), dict) else {} metrics = overview.get("metrics") if isinstance(overview, dict) else {} raw_campaigns = context.get("campaigns") campaigns = [item for item in raw_campaigns if isinstance(item, dict)] if isinstance(raw_campaigns, list) else [] total_campaigns = int(metrics.get("campaigns_total", 0) or 0) if isinstance(metrics, dict) else 0 active_campaigns = int(metrics.get("campaigns_active", 0) or 0) if isinstance(metrics, dict) else 0 consents_active = int(metrics.get("consents_active", 0) or 0) if isinstance(metrics, dict) else 0 messages_24h = int(metrics.get("messages_24h", 0) or 0) if isinstance(metrics, dict) else 0 summary = ( f"Plan de marketing preparado con contexto de chatbot: {total_campaigns} campaña(s) detectadas, " f"{active_campaigns} activa(s), {consents_active} consentimiento(s) y {messages_24h} mensajes en 24h." ) if bridge_error: summary = "Plan de marketing preparado con contexto parcial: el bridge del chatbot no respondió y se ha usado una estrategia conservadora." analysis_parts = [] if bridge_error: analysis_parts.append( "No he podido leer el bridge del chatbot, así que la propuesta se apoya en el brief y en heurística segura." ) else: analysis_parts.append( "El supervisor ha leído métricas y campañas recientes del chatbot para orientar la preparación de acciones comerciales." ) analysis_parts.append( f"Contexto actual: campañas totales={total_campaigns}, activas={active_campaigns}, consentimientos={consents_active}, mensajes_24h={messages_24h}." ) if campaigns: analysis_parts.append( "Campañas recientes detectadas: " + ", ".join( f"{item.get('campaign_ref', 'n/a')} ({item.get('status', 'n/a')})" for item in campaigns[:5] ) ) analysis_parts.append( "La recomendación inicial es preparar segmentación, copy, priorización y criterio de aprobación antes de lanzar nada automáticamente." ) proposals = [ { "agent_name": "growth", "title": "Preparar acciones de marketing y campañas", "summary": "Definir segmentación, mensajes, hipótesis, secuencia de seguimiento y próximos pasos usando el contexto actual del chatbot.", "action_type": "campaign_plan", "payload": { "brief": brief, "source": "chatbot_bridge", "metrics_snapshot": metrics if isinstance(metrics, dict) else {}, "recent_campaign_refs": [ str(item.get("campaign_ref")) for item in campaigns[:8] if item.get("campaign_ref") ], "deliverables": [ "segmentación", "copy principal y variantes", "riesgos y consentimiento", "prioridades por impacto", ], }, "risk_level": "low", }, { "agent_name": "research", "title": "Validar riesgos y dependencias de la activación marketing", "summary": "Revisar consentimientos, créditos, canales activos y posibles bloqueos antes de pasar de planificación a ejecución.", "action_type": "research_brief", "payload": { "brief": brief, "context": { "campaigns_total": total_campaigns, "campaigns_active": active_campaigns, "consents_active": consents_active, "messages_24h": messages_24h, }, }, "risk_level": "low", }, ] return { "summary": summary, "analysis": " ".join(part for part in analysis_parts if part).strip(), "recommended_agents": ["growth", "research"], "proposals": proposals, } async def _maybe_enrich_marketing_with_llm( self, brief: str, fallback: dict[str, Any], context: dict[str, Any], bridge_error: str | None = None, ) -> dict[str, Any]: if self.provider.name == "heuristic": return fallback _s = get_settings() system_prompt = ( f"Eres el supervisor growth de '{_s.project_name}'. Devuelve solo JSON válido. " "No inventes shell ni ejecuciones directas. Propón únicamente planificación y revisión segura." ) user_prompt = ( "Prepara un estudio y propuestas de marketing/campañas usando el contexto disponible del chatbot.\n\n" f"Brief: {brief}\n\n" f"Contexto chatbot: {json.dumps(context, ensure_ascii=False)}\n\n" f"Error bridge (si existe): {bridge_error or 'ninguno'}\n\n" f"Fallback heurístico: {json.dumps(fallback, ensure_ascii=False)}\n\n" "Responde JSON con esta forma:\n" "{\n" ' "summary": "...",\n' ' "analysis": "...",\n' ' "recommended_agents": ["growth", "research"],\n' ' "proposals": [\n' " {\n" ' "agent_name": "growth|research|ops",\n' ' "title": "...",\n' ' "summary": "...",\n' ' "action_type": "campaign_plan|research_brief|ops_review",\n' ' "risk_level": "low|medium|high",\n' ' "payload": {}\n' " }\n" " ]\n" "}" ) try: raw = await self.provider.complete(system_prompt, user_prompt) parsed = extract_json_document(raw) if not parsed: return fallback proposals = self._normalize_proposals(parsed.get("proposals", []), fallback["proposals"]) proposals = [ proposal for proposal in proposals if proposal.get("action_type") in {"campaign_plan", "research_brief", "ops_review"} ] return { "summary": str(parsed.get("summary") or fallback["summary"]), "analysis": str(parsed.get("analysis") or fallback["analysis"]), "recommended_agents": self._normalize_agents(parsed.get("recommended_agents"), fallback["recommended_agents"]), "proposals": proposals or fallback["proposals"], } except Exception: return fallback def _heuristic_plan(self, text: str) -> dict[str, Any]: lowered = text.lower() recommended_agents: list[str] = [] proposals: list[dict[str, Any]] = [] ops_added = False for command_key, (keywords, title, summary, risk) in OPS_KEYWORDS.items(): if any(keyword in lowered for keyword in keywords): ops_added = True recommended_agents.append("ops") proposals.append( { "agent_name": "ops", "title": title, "summary": summary, "action_type": "ssh_command", "payload": { "profile": "siteground-prod", "command_key": command_key, }, "risk_level": risk, } ) break if not ops_added and any(token in lowered for token in ["ssh", "servidor", "siteground", "producción", "produccion", "deploy"]): recommended_agents.append("ops") proposals.append( { "agent_name": "ops", "title": "Revisión operativa manual", "summary": "La petición parece técnica, pero no la he podido mapear a una clave SSH permitida. Hace falta revisión humana.", "action_type": "ops_review", "payload": {"original_request": text}, "risk_level": "medium", } ) if any(keyword in lowered for keyword in GROWTH_WORDS): recommended_agents.append("growth") proposals.append( { "agent_name": "growth", "title": "Plan de captación y reactivación", "summary": "Preparar secuencia comercial, segmentación, mensajes y siguientes pasos para captar o reactivar anunciantes.", "action_type": "campaign_plan", "payload": { "brief": text, "deliverables": [ "segmentación", "argumentario", "secuencia de seguimiento", "prioridades por valor potencial", ], }, "risk_level": "low", } ) if any(keyword in lowered for keyword in RESEARCH_WORDS) or not proposals: recommended_agents.append("research") proposals.insert( 0, { "agent_name": "research", "title": "Estudio previo y mapa de decisiones", "summary": "Generar un estudio corto con riesgos, impacto, dependencias y propuesta de ejecución bajo supervisión.", "action_type": "research_brief", "payload": { "brief": text, }, "risk_level": "low", }, ) unique_agents = list(dict.fromkeys(recommended_agents)) summary = f"Petición clasificada para: {', '.join(unique_agents)}. Se ha generado un estudio inicial y {len(proposals)} propuesta(s) para revisión." analysis = ( "El supervisor ha traducido la orden a un flujo controlado: primero estudio, luego propuestas separadas por rol, y ejecución solo tras aprobación. " "Si una acción técnica no coincide con la allowlist SSH, queda en revisión manual en vez de improvisar shell arbitrario." ) return { "summary": summary, "analysis": analysis, "recommended_agents": unique_agents, "proposals": proposals, } async def _maybe_enrich_with_llm(self, text: str, fallback: dict[str, Any]) -> dict[str, Any]: if self.provider.name == "heuristic": return fallback # Detectar agente más adecuado y usar su prompt especializado agent_kind = route_to_agent(text) system_prompt = get_agent_system_prompt(agent_kind) agent_name = agent_label(agent_kind) allowed_commands = self.runner.list_allowed_commands() user_prompt = ( f"Eres el agente activo: {agent_name}.\n\n" "Convierte la orden del operador en un estudio y propuestas accionables.\n\n" f"Orden: {text}\n\n" f"Fallback heurístico de respaldo: {json.dumps(fallback, ensure_ascii=False)}\n\n" f"Command keys SSH permitidas: {json.dumps(allowed_commands, ensure_ascii=False)}\n\n" "Responde solo JSON con esta forma exacta:\n" "{\n" ' "summary": "...",\n' ' "analysis": "...",\n' ' "recommended_agents": ["ops", "growth", "research", "seo", "code", "security"],\n' ' "proposals": [\n' " {\n" ' "agent_name": "ops|growth|research|seo|code|security",\n' ' "title": "...",\n' ' "summary": "...",\n' ' "action_type": "ssh_command|campaign_plan|research_brief|ops_review|seo_audit|code_review|security_audit",\n' ' "risk_level": "low|medium|high",\n' ' "payload": {}\n' " }\n" " ]\n" "}" ) # ── Inyectar contexto RAG si está disponible ── # rag_context = await rag_store.query(text, top_k=4) if rag_context: user_prompt = ( f"Contexto de base de conocimiento (RAG):\n{rag_context}\n\n" + user_prompt ) try: raw = await self.provider.complete(system_prompt, user_prompt) parsed = extract_json_document(raw) parsed = extract_json_document(raw) if not parsed: return fallback proposals = self._normalize_proposals(parsed.get("proposals", []), fallback["proposals"]) return { "summary": str(parsed.get("summary") or fallback["summary"]), "analysis": str(parsed.get("analysis") or fallback["analysis"]), "recommended_agents": self._normalize_agents(parsed.get("recommended_agents"), fallback["recommended_agents"]), "proposals": proposals or fallback["proposals"], } except Exception: return fallback def _normalize_agents(self, candidate: Any, fallback: list[str]) -> list[str]: if not isinstance(candidate, list): return fallback cleaned = [str(item).strip().lower() for item in candidate if str(item).strip()] valid = [item for item in cleaned if item in {"research", "ops", "growth"}] return list(dict.fromkeys(valid)) or fallback def _normalize_proposals(self, candidate: Any, fallback: list[dict[str, Any]]) -> list[dict[str, Any]]: if not isinstance(candidate, list): return fallback allowed_commands = self.runner.list_allowed_commands() normalized: list[dict[str, Any]] = [] for item in candidate: if not isinstance(item, dict): continue agent_name = str(item.get("agent_name", "")).strip().lower() action_type = str(item.get("action_type", "research_brief")).strip() title = str(item.get("title", "Propuesta")).strip() summary = str(item.get("summary", "")).strip() or title risk_level = str(item.get("risk_level", "medium")).strip().lower() raw_payload = item.get("payload") payload: dict[str, Any] = raw_payload if isinstance(raw_payload, dict) else {} if agent_name not in {"research", "ops", "growth"}: continue if risk_level not in {"low", "medium", "high"}: risk_level = "medium" if action_type == "ssh_command": command_key = str(payload.get("command_key", "")).strip() if command_key not in allowed_commands: continue payload = { "profile": str(payload.get("profile", "siteground-prod")).strip() or "siteground-prod", "command_key": command_key, } normalized.append( { "agent_name": agent_name, "title": title, "summary": summary, "action_type": action_type, "payload": payload, "risk_level": risk_level, } ) return normalized def _feedback_label(self, score: int) -> str: if score >= 5: return "excelente" if score == 4: return "util" if score == 3: return "neutral" if score == 2: return "mejorable" return "descartar" async def approve_proposal(self, proposal_id: str, actor: str, note: str | None = None) -> dict[str, Any]: proposal = self.store.get_proposal(proposal_id) if proposal is None: raise KeyError("Propuesta no encontrada") updated = self.store.decide_proposal(proposal_id, "approved", actor, note) self.store.add_audit("proposal.approved", {"proposal_id": proposal_id, "actor": actor, "channel": "shared"}) if proposal.get("action_type") == "ssh_command": try: execution_log = await self.execute_proposal(updated or proposal) updated = self.store.set_proposal_execution(proposal_id, "executed", execution_log) self.store.add_audit("proposal.executed", {"proposal_id": proposal_id, "actor": actor, "channel": "shared"}) except Exception as exc: updated = self.store.set_proposal_execution(proposal_id, "failed", str(exc)) self.store.add_audit( "proposal.failed", {"proposal_id": proposal_id, "actor": actor, "channel": "shared", "error": str(exc)}, ) if updated is None: raise RuntimeError("No se pudo actualizar la propuesta") return updated def reject_proposal(self, proposal_id: str, actor: str, note: str | None = None) -> dict[str, Any]: proposal = self.store.get_proposal(proposal_id) if proposal is None: raise KeyError("Propuesta no encontrada") updated = self.store.decide_proposal(proposal_id, "rejected", actor, note) self.store.add_audit("proposal.rejected", {"proposal_id": proposal_id, "actor": actor, "channel": "shared"}) if updated is None: raise RuntimeError("No se pudo actualizar la propuesta") return updated def memorize_study(self, study_id: str, actor: str, note: str | None = None) -> dict[str, Any]: memory = self.store.create_memory_from_study(study_id, actor, note) if memory is None: raise KeyError("Estudio no encontrado") self.store.add_audit("study.memorized", {"study_id": study_id, "actor": actor, "channel": "shared"}) return memory async def execute_proposal(self, proposal: dict[str, Any]) -> str: action_type = proposal.get("action_type") if action_type == "ssh_command": payload = proposal.get("payload") or {} command_key = payload.get("command_key") profile = payload.get("profile") if not command_key: raise RuntimeError("La propuesta SSH no incluye command_key") return await self.runner.execute(str(command_key), str(profile) if profile else None) if action_type == "whatsapp_campaign": payload = proposal.get("payload") or {} contacts = payload.get("contacts") or [] template_name = payload.get("template_name", "") contacts_count = len(contacts) return ( f"Campaña WhatsApp lista para envío: {contacts_count} contactos, plantilla '{template_name}'.\n" f"Ejecuta POST /api/whatsapp/campaign/{proposal.get('id')}/send para iniciar el envío." ) return "Propuesta aprobada. No requiere ejecución automática adicional."