Proyecto Evolutivo: Sistema Multi-Agente (v5)

Descripción del proyecto

En el Módulo 9, construiste la v4 del AI Research Assistant: un agente supervisado que muestra su plan antes de ejecutar, pide aprobación para fuentes de pago, acepta feedback iterativo sobre borradores, y permite corregir datos factuales. Es un solo agente que hace de todo — planifica, busca, analiza, escribe — y un humano lo supervisa.

Pero ese "hace de todo" es el problema. Su system prompt tiene 4 roles mezclados. Sus 6 herramientas compiten entre sí (¿uso web_search o arxiv_search? ¿cuándo uso format_report?). Cuando el reporte final tiene un error, debuggear es doloroso: ¿el dato malo vino de la búsqueda, del análisis, o de la generación del texto? Todo pasa en un solo pipeline monolítico.

La v5 aplica el principio que define este módulo: especialización produce calidad. En vez de un agente generalista, ahora hay 4 agentes enfocados: un researcher que solo busca, un analyst que solo analiza, un writer que solo escribe, y un supervisor que coordina. Cada agente tiene su propio prompt optimizado para una sola tarea, sus propias herramientas relevantes, y su propio modelo (el analyst usa gpt-4.1 para razonamiento complejo; los demás usan gpt-4.1-mini por velocidad y costo).

El momento que define esta versión: ejecutas la misma consulta en v4 y v5 lado a lado. El reporte de v5 tiene hallazgos más precisos (porque el researcher se enfoca solo en buscar), análisis más profundo (porque el analyst tiene toda su capacidad dedicada a razonar), y mejor redacción (porque el writer no está distraído decidiendo qué buscar). Y cuando algo sale mal, sabes exactamente qué agente falló — el logging por agente y el trace del flujo te dicen "el analyst recibió datos correctos pero su conclusión fue incorrecta." Eso es debugging preciso, no arqueología.

---

Objetivo del proyecto

Evolucionar el AI Research Assistant de v4 (un solo agente supervisado) a v5 (sistema multi-agente con especialización), demostrando que la división en agentes produce reportes de mayor calidad y un sistema más fácil de debuggear.

Al completar este proyecto:

• 🔧 Crearás un Researcher agent con herramientas de búsqueda y prompt enfocado en recopilar información
• 🔧 Crearás un Analyst agent con herramientas de análisis y modelo más potente (gpt-4.1)
• 🔧 Crearás un Writer agent con herramientas de formato y prompt especializado en redacción clara
• 🔧 Implementarás un Supervisor que coordina el flujo researcher → analyst → writer, con capacidad de loop-back
• 🔧 Conectarás todo con StateGraph, estado compartido y HITL centralizado en el supervisor
• 🔧 Agregarás logging por agente y tracing del flujo completo para debugging

---

Antes y después

v4 (Módulo 9): un agente que hace todo

Usuario: "Investiga AI agent frameworks"
Agente: [system prompt de 500 tokens intentando cubrir búsqueda + análisis + escritura]
Agente: [tools: web_search, arxiv_search, news_search, calculator, format_report]
Agente: [busca, analiza, escribe — todo en el mismo contexto]
Agente: "Aquí está tu reporte."
Debug: "El dato está mal... ¿fue la búsqueda? ¿el análisis? ¿la redacción?"

v5 (Este módulo): agentes especializados coordinados

Usuario: "Investiga AI agent frameworks"
Supervisor: "Plan: researcher → analyst → writer. Costo: $0. ¿Procedo?"
Usuario: "Sí"
Researcher: [prompt: "Busca información relevante"] [tools: web_search, arxiv_search, news_search]
  → 8 hallazgos en structured format
Analyst: [prompt: "Analiza y sintetiza"] [tools: compare_sources, detect_patterns]
  → 4 patrones, 0 contradicciones, 3 insights
Writer: [prompt: "Redacta un reporte claro"] [tools: format_report, generate_summary]
  → Reporte final con estructura y claridad
Debug: "[analyst] Recibió 8 hallazgos correctos pero identificó un patrón incorrecto"
  → Fix the analyst prompt, no tocar researcher ni writer

---

Especificaciones técnicas

Componente	Versión	Propósito
Python	3.11+	Runtime
LangChain	v1.2+	Framework de LLMs
LangGraph	v1.0+	StateGraph + supervisión
langchain-openai	latest	Modelos OpenAI
pydantic	v2+	Modelos de datos

Estructura del proyecto

research-assistant-v5/
├── .env
├── requirements.txt
├── agents/
│   ├── researcher.py          # NUEVO — agente especializado en búsqueda
│   ├── analyst.py             # NUEVO — agente especializado en análisis
│   ├── writer.py              # NUEVO — agente especializado en redacción
│   └── supervisor.py          # NUEVO — coordinador del sistema
├── tools/
│   ├── search_tools.py        # web_search, arxiv_search, news_search
│   ├── analysis_tools.py      # NUEVO — compare_sources, detect_patterns
│   └── writing_tools.py       # NUEVO — format_report, generate_summary
├── state/
│   └── multi_agent_state.py   # NUEVO — estado compartido multi-agente
├── tracing/
│   ├── agent_logger.py        # NUEVO — logging por agente con prefijo
│   └── flow_tracer.py         # NUEVO — tracing del flujo entre agentes
├── config/
│   └── settings.py            # EXTENDIDO — modelos por agente
└── main.py                    # NUEVO — orquestación y CLI

---

