Orquestación multiagente: LangGraph, CrewAI y Mastra frente a Claude Code

# Ejemplo de un agente mínimo en LangGraph (Python)
from typing import TypedDict
from langgraph.graph import StateGraph, END

# Definimos el estado
class ResearchState(TypedDict):
    query: str
    sources: list[str]
    draft: str
    final: str
    iteration: int

# Los nodos son funciones o agentes
def researcher(state: ResearchState) -> ResearchState:
    # aquí el agente busca información
    return {"sources": ["source1", "source2"], "iteration": state["iteration"] + 1}

def writer(state: ResearchState) -> ResearchState:
    # aquí el agente escribe un borrador
    return {"draft": "Draft based on sources..."}

def should_revise(state: ResearchState) -> str:
    # Arista condicional: ir a revisión o terminar
    if state["iteration"] < 3 and len(state["draft"]) < 500:
        return "revise"
    return "finalize"

# Construimos el grafo
graph = StateGraph(ResearchState)
graph.add_node("researcher", researcher)
graph.add_node("writer", writer)

graph.set_entry_point("researcher")
graph.add_edge("researcher", "writer")
graph.add_conditional_edges("writer", should_revise, {
    "revise": "researcher",  # ¡ciclo hacia atrás!
    "finalize": END
})

app = graph.compile()
result = app.invoke({"query": "AI trends", "iteration": 0})

# Ejemplo: equipo de investigación con CrewAI
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# Agente 1: investigador de mercado
market_researcher = Agent(
    role="Market Research Analyst",
    goal="Encontrar y analizar los datos más recientes del mercado de herramientas de IA para desarrolladores",
    backstory="""Eres un analista de mercado con 10 años de experiencia.
    Te especializas en mercados tecnológicos.
    Siempre respaldas tus conclusiones con datos y cifras concretas.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, web_tool],
    llm="anthropic/claude-sonnet-5-5"  # CrewAI soporta Claude; el modelo se indica como «proveedor/modelo»; modelos vigentes: página «Lo vigente»
)

# Agente 2: experto técnico
tech_expert = Agent(
    role="Technical Expert",
    goal="Evaluar las capacidades y limitaciones técnicas de las herramientas de IA",
    backstory="""Eres un ingeniero senior con experiencia profunda en IA/ML.
    Sabes explicar conceptos técnicos complejos en lenguaje sencillo.
    Eres escéptico ante las afirmaciones de marketing y miras los benchmarks reales.""",
    verbose=True,
    allow_delegation=False,
    tools=[web_tool],
    llm="anthropic/claude-sonnet-5-5"
)

# Agente 3: analista financiero
financial_analyst = Agent(
    role="Financial Analyst",
    goal="Evaluar el ROI y el potencial de negocio de las herramientas de IA",
    backstory="""Eres un analista financiero especializado en startups tecnológicas.
    Sabes armar unit economics y evaluar el potencial de mercado.
    Siempre revisas el modelo de precios y el margen.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool],
    llm="anthropic/claude-haiku-4-5"  # La tarea es más simple: usamos un modelo más barato
)

# Agente 4: redactor que sintetiza
report_writer = Agent(
    role="Strategic Report Writer",
    goal="Sintetizar las investigaciones en un reporte claro y accionable",
    backstory="""Eres un redactor técnico y consultor estratégico con experiencia.
    Sabes convertir datos analíticos complejos en recomendaciones claras.
    Escribes para fundadores técnicos y CTOs.""",
    verbose=True,
    allow_delegation=False,
    tools=[],
    llm="anthropic/claude-sonnet-5-5"
)

# Tareas
market_task = Task(
    description="""Investiga el mercado de herramientas de IA para desarrolladores en 2026.
    Encuentra: los 5 principales jugadores, el tamaño del mercado, la tasa de crecimiento, las tendencias clave.
    Usa datos actuales, de no más de 6 meses.""",
    expected_output="Análisis de mercado estructurado: tamaño, jugadores, tendencias, cifras",
    agent=market_researcher,
    async_execution=True  # ¡En paralelo con las demás tareas!
)

