CrewAI y Workflows Multi-Agente

¿Qué es un Sistema Multi-Agente?

Un sistema multi-agente divide una tarea compleja entre varios agentes especializados que colaboran. Cada agente tiene un rol, objetivo y conjunto de herramientas específico.

                    ┌──────────────┐
                    │ Orchestrator │
                    └──────┬───────┘
                           │
              ┌────────────┼────────────┐
              ▼            ▼            ▼
        ┌──────────┐ ┌──────────┐ ┌──────────┐
        │Researcher│ │ Analyst  │ │  Writer  │
        │  Agent   │ │  Agent   │ │  Agent   │
        └────┬─────┘ └────┬─────┘ └────┬─────┘
             │            │            │
        [Web Search] [SQL Query]  [Templates]
        [RAG]        [Calculator] [Formatter]

¿Cuándo Usar Multi-Agente vs Single Agent?

Criterio	Single Agent	Multi-Agente
Complejidad de tarea	Baja-media	Alta
Herramientas necesarias	< 5	> 5
Roles distintos	No	Sí
Necesidad de paralelismo	No	Sí
Overheard aceptable	Bajo	Medio-alto

CrewAI: Framework Multi-Agente

Conceptos Clave

Crew = Equipo de agentes que trabajan juntos

Agent  → Rol + Objetivo + Herramientas + LLM
Task   → Descripción + Expected output + Agent asignado
Crew   → Lista de agentes + Lista de tareas + Proceso (secuencial/jerárquico)

Ejemplo: Equipo de Análisis de Competencia

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

# ── Herramientas ──────────────────────────────────────
search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# ── Agentes ───────────────────────────────────────────
researcher = Agent(
    role="Investigador de Mercado Senior",
    goal="Encontrar información actualizada y relevante sobre competidores y tendencias",
    # backstory: contexto/personalidad que se inyecta en el system prompt del LLM.
    # Cuanto más específico, mejor se comporta el agente en su rol.
    backstory=(
        "Eres un investigador de mercado con 15 años de experiencia. "
        "Tienes un talento especial para encontrar datos relevantes y "
        "distinguir información confiable de ruido."
    ),
    tools=[search_tool, web_tool],
    llm="gpt-4o",
    verbose=True,   # Imprime el razonamiento paso a paso (útil para debug)
    max_iter=5,     # Máx iteraciones del loop ReAct (previene loops infinitos)
)

analyst = Agent(
    role="Analista de Estrategia",
    goal="Analizar datos de mercado y generar insights accionables",
    backstory=(
        "Eres un analista estratégico que convierte datos raw en "
        "insights de negocio claros. Especialista en análisis FODA "
        "y posicionamiento competitivo."
    ),
    llm="gpt-4o",
    verbose=True,
)

writer = Agent(
    role="Redactor de Reportes Ejecutivos",
    goal="Crear reportes claros, concisos y accionables para C-level",
    backstory=(
        "Eres un redactor que transforma análisis complejos en "
        "documentos ejecutivos que cualquier directivo puede entender "
        "y actuar sobre ellos."
    ),
    llm="gpt-4o",
    verbose=True,
)

# ── Tareas ────────────────────────────────────────────
research_task = Task(
    # {company} y {sector} son template variables: se reemplazan
    # automáticamente con los valores pasados en crew.kickoff(inputs={...})
    description=(
        "Investiga a los principales competidores de {company} en el sector {sector}. "
        "Encuentra: productos, precios, funding, equipo, tecnología y movimientos recientes."
    ),
    # expected_output: le indica al agente qué formato/calidad se espera.
    # Funciona como criterio de aceptación para el output.
    expected_output="Lista detallada de 5+ competidores con datos verificados.",
    agent=researcher,
)

analysis_task = Task(
    description=(
        "Analiza los datos de competidores y genera: "
        "1. Matriz comparativa de features "
        "2. Análisis FODA de {company} vs competidores "
        "3. Oportunidades y amenazas clave "
        "4. Recomendaciones estratégicas"
    ),
    expected_output="Análisis FODA completo con recomendaciones priorizadas.",
    agent=analyst,
    # context: recibe automáticamente el OUTPUT de la(s) tarea(s) indicada(s).
    # El agente analyst ve el resultado de research_task en su prompt.
    context=[research_task],
)