Arquitectura del sistema

                    ┌─────────────┐
                    │  Supervisor  │ ← HITL: aprobación del plan
                    │  (gpt-4.1-  │
                    │    mini)    │
                    └──────┬──────┘
                           │
              ┌────────────┼────────────┐
              │            │            │
              ▼            ▼            ▼
       ┌────────────┐ ┌──────────┐ ┌──────────┐
       │ Researcher  │ │ Analyst  │ │  Writer  │
       │ (gpt-4.1-  │ │(gpt-4.1)│ │(gpt-4.1- │
       │   mini)    │ │          │ │  mini)   │
       └────────────┘ └──────────┘ └──────────┘
       Tools:          Tools:       Tools:
       - web_search    - compare    - format
       - arxiv_search  - patterns   - summary
       - news_search

El flujo principal es: Supervisor → Researcher → Analyst → Writer → Supervisor.

Si el Analyst determina que necesita más datos, el Supervisor envía al Researcher de vuelta (loop-back). El Supervisor es el único punto de HITL: aprueba el plan antes de iniciar y revisa el output final.

---

Paso 1: Estado compartido multi-agente (state/multi_agent_state.py)

El estado es el contrato entre agentes. Cada agente lee lo que necesita y escribe su output en campos separados.

"""
state/multi_agent_state.py
Estado compartido para el sistema multi-agente v5.
"""

import operator
from typing import TypedDict, Annotated
from pydantic import BaseModel, Field
from datetime import datetime


class Finding(BaseModel):
    title: str
    source: str
    source_type: str
    content: str
    relevance: float = Field(ge=0.0, le=1.0)


class AnalysisResult(BaseModel):
    patterns: list[str]
    contradictions: list[str]
    insights: list[str]
    confidence: float = Field(ge=0.0, le=1.0)
    needs_more_data: bool = False
    data_gaps: list[str] = Field(default_factory=list)


class Report(BaseModel):
    title: str
    summary: str
    sections: list[dict]
    sources_count: int
    confidence: float
    generated_at: str = Field(default_factory=lambda: datetime.now().isoformat())
    version: str = "v5"


class MultiAgentState(TypedDict):
    query: str
    plan: dict
    plan_approved: bool

    findings: list[dict]
    analysis: dict
    report: dict

    current_agent: str
    iteration: int
    max_iterations: int
    status: str

    trace: Annotated[list[str], operator.add]
    agent_logs: Annotated[list[dict], operator.add]

Los campos clave:

• ✅ findings — el Researcher escribe aquí, el Analyst lee
• ✅ analysis — el Analyst escribe aquí, el Writer lee
• ✅ report — el Writer escribe aquí, el Supervisor valida
• ✅ trace — todos escriben, acumulativo con operator.add
• ✅ iteration — el Supervisor lo usa para prevenir loops infinitos

---

Paso 2: Logging por agente y tracing (tracing/)

agent_logger.py

"""
tracing/agent_logger.py
Logger con prefijo por agente para debugging multi-agente.
"""

import logging
import time


class AgentLogger:
    def __init__(self, agent_name: str):
        self.agent_name = agent_name
        self.logger = logging.getLogger(f"multiagent.{agent_name}")

        if not self.logger.handlers:
            handler = logging.StreamHandler()
            formatter = logging.Formatter(
                f"%(asctime)s | %(levelname)-5s | [{agent_name}] %(message)s",
                datefmt="%H:%M:%S",
            )
            handler.setFormatter(formatter)
            self.logger.addHandler(handler)
            self.logger.setLevel(logging.DEBUG)

    def info(self, msg: str):
        self.logger.info(msg)

    def debug(self, msg: str):
        self.logger.debug(msg)

    def warning(self, msg: str):
        self.logger.warning(msg)

    def error(self, msg: str):
        self.logger.error(msg)

    def log_entry(self, event: str, data: dict | None = None) -> dict:
        """Genera un entry para el campo agent_logs del estado."""
        entry = {
            "agent": self.agent_name,
            "event": event,
            "timestamp": time.time(),
            "data": data or {},
        }
        self.info(f"{event}: {data}" if data else event)
        return entry

flow_tracer.py

"""
tracing/flow_tracer.py
Tracing del flujo entre agentes: quién manejó qué y cuándo.
"""

import time


class FlowTracer:
    def __init__(self):
        self.events: list[dict] = []
        self.start_time = time.time()

    def record(self, agent: str, event: str, details: str = ""):
        elapsed = (time.time() - self.start_time) * 1000
        self.events.append({
            "agent": agent,
            "event": event,
            "details": details,
            "elapsed_ms": round(elapsed),
        })

    def print_timeline(self):
        print(f"\n{'─' * 70}")
        print(f"  {'Agent':<15} {'Event':<20} {'Time':>8}  Details")
        print(f"{'─' * 70}")
        for e in self.events:
            print(f"  {e['agent']:<15} {e['event']:<20} {e['elapsed_ms']:>6}ms  {e['details']}")
        print(f"{'─' * 70}")

    def find_bottleneck(self) -> dict | None:
        starts: dict[str, int] = {}
        ends: dict[str, int] = {}
        for e in self.events:
            if e["event"] == "started":
                starts[e["agent"]] = e["elapsed_ms"]
            elif e["event"] == "completed":
                ends[e["agent"]] = e["elapsed_ms"]

        durations = {a: ends[a] - starts[a] for a in starts if a in ends}
        if not durations:
            return None

        slowest = max(durations, key=durations.get)
        return {"agent": slowest, "duration_ms": durations[slowest]}

    def total_time_ms(self) -> int:
        if not self.events:
            return 0
        return self.events[-1]["elapsed_ms"]


tracer = FlowTracer()

---

Paso 3: Herramientas especializadas por agente (tools/)

search_tools.py

"""
tools/search_tools.py
Herramientas de búsqueda — asignadas exclusivamente al Researcher.
"""

import time
import random