tech_task = Task(
    description="""Haz un análisis técnico de los 3 principales asistentes de código con IA (Claude Code, Cursor, GitHub Copilot).
    Compara: lenguajes soportados, ventana de contexto, integraciones, latencia, precisión.""",
    expected_output="Tabla comparativa con características técnicas y benchmarks",
    agent=tech_expert,
    async_execution=True  # ¡En paralelo!
)

financial_task = Task(
    description="""Analiza los precios y el ROI de los asistentes de código con IA.
    Calcula: costo por desarrollador al mes, ahorro de tiempo, periodo de recuperación.
    Compara modelos de precios: suscripción vs. pago por uso.""",
    expected_output="Modelo financiero con cálculo de ROI y comparación de precios",
    agent=financial_analyst,
    async_execution=True  # ¡En paralelo!
)

synthesis_task = Task(
    description="""Sintetiza las tres investigaciones en un reporte estratégico final.
    Estructura: Executive Summary (1 pág.), Market Overview, Tech Comparison, Financial Analysis, Recommendations.
    Público objetivo: un fundador técnico que decide si adopta herramientas de IA.""",
    expected_output="Reporte estratégico completo de 5-7 páginas, listo para presentar",
    agent=report_writer,
    context=[market_task, tech_task, financial_task]  # Depende de las anteriores
)

# Armamos el equipo
crew = Crew(
    agents=[market_researcher, tech_expert, financial_analyst, report_writer],
    tasks=[market_task, tech_task, financial_task, synthesis_task],
    process=Process.sequential,  # o Process.hierarchical con un agente manager
    verbose=True,
    memory=True,  # Los agentes recuerdan las interacciones anteriores
    max_rpm=10    # Rate limiting para las llamadas a la API
)

# Ejecutamos
result = crew.kickoff()
print(result.raw)

pip install crewai crewai-tools

// Ejemplo: agente de Mastra con un workflow (TypeScript)
import { Mastra } from "@mastra/core";
import { Agent } from "@mastra/core/agent";
import { createTool } from "@mastra/core/tools";
import { createStep, createWorkflow } from "@mastra/core/workflows";
import { z } from "zod";
import Anthropic from "@anthropic-ai/sdk";

const anthropic = new Anthropic();

// Definimos las herramientas
const searchTool = createTool({
  id: "web-search",
  description: "Búsqueda de información actual en internet",
  inputSchema: z.object({
    query: z.string().describe("Consulta de búsqueda"),
  }),
  outputSchema: z.object({ results: z.array(z.string()) }),
  execute: async ({ query }) => {
    // Aquí va la búsqueda real con Serper, Tavily, etc.
    return { results: [`Resultados para: ${query}`] };
  },
});

// Creamos el agente investigador. El modelo se indica con la cadena «proveedor/modelo»
const researchAgent = new Agent({
  id: "research-agent",
  name: "Research Agent",
  instructions: `Eres un investigador profesional.
  Encuentras información actual sobre la consulta, la estructuras
  y destacas los datos clave con sus fuentes.`,
  model: "anthropic/claude-sonnet-5-5",   // modelos vigentes: página «Lo vigente»
  tools: { searchTool },
});

// Pasos del workflow: cada paso describe sus datos de entrada y salida
const researchStep = createStep({
  id: "research",
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: z.object({ research: z.string() }),
  execute: async ({ inputData, mastra }) => {
    const agent = mastra.getAgent("researchAgent");
    const result = await agent.generate(
      `Investiga el tema: ${inputData.topic}. Encuentra datos y tendencias actuales.`
    );
    return { research: result.text };
  },
});