report_task = Task(
    description=(
        "Crea un reporte ejecutivo de máximo 2 páginas que incluya: "
        "Executive summary, hallazgos clave, matriz competitiva, "
        "recomendaciones top 3 con timeline."
    ),
    expected_output="Reporte ejecutivo en formato markdown profesional.",
    agent=writer,
    context=[research_task, analysis_task],
    output_file="competitive_report.md",  # Guarda el resultado en archivo automáticamente
)

# ── Crew ──────────────────────────────────────────────
crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.sequential,  # o Process.hierarchical
    verbose=True,
)

# Ejecutar
result = crew.kickoff(inputs={
    "company": "Acme AI",
    "sector": "Enterprise AI automation",
})

Proceso Jerárquico (Manager Agent)

crew = Crew(
    agents=[researcher, analyst, writer],
    tasks=[research_task, analysis_task, report_task],
    process=Process.hierarchical,
    manager_llm="gpt-4o",  # Manager que coordina
    # El manager decide qué agente trabaja en qué y cuándo
)

LangGraph para Multi-Agente

Para control más fino que CrewAI, usa LangGraph directamente.

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated, Literal
import operator

# Patrón de estado compartido: cada campo se llena progresivamente.
# Los agentes leen campos previos y escriben su campo correspondiente.
class MultiAgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_agent: str
    research_data: str   # Llenado por researcher
    analysis: str        # Llenado por analyst (lee research_data)
    final_report: str    # Llenado por writer (lee research_data + analysis)

def router(state: MultiAgentState) -> Literal["researcher", "analyst", "writer", "end"]:
    """Decide qué agente actúa siguiente.
    Lógica del router: revisa qué campos están vacíos para determinar
    la secuencia. Así se controla el flujo sin hardcodear el orden."""
    if not state.get("research_data"):
        return "researcher"
    elif not state.get("analysis"):
        return "analyst"
    elif not state.get("final_report"):
        return "writer"
    return "end"

# Cada nodo lee del estado compartido y escribe su resultado en él.
# Así los agentes se pasan datos sin acoplamiento directo.
def researcher_node(state: MultiAgentState):
    result = researcher_chain.invoke(state["messages"])
    return {"research_data": result, "messages": [{"role": "assistant", "content": result}]}

def analyst_node(state: MultiAgentState):
    result = analyst_chain.invoke({
        "research": state["research_data"],  # Lee output de researcher
        "messages": state["messages"],
    })
    return {"analysis": result, "messages": [{"role": "assistant", "content": result}]}

def writer_node(state: MultiAgentState):
    result = writer_chain.invoke({
        "research": state["research_data"],  # Lee output de researcher
        "analysis": state["analysis"],       # Lee output de analyst
    })
    return {"final_report": result}

# Construir grafo
graph = StateGraph(MultiAgentState)
graph.add_node("researcher", researcher_node)
graph.add_node("analyst", analyst_node)
graph.add_node("writer", writer_node)

# add_conditional_edges: el router se evalúa DESPUÉS de cada nodo.
# Tras ejecutar un nodo, el router inspecciona el estado y decide el siguiente.
graph.add_conditional_edges(START, router)
graph.add_conditional_edges("researcher", router)
graph.add_conditional_edges("analyst", router)
graph.add_edge("writer", END)  # writer siempre termina el flujo

multi_agent = graph.compile()

Patrones de Comunicación Multi-Agente

1. Secuencial (Pipeline)

Agent A ──► Agent B ──► Agent C ──► Output

Cada agente recibe el output del anterior. Simple y predecible.

2. Jerárquico (Manager)

Manager ──┬── Asigna a Agent A
          ├── Revisa output
          ├── Asigna a Agent B  
          └── Compila resultado final

Un agente manager coordina a los demás.

3. Debate/Consenso

Agent A ──► Propuesta ──┐
Agent B ──► Propuesta ──┼── Evaluador ──► Mejor propuesta
Agent C ──► Propuesta ──┘