def web_search(query: str) -> list[dict]:
    time.sleep(0.1)
    return [
        {
            "title": f"Web: {query} - resultado {i+1}",
            "source": f"https://example.com/{query.replace(' ', '-')}-{i+1}",
            "source_type": "web",
            "content": f"Información web relevante sobre {query}. Hallazgo {i+1} de 3.",
            "relevance": round(random.uniform(0.6, 0.95), 2),
        }
        for i in range(3)
    ]


def arxiv_search(query: str) -> list[dict]:
    time.sleep(0.15)
    return [
        {
            "title": f"arXiv: {query} - paper {i+1}",
            "source": f"https://arxiv.org/abs/2026.{random.randint(10000, 99999)}",
            "source_type": "academic",
            "content": f"Paper académico sobre {query}. Metodología y resultados del estudio {i+1}.",
            "relevance": round(random.uniform(0.7, 0.98), 2),
        }
        for i in range(2)
    ]


def news_search(query: str) -> list[dict]:
    time.sleep(0.08)
    return [
        {
            "title": f"News: {query} - último desarrollo",
            "source": "https://technews.example.com/latest",
            "source_type": "news",
            "content": f"Noticia reciente sobre {query}. Desarrollos de las últimas 48 horas.",
            "relevance": round(random.uniform(0.5, 0.85), 2),
        }
    ]

analysis_tools.py

"""
tools/analysis_tools.py
Herramientas de análisis — asignadas exclusivamente al Analyst.
"""


def compare_sources(findings: list[dict]) -> dict:
    source_types = set(f.get("source_type", "unknown") for f in findings)
    contradictions = []

    for i, f1 in enumerate(findings):
        for f2 in findings[i + 1:]:
            if f1.get("source_type") != f2.get("source_type"):
                pass

    avg_relevance = sum(f.get("relevance", 0.5) for f in findings) / max(len(findings), 1)

    return {
        "total_sources": len(findings),
        "source_types": list(source_types),
        "contradictions": contradictions,
        "avg_relevance": round(avg_relevance, 2),
        "diversity_score": round(len(source_types) / max(len(findings), 1), 2),
    }


def detect_patterns(findings: list[dict]) -> list[str]:
    patterns = []

    if len(findings) >= 3:
        patterns.append(f"Consenso entre {len(findings)} fuentes sobre el tema principal")

    source_types = [f.get("source_type") for f in findings]
    if "academic" in source_types and "web" in source_types:
        patterns.append("Convergencia entre investigación académica y contenido web")

    high_relevance = [f for f in findings if f.get("relevance", 0) > 0.8]
    if len(high_relevance) >= 2:
        patterns.append(f"{len(high_relevance)} fuentes de alta relevancia (>80%)")

    if "news" in source_types:
        patterns.append("Tema con cobertura mediática activa")

    return patterns

writing_tools.py

"""
tools/writing_tools.py
Herramientas de redacción — asignadas exclusivamente al Writer.
"""


def format_report(title: str, summary: str, sections: list[dict], sources_count: int) -> dict:
    formatted_sections = []
    for i, section in enumerate(sections, 1):
        formatted_sections.append({
            "number": i,
            "heading": section.get("heading", f"Sección {i}"),
            "content": section.get("content", ""),
        })

    return {
        "title": title,
        "summary": summary,
        "sections": formatted_sections,
        "sources_count": sources_count,
        "format": "structured",
    }


def generate_summary(findings: list[dict], analysis: dict) -> str:
    num_findings = len(findings)
    num_patterns = len(analysis.get("patterns", []))
    confidence = analysis.get("confidence", 0.5)
    contradictions = len(analysis.get("contradictions", []))

    parts = [
        f"Investigación basada en {num_findings} fuentes.",
        f"Se identificaron {num_patterns} patrones principales.",
    ]

    if contradictions > 0:
        parts.append(f"Se encontraron {contradictions} contradicciones entre fuentes.")
    else:
        parts.append("Las fuentes son consistentes entre sí.")

    parts.append(f"Nivel de confianza general: {confidence:.0%}.")

    return " ".join(parts)

---

Paso 4: Agentes especializados (agents/)

researcher.py

"""
agents/researcher.py
Researcher Agent — especializado en búsqueda de información.
Modelo: gpt-4.1-mini (rápido, económico).
"""

import json
from langchain.chat_models import init_chat_model

import sys
sys.path.insert(0, ".")

from tools.search_tools import web_search, arxiv_search, news_search
from tracing.agent_logger import AgentLogger

logger = AgentLogger("researcher")
model = init_chat_model("openai:gpt-4.1-mini", temperature=0.1)

RESEARCHER_PROMPT = (
    "Eres un investigador especializado. Tu ÚNICA tarea es buscar información relevante "
    "sobre el tema dado. NO analices, NO saques conclusiones, NO escribas reportes. "
    "Solo busca y devuelve los hallazgos en formato estructurado."
)


def run_researcher(query: str, sources: list[str] | None = None) -> dict:
    """Ejecuta el researcher agent sobre un query."""
    if sources is None:
        sources = ["web", "arxiv", "news"]

    logger.info(f"Iniciando búsqueda: '{query}' en {sources}")

    all_findings = []

    search_fns = {
        "web": web_search,
        "arxiv": arxiv_search,
        "news": news_search,
    }

    for source in sources:
        fn = search_fns.get(source)
        if fn:
            try:
                results = fn(query)
                all_findings.extend(results)
                logger.debug(f"{source}: {len(results)} resultados OK")
            except Exception as e:
                logger.error(f"{source}: error — {e}")
        else:
            logger.warning(f"Fuente desconocida: {source}")

    all_findings.sort(key=lambda f: f.get("relevance", 0), reverse=True)

    logger.info(f"Búsqueda completada: {len(all_findings)} hallazgos de {len(sources)} fuentes")

    return {
        "findings": [f for f in all_findings],
        "sources_searched": sources,
        "total_results": len(all_findings),
    }