const summarizeStep = createStep({
  id: "summarize",
  inputSchema: z.object({ research: z.string() }),
  outputSchema: z.object({ summary: z.string() }),
  execute: async ({ inputData }) => {
    const response = await anthropic.messages.create({
      model: "claude-haiku-4-5",
      max_tokens: 500,
      messages: [
        {
          role: "user",
          content: `Resume esta investigación en 3 conclusiones clave:\n\n${inputData.research}`,
        },
      ],
    });
    const summary = response.content.map((b) => (b.type === "text" ? b.text : "")).join("");
    return { summary };
  },
});

const researchWorkflow = createWorkflow({
  id: "research-workflow",
  inputSchema: z.object({ topic: z.string() }),
  outputSchema: z.object({ summary: z.string() }),
})
  .then(researchStep)
  .then(summarizeStep)
  .commit();

// Inicializamos Mastra
const mastra = new Mastra({
  agents: { researchAgent },
  workflows: { researchWorkflow },
});

// Ejecutamos (revisa la forma de ejecución en la documentación de tu versión de Mastra)
const run = await mastra.getWorkflow("researchWorkflow").createRun();
const result = await run.start({ inputData: { topic: "AI agents" } });
console.log(result);

# crew_service.py — envolvemos CrewAI como un servicio FastAPI
from fastapi import FastAPI
from pydantic import BaseModel
from crewai import Agent, Task, Crew, Process

app = FastAPI()

class ResearchRequest(BaseModel):
    topic: str
    output_format: str = "report"

@app.post("/research")
async def run_research_crew(request: ResearchRequest):
    """
    Este endpoint lo llama el orchestrator de Claude Code.
    Dentro se lanza un equipo de CrewAI en paralelo.
    """
    researcher = Agent(
        role="Senior Researcher",
        goal=f"Encuentra todo lo importante sobre: {request.topic}",
        backstory="Analista con experiencia y acceso a internet",
        llm="anthropic/claude-sonnet-5-5",
        verbose=False  # En producción desactivamos verbose
    )

    writer = Agent(
        role="Technical Writer",
        goal="Convertir la investigación en un documento claro y estructurado",
        backstory="Especialista en documentación técnica",
        llm="anthropic/claude-haiku-4-5",
        verbose=False
    )

    research_task = Task(
        description=f"Investiga: {request.topic}. Encuentra hechos, cifras y tendencias.",
        expected_output="Datos estructurados con fuentes",
        agent=researcher
    )

    write_task = Task(
        description=f"Escribe un {request.output_format} basado en la investigación",
        expected_output=f"Un {request.output_format} listo para el cliente",
        agent=writer,
        context=[research_task]
    )

    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, write_task],
        process=Process.sequential,
        verbose=False
    )

    result = crew.kickoff()
    return {"result": result.raw, "topic": request.topic}

# En .claude/agents/research-orchestrator.md
# Claude Code llama a CrewAI como una API externa
curl -X POST http://localhost:8000/research \
  -H "Content-Type: application/json" \
  -d '{"topic": "AI orchestration frameworks", "output_format": "report"}'

¿La tarea es simple, lineal, sin ramas?
  → Claude Code nativo. No lo compliques.

¿Necesitas ciclos "intentar → corregir → repetir"?
  → LangGraph

¿Necesitas varios expertos en paralelo sobre un mismo tema?
  → CrewAI

¿Todo el stack es TypeScript y despliegas en Vercel?
  → Mastra

¿Necesitas confiabilidad corporativa y mucho cumplimiento normativo?
  → Microsoft Agent Framework (sucesor de AutoGen)

¿Quieres un enfoque mixto?
  → Claude Code como orchestrator + CrewAI/LangGraph como servicio externo

# Creamos el proyecto
mkdir multiagent-crew && cd multiagent-crew
python -m venv venv && source venv/bin/activate  # Windows: venv\Scripts\activate

# Instalamos las dependencias
pip install crewai crewai-tools anthropic python-dotenv fastapi uvicorn

