Upload 6 files

agents/specialized/analyst.py

@@ -0,0 +1,19 @@
+# agents/specialized/analyst.py
+from agents.specialized.base import build_result, call_llm
+
+ROLE = """
+Eres analista de negocios. REGLAS:
+1. Solo haz lo que el manager delegó: revisar documentos, evaluar viabilidad, analizar riesgos.
+2. NUNCA describas imágenes ni hagas trabajo de otros agentes.
+3. Si la tarea no requiere análisis → responde: {"skip":"no analysis needed"}
+"""
+
+
+async def run(task: str, context: dict = None) -> dict:
+    result = build_result("analyst")
+    try:
+        result["response"] = await call_llm("analyst", ROLE, task, context)
+    except Exception as e:
+        result["success"] = False
+        result["error"] = str(e)
+    return result

agents/specialized/backend_dev.py

@@ -0,0 +1,41 @@
+# agents/specialized/backend_dev.py
+import re
+import json
+from agents.specialized.base import build_result, call_llm
+from core.file_builder import build_excel
+
+ROLE = """
+Eres programador backend senior. REGLAS ABSOLUTAS:
+1. Entrega SOLO el código pedido, sin explicaciones innecesarias.
+2. Para excel/planilla → responde con EXCEL_TEMPLATE:{"title":"...","sheet_name":"...","headers":[...],"sample_rows":[[...],[...]]}
+3. Python → entrega código Python puro y funcional.
+4. Groovy/Jenkins → entrega el script completo.
+5. Si hay frontend_dev en el equipo, TÚ haces servidor/backend, él hace HTML.
+6. Si la tarea no requiere backend → responde: {"skip":"no backend needed"}
+"""
+
+
+async def run(task: str, context: dict = None) -> dict:
+    result = build_result("backend_dev")
+    try:
+        response = await call_llm("backend_dev", ROLE, task, context)
+        result["response"] = response
+
+        # Detectar si el agente generó un EXCEL_TEMPLATE
+        match = re.search(r'EXCEL_TEMPLATE:\s*(\{.*\})', response, re.DOTALL | re.IGNORECASE)
+        if match:
+            try:
+                excel_data = json.loads(match.group(1))
+                file_path = build_excel(task, excel_data)
+                if file_path:
+                    result["file_path"] = file_path.name
+                    result["file_type"] = "xlsx"
+                    result["response"] = response.split("EXCEL_TEMPLATE:")[0].strip()
+            except Exception as e:
+                result["file_error"] = str(e)
+
+    except Exception as e:
+        result["success"] = False
+        result["error"] = str(e)
+
+    return result

agents/specialized/base.py

@@ -0,0 +1,30 @@
+# agents/specialized/base.py
+from core.llm_client import llm
+from utils.prompts import get_today_prefix, get_chat_style
+
+
+def build_result(agent_key: str) -> dict:
+    """Estructura base de resultado para cualquier agente."""
+    return {
+        "agent": agent_key,
+        "response": "",
+        "success": True,
+        "file_path": None,
+        "file_type": None,
+        "image_urls": [],
+        "queries": None,
+    }
+
+
+async def call_llm(agent_key: str, role: str, task: str, context: dict = None) -> str:
+    """Llama al LLM con el rol, tarea y contexto dados."""
+    today = get_today_prefix()
+    full_prompt = today + role + get_chat_style() + f"\n\nTarea asignada: {task}"
+
+    if context:
+        full_prompt += "\n\nContexto de otros agentes:\n" + "\n".join(
+            [f"[{k.upper()}] {v[:300]}..." for k, v in context.items()]
+        )
+
+    messages = [{"role": "user", "content": full_prompt}]
+    return await llm.call(agent_key, messages, temperature=0.45)

agents/specialized/frontend_dev.py

@@ -0,0 +1,20 @@
+# agents/specialized/frontend_dev.py
+from agents.specialized.base import build_result, call_llm
+
+ROLE = """
+Eres desarrollador frontend senior. REGLAS ABSOLUTAS:
+1. Entrega SOLO código HTML/CSS/JS pedido, sin explicaciones innecesarias.
+2. Si hay backend_dev, TÚ haces HTML/interfaz, él hace servidor/lógica.
+3. Si la tarea NO requiere frontend → responde: {"skip":"no frontend needed"}
+4. Entrega siempre HTML completo y funcional con los estilos incluidos.
+"""
+
+
+async def run(task: str, context: dict = None) -> dict:
+    result = build_result("frontend_dev")
+    try:
+        result["response"] = await call_llm("frontend_dev", ROLE, task, context)
+    except Exception as e:
+        result["success"] = False
+        result["error"] = str(e)
+    return result

agents/specialized/image_agent.py