analyst.py

"""
agents/analyst.py
Analyst Agent — especializado en análisis y síntesis.
Modelo: gpt-4.1 (potente, para razonamiento complejo).
"""

import json
from langchain.chat_models import init_chat_model

import sys
sys.path.insert(0, ".")

from tools.analysis_tools import compare_sources, detect_patterns
from tracing.agent_logger import AgentLogger

logger = AgentLogger("analyst")
model = init_chat_model("openai:gpt-4.1", temperature=0.2)

ANALYST_PROMPT = (
    "Eres un analista especializado. Tu ÚNICA tarea es analizar hallazgos de investigación, "
    "identificar patrones, detectar contradicciones, y generar insights. "
    "NO busques información nueva, NO escribas reportes finales. "
    "Si los datos son insuficientes, indícalo claramente."
)


def run_analyst(findings: list[dict], query: str) -> dict:
    """Ejecuta el analyst agent sobre los hallazgos del researcher."""
    logger.info(f"Analizando {len(findings)} hallazgos sobre '{query}'")

    if len(findings) < 2:
        logger.warning(f"Solo {len(findings)} hallazgo(s) — datos insuficientes")
        return {
            "patterns": [],
            "contradictions": [],
            "insights": [f"Datos insuficientes: solo {len(findings)} hallazgo(s)"],
            "confidence": 0.3,
            "needs_more_data": True,
            "data_gaps": ["Se necesitan más fuentes para un análisis confiable"],
        }

    comparison = compare_sources(findings)
    logger.debug(f"Comparación: {comparison['total_sources']} fuentes, diversidad {comparison['diversity_score']}")

    patterns = detect_patterns(findings)
    logger.debug(f"Patrones identificados: {len(patterns)}")

    findings_text = "\n".join(
        f"- [{f.get('source_type', '?')}] {f.get('title', '?')}: {f.get('content', '')[:100]}"
        for f in findings[:8]
    )

    response = model.invoke(
        f"{ANALYST_PROMPT}\n\n"
        f"Tema: {query}\n"
        f"Hallazgos ({len(findings)}):\n{findings_text}\n\n"
        f"Patrones detectados automáticamente: {patterns}\n"
        f"Comparación de fuentes: {comparison}\n\n"
        f"Genera 2-4 insights concisos sobre lo que revelan estos datos. "
        f"Responde en JSON: {{\"insights\": [\"...\"], \"confidence\": 0.8, "
        f"\"needs_more_data\": false, \"data_gaps\": []}}\n"
        f"Solo JSON."
    )

    try:
        llm_analysis = json.loads(response.content)
    except json.JSONDecodeError:
        logger.warning("LLM no devolvió JSON válido — usando análisis automático")
        llm_analysis = {
            "insights": [f"Análisis general de {len(findings)} fuentes sobre {query}"],
            "confidence": 0.6,
            "needs_more_data": False,
            "data_gaps": [],
        }

    result = {
        "patterns": patterns,
        "contradictions": comparison["contradictions"],
        "insights": llm_analysis.get("insights", []),
        "confidence": llm_analysis.get("confidence", 0.6),
        "needs_more_data": llm_analysis.get("needs_more_data", False),
        "data_gaps": llm_analysis.get("data_gaps", []),
        "source_comparison": comparison,
    }

    logger.info(
        f"Análisis completo: {len(patterns)} patrones, "
        f"{len(result['insights'])} insights, "
        f"confianza {result['confidence']:.0%}, "
        f"más datos: {'sí' if result['needs_more_data'] else 'no'}"
    )

    return result

writer.py

"""
agents/writer.py
Writer Agent — especializado en redacción de reportes.
Modelo: gpt-4.1-mini (rápido, buena redacción).
"""

import json
from langchain.chat_models import init_chat_model

import sys
sys.path.insert(0, ".")

from tools.writing_tools import format_report, generate_summary
from tracing.agent_logger import AgentLogger

logger = AgentLogger("writer")
model = init_chat_model("openai:gpt-4.1-mini", temperature=0.3)

WRITER_PROMPT = (
    "Eres un redactor especializado en reportes de investigación. "
    "Tu ÚNICA tarea es tomar los hallazgos y el análisis, y producir un reporte "
    "claro, estructurado y profesional. NO investigues, NO analices. Solo escribe."
)


def run_writer(query: str, findings: list[dict], analysis: dict) -> dict:
    """Ejecuta el writer agent para generar el reporte final."""
    logger.info(f"Generando reporte para '{query}'")
    logger.debug(f"Input: {len(findings)} hallazgos, {len(analysis.get('insights', []))} insights")

    auto_summary = generate_summary(findings, analysis)

    findings_text = "\n".join(
        f"- [{f.get('source_type', '?')}] {f.get('title', '?')}: {f.get('content', '')[:80]}"
        for f in findings[:6]
    )
    insights_text = "\n".join(f"- {ins}" for ins in analysis.get("insights", []))
    patterns_text = "\n".join(f"- {p}" for p in analysis.get("patterns", []))

    response = model.invoke(
        f"{WRITER_PROMPT}\n\n"
        f"Tema: {query}\n"
        f"Hallazgos:\n{findings_text}\n\n"
        f"Patrones:\n{patterns_text}\n\n"
        f"Insights del análisis:\n{insights_text}\n\n"
        f"Confianza general: {analysis.get('confidence', 0.5):.0%}\n\n"
        f"Genera un reporte con 2-3 secciones. Responde en JSON:\n"
        f'{{"title": "...", "summary": "...", "sections": ['
        f'{{"heading": "...", "content": "..."}}]}}\n'
        f"Solo JSON."
    )

    try:
        report_data = json.loads(response.content)
    except json.JSONDecodeError:
        logger.warning("LLM no devolvió JSON válido — usando reporte automático")
        report_data = {
            "title": f"Reporte: {query}",
            "summary": auto_summary,
            "sections": [
                {"heading": "Hallazgos", "content": findings_text},
                {"heading": "Análisis", "content": insights_text},
            ],
        }

    report = format_report(
        title=report_data.get("title", f"Reporte: {query}"),
        summary=report_data.get("summary", auto_summary),
        sections=report_data.get("sections", []),
        sources_count=len(findings),
    )

    report["confidence"] = analysis.get("confidence", 0.5)

    logger.info(f"Reporte generado: '{report['title']}' ({len(report['sections'])} secciones)")

    return report