# Archivo .env
echo "ANTHROPIC_API_KEY=sk-ant-your-key-here" > .env
echo "SERPER_API_KEY=your-serper-key" >> .env  # para buscar en internet

# simple_crew.py — empezamos con lo mínimo
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew

load_dotenv()

# Configuramos el modelo para CrewAI
os.environ["ANTHROPIC_API_KEY"] = os.getenv("ANTHROPIC_API_KEY")

# Dos agentes
researcher = Agent(
    role="Researcher",
    goal="Encontrar la información clave sobre el tema",
    backstory="Analista con experiencia en la industria tecnológica",
    llm="anthropic/claude-haiku-4-5",  # Modelo barato para investigar
    verbose=True
)

writer = Agent(
    role="Writer",
    goal="Escribir contenido claro y útil basado en la investigación",
    backstory="Redactor técnico especializado en temas de IA",
    llm="anthropic/claude-sonnet-5-5",  # Modelo más listo para el texto final
    verbose=True
)

# Tareas
research_task = Task(
    description="Encuentra 5 tendencias clave en sistemas de IA multiagente en 2026",
    expected_output="Lista de 5 tendencias con una descripción breve de cada una (50-100 palabras por tendencia)",
    agent=researcher
)

write_task = Task(
    description="""Con base en la investigación, escribe un artículo corto (500 palabras)
    para desarrolladores. Tono: profesional, sin relleno, con datos concretos.""",
    expected_output="Artículo de 500 palabras listo, en formato markdown",
    agent=writer,
    context=[research_task]
)

# Crew
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, write_task],
    verbose=True
)

# Ejecución
print("Lanzando el equipo...")
result = crew.kickoff()
print("\n" + "="*50)
print("RESULTADO:")
print("="*50)
print(result.raw)

# parallel_crew.py — un equipo real en paralelo
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

load_dotenv()

search = SerperDevTool()

# Cuatro expertos especializados
market_analyst = Agent(
    role="Market Analyst",
    goal="Analizar la situación del mercado y la competencia",
    backstory="Analista de mercado enfocado en SaaS y herramientas de IA",
    tools=[search],
    llm="anthropic/claude-haiku-4-5",
    verbose=False
)

tech_reviewer = Agent(
    role="Technical Reviewer",
    goal="Evaluar el lado técnico y las integraciones",
    backstory="Desarrollador senior que conoce todos los frameworks de IA populares",
    tools=[search],
    llm="anthropic/claude-haiku-4-5",
    verbose=False
)

pricing_expert = Agent(
    role="Pricing Expert",
    goal="Entender los precios y el ROI",
    backstory="Analista financiero especializado en métricas de SaaS",
    tools=[search],
    llm="anthropic/claude-haiku-4-5",
    verbose=False
)

strategy_writer = Agent(
    role="Strategy Consultant",
    goal="Sintetizar todos los datos en recomendaciones accionables",
    backstory="Consultor de nivel McKinsey que escribe executive briefs claros",
    llm="anthropic/claude-sonnet-5-5",  # Modelo fuerte para la síntesis final
    verbose=True
)

# Tareas en paralelo (async_execution=True)
market_task = Task(
    description="Investiga el mercado de frameworks de IA multiagente: tamaño, jugadores, tendencias 2026",
    expected_output="Análisis de mercado: 3-5 jugadores clave, tendencias, tamaño del mercado",
    agent=market_analyst,
    async_execution=True  # EN PARALELO
)

tech_task = Task(
    description="Compara técnicamente LangGraph, CrewAI y Mastra: capacidades, limitaciones, ecosistema",
    expected_output="Comparación técnica con ejemplos concretos de casos de uso",
    agent=tech_reviewer,
    async_execution=True  # EN PARALELO
)