Múltiples agentes proponen soluciones, un evaluador elige.

4. Supervisor con Workers

from langgraph.prebuilt import create_react_agent

# Workers especializados
search_agent = create_react_agent(llm, tools=[search_tool])
sql_agent = create_react_agent(llm, tools=[sql_tool])
code_agent = create_react_agent(llm, tools=[code_tool])

# Supervisor que delega
def supervisor(state):
    response = supervisor_llm.invoke(
        f"¿Qué agente debe manejar esto? Opciones: search, sql, code, FINISH\n"
        f"Tarea: {state['messages'][-1]}"
    )
    return {"next": response.content.strip().lower()}

Costos y Latencia en Multi-Agente

Single Agent:
  1 LLM call × $0.01 = $0.01, 2s latencia

Multi-Agente (3 agentes secuenciales):
  3 agentes × 3 iteraciones × $0.01 = $0.09, 18s latencia
  (9x más caro, 9x más lento)

Multi-Agente (3 agentes paralelos):
  3 agentes × 2 iteraciones × $0.01 = $0.06, 4s latencia
  (6x más caro, 2x más lento)

Regla: Solo usa multi-agente cuando la tarea lo justifica. La complejidad y el costo crecen rápido.

Resumen

CrewAI — Framework high-level para equipos de agentes
LangGraph — Control granular de workflows multi-agente
Patrones: secuencial, jerárquico, debate, supervisor
Trade-offs: costo/latencia vs capacidad de resolver problemas complejos
Siempre medir si multi-agente mejora realmente sobre single-agent

🧠 Preguntas de Repaso

1. Comparado con un single agent, un sistema multi-agente secuencial de 3 agentes con 3 iteraciones cada uno es aproximadamente:

A) 3x más caro y 3x más lento
B) 9x más caro y 9x más lento
C) Igual de caro pero 3x más lento
D) 2x más caro y más rápido gracias al paralelismo

Respuesta: B) — Un sistema multi-agente secuencial (3 agentes × 3 iteraciones) cuesta ~$0.09 vs $0.01 del single agent (9x más caro) y tarda ~18s vs 2s (9x más lento). Este overhead significativo justifica usar multi-agente solo cuando la calidad lo amerita.

2. En CrewAI, ¿qué diferencia hay entre Process.sequential y Process.hierarchical?

A) Sequential es más rápido, hierarchical es más preciso
B) En sequential las tareas se ejecutan en orden fijo, en hierarchical un manager_llm coordina y asigna tareas dinámicamente
C) Sequential usa un solo agente, hierarchical usa múltiples
D) No hay diferencia funcional, solo en nomenclatura

Respuesta: B) — En Process.sequential las tareas se ejecutan en orden predefinido pasando output de una a otra. En Process.hierarchical se especifica un manager_llm (ej: "gpt-4o") que coordina dinámicamente a los agentes, asigna tareas y revisa resultados.

3. ¿Cuándo se recomienda usar un sistema multi-agente en vez de un agente único?

A) Siempre, porque múltiples agentes son más precisos
B) Cuando se necesitan más de 5 herramientas, hay roles claramente distintos, y el overhead adicional es aceptable
C) Solo cuando el presupuesto es ilimitado
D) Cuando la tarea es simple y bien definida

Respuesta: B) — Multi-agente se justifica cuando: se necesitan >5 herramientas, hay roles claramente distintos (investigador, analista, escritor), y el overhead en costo/latencia es aceptable dado el valor del resultado.

4. En el patrón de debate/consenso para multi-agente, ¿cómo se selecciona la mejor propuesta?

A) Se elige al azar
B) Múltiples agentes generan propuestas independientes y un evaluador selecciona la mejor basándose en criterios definidos
C) Se promedian todas las respuestas
D) El primer agente que responda gana

Respuesta: B) — En el patrón de debate, múltiples agentes generan propuestas independientes sobre el mismo problema, y un agente evaluador analiza todas las propuestas y selecciona la mejor basándose en criterios de calidad predefinidos.

CrewAI y Sistemas Multi-Agente

Estás en modo lectura