Как запустить несколько coding-агентов параллельно: техника для максимальной продуктивности

Один агент — это медленно. Три — в три раза быстрее?

Нет. Три агента работающие параллельно — это не в три раза быстрее. Это в пять раз эффективнее.

Разница между линейным и параллельным выполнением — это разница между ожиданием и действием. Когда один агент пишет код, второй рефакторит, а третий ищет баги — вы не просто ускоряете процесс. Вы меняете саму природу работы с ИИ.

Представьте: вы даёте задачу "создать REST API для пользователей". Обычный workflow: агент пишет код → вы проверяете → агент исправляет → вы тестируете → агент добавляет документацию. Цикл за циклом. Контекстное переключение убивает вашу концентрацию.

Параллельный подход: все эти шаги происходят одновременно. Пока вы пьёте кофе.

Почему последовательность убивает эффективность

Давайте посчитаем. Типичный coding-агент:

• Думает 15-30 секунд на каждый ответ
• Требует вашего внимания для каждого следующего шага
• Работает в одном контексте за раз
• Простаивает, пока вы проверяете его работу

Время простоя — это не просто потерянные секунды. Это сломанный flow. Вы отвлекаетесь, теряете нить, возвращаетесь — и тратите минуты на восстановление контекста.

Архитектура параллельных агентов: не как в учебнике

Теория говорит: "Используйте очереди сообщений и блокировки". Практика показывает: это слишком тяжело для coding-агентов.

Вот что работает на самом деле:

Я начинал с оркестратора. Потом отказался. Слишком много кода для управления, слишком мало отдачи. Сейчас использую гибрид: изолированные процессы + простая файловая синхронизация.

Пошаговый план: от нуля до трёх параллельных агентов

1 Выбираем инструменты (не очевидный выбор)

Не берите сложные фреймворки. Они добавят абстракций, но не ценности.

Мой стек:

• Python + asyncio для оркестрации
• vLLM для запуска моделей (если есть GPU) — см. тесты моделей для coding-агентов
• LM Studio или Ollama для локального запуска
• Watchdog для отслеживания изменений файлов

Важный момент: не пытайтесь использовать одну модель для всех агентов. Разные задачи — разные модели.

2 Создаём базовую структуру проекта

Вот как выглядит минимальная структура:

parallel-agents/
├── agents/
│   ├── coder.py      # Агент для написания кода
│   ├── refactor.py   # Агент для рефакторинга
│   └── tester.py     # Агент для тестирования
├── shared/
│   ├── context.json  # Общий контекст
│   └── tasks.json    # Очередь задач
├── outputs/
│   ├── coder/
│   ├── refactor/
│   └── tester/
└── orchestrator.py   # Главный скрипт

3 Пишем простой оркестратор (без over-engineering)

Вот код, который работает. Не идеальный, но рабочий:

import asyncio
import subprocess
import json
from pathlib import Path
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class AgentOrchestrator:
    def __init__(self):
        self.agents = {
            'coder': 'python agents/coder.py',
            'refactor': 'python agents/refactor.py',
            'tester': 'python agents/tester.py'
        }
        self.processes = {}
        
    async def start_agent(self, name, command):
        """Запускаем агента в отдельном процессе"""
        process = await asyncio.create_subprocess_shell(
            command,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE
        )
        self.processes[name] = process
        print(f"[+] Агент {name} запущен")
        
    async def run_all(self):
        """Запускаем всех агентов параллельно"""
        tasks = []
        for name, cmd in self.agents.items():
            tasks.append(self.start_agent(name, cmd))
        
        await asyncio.gather(*tasks)
        
        # Мониторим выходы
        while True:
            for name, process in self.processes.items():
                if process.returncode is not None:
                    print(f"[-] Агент {name} завершился")
                    # Перезапускаем при необходимости
                    await self.start_agent(name, self.agents[name])
            await asyncio.sleep(5)

if __name__ == "__main__":
    orchestrator = AgentOrchestrator()
    asyncio.run(orchestrator.run_all())

4 Настраиваем синхронизацию через файлы (проще, чем кажется)

Агенты не общаются напрямую. Они пишут в общие файлы и читают из них.

# agents/coder.py - упрощённая версия
import json
import time
from pathlib import Path

class CoderAgent:
    def __init__(self):
        self.context_file = Path("shared/context.json")
        self.output_dir = Path("outputs/coder")
        self.output_dir.mkdir(exist_ok=True)
        
    def read_context(self):
        """Читаем общий контекст"""
        if self.context_file.exists():
            with open(self.context_file) as f:
                return json.load(f)
        return {}
        
    def write_output(self, task_id, code):
        """Пишем результат в свой каталог"""
        output_file = self.output_dir / f"{task_id}.py"
        output_file.write_text(code)
        
        # Обновляем общий контекст
        context = self.read_context()
        context['last_coder_output'] = task_id
        with open(self.context_file, 'w') as f:
            json.dump(context, f)
        
    def run(self):
        """Основной цикл агента"""
        while True:
            # Проверяем новые задачи
            tasks_file = Path("shared/tasks.json")
            if tasks_file.exists():
                with open(tasks_file) as f:
                    tasks = json.load(f)
                
                for task in tasks.get('coding', []):
                    if not task.get('processed'):
                        # Генерируем код (здесь вызов LLM)
                        generated_code = self.generate_code(task['description'])
                        self.write_output(task['id'], generated_code)
                        
                        # Помечаем как обработанную
                        task['processed'] = True
                        
                # Сохраняем обновлённые задачи
                with open(tasks_file, 'w') as f:
                    json.dump(tasks, f)
            
            time.sleep(2)  # Не спамим
            
    def generate_code(self, description):
        """Здесь реальный вызов LLM"""
        # Для примера возвращаем заглушку
        return f"# Код для: {description}\nprint('Hello World')"

