Spaces:
Sleeping
Sleeping
| #!/usr/bin/env python | |
| # coding: utf-8 | |
| # # 🤖 StratGen AGENT v4 — Azure OpenAI Multi-Model | |
| # | |
| # > **Arquitectura Azure**: Todos los modelos desde Azure OpenAI · Sin dependencia de Ollama local | |
| # | |
| # ``` | |
| # gpt-4o-2024-11-20 │ Planificador — JSON mode · Orquestador principal | |
| # gpt-4o-2024-08-06 │ Analista — Análisis de mercado (Paso 2) | |
| # gpt-4o-2024-11-20 │ Estratega — Consolidación estratégica (Paso 3) | |
| # gpt-4o-mini-2024-07-18│ Generador — Iniciativas accionables (Paso 4) | |
| # ``` | |
| # | |
| # **Antes de empezar:** crea un archivo `.env` en el mismo directorio con: | |
| # ``` | |
| # AZURE_OPENAI_KEY=<tu_key> | |
| # AZURE_OPENAI_ENDPOINT=https://<tu-recurso>.openai.azure.com/ | |
| # AZURE_OPENAI_API_VERSION=2024-10-21 | |
| # ``` | |
| # | |
| # ## 📦 Celda 1 — Instalación | |
| # In[1]: | |
| #!pip install openai gradio python-docx python-dotenv -q | |
| # ## ⚙️ Celda 2 — Clientes Azure OpenAI y modelos | |
| # In[2]: | |
| from openai import AzureOpenAI | |
| from dotenv import load_dotenv | |
| import json, os, re, io, tempfile, unicodedata | |
| from datetime import datetime | |
| from dataclasses import dataclass, field | |
| from typing import List | |
| load_dotenv() | |
| AZURE_KEY = os.getenv("AZURE_OPENAI_KEY") | |
| AZURE_ENDPOINT = os.getenv("AZURE_OPENAI_ENDPOINT") | |
| AZURE_VERSION = os.getenv("AZURE_OPENAI_API_VERSION", "2024-10-21") | |
| if not AZURE_KEY: | |
| print("⚠️ AZURE_OPENAI_KEY no encontrada en .env") | |
| elif not AZURE_ENDPOINT: | |
| print("⚠️ AZURE_OPENAI_ENDPOINT no encontrada en .env") | |
| else: | |
| print(f"✅ Azure OpenAI key cargada ({AZURE_KEY[:8]}...)") | |
| print(f"✅ Endpoint: {AZURE_ENDPOINT}") | |
| print(f"✅ API Version: {AZURE_VERSION}") | |
| # Cliente único Azure OpenAI (todos los modelos pasan por aquí) | |
| client_azure = AzureOpenAI( | |
| api_key=AZURE_KEY, | |
| api_version=AZURE_VERSION, | |
| azure_endpoint=AZURE_ENDPOINT, | |
| ) | |
| # ─── Asignación de modelos por rol ─────────────────────────────────────────── | |
| # 2 deployments activos: gpt-4o (planner+estratega) y gpt-4.1-mini (analista+generador) | |
| MODELS = { | |
| "planner": "gpt-4o", # Orquestador principal — JSON mode | |
| "analyst": "gpt-4.1-mini", # Análisis de mercado — rápido y económico | |
| "strategist": "gpt-4o", # Consolidación — paso más crítico | |
| "initiator": "gpt-4.1-mini", # Iniciativas — velocidad sobre potencia | |
| } | |
| def llamar_azure(deployment: str, messages: list, | |
| temperature: float = 0.4, | |
| json_mode: bool = False) -> str: | |
| """Llamada genérica a Azure OpenAI. json_mode=True activa response_format JSON.""" | |
| try: | |
| kwargs = dict( | |
| model=deployment, | |
| messages=messages, | |
| temperature=temperature, | |
| max_tokens=2500, | |
| ) | |
| if json_mode: | |
| kwargs["response_format"] = {"type": "json_object"} | |
| resp = client_azure.chat.completions.create(**kwargs) | |
| return resp.choices[0].message.content.strip() | |
| except Exception as e: | |
| if json_mode: | |
| return json.dumps({"tool": "ask_user", "args": {"question": f"Error Azure: {str(e)[:80]}"}}) | |
| return f"⚠️ Error Azure ({deployment}): {str(e)}" | |
| # Función específica para el planificador (JSON mode) | |
| def llamar_planner(messages: list, temperature: float = 0.1) -> str: | |
| return llamar_azure(MODELS["planner"], messages, temperature, json_mode=True) | |
| # Función específica para modelos especialistas (texto libre) | |
| def llamar_especialista(rol: str, messages: list, temperature: float = 0.5) -> str: | |
| return llamar_azure(MODELS[rol], messages, temperature, json_mode=False) | |
| print(f"\n✅ Planificador : {MODELS['planner']} (Azure — JSON mode)") | |
| print(f"✅ Analista : {MODELS['analyst']} (Azure)") | |
| print(f"✅ Estratega : {MODELS['strategist']} (Azure)") | |
| print(f"✅ Generador : {MODELS['initiator']} (Azure)") | |
| # ## 🗄️ Celda 3 — Modelo de datos | |
| # In[3]: | |
| class AreaOrganizacional: | |
| name: str = "" | |
| objectives: str = "" | |
| challenges: str = "" | |
| def is_complete(self) -> bool: | |
| return bool(self.name and self.objectives and self.challenges) | |
| class ProyectoEstrategico: | |
| company_name: str = "" | |
| areas: List[AreaOrganizacional] = field(default_factory=list) | |
| market_analysis: str = "" | |
| consolidated_strategy: str = "" | |
| strategic_initiatives: str = "" | |
| step1_complete: bool = False | |
| step2_complete: bool = False | |
| step3_complete: bool = False | |
| step4_complete: bool = False | |
| def current_step(self) -> int: | |
| if not self.step1_complete: return 1 | |
| if not self.step2_complete: return 2 | |
| if not self.step3_complete: return 3 | |
| if not self.step4_complete: return 4 | |
| return 5 | |
| def progress_pct(self) -> int: | |
| return (self.current_step() - 1) * 25 | |
| def areas_completas(self) -> bool: | |
| return len(self.areas) >= 3 and all(a.is_complete() for a in self.areas) | |
| def state_for_planner(self) -> str: | |
| areas_detail = [ | |
| { | |
| "index": i, | |
| "name": a.name, | |
| "has_objectives": bool(a.objectives), | |
| "has_challenges": bool(a.challenges), | |
| "complete": a.is_complete(), | |
| } | |
| for i, a in enumerate(self.areas) | |
| ] | |
| return json.dumps({ | |
| "company_name": self.company_name or None, | |
| "areas": areas_detail, | |
| "step1_complete": self.step1_complete, | |
| "step2_complete": self.step2_complete, | |
| "step3_complete": self.step3_complete, | |
| "step4_complete": self.step4_complete, | |
| "all_done": self.current_step() == 5, | |
| }, ensure_ascii=False) | |
| def proyecto_a_dict(p: ProyectoEstrategico, existing: dict = None) -> dict: | |
| d = { | |
| "company_name": p.company_name, | |
| "areas": [a.__dict__.copy() for a in p.areas], | |
| "market_analysis": p.market_analysis, | |
| "consolidated_strategy": p.consolidated_strategy, | |
| "strategic_initiatives": p.strategic_initiatives, | |
| "step1_complete": p.step1_complete, | |
| "step2_complete": p.step2_complete, | |
| "step3_complete": p.step3_complete, | |
| "step4_complete": p.step4_complete, | |
| "doc_p1": (existing or {}).get("doc_p1"), | |
| "doc_p2": (existing or {}).get("doc_p2"), | |
| "doc_p3": (existing or {}).get("doc_p3"), | |
| "doc_p4": (existing or {}).get("doc_p4"), | |
| } | |
| for k in ("doc_p1", "doc_p2", "doc_p3", "doc_p4"): | |
| if (existing or {}).get(k): | |
| d[k] = existing[k] | |
| return d | |
| def dict_a_proyecto(d: dict) -> ProyectoEstrategico: | |
| return ProyectoEstrategico( | |
| company_name = d.get("company_name", ""), | |
| areas = [AreaOrganizacional(**a) for a in d.get("areas", [])], | |
| market_analysis = d.get("market_analysis", ""), | |
| consolidated_strategy = d.get("consolidated_strategy", ""), | |
| strategic_initiatives = d.get("strategic_initiatives", ""), | |
| step1_complete = d.get("step1_complete", False), | |
| step2_complete = d.get("step2_complete", False), | |
| step3_complete = d.get("step3_complete", False), | |
| step4_complete = d.get("step4_complete", False), | |
| ) | |
| print("✅ Modelo de datos configurado") | |
| # ## 🔧 Celda 4 — Herramientas del agente | |
| # In[4]: | |
| class ToolResult: | |
| def __init__(self, success: bool, message: str, data: dict = None): | |
| self.success = success | |
| self.message = message | |
| self.data = data or {} | |
| def __str__(self): return self.message | |
| def tool_ask_user(question: str, proyecto) -> ToolResult: | |
| if not question.strip(): | |
| return ToolResult(False, "Pregunta vacía.") | |
| return ToolResult(True, question.strip(), {"type": "ask_user"}) | |
| def tool_save_company(name: str, proyecto) -> ToolResult: | |
| name = name.strip() | |
| if not name or len(name) < 2: | |
| return ToolResult(False, "Nombre de empresa inválido.") | |
| proyecto.company_name = name | |
| return ToolResult(True, | |
| f"✅ Empresa registrada: **{name}**\n\n" | |
| "Ahora necesito las áreas organizativas (mínimo 3)." | |
| ) | |
| def tool_save_areas(areas_raw: str, proyecto) -> ToolResult: | |
| partes = re.split(r"[,;/|\n]|\s+[yY]\s+", areas_raw) | |
| areas = [p.strip() for p in partes if p.strip() and len(p.strip()) > 1] | |
| seen, unique = set(), [] | |
| for a in areas: | |
| k = a.lower() | |
| if k not in seen: | |
| seen.add(k) | |
| unique.append(a) | |
| if len(unique) < 3: | |
| return ToolResult(False, | |
| f"Necesito al menos 3 áreas. Has proporcionado {len(unique)}: " | |
| f"{unique}. Por favor añade más." | |
| ) | |
| proyecto.areas = [AreaOrganizacional(name=a) for a in unique] | |
| lista = "\n".join(f" [{i}] {a}" for i, a in enumerate(unique)) | |
| return ToolResult(True, | |
| f"✅ {len(unique)} áreas registradas:\n{lista}\n\n" | |
| f"Empecemos con **[0] {unique[0]}** — ¿cuáles son sus objetivos principales?" | |
| ) | |
| def tool_save_area_detail(area_index: int, objectives: str, | |
| challenges: str, proyecto) -> ToolResult: | |
| objectives = objectives.strip() | |
| challenges = challenges.strip() | |
| if not (0 <= area_index < len(proyecto.areas)): | |
| indices = [f"[{i}] {a.name}" for i, a in enumerate(proyecto.areas)] | |
| return ToolResult(False, f"Índice {area_index} inválido. Áreas: {indices}") | |
| target = proyecto.areas[area_index] | |
| if objectives: | |
| target.objectives = objectives | |
| if challenges: | |
| target.challenges = challenges | |
| completas = sum(1 for a in proyecto.areas if a.is_complete()) | |
| pendientes = [a for a in proyecto.areas if not a.is_complete()] | |
| if proyecto.areas_completas(): | |
| return ToolResult(True, | |
| f"✅ **{target.name}** completada.\n\n" | |
| f"**Todas las áreas listas** ({len(proyecto.areas)}/{len(proyecto.areas)}).\n\n" | |
| "Escribe **confirmar** para lanzar el análisis completo.", | |
| {"all_complete": True} | |
| ) | |
| sig = pendientes[0] | |
| sig_idx = proyecto.areas.index(sig) | |
| falta = "objetivos" if not sig.objectives else "desafíos" | |
| return ToolResult(True, | |
| f"✅ **{target.name}** actualizada ({completas}/{len(proyecto.areas)} completas).\n\n" | |
| f"Siguiente → **[{sig_idx}] {sig.name}**: ¿cuáles son sus {falta}?" | |
| ) | |
| def tool_confirm_step1(proyecto) -> ToolResult: | |
| if not proyecto.company_name: | |
| return ToolResult(False, "Falta el nombre de la empresa.") | |
| if not proyecto.areas_completas(): | |
| incompletas = [ | |
| f"[{i}] {a.name}" for i, a in enumerate(proyecto.areas) | |
| if not a.is_complete() | |
| ] | |
| return ToolResult(False, | |
| f"Áreas incompletas: {incompletas}. Completa objetivos y desafíos primero." | |
| ) | |
| proyecto.step1_complete = True | |
| return ToolResult(True, | |
| "✅ **Diagnóstico confirmado.**\n\n" | |
| "⏳ Iniciando análisis automático (Pasos 2-3-4)...", | |
| {"trigger_auto": True} | |
| ) | |
| print("✅ Tools: ask_user / save_company / save_areas / save_area_detail / confirm_step1") | |
| # ## 🧠 Celda 5 — Modelos especialistas Azure (Pasos 2-4) | |
| # In[5]: | |
| def ejecutar_analisis_mercado(proyecto) -> str: | |
| areas_str = "\n".join( | |
| f"- {a.name}: Objetivos: {a.objectives} | Desafíos: {a.challenges}" | |
| for a in proyecto.areas | |
| ) | |
| print(f" → Llamando a {MODELS['analyst']} para análisis de mercado...") | |
| return llamar_especialista("analyst", [ | |
| {"role": "system", "content": ( | |
| "Eres un analista estratégico senior. Responde SIEMPRE en español.\n" | |
| "Genera SOLO el análisis. Sin saludos ni meta-comentarios." | |
| )}, | |
| {"role": "user", "content": ( | |
| f"Empresa: {proyecto.company_name}\n\nDiagnóstico:\n{areas_str}\n\n" | |
| "Genera el análisis con:\n" | |
| "1. CONTEXTO SECTORIAL\n2. TENDENCIAS GLOBALES\n" | |
| "3. COMPARACIÓN INTERNO vs. EXTERNO\n4. INSIGHTS ESTRATÉGICOS (mín. 4)\n" | |
| "5. OPORTUNIDADES Y AMENAZAS" | |
| )}, | |
| ], temperature=0.5) | |
| def ejecutar_consolidacion(proyecto) -> str: | |
| areas_str = "\n".join( | |
| f"- {a.name}: {a.objectives} / {a.challenges}" for a in proyecto.areas | |
| ) | |
| print(f" → Llamando a {MODELS['strategist']} para consolidación...") | |
| return llamar_especialista("strategist", [ | |
| {"role": "system", "content": ( | |
| "Eres un consultor estratégico senior. Responde SIEMPRE en español.\n" | |
| "Genera SOLO la consolidación. Sin saludos ni meta-comentarios." | |
| )}, | |
| {"role": "user", "content": ( | |
| f"Empresa: {proyecto.company_name}\n\nDIAGNÓSTICO:\n{areas_str}\n\n" | |
| f"ANÁLISIS DE MERCADO:\n{proyecto.market_analysis}\n\n" | |
| "Genera la consolidación con:\n" | |
| "1. TOP 3-5 PRIORIDADES ESTRATÉGICAS\n2. GAPS ORGANIZACIONALES\n" | |
| "3. OPORTUNIDADES ALINEADAS\n4. DIRECCIÓN ESTRATÉGICA RECOMENDADA" | |
| )}, | |
| ], temperature=0.4) | |
| def ejecutar_iniciativas(proyecto) -> str: | |
| print(f" → Llamando a {MODELS['initiator']} para iniciativas...") | |
| return llamar_especialista("initiator", [ | |
| {"role": "system", "content": ( | |
| "Eres un consultor estratégico senior. Responde SIEMPRE en español.\n" | |
| "Genera SOLO las iniciativas. Sin saludos ni meta-comentarios." | |
| )}, | |
| {"role": "user", "content": ( | |
| f"Empresa: {proyecto.company_name}\n\n" | |
| f"ESTRATEGIA:\n{proyecto.consolidated_strategy}\n\n" | |
| f"ÁREAS: {', '.join(a.name for a in proyecto.areas)}\n\n" | |
| "Genera 5-8 iniciativas. Para cada una:\n" | |
| "INICIATIVA [N]: [NOMBRE]\nDescripción: ...\nJustificación: ...\n" | |
| "Impacto esperado: ...\nHorizonte: Corto/Medio/Largo\nÁrea(s): ..." | |
| )}, | |
| ], temperature=0.6) | |
| def ejecutar_pasos_automaticos(proyecto, estado: dict) -> list: | |
| msgs = [] | |
| msgs.append(f"⏳ **Paso 2** — Análisis de mercado ({MODELS['analyst']})...") | |
| proyecto.market_analysis = ejecutar_analisis_mercado(proyecto) | |
| proyecto.step2_complete = True | |
| estado["doc_p2"] = generar_docx_paso2(proyecto) | |
| msgs.append("✅ Análisis de mercado completado.") | |
| msgs.append(f"⏳ **Paso 3** — Consolidación estratégica ({MODELS['strategist']})...") | |
| proyecto.consolidated_strategy = ejecutar_consolidacion(proyecto) | |
| proyecto.step3_complete = True | |
| estado["doc_p3"] = generar_docx_paso3(proyecto) | |
| msgs.append("✅ Consolidación estratégica completada.") | |
| msgs.append(f"⏳ **Paso 4** — Iniciativas ({MODELS['initiator']})...") | |
| proyecto.strategic_initiatives = ejecutar_iniciativas(proyecto) | |
| proyecto.step4_complete = True | |
| estado["doc_p4"] = generar_docx_paso4(proyecto) | |
| msgs.append("✅ Iniciativas generadas.") | |
| return msgs | |
| print("✅ Modelos especialistas Azure configurados") | |
| # ## 📄 Celda 6 — Generadores .docx in-memory | |
| # In[6]: | |
| from docx import Document as DocxDocument | |
| from docx.shared import Pt, RGBColor | |
| from docx.enum.text import WD_ALIGN_PARAGRAPH | |
| def _doc_bytes(doc) -> bytes: | |
| buf = io.BytesIO(); doc.save(buf); return buf.getvalue() | |
| def _h(doc, txt, lvl=1): | |
| p = doc.add_heading(txt, level=lvl) | |
| for r in p.runs: r.font.color.rgb = RGBColor(0x1F, 0x45, 0x7C) | |
| def _p(doc, txt, bold=False): | |
| p = doc.add_paragraph(); r = p.add_run(txt) | |
| r.bold = bold; r.font.size = Pt(11) | |
| def _pie(doc, empresa, titulo): | |
| s = doc.sections[0] | |
| h = s.header.paragraphs[0] | |
| h.text = f"{empresa} — {titulo}" | |
| h.alignment = WD_ALIGN_PARAGRAPH.RIGHT | |
| if h.runs: h.runs[0].font.size = Pt(9) | |
| f = s.footer.paragraphs[0] | |
| f.text = f"StratGen Agent v4 | Azure OpenAI | {datetime.now().strftime('%d/%m/%Y')}" | |
| f.alignment = WD_ALIGN_PARAGRAPH.CENTER | |
| if f.runs: f.runs[0].font.size = Pt(8) | |
| def generar_docx_paso1(p) -> tuple: | |
| doc = DocxDocument() | |
| n = p.company_name.replace(" ", "_") | |
| _pie(doc, p.company_name, "Diagnóstico Organizacional") | |
| _h(doc, "DIAGNÓSTICO ORGANIZACIONAL"); _h(doc, p.company_name, 2) | |
| doc.add_paragraph(f"Fecha: {datetime.now().strftime('%d de %B de %Y')}") | |
| doc.add_paragraph() | |
| _h(doc, "1. RESUMEN EJECUTIVO", 2) | |
| _p(doc, f"Diagnóstico de {p.company_name}: {len(p.areas)} áreas analizadas.") | |
| doc.add_paragraph() | |
| _h(doc, "2. ÁREAS ANALIZADAS", 2) | |
| for i, a in enumerate(p.areas, 1): | |
| _h(doc, f"2.{i} {a.name}", 3) | |
| _p(doc, "Objetivos:", bold=True); _p(doc, a.objectives) | |
| _p(doc, "Desafíos:", bold=True); _p(doc, a.challenges) | |
| doc.add_paragraph() | |
| _h(doc, "3. TABLA RESUMEN", 2) | |
| t = doc.add_table(rows=1, cols=3); t.style = "Table Grid" | |
| for txt, c in zip(["Área", "Objetivos", "Desafíos"], t.rows[0].cells): | |
| c.text = txt | |
| if c.paragraphs[0].runs: c.paragraphs[0].runs[0].bold = True | |
| for a in p.areas: | |
| r = t.add_row().cells | |
| r[0].text = a.name; r[1].text = a.objectives[:200]; r[2].text = a.challenges[:200] | |
| return _doc_bytes(doc), f"StratGen_P1_Diagnostico_{n}.docx" | |
| def generar_docx_paso2(p) -> tuple: | |
| doc = DocxDocument(); n = p.company_name.replace(" ", "_") | |
| _pie(doc, p.company_name, "Análisis de Mercado") | |
| _h(doc, "ANÁLISIS DE MERCADO"); _h(doc, p.company_name, 2) | |
| doc.add_paragraph(f"Fecha: {datetime.now().strftime('%d de %B de %Y')}") | |
| doc.add_paragraph(); _h(doc, "ANÁLISIS", 2); _p(doc, p.market_analysis) | |
| return _doc_bytes(doc), f"StratGen_P2_Mercado_{n}.docx" | |
| def generar_docx_paso3(p) -> tuple: | |
| doc = DocxDocument(); n = p.company_name.replace(" ", "_") | |
| _pie(doc, p.company_name, "Estrategia Consolidada") | |
| _h(doc, "ESTRATEGIA CONSOLIDADA"); _h(doc, p.company_name, 2) | |
| doc.add_paragraph(f"Fecha: {datetime.now().strftime('%d de %B de %Y')}") | |
| doc.add_paragraph(); _h(doc, "SÍNTESIS", 2); _p(doc, p.consolidated_strategy) | |
| return _doc_bytes(doc), f"StratGen_P3_Estrategia_{n}.docx" | |
| def generar_docx_paso4(p) -> tuple: | |
| doc = DocxDocument(); n = p.company_name.replace(" ", "_") | |
| _pie(doc, p.company_name, "Plan de Iniciativas") | |
| _h(doc, "PLAN DE INICIATIVAS"); _h(doc, p.company_name, 2) | |
| doc.add_paragraph(f"Fecha: {datetime.now().strftime('%d de %B de %Y')}") | |
| doc.add_paragraph(); _h(doc, "INICIATIVAS", 2); _p(doc, p.strategic_initiatives) | |
| return _doc_bytes(doc), f"StratGen_P4_Iniciativas_{n}.docx" | |
| print("✅ Generadores .docx in-memory configurados") | |
| # ## 🧠 Celda 7 — Planner Azure (gpt-4o-2024-11-20) + Orquestador | |
| # | |
| # El planificador usa `response_format=json_object` sobre Azure OpenAI. | |
| # Las áreas se identifican por **índice numérico** para evitar problemas con acentos/encoding. | |
| # In[7]: | |
| PLANNER_SYSTEM = ( | |
| "Eres el planificador de StratGen, un asistente de consultoría estratégica.\n" | |
| "Tu salida debe ser SIEMPRE un JSON válido con esta estructura:\n" | |
| '{"tool": "<nombre>", "args": {<argumentos>}}\n\n' | |
| "HERRAMIENTAS DISPONIBLES:\n\n" | |
| "ask_user\n" | |
| ' args: {"question": "texto"}\n' | |
| " Úsala cuando necesites información que el usuario no ha dado.\n\n" | |
| "save_company\n" | |
| ' args: {"name": "nombre exacto"}\n' | |
| " Copia el nombre EXACTAMENTE como lo escribió el usuario. No corrijas.\n\n" | |
| "save_areas\n" | |
| ' args: {"areas_raw": "Area1, Area2, Area3"}\n' | |
| " Cuando el usuario liste las áreas (mínimo 3).\n\n" | |
| "save_area_detail\n" | |
| ' args: {"area_index": 0, "objectives": "...", "challenges": ""}\n' | |
| " SIEMPRE usa area_index (entero). Nunca uses el nombre del área.\n" | |
| ' Si solo tienes un campo, deja el otro como "".\n\n' | |
| "confirm_step1\n" | |
| " args: {}\n" | |
| " Úsala cuando el usuario confirme Y todas las áreas estén completas.\n\n" | |
| "REGLAS:\n" | |
| "- Responde ÚNICAMENTE con el JSON. Sin texto adicional.\n" | |
| "- save_company: copia el nombre EXACTO del usuario.\n" | |
| "- save_area_detail: usa SIEMPRE area_index, nunca el nombre.\n" | |
| "- objectives/challenges: copia el texto EXACTO del usuario.\n" | |
| "- Un solo tool call por respuesta.\n" | |
| "- NO inventes datos. Si no tienes la info, usa ask_user.\n" | |
| "- NO llames confirm_step1 si algún área está incompleta.\n" | |
| ) | |
| def construir_messages(proyecto, history: list, mensaje: str) -> list: | |
| areas_ctx = "" | |
| if proyecto.areas: | |
| areas_ctx = "\nÁREAS (usa area_index en save_area_detail):\n" | |
| for i, a in enumerate(proyecto.areas): | |
| if a.is_complete(): | |
| st = "✅ completa" | |
| elif not a.objectives: | |
| st = "⚠️ falta objetivos" | |
| else: | |
| st = "⚠️ falta desafíos" | |
| areas_ctx += f" [{i}] {a.name} — {st}\n" | |
| system = ( | |
| PLANNER_SYSTEM | |
| + f"\nESTADO ACTUAL:\n{proyecto.state_for_planner()}" | |
| + areas_ctx | |
| ) | |
| messages = [{"role": "system", "content": system}] | |
| recent = [m for m in history if m.get("role") in ("user", "assistant")][-6:] | |
| messages.extend(recent) | |
| messages.append({"role": "user", "content": mensaje}) | |
| return messages | |
| def ejecutar_tool(tool_name: str, args: dict, proyecto, estado: dict): | |
| auto = False | |
| if tool_name == "ask_user": | |
| result = tool_ask_user(args.get("question", "¿Puedes repetir?"), proyecto) | |
| elif tool_name == "save_company": | |
| result = tool_save_company(args.get("name", ""), proyecto) | |
| elif tool_name == "save_areas": | |
| result = tool_save_areas(args.get("areas_raw", ""), proyecto) | |
| elif tool_name == "save_area_detail": | |
| try: | |
| idx = int(args.get("area_index", -1)) | |
| except (TypeError, ValueError): | |
| idx = -1 | |
| result = tool_save_area_detail( | |
| area_index=idx, | |
| objectives=args.get("objectives", ""), | |
| challenges=args.get("challenges", ""), | |
| proyecto=proyecto, | |
| ) | |
| elif tool_name == "confirm_step1": | |
| result = tool_confirm_step1(proyecto) | |
| if result.success: | |
| estado["doc_p1"] = generar_docx_paso1(proyecto) | |
| auto = True | |
| else: | |
| print(f"[WARN] Tool desconocida: {tool_name}") | |
| result = tool_ask_user("¿Puedes repetir?", proyecto) | |
| return result, auto | |
| def _estado_inicial() -> dict: | |
| return { | |
| "company_name": "", "areas": [], | |
| "market_analysis": "", "consolidated_strategy": "", "strategic_initiatives": "", | |
| "step1_complete": False, "step2_complete": False, | |
| "step3_complete": False, "step4_complete": False, | |
| "doc_p1": None, "doc_p2": None, "doc_p3": None, "doc_p4": None, | |
| } | |
| def _barra(pct: int) -> str: | |
| f = int(pct / 10) | |
| return f"[{'█' * f}{'░' * (10 - f)}] {pct}%" | |
| def stratgen_agente(mensaje: str, history: list, estado: dict): | |
| if not estado: | |
| estado = _estado_inicial() | |
| proyecto = dict_a_proyecto(estado) | |
| if proyecto.current_step() == 5: | |
| respuesta = ( | |
| f"🏆 Proceso completado para **{proyecto.company_name}**.\n" | |
| "Los 4 documentos están disponibles en el panel de descarga.\n" | |
| "Pulsa **Nuevo proyecto** para empezar de nuevo." | |
| ) | |
| history.append({"role": "user", "content": mensaje}) | |
| history.append({"role": "assistant", "content": respuesta}) | |
| return respuesta, history, estado, None | |
| messages = construir_messages(proyecto, history, mensaje) | |
| raw = llamar_planner(messages, temperature=0.1) | |
| print(f"[PLANNER] {raw[:200]}") | |
| try: | |
| decision = json.loads(raw) | |
| tool_name = decision.get("tool", "ask_user") | |
| args = decision.get("args", {}) | |
| if not isinstance(args, dict): | |
| args = {} | |
| except json.JSONDecodeError: | |
| print(f"[PARSER WARN] {raw[:100]}") | |
| tool_name = "ask_user" | |
| args = {"question": "No entendí. ¿Puedes repetir?"} | |
| print(f"[AGENT] tool={tool_name} | args={json.dumps(args, ensure_ascii=False)[:120]}") | |
| result, trigger = ejecutar_tool(tool_name, args, proyecto, estado) | |
| print(f"[AGENT] ok={result.success} | trigger={trigger}") | |
| step = proyecto.current_step() | |
| names = { | |
| 1: "Diagnóstico Interno", 2: "Análisis de Mercado", | |
| 3: "Consolidación", 4: "Iniciativas", 5: "COMPLETADO" | |
| } | |
| areas_lines = "" | |
| if proyecto.areas: | |
| areas_lines = "\n" + "\n".join( | |
| f" {'✅' if a.is_complete() else '🔄'} {a.name}" | |
| for a in proyecto.areas | |
| ) | |
| header = ( | |
| f"📊 **Paso {min(step, 4)}/4** — {names.get(step, '?')} | {_barra(proyecto.progress_pct())}\n" | |
| f"🏢 **{proyecto.company_name or '(empresa pendiente)'}** | " | |
| f"Áreas: {len(proyecto.areas)}{areas_lines}" | |
| ) | |
| partes = [header, "\n---\n", result.message] | |
| if trigger: | |
| prog = ejecutar_pasos_automaticos(proyecto, estado) | |
| partes.append("\n\n" + "\n".join(prog)) | |
| partes.append( | |
| f"\n\n{'=' * 50}\n🏆 **PROCESO STRATGEN COMPLETADO**\n{'=' * 50}\n\n" | |
| f"{_barra(100)}\n\n" | |
| "📄 **4 documentos Word** listos para descargar:\n" | |
| " • P1 — Diagnóstico • P2 — Análisis • P3 — Estrategia • P4 — Iniciativas" | |
| ) | |
| respuesta = "\n".join(partes) | |
| estado_nuevo = proyecto_a_dict(proyecto, existing=estado) | |
| for k in ("doc_p1", "doc_p2", "doc_p3", "doc_p4"): | |
| if estado.get(k): | |
| estado_nuevo[k] = estado[k] | |
| history.append({"role": "user", "content": mensaje}) | |
| history.append({"role": "assistant", "content": respuesta}) | |
| return respuesta, history, estado_nuevo, None | |
| print(f"✅ Planner Azure configurado ({MODELS['planner']} — JSON mode)") | |
| # ## 🎨 Celda 8 — Interfaz Gradio | |
| # In[8]: | |
| import gradio as gr | |
| CSS = ( | |
| ".gradio-container {" | |
| "font-family: 'Segoe UI', Helvetica, sans-serif;" | |
| "max-width: 960px !important; margin: auto !important;}" | |
| ".send-btn { background: #2E75B6 !important; color: white !important; }" | |
| ".reset-btn { background: #dc3545 !important; color: white !important; }" | |
| ) | |
| MSG_BIENVENIDA = ( | |
| "🤖 **Bienvenido a StratGen AGENT v4**\n\n" | |
| "Soy un agente estratégico multi-modelo sobre **Azure OpenAI**.\n\n" | |
| "```\n" | |
| f"Planificador │ {MODELS['planner']:<28} │ JSON mode · Orquestador\n" | |
| f"Analista │ {MODELS['analyst']:<28} │ Análisis de mercado (P2)\n" | |
| f"Estratega │ {MODELS['strategist']:<28} │ Consolidación (P3)\n" | |
| f"Generador │ {MODELS['initiator']:<28} │ Iniciativas (P4)\n" | |
| "```\n\n" | |
| "Al finalizar recibirás **4 documentos Word** descargables.\n\n" | |
| "---\n\n" | |
| "🚀 **¿Con qué empresa trabajamos hoy?**" | |
| ) | |
| def _bytes_to_tempfile(data) -> str | None: | |
| if not data: | |
| return None | |
| content, filename = data | |
| fpath = os.path.join(tempfile.gettempdir(), filename) | |
| with open(fpath, "wb") as f: | |
| f.write(content) | |
| return fpath | |
| def responder_chat(mensaje, history, estado): | |
| if not mensaje.strip(): | |
| return "", history, estado, None, None, None, None | |
| if not estado: | |
| estado = _estado_inicial() | |
| _, history, estado, _ = stratgen_agente(mensaje, history, estado) | |
| return ( | |
| "", | |
| history, | |
| estado, | |
| _bytes_to_tempfile(estado.get("doc_p1")), | |
| _bytes_to_tempfile(estado.get("doc_p2")), | |
| _bytes_to_tempfile(estado.get("doc_p3")), | |
| _bytes_to_tempfile(estado.get("doc_p4")), | |
| ) | |
| def resetear_sesion(): | |
| h = [{"role": "assistant", "content": MSG_BIENVENIDA}] | |
| return h, _estado_inicial(), None, None, None, None, "⏳ Sin proyecto activo" | |
| def panel_estado(estado): | |
| if not estado: | |
| return "⏳ Sin proyecto activo" | |
| compl = sum(estado.get(f"step{i}_complete", False) for i in range(1, 5)) | |
| areas = estado.get("areas", []) | |
| a_ok = sum(1 for a in areas if a.get("objectives") and a.get("challenges")) | |
| docs_ok = sum(1 for k in ["doc_p1", "doc_p2", "doc_p3", "doc_p4"] if estado.get(k)) | |
| return ( | |
| f"**Empresa:** {estado.get('company_name') or '(pendiente)'}\n\n" | |
| f"**Progreso:** {compl * 25}%\n\n" | |
| f"**Áreas:** {len(areas)} ({a_ok} completas)\n\n" | |
| f"**Pasos:** {compl}/4\n\n" | |
| f"**Docs listos:** {docs_ok}/4" | |
| ) | |
| with gr.Blocks(title="StratGen Agent v4") as demo: | |
| estado_sesion = gr.State(_estado_inicial()) | |
| gr.HTML( | |
| '<div style="background:linear-gradient(135deg,#1F457C,#2E75B6);color:white;' | |
| 'padding:20px 30px;border-radius:10px;margin-bottom:10px;">' | |
| '<h1 style="margin:0;font-size:1.8em;">🤖 StratGen AGENT v4</h1>' | |
| '<p style="margin:5px 0 0;opacity:.85;">' | |
| "Azure OpenAI Multi-Model · GPT-4o Planner · Gradio</p></div>" | |
| ) | |
| with gr.Row(): | |
| with gr.Column(scale=3): | |
| chatbot = gr.Chatbot( | |
| value=[{"role": "assistant", "content": MSG_BIENVENIDA}], | |
| height=520, | |
| label="💬 StratGen Agent v4", | |
| ) | |
| with gr.Row(): | |
| txt_input = gr.Textbox( | |
| placeholder="Escribe aquí...", | |
| show_label=False, scale=5, lines=2, | |
| ) | |
| btn_enviar = gr.Button( | |
| "Enviar ➤", scale=1, variant="primary", | |
| elem_classes="send-btn" | |
| ) | |
| with gr.Column(scale=1): | |
| gr.Markdown("### 📊 Estado") | |
| estado_display = gr.Markdown("⏳ Sin proyecto activo") | |
| gr.Markdown("---") | |
| gr.Markdown("### 🤖 Modelos activos") | |
| gr.Markdown( | |
| f"🧠 **Planner:** `{MODELS['planner']}`\n\n" | |
| f"🔍 **Analista:** `{MODELS['analyst']}`\n\n" | |
| f"🎯 **Estratega:** `{MODELS['strategist']}`\n\n" | |
| f"⚡ **Generador:** `{MODELS['initiator']}`" | |
| ) | |
| gr.Markdown("---") | |
| gr.Markdown("### 📥 Documentos") | |
| dl_p1 = gr.File(label="📄 P1 — Diagnóstico", interactive=False) | |
| dl_p2 = gr.File(label="📄 P2 — Análisis Mercado", interactive=False) | |
| dl_p3 = gr.File(label="📄 P3 — Estrategia", interactive=False) | |
| dl_p4 = gr.File(label="📄 P4 — Iniciativas", interactive=False) | |
| gr.Markdown("---") | |
| btn_act = gr.Button("🔄 Actualizar estado", size="sm") | |
| btn_reset = gr.Button( | |
| "🗑️ Nuevo proyecto", size="sm", | |
| variant="stop", elem_classes="reset-btn" | |
| ) | |
| gr.Markdown("### 💬 Ejemplos") | |
| gr.Examples( | |
| examples=[ | |
| ["Nestlé"], | |
| ["Comercial, Operaciones, TI, RRHH"], | |
| ["Aumentar ventas 20% en canales digitales"], | |
| ["Alta rotación de clientes y ciclos de venta largos"], | |
| ["Sí, confirmo"], | |
| ], | |
| inputs=txt_input, | |
| ) | |
| _outs = [txt_input, chatbot, estado_sesion, dl_p1, dl_p2, dl_p3, dl_p4] | |
| btn_enviar.click( | |
| responder_chat, [txt_input, chatbot, estado_sesion], _outs | |
| ).then(panel_estado, [estado_sesion], [estado_display]) | |
| txt_input.submit( | |
| responder_chat, [txt_input, chatbot, estado_sesion], _outs | |
| ).then(panel_estado, [estado_sesion], [estado_display]) | |
| btn_act.click(panel_estado, [estado_sesion], [estado_display]) | |
| btn_reset.click( | |
| resetear_sesion, [], | |
| [chatbot, estado_sesion, dl_p1, dl_p2, dl_p3, dl_p4, estado_display] | |
| ) | |
| print("✅ Interfaz Gradio v4 construida") | |
| # ## 🚀 Celda 9 — Lanzar la aplicación | |
| # In[ ]: | |
| print("=" * 60) | |
| print("🤖 STRATGEN AGENT v4 — Azure OpenAI Multi-Model") | |
| print("=" * 60) | |
| for rol, modelo in MODELS.items(): | |
| print(f" {rol:<15} → {modelo}") | |
| print("=" * 60) | |
| demo.launch(share=False) | |