supervisor.py — el coordinador

"""
agents/supervisor.py
Supervisor Agent — coordina el flujo researcher → analyst → writer.
Implementa HITL centralizado y manejo de loop-back.
"""

import sys
sys.path.insert(0, ".")

from tracing.agent_logger import AgentLogger

logger = AgentLogger("supervisor")


def create_plan(query: str, sources: list[str] | None = None) -> dict:
    if sources is None:
        sources = ["web", "arxiv", "news"]

    plan = {
        "query": query,
        "sources": sources,
        "pipeline": ["researcher", "analyst", "writer"],
        "estimated_cost": 0.0,
        "max_iterations": 3,
    }

    logger.info(f"Plan creado: {plan['pipeline']} con fuentes {sources}")
    return plan


def evaluate_analysis(analysis: dict, iteration: int, max_iter: int) -> str:
    """Decide si continuar a writer o enviar de vuelta a researcher."""
    if analysis.get("needs_more_data", False) and iteration < max_iter:
        gaps = analysis.get("data_gaps", [])
        logger.warning(f"Analyst solicita más datos (iteración {iteration}/{max_iter}): {gaps}")
        return "loop_back"

    if analysis.get("confidence", 0) < 0.4 and iteration < max_iter:
        logger.warning(f"Confianza baja ({analysis.get('confidence', 0):.0%}) — loop back")
        return "loop_back"

    logger.info(f"Análisis aceptado (confianza {analysis.get('confidence', 0):.0%}) → writer")
    return "continue"

---

Paso 5: Orquestación con StateGraph (main.py)

Aquí es donde todo se conecta. El StateGraph define el flujo completo con HITL centralizado en el supervisor.

"""
main.py
Sistema multi-agente v5 — orquestación completa.
"""

import json
import time
import operator
from typing import TypedDict, Annotated

from dotenv import load_dotenv
load_dotenv()

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

import sys
sys.path.insert(0, ".")

from agents.researcher import run_researcher
from agents.analyst import run_analyst
from agents.writer import run_writer
from agents.supervisor import create_plan, evaluate_analysis, logger as sup_logger
from tracing.agent_logger import AgentLogger
from tracing.flow_tracer import FlowTracer

pipeline_logger = AgentLogger("pipeline")


class MultiAgentState(TypedDict):
    query: str
    plan: dict
    plan_approved: bool
    findings: list[dict]
    analysis: dict
    report: dict
    current_agent: str
    iteration: int
    max_iterations: int
    status: str
    trace: Annotated[list[str], operator.add]


def supervisor_plan_node(state: MultiAgentState) -> dict:
    plan = create_plan(state["query"])
    return {
        "plan": plan,
        "max_iterations": plan["max_iterations"],
        "current_agent": "supervisor",
        "trace": [f"[supervisor] Plan: {plan['pipeline']} | fuentes: {plan['sources']}"],
    }


def hitl_approve_node(state: MultiAgentState) -> dict:
    """HITL centralizado: el supervisor pide aprobación antes de ejecutar."""
    plan = state["plan"]

    response = interrupt({
        "type": "plan_approval",
        "message": (
            f"Plan de investigación multi-agente:\n"
            f"  Query: {plan['query']}\n"
            f"  Pipeline: {' → '.join(plan['pipeline'])}\n"
            f"  Fuentes: {plan['sources']}\n"
            f"  Costo estimado: ${plan['estimated_cost']:.2f}\n"
            f"  Max iteraciones: {plan['max_iterations']}\n"
            f"¿Aprobar? (approve / cancel)"
        ),
        "plan": plan,
    })

    action = response if isinstance(response, str) else response.get("action", "approve")

    if action == "cancel":
        return {
            "plan_approved": False,
            "status": "cancelled",
            "trace": [f"[supervisor:hitl] Plan cancelado por el usuario"],
        }

    return {
        "plan_approved": True,
        "status": "approved",
        "trace": [f"[supervisor:hitl] Plan aprobado — ejecutando pipeline"],
    }


def route_after_approval(state: MultiAgentState) -> str:
    if state.get("status") == "cancelled":
        return "end"
    return "researcher"


def researcher_node(state: MultiAgentState) -> dict:
    sources = state["plan"].get("sources", ["web", "arxiv", "news"])
    result = run_researcher(state["query"], sources)

    return {
        "findings": result["findings"],
        "current_agent": "researcher",
        "trace": [
            f"[researcher] {result['total_results']} hallazgos de {result['sources_searched']}",
        ],
    }


def analyst_node(state: MultiAgentState) -> dict:
    result = run_analyst(state["findings"], state["query"])

    return {
        "analysis": result,
        "current_agent": "analyst",
        "trace": [
            f"[analyst] {len(result.get('patterns', []))} patrones, "
            f"{len(result.get('insights', []))} insights, "
            f"confianza {result.get('confidence', 0):.0%}"
            f"{' — SOLICITA MÁS DATOS' if result.get('needs_more_data') else ''}",
        ],
    }