if __name__ == "__main__":
    agent = CoderAgent()
    agent.run()

5 Добавляем простой workflow менеджер

Этот скрипт превращает вашу задачу в набор подзадач для агентов:

# workflow_manager.py
import json
import uuid
from datetime import datetime

class WorkflowManager:
    def create_workflow(self, main_task):
        """Разбиваем главную задачу на подзадачи"""
        task_id = str(uuid.uuid4())[:8]
        
        workflow = {
            'id': task_id,
            'main_task': main_task,
            'created_at': datetime.now().isoformat(),
            'subtasks': {
                'coding': [
                    {'id': f'{task_id}_code', 'description': main_task, 'processed': False}
                ],
                'refactor': [
                    {'id': f'{task_id}_refactor', 'description': f'Рефакторинг кода для: {main_task}', 'processed': False}
                ],
                'testing': [
                    {'id': f'{task_id}_test', 'description': f'Тесты для: {main_task}', 'processed': False}
                ]
            }
        }
        
        # Сохраняем в общий файл задач
        with open('shared/tasks.json', 'w') as f:
            json.dump(workflow['subtasks'], f)
        
        print(f"[+] Workflow {task_id} создан: {main_task}")
        return workflow

# Использование:
if __name__ == "__main__":
    manager = WorkflowManager()
    manager.create_workflow(
        "Создать REST API эндпоинт для пользователей с аутентификацией"
    )

Типичные ошибки (совершал их все)

Оптимизация: как сделать систему быстрее

Базовая версия работает. Но можно сделать лучше.

Кэширование контекста

Агенты постоянно читают context.json. Добавьте in-memory кэш:

import time

class CachedContext:
    def __init__(self):
        self.cache = {}
        self.last_read = 0
        self.cache_ttl = 3  # секунды
        
    def get_context(self):
        """Читаем из файла только если кэш устарел"""
        now = time.time()
        if now - self.last_read > self.cache_ttl:
            self._refresh_cache()
            self.last_read = now
        return self.cache
        
    def _refresh_cache(self):
        # Чтение из файла
        pass

Приоритизация задач

Не все задачи равны. Добавьте приоритеты:

{
  "coding": [
    {
      "id": "task_1",
      "description": "Срочно: пофиксить баг с аутентификацией",
      "priority": "high",
      "processed": false
    },
    {
      "id": "task_2", 
      "description": "Добавить документацию",
      "priority": "low",
      "processed": false
    }
  ]
}

Балансировка нагрузки

Если один агент завален задачами, а другой простаивает — перераспределите задачи. Простейший алгоритм:

def balance_tasks(tasks):
    """Распределяем задачи между агентами"""
    # Считаем нагрузку
    coder_load = len([t for t in tasks['coding'] if not t['processed']])
    refactor_load = len([t for t in tasks['refactor'] if not t['processed']])
    
    # Если кодер перегружен, а рефактор свободен
    if coder_load > 5 and refactor_load < 2:
        # Переносим некоторые задачи рефакторинга кодеру
        pass

Реальные цифры: что даёт параллелизм

Я замерял на проекте среднего размера (10k строк кода):

Обратите внимание: ускорение не линейное. Для сложных задач с зависимостями выигрыш больше, потому что агенты работают над разными частями одновременно.

Что дальше? Эволюция системы

Начали с простого файлового обмена. Куда двигаться:

1. Добавить мониторинг — сколько времени каждый агент тратит на задачи, какие ошибки возникают
2. Автоматическое масштабирование — запускать больше агентов при высокой нагрузке
3. Умное планирование — предсказывать, какие задачи будут следующими
4. Интеграция с CI/CD — агенты запускают тесты и деплоят код

Но главное — не усложнять раньше времени. Сначала сделайте работающую систему с тремя агентами. Потом оптимизируйте.

Финальный совет: начинайте с малого

Не пытайтесь построить perfect system с первого раза. Возьмите код из этой статьи, запустите двух агентов: один пишет код, второй — комментарии. Посмотрите, как они взаимодействуют.

Потом добавьте третьего. Потом оптимизируйте.

Через неделю у вас будет система, которая делает за час то, на что раньше уходило полдня. И вы поймёте, почему обратно к одному агенту возвращаться уже не захочется.

P.S. Если ваши агенты начинают конфликтовать слишком часто — возможно, вы даёте им слишком похожие задачи. Разделите ответственность чётче. Кодер пишет, рефактор улучшает, тестировщик ломает. Каждый в своей песочнице.