Multi-agent-система — это архитектура, в которой несколько независимых AI-агентов работают вместе над одной задачей. Каждый агент — это изолированная LLM-сессия со своим промптом, ролью и набором инструментов. В отличие от монолитного чат-бота, multi-agent-пайплайн разделяет ответственность: один агент ищет данные, второй анализирует, третий пишет код, четвёртый проверяет результат.
▸ АРХИТЕКТУРА КОМАНДЫ (пример):
┌─────────────┐
│ ORCHESTRATOR│ ← Раздаёт задачи, собирает результаты
└──────┬──────┘
│
┌──────┼──────┬──────────┐
▼ ▼ ▼ ▼
┌────┐ ┌────┐ ┌────┐ ┌────┐
│🔍 │ │📝 │ │🐍 │ │✅ │
│Rsch│ │Wrt │ │Code│ │Rev │
└────┘ └────┘ └────┘ └────┘
Researcher Writer Coder Reviewer
Ключевое преимущество — специализация. Агент-исследователь получает длинный системный промпт про фактчекинг, агент-кодер — про чистый код. Вместе они выдают результат, который одиночный LLM не осилит. Плюс — параллелизм: Researcher и Coder могут работать одновременно, сокращая общее время выполнения в 2–4 раза.
Для начала нам понадобится Python 3.11+, несколько библиотек и API-ключ к LLM-провайдеру. Я использую CrewAI — самую зрелую библиотеку для multi-agent-систем на середину 2026. Альтернативы: AutoGen (Microsoft), LangGraph (LangChain), TaskWeaver.
▸ Шаг 1: Установка зависимостей
pip install crewai==3.2.0 crewai-tools==1.4.1 pip install litellm python-dotenv rich loguru # Для локальных моделей (опционально): pip install llama-cpp-python chromadb
▸ Шаг 2: Переменные окружения
Создайте файл .env в корне проекта. CrewAI поддерживает OpenAI, Anthropic, Groq, и локальные модели через LiteLLM.
# .env OPENAI_API_KEY="sk-your-key-here" OPENAI_MODEL_NAME="gpt-4o" # Альтернативно — Groq (быстрее, дешевле): # GROQ_API_KEY="gsk_your_key" # MODEL_NAME="groq/llama-4-70b" # Логирование CREWAI_VERBOSE="true"
Агент в CrewAI — это объект с тремя обязательными атрибутами: role (роль), goal (цель) и backstory (контекст). К ним добавляются инструменты (tools) — функции, которые агент может вызывать: поиск в интернете, чтение файлов, запросы к API.
# agents.py — определение агентов from crewai import Agent, Task from crewai_tools import SerperDevTool, FileReadTool # Инструменты search_tool = SerperDevTool() # поиск в Google file_reader = FileReadTool() # чтение файлов с диска # ── Агент-исследователь ── researcher = Agent( role="Старший исследователь", goal="Найти релевантные данные по запросу. " "Всегда проверять минимум 3 источника.", backstory=( "Ты — исследователь с 15-летним опытом. " "Твоя специализация: фактчекинг и агрегация данных. " "Ты не делаешь выводов — только собираешь факты." ), tools=[search_tool, file_reader], verbose=True, allow_delegation=False, llm="gpt-4o" # можно указать модель явно ) # ── Агент-писатель ── writer = Agent( role="Технический писатель", goal="Превратить сырые данные в структурированный " "текст на русском языке.", backstory=( "Ты — автор технической документации. " "Твой стиль: чётко, по делу, с примерами. " "Ты адаптируешь сложные концепции для широкой аудитории." ), verbose=True, allow_delegation=True, ) # ── Агент-кодер ── coder = Agent( role="Senior Python разработчик", goal="Написать рабочий, протестированный код " "по спецификации.", backstory=( "Ты — Python-разработчик с фокусом на качество. " "Ты всегда добавляешь type hints, docstrings и тесты. " "Ты не генерируешь boilerplate, если он не нужен." ), verbose=True, allow_delegation=False, ) print("[OK] Агенты созданы: researcher, writer, coder")
Обратите внимание: каждому агенту задана понятная роль, конкретная цель и детальный бэкграунд. Это критически важно — чем точнее промпт, тем качественнее результат. Параметр allow_delegation разрешает агенту передавать подзадачи коллегам.
Оркестратор — это связующее звено. Он создаёт задачи (Tasks), привязывает их к агентам и определяет порядок выполнения. CrewAI поддерживает два режима: sequential (последовательно) и hierarchical (с менеджером-агентом).
# crew.py — оркестрация команды from crewai import Crew, Process, Task from agents import researcher, writer, coder # ── Задача 1: Исследование ── research_task = Task( description=( "Исследуй тему: 'Лучшие практики multi-agent систем в 2026'. " "Найди минимум 5 релевантных источников. " "Выдели ключевые паттерны, фреймворки и антипаттерны." ), expected_output="Структурированный отчёт с источниками.", agent=researcher, async_execution=False, ) # ── Задача 2: Кодинг ── coding_task = Task( description=( "Напиши Python-класс MultiAgentBenchmark для " "сравнения производительности одиночного агента и команды. " "Должен поддерживать метрики: latency, token_usage, accuracy. " "Используй asyncio для параллельных замеров." ), expected_output="Файл benchmark.py с полной реализацией.", agent=coder, async_execution=True, # параллельно с research_task! ) # ── Задача 3: Документирование (ждёт обе предыдущие) ── writing_task = Task( description=( "На основе отчёта исследователя и кода разработчика " "напиши README.md и архитектурный обзор. " "Объём: 400-600 слов. Язык: русский." ), expected_output="Готовый README.md.", agent=writer, context=[research_task, coding_task], # зависимость ) # ── Сборка Crew ── crew = Crew( agents=[researcher, writer, coder], tasks=[research_task, coding_task, writing_task], process=Process.sequential, # или Process.hierarchical verbose=True, memory=True, # сохраняет контекст между запусками planning=True, # автоматическое планирование шагов ) print("[OK] Crew собран: 3 агента, 3 задачи")
Ключевая деталь: async_execution=True у coding_task. Пока Researcher ищет данные, Coder параллельно пишет код. Это сокращает общее время пайплайна на 40-60%. Параметр context указывает, что задача Writer должна дождаться результатов обеих предыдущих.
Точка входа — минималистичный скрипт, который загружает переменные окружения, собирает Crew и запускает его. Результаты каждого агента возвращаются в structured-формате.
# main.py — точка входа import os import time from pathlib import Path from dotenv import load_dotenv from crew import crew from rich.console import Console console = Console() load_dotenv() def main(): console.rule("[bold red]ЗАПУСК MULTI-AGENT CREW[/bold red]") start = time.monotonic() # Запуск — один вызов! result = crew.kickoff() elapsed = time.monotonic() - start # Сохраняем результаты output_dir = Path("output") output_dir.mkdir(exist_ok=True) (output_dir / "report.md").write_text( result.final_output, encoding="utf-8" ) console.print(f"\n[green]✓ ЗАВЕРШЕНО за {elapsed:.1f}с[/green]") console.print(f"[dim]Токенов использовано: {result.token_usage}[/dim]") console.print(f"[dim]Результат сохранён в output/report.md[/dim]") return result if __name__ == "__main__": main()
▸ Запуск:
python main.py # ── Вывод ── # ═══════════ ЗАПУСК MULTI-AGENT CREW ═══════════ # [Researcher] Начинаю поиск... # [Coder] Пишу benchmark.py... # [Researcher] Найдено 7 источников # [Coder] Код готов: 148 строк # [Writer] Формирую документацию... # ✓ ЗАВЕРШЕНО за 17.3с # Токенов использовано: 12_847
Весь пайплайн запускается одним вызовом crew.kickoff(). CrewAI сам управляет параллелизмом, передачей контекста и сбором ошибок. Результат содержит final_output (итоговый текст), token_usage (расход токенов) и tasks_output (результаты каждой задачи по отдельности).
В продакшене недостаточно просто запустить агентов — нужно видеть, что происходит внутри. CrewAI предоставляет колбэки на каждом шаге, а Loguru даёт структурированные логи.
# monitoring.py — отслеживание и логирование import json from datetime import datetime, timezone from pathlib import Path from loguru import logger from crewai import Crew # Настройка Loguru logger.add( "logs/crew_{time:YYYY-MM-DD}.log", rotation="10 MB", retention="7 days", level="DEBUG", format="{time:HH:mm:ss} | {level} | {message}", ) class CrewMonitor: def __init__(self, run_id: str): self.run_id = run_id self.metrics: dict = { "started": datetime.now(timezone.utc).isoformat(), "agent_calls": [], "errors": [], } def on_agent_start(self, agent_name: str, task: str): """Колбэк при старте агента.""" logger.info( "[{}] Агент '{}' начал задачу: {}", self.run_id, agent_name, task[:80] ) self.metrics["agent_calls"].append({ "agent": agent_name, "task": task, "started": datetime.now(timezone.utc).isoformat(), }) def on_agent_finish(self, agent_name: str, output: str): """Колбэк при завершении агента.""" output_len = len(output) logger.success( "[{}] Агент '{}' завершил ({} символов)", self.run_id, agent_name, output_len ) def on_error(self, agent_name: str, error: Exception): """Колбэк при ошибке.""" logger.error( "[{}] ОШИБКА в '{}': {}", self.run_id, agent_name, str(error) ) self.metrics["errors"].append({ "agent": agent_name, "error": str(error), "time": datetime.now(timezone.utc).isoformat(), }) def flush(self, path: str = "metrics.json"): """Сохранить метрики в JSON.""" self.metrics["finished"] = datetime.now(timezone.utc).isoformat() Path(path).write_text(json.dumps( self.metrics, ensure_ascii=False, indent=2 ), encoding="utf-8") logger.info("Метрики сохранены в {}", path) # Использование: # monitor = CrewMonitor(run_id="prod-001") # crew.kickoff() # колбэки сработают автоматически # monitor.flush("output/metrics.json")
Loguru пишет структурированные логи с ротацией (каждые 10 МБ, хранение — 7 дней). Run ID позволяет отслеживать конкретный прогон. Файл metrics.json накапливает статистику для последующего анализа в Grafana или Kibana.
▸ ВЕРДИКТ ◂
Multi-agent-системы — это не хайп, а реальный инструмент для сложных задач. За 30 минут мы прошли путь от нуля до работающей команды из трёх специализированных агентов с параллельным выполнением, оркестрацией и мониторингом. Главные выводы: (1) специализация агентов даёт качество, (2) параллелизм ускоряет пайплайн в 2–4 раза, (3) продакшен-мониторинг — обязателен с первого дня. Следующий шаг: замените GPT-4o на локальную Llama-4 и получите полностью приватный AI-воркфлоу.
ГОТОВЫ СОБРАТЬ СВОЮ КОМАНДУ?
Подпишитесь на qantcore.space — еженедельные гайды по AI-агентам, RAG-пайплайнам и автономным воркфлоу. Без воды, только код и архитектура.
QANTCORE.SPACE / GUIDE-042 / MULTI-AGENT / 2026-08-05 / CREWAI v3.2.0 / PYTHON 3.12+