def supervisor_evaluate_node(state: MultiAgentState) -> dict:
    """El supervisor decide: continuar a writer o loop back a researcher."""
    decision = evaluate_analysis(
        state["analysis"],
        state["iteration"],
        state["max_iterations"],
    )

    new_iteration = state["iteration"] + (1 if decision == "loop_back" else 0)

    return {
        "status": decision,
        "iteration": new_iteration,
        "trace": [
            f"[supervisor] Evaluación: {decision} (iteración {new_iteration}/{state['max_iterations']})",
        ],
    }


def route_after_evaluation(state: MultiAgentState) -> str:
    if state.get("status") == "loop_back":
        return "researcher"
    return "writer"


def writer_node(state: MultiAgentState) -> dict:
    result = run_writer(state["query"], state["findings"], state["analysis"])

    return {
        "report": result,
        "current_agent": "writer",
        "status": "completed",
        "trace": [
            f"[writer] Reporte generado: '{result.get('title', '?')}' "
            f"({len(result.get('sections', []))} secciones)",
        ],
    }


builder = StateGraph(MultiAgentState)

builder.add_node("plan", supervisor_plan_node)
builder.add_node("approve", hitl_approve_node)
builder.add_node("researcher", researcher_node)
builder.add_node("analyst", analyst_node)
builder.add_node("evaluate", supervisor_evaluate_node)
builder.add_node("writer", writer_node)

builder.add_edge(START, "plan")
builder.add_edge("plan", "approve")
builder.add_conditional_edges("approve", route_after_approval, {
    "researcher": "researcher",
    "end": END,
})
builder.add_edge("researcher", "analyst")
builder.add_edge("analyst", "evaluate")
builder.add_conditional_edges("evaluate", route_after_evaluation, {
    "researcher": "researcher",
    "writer": "writer",
})
builder.add_edge("writer", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)


def format_v5_report(result: dict) -> str:
    report = result.get("report", {})
    if not report:
        return f"  ❌ {result.get('status', 'Error desconocido')}"

    lines = [
        "",
        "╔" + "═" * 58 + "╗",
        "║" + f"  📄 {report.get('title', 'Reporte v5')}".center(58) + "║",
        "╚" + "═" * 58 + "╝",
        f"\n🎯 Confianza: {report.get('confidence', 0):.0%}",
        f"📚 Fuentes: {report.get('sources_count', 0)}",
        f"📝 Versión: v5 (multi-agente)",
        f"\n{'─' * 60}",
        "📋 RESUMEN",
        f"{'─' * 60}",
        report.get("summary", "Sin resumen."),
    ]

    sections = report.get("sections", [])
    for section in sections:
        lines.extend([
            f"\n{'─' * 60}",
            f"📌 {section.get('heading', 'Sección')}",
            f"{'─' * 60}",
            section.get("content", ""),
        ])

    lines.append(f"\n{'═' * 60}")
    return "\n".join(lines)


def run_v5(query: str, thread_id: str = "v5-001"):
    """Ejecuta el pipeline multi-agente completo con manejo de HITL."""
    config = {"configurable": {"thread_id": thread_id}}
    tracer = FlowTracer()

    tracer.record("pipeline", "started", query)
    result = graph.invoke(
        {
            "query": query,
            "plan": {},
            "plan_approved": False,
            "findings": [],
            "analysis": {},
            "report": {},
            "current_agent": "",
            "iteration": 0,
            "max_iterations": 3,
            "status": "",
            "trace": [],
        },
        config,
    )

    while True:
        state = graph.get_state(config)
        if not state.next:
            break

        task_data = state.tasks
        for task_item in task_data:
            if hasattr(task_item, "interrupts") and task_item.interrupts:
                interrupt_data = task_item.interrupts[0].value
                print(f"\n{'─' * 58}")
                print(f"  🔔 {interrupt_data.get('type', 'interrupt')}")
                print(f"{'─' * 58}")
                print(f"  {interrupt_data.get('message', '')}")
                print(f"{'─' * 58}")

                choice = input("\n  → ").strip().lower()
                if choice in ("cancel", "no", "n"):
                    result = graph.invoke(Command(resume="cancel"), config)
                else:
                    result = graph.invoke(Command(resume="approve"), config)

    tracer.record("pipeline", "completed")

    print("\n=== TRACE DEL FLUJO ===")
    for step in result.get("trace", []):
        print(f"  {step}")

    print(format_v5_report(result))

    return result


if __name__ == "__main__":
    print("=" * 60)
    print("  🔬 AI Research Assistant v5 — Multi-Agent System")
    print("=" * 60)

    while True:
        try:
            query = input("\n🔎 Query (o 'salir'): ").strip()
        except (KeyboardInterrupt, EOFError):
            break

        if not query or query.lower() in ("salir", "exit", "quit"):
            break

        import uuid
        thread_id = f"v5-{uuid.uuid4().hex[:8]}"
        run_v5(query, thread_id)

    print("\n  ¡Hasta luego!")

---

Paso 6: Logging por agente y tracing en acción

Cuando ejecutas el sistema, cada agente loguea con su prefijo. Esto es lo que ves en la consola:

14:30:01 | INFO  | [supervisor] Plan creado: ['researcher', 'analyst', 'writer'] con fuentes ['web', 'arxiv', 'news']
14:30:05 | INFO  | [researcher] Iniciando búsqueda: 'AI agent frameworks' en ['web', 'arxiv', 'news']
14:30:05 | DEBUG | [researcher] web: 3 resultados OK
14:30:05 | DEBUG | [researcher] arxiv: 2 resultados OK
14:30:05 | DEBUG | [researcher] news: 1 resultados OK
14:30:05 | INFO  | [researcher] Búsqueda completada: 6 hallazgos de 3 fuentes
14:30:06 | INFO  | [analyst] Analizando 6 hallazgos sobre 'AI agent frameworks'
14:30:06 | DEBUG | [analyst] Comparación: 6 fuentes, diversidad 0.50
14:30:06 | DEBUG | [analyst] Patrones identificados: 4
14:30:08 | INFO  | [analyst] Análisis completo: 4 patrones, 3 insights, confianza 82%, más datos: no
14:30:08 | INFO  | [supervisor] Análisis aceptado (confianza 82%) → writer
14:30:08 | INFO  | [writer] Generando reporte para 'AI agent frameworks'
14:30:08 | DEBUG | [writer] Input: 6 hallazgos, 3 insights
14:30:10 | INFO  | [writer] Reporte generado: 'AI Agent Frameworks: Análisis Comparativo' (3 secciones)