pricing_task = Task(
    description="Analiza los modelos de precios y el ROI: cuánto cuesta usarlos, qué ahorro hay",
    expected_output="Análisis financiero con cifras concretas y cálculo del ROI",
    agent=pricing_expert,
    async_execution=True  # EN PARALELO
)

# Síntesis: espera las tres tareas en paralelo
synthesis_task = Task(
    description="""Sintetiza los tres análisis en un executive brief para el director técnico.
    Estructura: Situación → Conclusiones → Recomendaciones (las 3 más accionables).
    Máximo 1 página. Nada de relleno.""",
    expected_output="Executive brief en markdown, 1 página, listo para presentar",
    agent=strategy_writer,
    context=[market_task, tech_task, pricing_task]  # Depende de las tres
)

# Armamos el equipo
crew = Crew(
    agents=[market_analyst, tech_reviewer, pricing_expert, strategy_writer],
    tasks=[market_task, tech_task, pricing_task, synthesis_task],
    process=Process.sequential,
    verbose=True,
    memory=False  # En el ejemplo la desactivamos para ir más rápido
)

# Ejecución
print("Arranca el equipo en paralelo...")
print("market_analyst, tech_reviewer y pricing_expert trabajan al mismo tiempo")
print("strategy_writer empieza cuando los tres terminen")
print("-" * 50)

result = crew.kickoff()

# Guardamos el resultado
with open("report.md", "w", encoding="utf-8") as f:
    f.write(result.raw)

print("\nReporte guardado en report.md")

# Ejecutamos el equipo simple
python simple_crew.py

# Ejecutamos el equipo en paralelo
python parallel_crew.py

# Si hace falta depurar, pon verbose=True en los agentes y en el Crew dentro del código

# Revisamos el consumo de tokens
# Al final, CrewAI muestra el total tokens usage

# Error: modelo no encontrado
# Asegúrate de que ANTHROPIC_API_KEY esté en .env
# CrewAI busca la clave en os.environ

# Error: rate limit
# Agrega el parámetro max_rpm al Crew:
crew = Crew(..., max_rpm=5)  # máximo 5 solicitudes por minuto

# El resultado no llegó al context
# Asegúrate de tener context=[task1, task2] en la tarea dependiente

# api_service.py — CrewAI como microservicio
from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn
from crewai import Agent, Task, Crew

app = FastAPI(title="Research Crew API")

class ResearchRequest(BaseModel):
    topic: str
    style: str = "technical"  # technical, executive, casual

@app.post("/research")
async def research(req: ResearchRequest):
    """
    Claude Code llama a este endpoint con bash/curl.
    Dentro hay un equipo de CrewAI.
    """
    researcher = Agent(
        role="Researcher",
        goal=f"Investiga el tema a fondo: {req.topic}",
        backstory="Analista con experiencia",
        llm="anthropic/claude-haiku-4-5",
        verbose=False
    )

    writer = Agent(
        role="Writer",
        goal=f"Escribe con el estilo: {req.style}",
        backstory="Redactor técnico",
        llm="anthropic/claude-sonnet-5-5",
        verbose=False
    )

    research_task = Task(
        description=f"Investiga: {req.topic}",
        expected_output="Análisis detallado con datos",
        agent=researcher
    )

    write_task = Task(
        description=f"Escribe con el estilo '{req.style}' basándote en la investigación",
        expected_output="Documento listo",
        agent=writer,
        context=[research_task]
    )

    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, write_task]
    )

    result = crew.kickoff()
    return {
        "topic": req.topic,
        "result": result.raw,
        "tokens_used": result.token_usage.total_tokens if result.token_usage else 0
    }

@app.get("/health")
def health():
    return {"status": "ok"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8001)

# Levantamos el servicio
python api_service.py

# Ahora Claude Code puede llamarlo:
curl -X POST http://localhost:8001/research \
  -H "Content-Type: application/json" \
  -d '{"topic": "Multiagent orchestration trends", "style": "executive"}'