Паттерн оркестратора

Updated Jul 2026

Почему одиночные агенты достигают предела

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

Паттерн оркестратора использует одного агента-координатора для делегирования работы агентам-специалистам и объединения их результатов. Это наиболее структурированный и предсказуемый мультиагентный паттерн.

Архитектура

                    +------------------+
                    |   ORCHESTRATOR   |
                    |   (Coordinator)  |
                    +--------+---------+
                             |
              +--------------+--------------+
              |              |              |
        +-----+-----+  +-----+-----+  +-----+-----+
        |  Agent:   |  |  Agent:   |  |  Agent:   |
        |  UI Tests |  | API Tests |  | Perf Tests|
        +-----------+  +-----------+  +-----------+

Оркестратор:

1. Получает спецификацию функциональности или тест-план
2. Анализирует её, чтобы определить, какие типы тестов необходимы
3. Делегирует сценарии агентам-специалистам
4. Собирает и объединяет результаты
5. Формирует единый отчёт о тестировании

Реализация

class OrchestratorAgent:
    def __init__(self, specialists: dict[str, Agent]):
        self.specialists = specialists  # {"ui": UIAgent, "api": APIAgent, ...}
        self.llm = get_llm()

    def plan_and_execute(self, feature_spec: str) -> TestReport:
        # Step 1: Analyze the spec and create a test plan
        plan = self.llm.generate(f"""
        Given this feature specification:
        {feature_spec}

        Determine which test types are needed:
        - UI tests (if there are user-facing changes)
        - API tests (if there are endpoint changes)
        - Performance tests (if there are SLA requirements)
        - Security tests (if there are auth/data changes)

        Output a JSON plan: {{"ui": [...scenarios], "api": [...scenarios], ...}}
        """)

        # Step 2: Delegate to specialists
        results = {}
        for agent_type, scenarios in plan.items():
            if agent_type not in self.specialists:
                results[agent_type] = {"skipped": f"No specialist for {agent_type}"}
                continue
            specialist = self.specialists[agent_type]
            results[agent_type] = specialist.execute_scenarios(scenarios)

        # Step 3: Merge and resolve conflicts
        return self.merge_results(results)

    def merge_results(self, results: dict) -> TestReport:
        """Merge results from multiple specialists into a unified report."""
        all_tests = []
        all_failures = []
        all_coverage = {}

        for agent_type, agent_results in results.items():
            if isinstance(agent_results, dict) and "skipped" in agent_results:
                continue
            all_tests.extend(agent_results.tests)
            all_failures.extend(agent_results.failures)
            all_coverage[agent_type] = agent_results.coverage

        return TestReport(
            total_tests=len(all_tests),
            total_failures=len(all_failures),
            results_by_type=results,
            coverage=all_coverage,
            overall_status="FAIL" if all_failures else "PASS"
        )

Проектирование агентов-специалистов

Каждый агент-специалист оптимизирован для своей области:

Специалист по UI-тестам

class UITestSpecialist(Agent):
    def __init__(self, browser, llm):
        self.browser = browser
        self.llm = llm

    def execute_scenarios(self, scenarios: list[str]) -> AgentResults:
        results = []
        for scenario in scenarios:
            # Each scenario is a natural language description
            # e.g., "Verify the checkout button is disabled when cart is empty"
            result = self.react_loop(
                objective=scenario,
                tools=["navigate", "click", "type", "text", "screenshot"],
                max_steps=20
            )
            results.append(result)
        return AgentResults(tests=results, failures=[r for r in results if r.failed])

Специалист по API-тестам

class APITestSpecialist(Agent):
    def __init__(self, base_url: str, llm):
        self.base_url = base_url
        self.llm = llm

    def execute_scenarios(self, scenarios: list[str]) -> AgentResults:
        results = []
        for scenario in scenarios:
            # e.g., "Verify POST /orders returns 400 when quantity is 0"
            result = self.execute_api_test(scenario)
            results.append(result)
        return AgentResults(tests=results, failures=[r for r in results if r.failed])

    def execute_api_test(self, scenario: str) -> TestResult:
        # Use LLM to determine the HTTP request
        request_spec = self.llm.generate(f"""
        Scenario: {scenario}
        Base URL: {self.base_url}

        Generate the HTTP request as JSON:
        {{"method": "...", "path": "...", "headers": {{...}}, "body": {{...}}}}
        And the expected response:
        {{"status": ..., "body_contains": [...], "body_not_contains": [...]}}
        """)

        # Execute and evaluate
        response = httpx.request(
            method=request_spec["method"],
            url=f"{self.base_url}{request_spec['path']}",
            headers=request_spec.get("headers", {}),
            json=request_spec.get("body")
        )

        return self.evaluate_response(response, request_spec["expected"])

Протокол коммуникации оркестратора

Оркестратор и специалисты общаются через структурированные сообщения:

@dataclass
class TaskAssignment:
    """Message from orchestrator to specialist."""
    task_id: str
    agent_type: str         # "ui", "api", "perf", "security"
    scenarios: list[str]    # Natural language test scenarios
    constraints: dict       # Time budget, step limits, etc.
    priority: int           # 0=low, 1=normal, 2=high

@dataclass
class TaskResult:
    """Message from specialist to orchestrator."""
    task_id: str
    agent_type: str
    tests_executed: int
    tests_passed: int
    tests_failed: int
    failures: list[dict]    # {scenario, reason, screenshot}
    duration_seconds: float
    tokens_used: int

Когда использовать паттерн оркестратора

Лучше всего подходит для:

• Многоуровневого тестирования (UI + API + производительность за один прогон)
• Оркестрации тестов на уровне функциональности (одна функциональность, несколько типов тестов)
• Централизованной отчётности по разным областям тестирования
• Команд с различными специализациями в тестировании

Риски:

• Оркестратор становится узким местом, если принимает плохие решения о делегировании
• Единая точка отказа: если агент-оркестратор падает, всё тестирование останавливается
• Избыточная инженерия: для простых наборов тестов одиночный агент проще

Меры снижения рисков:

• Всегда логируйте обоснование делегирования для отладки
• Реализуйте запасной вариант: если специалист падает, оркестратор продолжает с остальными
• Установите таймауты для каждого специалиста, чтобы один медленный специалист не блокировал отчёт

Оркестратор vs ручная координация тестирования

Ключевой вывод

Паттерн оркестратора -- наиболее предсказуемая мультиагентная архитектура. Он естественно отображается на то, как QA-команды уже работают (лид + специалисты), но выполняется на машинной скорости. Ключевое проектное решение -- сколько интеллекта вложить в оркестратор (генерация плана, разрешение конфликтов) versus специалистов (доменная экспертиза, выполнение тестов).