Para filtrar por agente: grep "[analyst]" logs.txt muestra solo lo que hizo el analyst. Si el reporte tiene un error de análisis, sabes exactamente dónde buscar.

---

Ejecución: sesión completa

cd research-assistant-v5
python main.py

============================================================
  🔬 AI Research Assistant v5 — Multi-Agent System
============================================================

🔎 Query (o 'salir'): AI agent frameworks comparison 2026

──────────────────────────────────────────────────
  🔔 plan_approval
──────────────────────────────────────────────────
  Plan de investigación multi-agente:
    Query: AI agent frameworks comparison 2026
    Pipeline: researcher → analyst → writer
    Fuentes: ['web', 'arxiv', 'news']
    Costo estimado: $0.00
    Max iteraciones: 3
  ¿Aprobar? (approve / cancel)
──────────────────────────────────────────────────

  → approve

=== TRACE DEL FLUJO ===
  [supervisor] Plan: ['researcher', 'analyst', 'writer'] | fuentes: ['web', 'arxiv', 'news']
  [supervisor:hitl] Plan aprobado — ejecutando pipeline
  [researcher] 6 hallazgos de ['web', 'arxiv', 'news']
  [analyst] 4 patrones, 3 insights, confianza 82%
  [supervisor] Evaluación: continue (iteración 0/3)
  [writer] Reporte generado: 'AI Agent Frameworks 2026' (3 secciones)

╔══════════════════════════════════════════════════════════╗
║  📄 AI Agent Frameworks 2026                             ║
╚══════════════════════════════════════════════════════════╝

🎯 Confianza: 82%
📚 Fuentes: 6
📝 Versión: v5 (multi-agente)

────────────────────────────────────────────────────────────
📋 RESUMEN
────────────────────────────────────────────────────────────
Investigación basada en 6 fuentes. Se identificaron 4 patrones
principales. Las fuentes son consistentes entre sí. Nivel de
confianza general: 82%.

────────────────────────────────────────────────────────────
📌 Panorama General
────────────────────────────────────────────────────────────
Los principales frameworks para agentes de IA en 2026 incluyen
LangGraph, CrewAI, AutoGen y OpenAI Agents SDK...

────────────────────────────────────────────────────────────
📌 Comparación Técnica
────────────────────────────────────────────────────────────
LangGraph ofrece el mayor control granular sobre el flujo de
agentes. CrewAI simplifica la creación de equipos multi-agente...

────────────────────────────────────────────────────────────
📌 Recomendaciones
────────────────────────────────────────────────────────────
Para producción con requisitos de control: LangGraph. Para
prototipado rápido de equipos: CrewAI...

════════════════════════════════════════════════════════════

---

Criterios de éxito

• ✅ 4 agentes especializados — researcher, analyst, writer y supervisor, cada uno con su prompt y herramientas propias
• ✅ Modelos diferenciados — analyst usa gpt-4.1 (razonamiento), los demás usan gpt-4.1-mini (velocidad)
• ✅ Supervisor coordina — flujo researcher → analyst → writer con evaluación después del analyst
• ✅ Loop-back funciona — si el analyst pide más datos, el supervisor reenvía al researcher (máximo 3 iteraciones)
• ✅ HITL centralizado — el supervisor pide aprobación del plan antes de iniciar, no cada agente por separado
• ✅ Logging por agente — cada agente loguea con su nombre como prefijo, filtrable con grep
• ✅ Trace del flujo — el campo trace del estado muestra la cadena completa de decisiones
• ✅ Reporte de calidad — el output es un reporte estructurado con resumen, secciones, y metadata

---

Escenarios de prueba

Test 1: Flujo completo sin loop-back

🔎 AI agent frameworks comparison
→ approve plan
→ researcher: 6 hallazgos
→ analyst: confianza 80%, no necesita más datos
→ supervisor: continue → writer
→ writer: reporte con 3 secciones
Resultado: reporte completo en una sola pasada.

Test 2: Loop-back por datos insuficientes

🔎 quantum error correction recent advances
→ approve plan
→ researcher: 6 hallazgos (pero solo de web, poco académico)
→ analyst: confianza 40%, needs_more_data=True, gaps=["faltan papers recientes"]
→ supervisor: loop_back (iteración 1/3)
→ researcher: busca de nuevo (ahora con más enfoque)
→ analyst: confianza 75%, needs_more_data=False
→ supervisor: continue → writer
→ writer: reporte mejorado con más fuentes
Resultado: 2 pasadas del researcher produjeron un reporte más sólido.

Test 3: Plan cancelado

🔎 algo que cambié de opinión
→ cancel plan
Resultado: status="cancelled", sin ejecución de agentes.

---

Comparación de calidad: v4 vs v5

Esta es la prueba definitiva. La misma consulta, los dos sistemas:

Query: "Compare LangGraph vs CrewAI for production multi-agent systems"

v4 (un solo agente):

Resumen: LangGraph y CrewAI son frameworks para agentes de IA.
LangGraph ofrece control granular y CrewAI es más simple.
Ambos son opciones válidas.

