Как тестировать delegation paths (интеграционное тестирование multi-agent)?

Краткий тезис

Delegation paths — это цепочки вызовов между агентами в multi-agent системе. Интеграционное тестирование таких путей критически важно для надёжности: один сбой на любом шаге может разрушить весь сценарий. Основные подходы включают mock downstream (изоляцию агентов-исполнителей), fault injection (симуляцию ошибок), trace validation (проверку корректности передачи контекста) и coverage путей. Эти тестирования должны выполняться в CI/CD и дополняться регрессионными проверками при изменениях в любом агенте.

1. Термины и контекст

Delegation path — последовательность агентов, через которую проходит запрос. Например: RouterAgent → DataAgent → SummarizerAgent → User. Каждый переход (делегирование) — это вызов от одного агента к другому.

Multi-agent система — архитектура, где несколько автономных агентов обмениваются сообщениями, координируют действия и выполняют распределённые задачи. Тестирование таких систем сложнее, чем монолитного LLM, из-за асинхронности, недетерминизма LLM-ответов и распределённости компонентов.

Интеграционное тестирование — проверка взаимодействия между агентами, а не каждого агента в изоляции (unit-тесты) или всей системы целиком (end-to-end). Цель — убедиться, что delegation paths работают как единый конвейер.

2. Почему тестирование delegation paths — это сложно?

3. Mock downstream — изоляция агентов-исполнителей

Mock downstream — подмена реального агента-исполнителя на этапе тестирования его «заглушкой» (mock), которая возвращает заранее заданный ответ. Это позволяет проверить, как агент-руководитель (или маршрутизатор) обрабатывает ответы, не вызывая настоящий сервис.

Пример на Python с использованием unittest.mock в контексте LangGraph или простой очереди сообщений:

import pytest
from unittest.mock import AsyncMock, patch

# Предположим, у нас есть агент RouterAgent, который вызывает DataAgent
class RouterAgent:
    async def delegate(self, request: str) -> str:
        # Вызов реального DataAgent (внешний сервис)
        result = await call_data_agent(request)
        # post-processing
        return result

@pytest.mark.asyncio
async def test_router_agent_delegates_correctly():
    # Создаём mock для DataAgent
    mock_data_agent = AsyncMock(return_value="извлечённые данные")
    
    # Подменяем вызов call_data_agent на mock
    with patch('my_agent.call_data_agent', mock_data_agent):
        router = RouterAgent()
        response = await router.delegate("запрос")
        assert response == "извлечённые данные"
        mock_data_agent.assert_awaited_once_with("запрос")

Ключевые моменты:

• mock должен имитировать интерфейс реального агента (сигнатуру, типы ответов).
• Важно тестировать обработку граничных случаев: пустой ответ, очень длинный ответ, ответ с ошибкой внутри JSON.

4. Fault injection — симуляция сбоев

Fault injection — намеренное внесение ошибок в поведение downstream-агента для проверки устойчивости вышестоящих агентов. Типы сбоев:

Пример интеграционного теста с fault injection (используем pytest + httpx mock или собственный сервер-заглушку):

@pytest.mark.asyncio
async def test_fault_injection_timeout():
    # Настраиваем mock, который эмулирует тайм-аут
    mock_call = AsyncMock(side_effect=asyncio.TimeoutError("simulated timeout"))

    with patch('my_agent.call_downstream', mock_call):
        router = RouterAgent()
        with pytest.raises(TimeoutError):
            await router.delegate("запрос")
        # Или убеждаемся, что сработал механизм retry
    
    # Проверяем, что агент сделал 3 попытки (если настроен retry)
    assert mock_call.await_count == 3

5. Trace validation — проверка передачи контекста

Trace validation — проверка, что контекст (история сообщений, метаданные, собранные данные) корректно передаётся от агента к агенту. Для этого используют трассировки (OpenTelemetry, LangSmith, Weights & Biases) или явное логирование каждого шага.

В тесте можно анализировать структурированные логи:

def test_context_preserved_along_path():
    # Запускаем сценарий с известным контекстом
    # Собираем все trace events (например, из in-memory списка)
    events = run_scenario(initial_context={"user_id": 42})
    
    # Проверяем, что в каждом шаге присутствует user_id
    for event in events:
        assert "user_id" in event["context"]
        assert event["context"]["user_id"] == 42
    
    # Проверяем порядок агентов
    agents_sequence = [e["agent"] for e in events]
    assert agents_sequence == ["RouterAgent", "DataAgent", "SummarizerAgent"]

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

6. Тестовое покрытие delegation paths

Coverage — метрика, показывающая, какие пути делегирования были протестированы. Пути бывают:

Для каждого типа необходимо написать минимум один интеграционный тест. Используйте матрицу покрытия:

# coverage_matrix.yaml
paths:
  - name: "основной_сценарий"
    agents: [Router, Data, Summarizer]
    tested: true
    test_file: "test_main_path.py"
  - name: "fallback_при_ошибке_Data"
    agents: [Router, Data(fail), Summarizer]
    tested: false  # TODO

Такой файл можно парсить и генерировать отчёты в CI.

7. Регрессионное тестирование

Regression testing — повторный прогон всех тестов на delegation paths после любого изменения кода одного из агентов (изменение логики, обновление модели LLM, смена API). Поскольку агенты взаимозависимы, даже небольшое изменение в одном агенте может сломать соседние пути.

Правила:

• Каждый агент должен иметь свой набор unit-тестов.
• Интеграционные тесты на delegation paths должны запускаться при каждом коммите в репозиторий любого агента.
• Желательно использовать feature flags или canary deployments, чтобы быстро откатить изменения, если регрессионные тесты падают.

Автоматизация: настройте CI job, который собирает всех агентов, поднимает тестовую среду (например, Docker Compose) и запускает полный набор интеграционных тестов.

8. Инструменты для интеграционного тестирования multi-agent

9. Интеграция в CI/CD пайплайн

Идеальный пайплайн:

1. Code push → 2. Build images агентов → 3. Deploy test environment (Docker Compose) 
   → 4. Run unit tests (каждый агент) → 5. Run integration tests (delegation paths) 
   → 6. Run fault injection suite → 7. Collect coverage → 8. Откат, если падения критичны

В шаге 4-6 можно добавить параллельный запуск тестов для сокращения времени. Coverage отчёт выводится в PR.

10. Best practices

• Тестируйте обработку краевых случаев: пустые ответы, неверный формат, пропущенные поля.
• Используйте фиксированные сиды для LLM (temperature=0) для уменьшения флаки.
• Изолируйте тестовую среду: не используйте общие базы данных между тестами.
• Логируйте каждый шаг делегирования с уникальным trace ID для отладки.
• Делайте автоматические алерты при падении критических paths (например, основной путь router → data → summary).
• Пишите тесты на «человека в цикле» (HITL): эмулируйте решения человека через mock approval.

Пет-проект для закрепления

Задача: Разработать мини-систему из трёх агентов: Supervisor (принимает запрос), DataFetcher (запрашивает данные из внешнего API), Summarizer (обобщает). Написать набор интеграционных тестов для delegation paths.

Инструменты: Python, asyncio, pytest, unittest.mock, httpx (для mock вызова API), Docker (для изоляции).

Шаги:

1. Создайте три класса агентов с методами async def process(request, context) -> dict.
2. Supervisor делегирует DataFetcher, затем Summarizer.
3. Реализуйте fallback: если DataFetcher выкидывает timeout, Supervisor вызывает альтернативный LocalFetcher.
4. Напишите unit-тесты для каждого агента (mock внешнего API).
5. Напишите интеграционные тесты:
   • Основной путь (Supervisor → DataFetcher → Summarizer) — проверьте передачу контекста.
   • Fault injection: DataFetcher возвращает HTTP 500 — убедитесь, что срабатывает fallback.
   • Fault injection: Summarizer возвращает garbage response — проверьте, что Supervisor не крашится.
6. Добавьте coverage матрицу путей.
7. Настройте запуск тестов через pytest в CI (GitHub Actions).

Ожидаемый результат: Чёткое понимание, как изолировать агентов, симулировать сбои и проверять корректность делегирования. Проект станет основой для тестирования production multi-agent системы.

Связь с другими вопросами

Навигация

• Предыдущий: 768
• Следующий: 770
• Индекс: 00. Индекс разборов