Мультиагентная оркестрация — LangGraph, CrewAI и Mastra против Claude Code

# Пример минимального LangGraph агента (Python)
from typing import TypedDict
from langgraph.graph import StateGraph, END

# Определяем state
class ResearchState(TypedDict):
    query: str
    sources: list[str]
    draft: str
    final: str
    iteration: int

# Ноды — это функции или агенты
def researcher(state: ResearchState) -> ResearchState:
    # здесь агент ищет информацию
    return {"sources": ["source1", "source2"], "iteration": state["iteration"] + 1}

def writer(state: ResearchState) -> ResearchState:
    # здесь агент пишет черновик
    return {"draft": "Draft based on sources..."}

def should_revise(state: ResearchState) -> str:
    # Conditional edge: идти к ревизии или завершить
    if state["iteration"] < 3 and len(state["draft"]) < 500:
        return "revise"
    return "finalize"

# Строим граф
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",  # цикл назад!
    "finalize": END
})

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

# Пример: исследовательская команда CrewAI
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool, WebsiteSearchTool

search_tool = SerperDevTool()
web_tool = WebsiteSearchTool()

# Агент 1: исследователь рынка
market_researcher = Agent(
    role="Market Research Analyst",
    goal="Найти и проанализировать последние данные о рынке AI-инструментов для разработчиков",
    backstory="""Ты опытный аналитик рынка с 10-летним стажем. 
    Специализируешься на технологических рынках. 
    Всегда подкрепляешь выводы конкретными данными и цифрами.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool, web_tool],
    llm="anthropic/claude-sonnet-5-5"  # CrewAI поддерживает Claude, модель задаётся как «провайдер/модель»; актуальные модели: страница «Актуальное сейчас»
)

# Агент 2: технический эксперт
tech_expert = Agent(
    role="Technical Expert",
    goal="Оценить технические возможности и ограничения AI-инструментов",
    backstory="""Ты senior инженер с глубокой экспертизой в AI/ML.
    Умеешь объяснять сложные технические концепты простым языком.
    Скептически относишься к маркетинговым заявлениям, смотришь на реальные benchmarks.""",
    verbose=True,
    allow_delegation=False,
    tools=[web_tool],
    llm="anthropic/claude-sonnet-5-5"
)

# Агент 3: финансовый аналитик
financial_analyst = Agent(
    role="Financial Analyst",
    goal="Оценить ROI и бизнес-потенциал AI-инструментов",
    backstory="""Ты финансовый аналитик специализирующийся на tech-стартапах.
    Умеешь строить unit economics и оценивать рыночный потенциал.
    Всегда смотришь на pricing model и маржинальность.""",
    verbose=True,
    allow_delegation=False,
    tools=[search_tool],
    llm="anthropic/claude-haiku-4-5"  # Задача проще — используем более дешёвую модель
)

# Агент 4: писатель-синтезатор
report_writer = Agent(
    role="Strategic Report Writer",
    goal="Синтезировать исследования в чёткий actionable отчёт",
    backstory="""Ты опытный технический писатель и стратегический консультант.
    Умеешь превращать сложные аналитические данные в чёткие рекомендации.
    Пишешь для аудитории: tech-founders и CTOs.""",
    verbose=True,
    allow_delegation=False,
    tools=[],
    llm="anthropic/claude-sonnet-5-5"
)

# Задачи
market_task = Task(
    description="""Исследуй рынок AI-инструментов для разработчиков в 2026 году.
    Найди: топ-5 игроков, объём рынка, темп роста, ключевые тренды.
    Используй актуальные данные, не старше 6 месяцев.""",
    expected_output="Структурированный анализ рынка: размер, игроки, тренды, цифры",
    agent=market_researcher,
    async_execution=True  # Параллельно с другими задачами!
)

tech_task = Task(
    description="""Проведи технический анализ топ-3 AI coding assistants (Claude Code, Cursor, GitHub Copilot).
    Сравни: поддерживаемые языки, контекстное окно, интеграции, latency, точность.""",
    expected_output="Сравнительная таблица с техническими характеристиками и benchmarks",
    agent=tech_expert,
    async_execution=True  # Параллельно!
)

financial_task = Task(
    description="""Проанализируй ценообразование и ROI для AI coding assistants.
    Рассчитай: стоимость на разработчика/мес, экономия времени, payback period.
    Сравни pricing models: subscription vs usage-based.""",
    expected_output="Финансовая модель с ROI расчётом и сравнением pricing",
    agent=financial_analyst,
    async_execution=True  # Параллельно!
)

synthesis_task = Task(
    description="""Синтезируй все три исследования в итоговый стратегический отчёт.
    Структура: Executive Summary (1 стр), Market Overview, Tech Comparison, Financial Analysis, Recommendations.
    Целевая аудитория: tech-founder принимающий решение о внедрении AI-инструментов.""",
    expected_output="Полный стратегический отчёт 5-7 страниц готовый к презентации",
    agent=report_writer,
    context=[market_task, tech_task, financial_task]  # Зависит от предыдущих
)

# Собираем команду
crew = Crew(
    agents=[market_researcher, tech_expert, financial_analyst, report_writer],
    tasks=[market_task, tech_task, financial_task, synthesis_task],
    process=Process.sequential,  # или Process.hierarchical с manager агентом
    verbose=True,
    memory=True,  # Агенты помнят предыдущие взаимодействия
    max_rpm=10    # Rate limiting для API calls
)

# Запускаем
result = crew.kickoff()
print(result.raw)

pip install crewai crewai-tools

// Пример: Mastra agent с 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();

// Определяем инструменты
const searchTool = createTool({
  id: "web-search",
  description: "Поиск актуальной информации в интернете",
  inputSchema: z.object({
    query: z.string().describe("Поисковый запрос"),
  }),
  outputSchema: z.object({ results: z.array(z.string()) }),
  execute: async ({ query }) => {
    // Здесь реальный поиск через Serper, Tavily и т.д.
    return { results: [`Результаты для: ${query}`] };
  },
});

// Создаём агента-исследователя. Модель задаётся строкой «провайдер/модель»
const researchAgent = new Agent({
  id: "research-agent",
  name: "Research Agent",
  instructions: `Ты профессиональный исследователь. 
  Находишь актуальную информацию по запросу, структурируешь её, 
  выделяешь ключевые факты с источниками.`,
  model: "anthropic/claude-sonnet-5-5",   // актуальные модели: страница «Актуальное сейчас»
  tools: { searchTool },
});

// Шаги workflow: у каждого шага описаны входные и выходные данные
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(
      `Исследуй тему: ${inputData.topic}. Найди актуальные данные и тренды.`
    );
    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: `Сожми это исследование в 3 ключевых вывода:\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();

// Инициализируем Mastra
const mastra = new Mastra({
  agents: { researchAgent },
  workflows: { researchWorkflow },
});

// Запускаем (способ запуска сверь с документацией своей версии Mastra)
const run = await mastra.getWorkflow("researchWorkflow").createRun();
const result = await run.start({ inputData: { topic: "AI agents" } });
console.log(result);

# crew_service.py — оборачиваем CrewAI как 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):
    """
    Этот endpoint вызывается Claude Code orchestrator'ом.
    Внутри запускается параллельная CrewAI команда.
    """
    researcher = Agent(
        role="Senior Researcher",
        goal=f"Найди всё важное о: {request.topic}",
        backstory="Опытный аналитик с доступом к интернету",
        llm="anthropic/claude-sonnet-5-5",
        verbose=False  # В продакшене выключаем verbose
    )

    writer = Agent(
        role="Technical Writer",
        goal="Превратить исследование в чёткий структурированный документ",
        backstory="Специалист по технической документации",
        llm="anthropic/claude-haiku-4-5",
        verbose=False
    )

    research_task = Task(
        description=f"Исследуй: {request.topic}. Найди факты, цифры, тренды.",
        expected_output="Структурированные данные с источниками",
        agent=researcher
    )

    write_task = Task(
        description=f"Напиши {request.output_format} на основе исследования",
        expected_output=f"Готовый {request.output_format} для клиента",
        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}

# В .claude/agents/research-orchestrator.md
# Claude Code вызывает CrewAI как внешний API
curl -X POST http://localhost:8000/research \
  -H "Content-Type: application/json" \
  -d '{"topic": "AI orchestration frameworks", "output_format": "report"}'

Задача простая, линейная, без веток?
  → Claude Code native. Не усложняй.

Нужны циклы "попробовать → исправить → повторить"?
  → LangGraph

Нужно несколько параллельных экспертов над одной темой?
  → CrewAI

Весь стек TypeScript, деплоим на Vercel?
  → Mastra

Нужна корпоративная надёжность, много compliance?
  → Microsoft Agent Framework (наследник AutoGen)

Хочешь смешанный подход?
  → Claude Code orchestrator + CrewAI/LangGraph как внешний сервис

# Создаём проект
mkdir multiagent-crew && cd multiagent-crew
python -m venv venv && source venv/bin/activate  # Windows: venv\Scripts\activate

# Устанавливаем зависимости
pip install crewai crewai-tools anthropic python-dotenv fastapi uvicorn

# Файл .env
echo "ANTHROPIC_API_KEY=sk-ant-your-key-here" > .env
echo "SERPER_API_KEY=your-serper-key" >> .env  # для поиска в интернете

# simple_crew.py — начинаем с минимума
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew

load_dotenv()

# Настраиваем модель для CrewAI
os.environ["ANTHROPIC_API_KEY"] = os.getenv("ANTHROPIC_API_KEY")

# Два агента
researcher = Agent(
    role="Researcher",
    goal="Найти ключевую информацию по теме",
    backstory="Аналитик с опытом работы в технологической индустрии",
    llm="anthropic/claude-haiku-4-5",  # Дешёвая модель для исследований
    verbose=True
)

writer = Agent(
    role="Writer",
    goal="Написать чёткий и полезный контент на основе исследования",
    backstory="Технический писатель специализирующийся на AI-тематике",
    llm="anthropic/claude-sonnet-5-5",  # Более умная модель для финального текста
    verbose=True
)

# Задачи
research_task = Task(
    description="Найди 5 ключевых трендов в мультиагентных AI системах в 2026 году",
    expected_output="Список из 5 трендов с кратким описанием каждого (50-100 слов на тренд)",
    agent=researcher
)

write_task = Task(
    description="""На основе исследования напиши short-form статью (500 слов) 
    для технических разработчиков. Тон: профессиональный, без воды, с конкретикой.""",
    expected_output="Готовая статья 500 слов в markdown формате",
    agent=writer,
    context=[research_task]
)

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

# Запуск
print("Запускаем команду...")
result = crew.kickoff()
print("\n" + "="*50)
print("РЕЗУЛЬТАТ:")
print("="*50)
print(result.raw)

# parallel_crew.py — реальная параллельная команда
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

load_dotenv()

search = SerperDevTool()

# Четыре специализированных эксперта
market_analyst = Agent(
    role="Market Analyst",
    goal="Проанализировать рыночную ситуацию и конкурентов",
    backstory="Аналитик рынка с фокусом на SaaS и AI-инструменты",
    tools=[search],
    llm="anthropic/claude-haiku-4-5",
    verbose=False
)

tech_reviewer = Agent(
    role="Technical Reviewer",
    goal="Оценить техническую сторону и интеграции",
    backstory="Senior разработчик знающий все популярные AI фреймворки",
    tools=[search],
    llm="anthropic/claude-haiku-4-5",
    verbose=False
)

pricing_expert = Agent(
    role="Pricing Expert",
    goal="Разобраться в ценообразовании и ROI",
    backstory="Финансовый аналитик специализирующийся на SaaS метриках",
    tools=[search],
    llm="anthropic/claude-haiku-4-5",
    verbose=False
)

strategy_writer = Agent(
    role="Strategy Consultant",
    goal="Синтезировать все данные в actionable рекомендации",
    backstory="Консультант McKinsey уровня который пишет чёткие executive briefs",
    llm="anthropic/claude-sonnet-5-5",  # Сильная модель для финального синтеза
    verbose=True
)

# Параллельные задачи (async_execution=True)
market_task = Task(
    description="Исследуй рынок мультиагентных AI-фреймворков: размер, игроки, тренды 2026",
    expected_output="Рыночный анализ: 3-5 ключевых игроков, тренды, размер рынка",
    agent=market_analyst,
    async_execution=True  # ПАРАЛЛЕЛЬНО
)

tech_task = Task(
    description="Сравни технически: LangGraph, CrewAI, Mastra — возможности, ограничения, экосистема",
    expected_output="Техническое сравнение с конкретными примерами use cases",
    agent=tech_reviewer,
    async_execution=True  # ПАРАЛЛЕЛЬНО
)

pricing_task = Task(
    description="Разбери pricing models и ROI: сколько стоит использование, какая экономия",
    expected_output="Финансовый анализ с конкретными цифрами и расчётом ROI",
    agent=pricing_expert,
    async_execution=True  # ПАРАЛЛЕЛЬНО
)

# Синтез — ждёт все три параллельных задачи
synthesis_task = Task(
    description="""Синтезируй три анализа в executive brief для технического директора.
    Структура: Ситуация → Выводы → Рекомендации (топ-3 actionable).
    Максимум 1 страница. Никакой воды.""",
    expected_output="Executive brief в markdown, 1 страница, готовый к презентации",
    agent=strategy_writer,
    context=[market_task, tech_task, pricing_task]  # Зависит от всех трёх
)

# Собираем команду
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  # В примере отключаем для скорости
)

# Запуск
print("Параллельная команда запускается...")
print("market_analyst, tech_reviewer, pricing_expert работают одновременно")
print("strategy_writer начнёт когда все трое завершат")
print("-" * 50)

result = crew.kickoff()

# Сохраняем результат
with open("report.md", "w", encoding="utf-8") as f:
    f.write(result.raw)

print("\nОтчёт сохранён в report.md")

# Запускаем простую команду
python simple_crew.py

# Запускаем параллельную команду
python parallel_crew.py

# Если нужна отладка — ставим verbose=True у агентов и у Crew в коде

# Смотрим на расход токенов
# В конце CrewAI показывает total tokens usage

# Ошибка: модель не найдена
# Убедись что ANTHROPIC_API_KEY задан в .env
# CrewAI ищет ключ через os.environ

# Ошибка: rate limit
# Добавь max_rpm параметр в Crew:
crew = Crew(..., max_rpm=5)  # максимум 5 запросов в минуту

# Результат не сохранился в context
# Убедись что context=[task1, task2] в зависимой задаче

# api_service.py — CrewAI как микросервис
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 вызывает этот endpoint через bash/curl.
    Внутри — параллельная CrewAI команда.
    """
    researcher = Agent(
        role="Researcher",
        goal=f"Исследуй тему досконально: {req.topic}",
        backstory="Опытный аналитик",
        llm="anthropic/claude-haiku-4-5",
        verbose=False
    )

    writer = Agent(
        role="Writer",
        goal=f"Напиши в стиле: {req.style}",
        backstory="Технический писатель",
        llm="anthropic/claude-sonnet-5-5",
        verbose=False
    )

    research_task = Task(
        description=f"Исследуй: {req.topic}",
        expected_output="Детальный анализ с фактами",
        agent=researcher
    )

    write_task = Task(
        description=f"Напиши в стиле '{req.style}' на основе исследования",
        expected_output="Готовый документ",
        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)

# Запускаем сервис
python api_service.py

# Claude Code теперь может вызвать его:
curl -X POST http://localhost:8001/research \
  -H "Content-Type: application/json" \
  -d '{"topic": "Multiagent orchestration trends", "style": "executive"}'