Hallazgos: 4 hallazgos genéricos
Confianza: 72%

v5 (multi-agente):

Resumen: Investigación basada en 6 fuentes de 3 tipos.
Se identificaron 4 patrones principales incluyendo
convergencia entre investigación académica y contenido web.

Secciones:
1. Panorama General — contexto de cada framework
2. Comparación Técnica — tabla con 5 criterios
3. Análisis de Producción — métricas reales de empresas
4. Recomendación — cuándo usar cada uno y por qué

Confianza: 85%

La diferencia:

• ✅ Más hallazgos — el researcher dedicado busca mejor que un agente multitarea
• ✅ Análisis más profundo — el analyst con gpt-4.1 identifica patrones que el agente generalista pierde
• ✅ Mejor estructura — el writer dedicado produce secciones claras, no un bloque de texto
• ✅ Debugging preciso — si el dato de "métricas reales" es incorrecto, sabes que fue el researcher (búsqueda) no el analyst ni el writer

---

Errores comunes

1. Los agentes leen campos que no les corresponden

Causa: El researcher accede a state["analysis"] que todavía está vacío.

Solución: Cada agente solo lee sus inputs. El researcher lee query y plan. El analyst lee findings. El writer lee findings y analysis:

def researcher_node(state):
    result = run_researcher(state["query"], state["plan"]["sources"])

def analyst_node(state):
    result = run_analyst(state["findings"], state["query"])

def writer_node(state):
    result = run_writer(state["query"], state["findings"], state["analysis"])

2. El loop-back no termina (delegación infinita)

Causa: El analyst siempre dice needs_more_data=True y no hay límite.

Solución: El campo max_iterations y el check en evaluate_analysis:

def evaluate_analysis(analysis, iteration, max_iter):
    if analysis.get("needs_more_data") and iteration < max_iter:
        return "loop_back"
    return "continue"

3. El trace está vacío

Causa: Los nodos retornan "trace": "mensaje" en vez de "trace": ["mensaje"].

Solución: El trace usa Annotated[list[str], operator.add]. Cada nodo debe retornar una lista:

return {"trace": [f"[researcher] 6 hallazgos"]}

4. Los logs de agentes se mezclan y son ilegibles

Causa: Todos usan el mismo logger sin prefijo.

Solución: Cada agente crea su propio AgentLogger con nombre:

logger = AgentLogger("researcher")
logger = AgentLogger("analyst")
logger = AgentLogger("writer")

5. El analyst no recibe los hallazgos del researcher

Causa: El researcher escribe en un campo diferente al que el analyst lee.

Solución: Verifica que el campo es el mismo en ambos nodos. Convención: el researcher escribe en findings, el analyst lee de findings:

def researcher_node(state) -> dict:
    return {"findings": result["findings"]}

def analyst_node(state) -> dict:
    result = run_analyst(state["findings"], ...)

6. El supervisor no evalúa — va directo a writer

Causa: Falta el nodo evaluate en el grafo, o el edge va de analyst directo a writer.

Solución: El grafo debe tener: analyst → evaluate → (writer | researcher):

builder.add_edge("analyst", "evaluate")
builder.add_conditional_edges("evaluate", route_after_evaluation, {
    "researcher": "researcher",
    "writer": "writer",
})

7. El HITL se ejecuta en cada agente

Causa: Cada agente tiene su propio interrupt().

Solución: Solo el supervisor usa interrupt(). Los workers ejecutan sin interrumpir:

def hitl_approve_node(state):
    response = interrupt(...)

def researcher_node(state):
    result = run_researcher(...)

8. El reporte v5 no es mejor que el v4

Causa: Los agentes especializados usan el mismo modelo (gpt-4.1-mini) y prompts genéricos.

Solución: Diferencia modelos por complejidad. El analyst usa gpt-4.1 porque necesita razonamiento. Los prompts deben ser específicos y restrictivos:

RESEARCHER_PROMPT = "Tu ÚNICA tarea es buscar. NO analices, NO escribas."
ANALYST_PROMPT = "Tu ÚNICA tarea es analizar. NO busques, NO escribas."
WRITER_PROMPT = "Tu ÚNICA tarea es escribir. NO investigues, NO analices."

---

Lo que viene: Módulo 11 — Deep Agents

Tu Research Assistant v5 es un sistema multi-agente completo: 4 agentes especializados, supervisor con HITL, loop-back cuando el analyst necesita más datos, logging por agente, y reportes de calidad superior al v4. Construiste la orquestación manualmente — definiste cada nodo, cada edge, cada condición de routing.

El Módulo 11 introduce Deep Agents: la capa "batteries-included" de LangGraph. Planning automático con write_todos, filesystem virtual para que los agentes guarden artefactos intermedios, spawning de subagentes dinámicos, y memoria a largo plazo con backends pluggables. Reimplementarás el Research Assistant como Deep Agent para ver cómo el framework provee out-of-the-box lo que construiste manualmente en este módulo.

La diferencia clave: en v5 tú decides "researcher → analyst → writer." En v6 (Deep Agent), el sistema decide el plan y lo ejecuta dinámicamente. Entender v5 te da la base para evaluar cuándo la orquestación manual es mejor y cuándo el framework automático es suficiente.

---

Recursos del proyecto

1. LangGraph Multi-Agent Concepts — Arquitecturas multi-agente en LangGraph
2. LangGraph Supervisor Pattern — Implementación del supervisor
3. LangGraph Subgraphs — Agentes como subgrafos independientes
4. Multi-Agent Workflows (LangChain Blog) — Patrones y ejemplos de producción
5. LangGraph Visualization — draw_mermaid para debugging de grafos
6. Python logging Best Practices — Logging con prefijos y niveles

---

Módulo 10 — LangChain & LangGraph: From Chains to Agents