@@ -0,0 +1,122 @@
+# agents/specialized/image_agent.py
+import os
+import re
+import json
+import base64
+import requests
+from pathlib import Path
+from datetime import datetime
+from agents.specialized.base import build_result, call_llm
+from agents.registry import AGENT_REGISTRY
+
+ROLE = """
+Eres el agente de imágenes.
+Cuando se te pida generar, crear, mostrar o visualizar una imagen, responde SOLO con un JSON:
+{"image_queries": ["término en inglés 1", "término en inglés 2", "término en inglés 3"]}
+Los términos deben ser específicos, detallados y en inglés para obtener mejores resultados.
+Si la tarea no es sobre imágenes, responde: {"skip":"no image task"}
+"""
+
+DOCS_DIR = Path("data/docs")
+
+
+async def generate_with_gemini(queries: list, model_name: str) -> list:
+    # ⚠️ pendiente de implementar con Gemini imagen real
+    return []
+
+
+async def generate_with_hf(queries: list) -> list:
+    urls = []
+    for prompt in queries[:2]:
+        response = requests.post(
+            "https://router.huggingface.co/v1/images/generations",
+            headers={
+                "Authorization": f"Bearer {os.getenv('HF_API_TOKEN')}",
+                "Content-Type": "application/json",
+            },
+            json={
+                "model": "black-forest-labs/FLUX.1-schnell",
+                "prompt": prompt,
+                "size": "1024x1024",
+            },
+        )
+
+        if response.status_code == 200:
+            data = response.json()
+
+            # Algunos modelos devuelven b64_json, otros url directa
+            item = data["data"][0]
+            if "url" in item:
+                urls.append(item["url"])
+            elif "b64_json" in item:
+                image_bytes = base64.b64decode(item["b64_json"])
+                DOCS_DIR.mkdir(parents=True, exist_ok=True)
+                file_name = f"hf_image_{datetime.now().timestamp()}.png"
+                file_path = DOCS_DIR / file_name
+                with open(file_path, "wb") as f:
+                    f.write(image_bytes)
+                urls.append(f"/docs/{file_name}")
+        else:
+            print("HF image error:", response.text)
+
+    return urls
+
+
+async def run(task: str, context: dict = None) -> dict:
+    result = build_result("image_agent")
+    try:
+        response = await call_llm("image_agent", ROLE, task, context)
+        result["response"] = response
+
+        agent_config = AGENT_REGISTRY.get("image_agent", {})
+        provider = agent_config.get("provider")
+        model_name = agent_config.get("models", [None])[0]
+
+        print("USANDO PROVIDER:", provider)
+        print("USANDO MODELO:", model_name)
+
+        match = re.search(r'\{.*"image_queries".*\}', response, re.DOTALL)
+
+        if match:
+            try:
+                data = json.loads(match.group(0))
+                queries = data.get("image_queries", [])
+                result["queries"] = queries
+                result["response"] = "Ideas de imágenes generadas:\n" + "\n".join(queries)
+
+                image_urls = []
+
+                # Intento 1: provider principal
+                try:
+                    if provider == "gemini":
+                        image_urls = await generate_with_gemini(queries, model_name)
+                    elif provider == "huggingface":
+                        image_urls = await generate_with_hf(queries)
+                except Exception as e:
+                    print("Provider principal falló:", str(e))
+
+                # Fallback a HF
+                if not image_urls:
+                    try:
+                        print("Intentando fallback HF...")
+                        image_urls = await generate_with_hf(queries)
+                    except Exception as e:
+                        print("Fallback HF falló:", str(e))
+
+                # Fallback visual final
+                if not image_urls:
+                    image_urls = ["https://picsum.photos/400/300"]
+
+                result["image_urls"] = image_urls
+                result["response"] += "\n\nImagen generada ✔" if image_urls else "\n\n⚠️ No se pudo generar imagen"
+
+            except Exception as e:
+                result["response"] = f"Error procesando JSON: {str(e)}"
+        else:
+            result["response"] = "No se detectaron instrucciones claras para imágenes."
+
+    except Exception as e:
+        result["success"] = False
+        result["error"] = str(e)
+
+    return result

agents/specialized/writer.py

@@ -0,0 +1,34 @@
+# agents/specialized/writer.py
+from agents.specialized.base import build_result, call_llm
+from core.file_builder import build_docx
+
+ROLE = """
+Eres redactor experto. Escribe SOLO contenido real y extenso (500+ palabras).
+Sin placeholders. Usa ## para secciones y ### para subsecciones.
+Secciones: ## Resumen Ejecutivo, ### Introducción, ### Desarrollo,
+### Hallazgos, ### Conclusiones, ### Recomendaciones
+"""
+
+
+async def run(task: str, context: dict = None) -> dict:
+    result = build_result("writer")
+    try:
+        response = await call_llm("writer", ROLE, task, context)
+        result["response"] = response
+
+        # Si la respuesta es suficientemente larga, generar un .docx
+        if len(response.strip()) > 300:
+            title = task[:60].strip('"') or "Informe generado"
+            try:
+                file_path = build_docx(title, response, images=None)
+                if file_path:
+                    result["file_path"] = file_path.name
+                    result["file_type"] = "docx"
+            except Exception as e:
+                result["file_error"] = str(e)
+
+    except Exception as e:
+        result["success"] = False
+        result["error"] = str(e)
+
+    return result