ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Собрать agentic mesh из 3 агентов

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

Научиться конструировать цепочку (mesh) из трёх агентов с передачей контекста через handoff-протокол. Агенты выполняют последовательную обработку запроса: Агент А парсит ввод, Агент Б обогащает контекст, Агент В формирует финальный ответ. Ключевая задача — обеспечить полную передачу промежуточных данных без потерь при каждом handoff.

Ключевой результат Рабочий скрипт (Python), который принимает входной текст, последовательно обрабатывает его тремя агентами и возвращает консолидированный результат. Контекст (все поля, сгенерированные Агентом А) должен быть доступен Агенту В без изменений.

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

Если нет реального инструмента — симулируем:

1. Ollama — если нет возможности поставить, используем fastchat с эмулятором или заглушку: функция mock_llm(text) -> {"response": f"mock: {text}"}.
2. OpenTelemetry — можно заменить простым JSON-логгером (записывать входящий/исходящий контекст в файл).
3. CrewAI — если не ставится, пишем собственный минимальный фреймворк на dataclasses и asyncio (в этапе 1 дана заготовка).

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

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

Этап 1: Проектирование контрактов и моделей контекста (40 мин)

Действия

1. Определить структуру контекста — единый Pydantic model, который передаётся между агентами.

   from pydantic import BaseModel, Field
   
   class AgentContext(BaseModel):
       user_query: str
       agent_a_output: str | None = None
       agent_b_enriched: str | None = None
       agent_c_final: str | None = None
       meta: dict = Field(default_factory=dict)
   

2. Спроектировать HandoffContract — dataclass, который содержит контекст и управляющую информацию.

   from dataclasses import dataclass
   
   @dataclass
   class HandoffSignal:
       source: str          # имя агента-отправителя
       target: str          # имя агента-получателя
       context: AgentContext
   

3. Создать минимальную абстракцию агента:

   from abc import ABC, abstractmethod
   
   class BaseAgent(ABC):
       name: str
       @abstractmethod
       async def process(self, signal: HandoffSignal) -> HandoffSignal:
           ...
   

4. Подготовить заготовку для 3 агентов (заглушки с принтами). Агент А парсит запрос, Агент Б обогащает, Агент В формирует ответ.

Ожидаемый результат этапа Файл models.py с AgentContext, HandoffSignal, BaseAgent. Три класса-заглушки в agents/.

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

Действия

1. Написать класс AgentA, который:

   • Принимает HandoffSignal, где context.user_query заполнен.
   • Вызывает LLM (Ollama) с промптом: «Извлеки из запроса ключевые сущности (сущность, действие, объект). Ответь JSON-строкой».
   • Парсит ответ и записывает context.agent_a_output = ....
   • Возвращает новый HandoffSignal с target="B".

2. Подключить реальный LLM через универсальный интерфейс:

   class LLMClient:
       async def generate(self, prompt: str) -> str:
           # вызов Ollama или mock
   

   Создать llm_client.py.

3. Написать unit-тест для AgentA: передать фиктивный сигнал с query, проверить, что agent_a_output не None и содержит поля "entity", "action", "object".

Ожидаемый результат этапа agents/agent_a.py с полной реализацией. Тест проходит.

Этап 3: Реализация Агента Б и Агента В (1.5 часа)

Действия

1. AgentB (обогащение):

   • Промпт: «На основе данных от Агента А: {agent_a_output}. Дополни контекст: дай краткое описание объекта. Ответ в JSON: "enriched_object": ...».
   • Результат сохранить в context.agent_b_enriched.

2. AgentC (финальный ответ):

   • Промпт: «У тебя есть пользовательский запрос: {user_query}, сущности: {agent_a_output}, обогащение: {agent_b_enriched}. Сформируй понятный ответ для пользователя (1-2 предложения).»
   • Результат сохранить в context.agent_c_final.

3. Реализовать оркестратор — MeshOrchestrator:

   class MeshOrchestrator:
       def __init__(self, agents: dict[str, BaseAgent]):
           self.agents = agents
   
       async def run(self, user_query: str) -> AgentContext:
           ctx = AgentContext(user_query=user_query)
           signal = HandoffSignal(source="user", target="A", context=ctx)
   
           for step in ["A", "B", "C"]:
               agent = self.agents[step]
               signal = await agent.process(signal)
               # проверка, что target изменился на следующий
               assert signal.target == step  # или другой механизм
           return signal.context
   

4. Написать интеграционный тест — полный прогон с тремя агентами и проверкой всех полей контекста.

Ожидаемый результат этапа Работающий пайплайн. При запуске orchestrator.run("Как установить Python?") получаем цепочку вызовов LLM и финальный ответ.

Этап 4: Мониторинг передачи контекста (40 мин)

Действия

1. Добавить трассировку с OpenTelemetry:

   from opentelemetry import trace
   tracer = trace.get_tracer(__name__)
   # в каждом агенте: with tracer.start_as_current_span(f"agent_{self.name}") as span:
   #     span.set_attribute("context.size", len(context.json()))
   #     span.set_attribute("context.fields", str(context.dict().keys()))
   

2. Создать скрипт проверки потери данных — test_context_integrity.py:

   • Сохранить копию context после AgentA.
   • После AgentB — проверить, что agent_a_output не изменился.
   • После AgentC — проверить, что все поля до agent_c_final остались неизменными.

3. Запустить 5 разных запросов и собрать статистику (число полей, размер context до / после).

4. Визуализировать в Jaeger (опционально) или вывести JSON-лог в файл.

Ожидаемый результат этапа Логи / трассы, подтверждающие полную передачу контекста. Тест integrity проходит.

Этап 5: Рефакторинг и документирование (30 мин)

Действия

1. Вынести LLM-промпты в отдельные файлы prompts/agent_a.txt, prompts/agent_b.txt, prompts/agent_c.txt.
2. Добавить обработку ошибок: retry при падении LLM, timeout.
3. Написать README.md с инструкцией по запуску, примером вывода и архитектурой.

Ожидаемый результат этапа Код приведён к читаемому виду. README содержит пример запуска python main.py "Мой запрос".

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

• Реализованы 3 агента с handoff через HandoffSignal.
• Оркестратор последовательно вызывает агентов в порядке A → B → C.
• После прохождения всех агентов контекст содержит все три поля: agent_a_output, agent_b_enriched, agent_c_final.
• Набор из 5 тестовых запросов проходит интеграционный тест без изменения agent_a_output после AgentB и AgentC.
• Код покрыт юнит-тестами (минимум 1 на агента) — pytest, тесты проходят.
• В трассировке (или логах) видны точки handoff и размер контекста.
• README содержит команду запуска и описание архитектуры.
• Репозиторий Git с историей коммитов.

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

Основной артефакт Папка проекта agentic_mesh/ с файлами:

Опционально Docker-compose с Ollama и Jaeger для локального запуска.

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

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

Примечание: Если используется самописный фреймворк вместо CrewAI, время увеличится на 1 час (первый этап более детальный). При использовании готового CrewAI — экономия ~30 мин.

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

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

• Я спроектировал контракты (AgentContext, HandoffSignal) до написания кода.
• Каждый агент модифицирует только своё поле, не трогая чужие.
• Оркестратор не изменяет контекст вручную — только через handoff.
• Я запустил тест integrity на 5 разных запросах и убедился, что потерь нет.
• В README описаны шаги для запуска (установка зависимостей, запуск Ollama, выполнение main.py).
• Я добавил хотя бы один unit-тест на каждого агента (mock LLM).
• Код выложен в Git с минимум двумя коммитами (черновик и финальная версия).