ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Измерить cost делегирования

1. Цель задачи

Научиться инструментировать и анализировать стоимость (cost) каждого шага делегирования в multi-agent системе. Необходимо разработать систему логирования, которая фиксирует затраты токенов]] и латентность для каждого узла цепочки делегирования от инициатора до исполнителя-агента. Собранные данные агрегировать в итоговую таблицу Delegation Delegation cost per delegation path, позволяющую сравнивать разные маршруты выполнения запроса.

Ключевой результат Таблица (CSV/JSON) с полями: path_id, node_name, prompt_tokens, completion_tokens, total_tokens, latency_ms — позволяющая оптимизировать стратегии делегирования.

2. Исходные данные

Если нет реальной multi-agent системы — симулируем:

1. Напишите простой симулятор делегирования с 3–5 узлами (агентами-исполнителями) на Python.
2. Каждый узел при вызове генерирует случайное количество токенов (prompt 200–500, completion 50–200) и задержку (100–500 ms).
3. Оркестратор (рут-агент) делает последовательные или параллельные вызовы к дочерним узлам, логируя метрики через декоратор.
4. Запустите не менее 10 различных сценариев (delegation path), чтобы получить вариативность.

3. Технологический стек

4. Этапы выполнения

Этап 1: Проектирование схемы данных логирования (30 минут)

Действия

1. Определить обязательные поля для лога одного шага делегирования:

   Поле	Тип	Пример
   delegation_path_id	str	"path_001"
   step_order	int	1
   node_name	str	"research_agent"
   parent_node	Optional[str]	"orchestrator"
   prompt_tokens	int	350
   completion_tokens	int	120
   total_tokens	int	470
   latency_ms	float	450.23
   timestamp	datetime	2025-04-12T14:30:00Z

2. Выбрать формат: JSON Lines (каждая строка – один вызов) — прост для агрегации.

3. Определить уникальный ID для каждого path (например, f"path_{uuid4().hex[:8]}").

4. Подготовить конфиг с порогом тревоги (например, latency > 5000 ms).

Ожидаемый результат этапа Документированная схема данных и конфиг логирования.

Этап 2: Реализация декоратора / контекстного менеджера для замеров (1 час)

Действия

1. Написать декоратор @log_delegation_cost для функции-агента, который:

   • Принимает path_id, node_name, parent_name и объект-логгер.
   • Замеряет время выполнения (старт/стоп).
   • После вызова LLM получает токены из ответа API.
   • Если токены недоступны — эмулирует (в симуляторе).
   • Записывает лог-строку в JSON Lines файл.

2. Реализовать контекстный менеджер DelegationSpan для асинхронных вызовов:

   class DelegationSpan:
       def __init__(self, logger, path_id, node_name, parent_name=None):
           self.logger = logger
           ...
       def __enter__(self):
           self.start = time.monotonic()
           return self
       def __exit__(self, *args):
           self.latency = time.monotonic() - self.start
           self.logger.log(...)
   

3. Учесть вложенные вызовы (child spans) — для агрегации потом в DataFrame.

4. Добавить поле status (success / error) для учёта неудачных вызовов.

Ожидаемый результат этапа Работающий декоратор/менеджер, пример использования в тестовом скрипте.

Этап 3: Интеграция с multi-agent оркестратором и сбор данных (1.5 часа)

Действия

1. Создать простой граф делегирования с тремя типами агентов:
   • Orchestrator (root) – принимает запрос, решает кого вызвать.
   • ResearchAgent – ищет информацию.
   • WriterAgent – пишет ответ.
2. Реализовать симуляцию вызовов LLM (возвращать фиксированные или случайные токены).
3. Вставить декоратор/контекстный менеджер в каждый агент.
4. Написать скрипт run_delegations.py, который генерирует не менее 10 различных путей (с последовательными и параллельными вызовами).
5. Запустить скрипт, проверить, что логи пишутся в файл delegation_logs.jsonl.

# Пример структуры симуляции
agent_graph = {
    "orchestrator": {
        "children": ["research", "writer"],
        "parallel": False
    },
    "research": { "children": [], "function": research_llm_call },
    "writer": { "children": [], "function": writer_llm_call }
}

Ожидаемый результат этапа Собранные логи не менее чем от 10 делегационных путей.

Этап 4: Агрегация и визуализация (1 час)

Действия

1. Загрузить логи в Pandas DataFrame:

   import pandas as pd
   df = pd.read_json('delegation_logs.jsonl', lines=True)
   

2. Создать сводную таблицу cost_per_path:

   • Сгруппировать по delegation_path_id.
   • Агрегировать: сумма total_tokens, средняя/макс latency_ms, количество шагов.
   • Добавить столбец total_cost_est = sum_total_tokens * token_price_per_1k.

3. Построить таблицу cost_per_node – средние значения по каждому агенту.

4. Экспортировать обе таблицы в CSV (файлы path_cost.csv и node_cost.csv).

5. Опционально: построить простой график (latency vs tokens) с помощью matplotlib.

Ожидаемый результат этапа Два CSV-файла с агрегированными стоимостями.

Этап 5: Анализ и выводы (30 минут)

Действия

1. Ответить на вопросы:
   • Какие пути самые дорогие? Почему?
   • Где узкое место по латентности?
   • Есть ли корреляция между числом шагов и стоимостью?
2. Написать короткий отчёт с рекомендациями (например, «увеличить параллелизм для research/writer»).
3. Выгрузить отчёт в Markdown.

Ожидаемый результат этапа Файл analysis_report.md с выводами и метриками.

5. Критерии приемки (Definition of Done)

• Реализован декоратор/контекстный менеджер, логирующий токены и latency для каждого вызова агента.
• Логи записываются в структурированном формате (JSON Lines) с полями: path_id, node_name, parent_node, prompt_tokens, completion_tokens, total_tokens, latency_ms, status.
• Собранные данные позволяют однозначно восстановить граф делегирования по логу (chain of calls).
• Сгенерировано не менее 10 различных delegation paths с разной архитектурой (последовательные, параллельные, смешанные).
• Агрегированная таблица path_cost.csv содержит строки с суммой токенов и латентностью по каждому path.
• Агрегированная таблица node_cost.csv содержит средние затраты на каждый узел (агент).
• Написан краткий отчёт analysis_report.md с визуализацией (таблицы, графики опционально) и выводами.
• Код покрыт базовыми unit-тестами для логирования и агрегации (минимум 3 теста).
• Все скрипты запускаются одной командой python run_all.py.

6. Ожидаемый результат

Основные артефакты

1. delegation_logs.jsonl — сырые логи со всех вызовов.
2. path_cost.csv — таблица со стоимостью по каждому пути:
   • delegation_path_id, num_steps, total_tokens, avg_latency_ms, max_latency_ms, estimated_cost_usd.
3. node_cost.csv — таблица со стоимостью по каждому агенту:
   • node_name, total_calls, avg_tokens_per_call, avg_latency_ms, success_rate.
4. analysis_report.md — выводы и рекомендации для оптимизации.
5. Исходный код с декоратором, симулятором и пайплайном агрегации.

Опционально Дашборд Grafana или OpenTelemetry трассировка (при расширенной реализации).

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я убедился, что все поля логирования (токены, latency) записаны с корректными единицами измерения.
• Я проверил, что лог для каждого вызова содержит уникальный delegation_path_id и step_order.
• Я сгенерировал минимум 10 разных path, включая хотя бы один с параллельными вызовами.
• Я создал два CSV-файла с агрегированными данными; они открываются без ошибок.
• Я написал отчёт с конкретными цифрами (не просто "дорого", а "путь X дороже пути Y на 30%").