1. Введение
Обзор
В этой лабораторной работе вы выйдете за рамки простых чат-ботов и создадите распределенную многоагентную систему .
Хотя один специалист с дипломом магистра права может ответить на вопросы, в реальных условиях для решения сложных задач часто требуются специализированные роли. Вы же не поручаете бэкенд-разработчику проектировать пользовательский интерфейс, а дизайнеру — оптимизировать запросы к базе данных. Аналогично, мы можем создавать специализированных агентов искусственного интеллекта, которые сосредоточены на одной задаче и координируют свои действия для решения сложных проблем.
Вы разработаете систему создания курсов, состоящую из:
- Агент-исследователь : Использует
google_searchдля поиска актуальной информации. - Эксперт-оценщик : Критическая оценка исследования на предмет качества и полноты.
- Content Builder Agent : Превращение результатов исследования в структурированный курс.
- Агент-оркестратор : Управление рабочим процессом и коммуникацией между этими специалистами.
Предварительные требования
- Базовые знания Python.
- Знание Google Cloud Console.
Что вы будете делать
- Определите агента (
researcher), использующего инструмент и способного осуществлять поиск в интернете. - Реализуйте структурированный вывод с помощью Pydantic для
judge. - Подключайтесь к удаленным агентам, используя протокол Agent-to-Agent (A2A) .
- Создайте
LoopAgentдля формирования обратной связи между исследователем и судьей. - Запустите распределенную систему локально, используя ADK.
- Разверните многоагентную систему в Google Cloud Run .
Принципы архитектуры и оркестровки
Прежде чем писать код, давайте разберемся, как эти агенты взаимодействуют друг с другом. Мы создаем конвейер создания курсов .
Проектирование системы

Организация работы с агентами
Стандартные агенты (например, исследователь) работают. Агенты-оркестраторы (например, LoopAgent или SequentialAgent ) управляют другими агентами. У них нет собственных инструментов; их «инструмент» — делегирование.
-
LoopAgent: Этот класс работает как циклwhileв коде. Он запускает последовательность агентов многократно, пока не будет выполнено условие (или не будет достигнуто максимальное количество итераций). Мы используем его для цикла исследований :- Исследователь находит информацию.
- Судья дает ему критику.
- Если Judge выдает "Fail", EscalationChecker позволяет циклу продолжиться.
- Если судья говорит «Пройдено», то EscalationChecker прерывает цикл.
-
SequentialAgent: Это работает как стандартное выполнение скрипта. Он запускает агенты один за другим. Мы используем это для высокоуровневого конвейера :- Сначала запустите цикл исследования (пока он не завершится получением корректных данных).
- Затем запустите конструктор контента (для создания курса).
Сочетая эти элементы, мы создаем надежную систему, способную к самокоррекции перед формированием окончательного результата.
2. Настройка
Настройка среды
Откройте Cloud Shell : откройте новую вкладку и введите shell.cloud.google.com
Получите стартовый код
- Скопируйте стартовый репозиторий в свою домашнюю директорию:
cd ~ git clone https://github.com/amitkmaraj/prai-roadshow-lab-1-starter.git cd prai-roadshow-lab-1-starter - Запустите скрипт инициализации, чтобы связать средства, полученные в рамках программы пополнения счета, с выставлением счетов.
chmod +x ./init.sh ./init.sh - Откройте эту папку в вашем редакторе.
Включить API
Теперь, когда у вас есть новый проект, выполните следующую команду, чтобы включить необходимые сервисы Google Cloud:
gcloud services enable \
run.googleapis.com \
artifactregistry.googleapis.com \
cloudbuild.googleapis.com \
aiplatform.googleapis.com \
compute.googleapis.com
Это может занять несколько секунд.
Установите зависимости
Мы используем uv для быстрого управления зависимостями.
- Установите зависимости проекта:
# Ensure you have uv installed: pip install uv uv sync - Укажите идентификатор вашего проекта в Google Cloud.
- Совет : Вы можете найти идентификатор своего проекта на панели управления Cloud Console или выполнив команду
gcloud config get-value project.
export GOOGLE_CLOUD_PROJECT=$(gcloud config get-value project) - Совет : Вы можете найти идентификатор своего проекта на панели управления Cloud Console или выполнив команду
- Установите оставшиеся переменные среды:
Внимание: переменные окружения не сохраняются между новыми сеансами терминала. Если вы откроете новую вкладку терминала, вам потребуется повторно выполнить эти команды экспорта.export GOOGLE_CLOUD_LOCATION=us-central1 export GOOGLE_GENAI_USE_VERTEXAI=true
3. 🕵️ Агент-исследователь

Исследователь — это специалист. Его единственная задача — находить информацию. Для этого ему необходим инструмент: поиск Google.
Зачем отделять исследователя?
Подробный анализ: Почему бы не поручить все задачи одному агенту?
Небольшие, узкоспециализированные агенты проще оценивать и отлаживать . Если исследование проведено плохо, вы продолжаете работу над заданием исследователя. Если форматирование курса плохое, вы продолжаете работу над конструктором контента. В монолитном, универсальном задании исправление одной проблемы часто приводит к поломке другой.
- Если вы работаете в Cloud Shell, выполните следующую команду, чтобы открыть редактор Cloud Shell:
Если вы работаете в локальной среде, откройте свою любимую IDE.cloudshell workspace . - Откройте файл
agents/researcher/agent.py. - Вы увидите скелет с надписью TODO.
- Добавьте следующий код для определения агента-
researcher:# ... existing imports ... # Define the Researcher Agent researcher = Agent( name="researcher", model=MODEL, description="Gathers information on a topic using Google Search.", instruction=""" You are an expert researcher. Your goal is to find comprehensive and accurate information on the user's topic. Use the `google_search` tool to find relevant information. Summarize your findings clearly. If you receive feedback that your research is insufficient, use the feedback to refine your next search. """, tools=[google_search], ) root_agent = researcher
Ключевая концепция: Использование инструментов
Обратите внимание, что мы передаем tools=[google_search] . ADK берет на себя сложность описания этого инструмента для LLM. Когда модель решает, что ей нужна информация, она генерирует структурированный вызов инструмента, ADK выполняет функцию Python google_search и передает результат обратно модели.
4. ⚖️ Агент-судья

Исследователь усердно работает, а магистранты могут быть ленивыми. Нам нужен судья для проверки работы. Судья принимает исследование и выставляет структурированную оценку «зачет/незачет».
Структурированный вывод
Подробный анализ: Для автоматизации рабочих процессов нам необходимы предсказуемые результаты. Сложный текстовый обзор сложно программно обработать. Используя схему JSON (с помощью Pydantic), мы гарантируем, что Judge вернет логическое значение « pass или fail , на основе которого наш код сможет надежно реагировать.
- Откройте файл
agents/judge/agent.py. - Определите схему
JudgeFeedbackи агентаjudge.# 1. Define the Schema class JudgeFeedback(BaseModel): """Structured feedback from the Judge agent.""" status: Literal["pass", "fail"] = Field( description="Whether the research is sufficient ('pass') or needs more work ('fail')." ) feedback: str = Field( description="Detailed feedback on what is missing. If 'pass', a brief confirmation." ) # 2. Define the Agent judge = Agent( name="judge", model=MODEL, description="Evaluates research findings for completeness and accuracy.", instruction=""" You are a strict editor. Evaluate the 'research_findings' against the user's original request. If the findings are missing key info, return status='fail'. If they are comprehensive, return status='pass'. """, output_schema=JudgeFeedback, # Disallow delegation because it should only output the schema disallow_transfer_to_parent=True, disallow_transfer_to_peers=True, ) root_agent = judge
Ключевая концепция: Ограничивающее поведение агента
Мы установили disallow_transfer_to_parent=True и disallow_transfer_to_peers=True . Это заставляет Judge возвращать только структурированный JudgeFeedback . Он не может решить, «общаться» ли с пользователем в чате или делегировать задачу другому агенту. Это делает его детерминированным компонентом в нашей логической схеме.
5. 🧪 Тестирование в изоляции
Перед подключением мы можем убедиться в работоспособности каждого агента. ADK позволяет запускать агенты по отдельности.
Ключевая концепция: Интерактивная среда выполнения
adk run запускает облегченную среду, где вы выступаете в роли «пользователя». Это позволяет тестировать инструкции агента и использование инструментов в изолированном режиме. Если агент выдает ошибку здесь (например, не может использовать поиск Google), он обязательно выдаст ошибку и в процессе оркестрации.
- Запустите программу Researcher в интерактивном режиме. Обратите внимание, что мы указываем конкретный каталог агента:
# This runs the researcher agent in interactive mode uv run adk run agents/researcher - В командной строке чата введите:
Программа должна использовать инструмент поиска Google и вернуть ответ. Примечание: Если вы видите ошибку, указывающую на то, что проект, местоположение и использование вершин не заданы, убедитесь, что идентификатор вашего проекта задан, и выполните следующее:Find the population of Tokyo in 2020export GOOGLE_CLOUD_PROJECT=$(gcloud config get-value project) export GOOGLE_CLOUD_LOCATION=us-central1 export GOOGLE_GENAI_USE_VERTEXAI=true - Выйти из чата (Ctrl+C).
- Запустите судью в интерактивном режиме:
uv run adk run agents/judge - В поле чата имитируйте ввод следующих данных:
Функция должна возвращатьTopic: Tokyo. Findings: Tokyo is a city.status='fail'поскольку результаты слишком краткие.
6. ✍️ Агент для создания контента

Контент-менеджер — это творческий автор. Он берет утвержденные исследования и превращает их в учебный курс.
- Откройте файл
agents/content_builder/agent.py. - Определите агент
content_builder.content_builder = Agent( name="content_builder", model=MODEL, description="Transforms research findings into a structured course.", instruction=""" You are an expert course creator. Take the approved 'research_findings' and transform them into a well-structured, engaging course module. **Formatting Rules:** 1. Start with a main title using a single `#` (H1). 2. Use `##` (H2) for main section headings. 3. Use bullet points and clear paragraphs. 4. Maintain a professional but engaging tone. Ensure the content directly addresses the user's original request. """, ) root_agent = content_builder
Ключевая концепция: Распространение контекста
Вы можете задаться вопросом: «Откуда Content Builder узнает, что обнаружил Researcher?» В ADK агенты в конвейере совместно используют session.state . Позже, в Orchestrator, мы настроим Researcher и Judge так, чтобы они сохраняли свои результаты в это общее состояние. Запрос Content Builder фактически имеет доступ к этой истории.
7. 🎻 Оркестратор

Оркестратор — это руководитель нашей многоагентной команды. В отличие от агентов-специалистов (Исследователь, Судья, Контент-менеджер), выполняющих конкретные задачи, работа Оркестратора заключается в координации рабочего процесса и обеспечении корректного обмена информацией между ними.
🌐 Архитектура: взаимодействие между агентами (A2A)

В этой лабораторной работе мы создаём распределённую систему . Вместо запуска всех агентов в одном процессе Python, мы развертываем их как независимые микросервисы. Это позволяет каждому агенту масштабироваться независимо и выходить из строя без сбоя всей системы.
Для этого мы используем протокол «агент-агент» (A2A) .
Протокол A2A
Подробное описание: В производственной системе агенты работают на разных серверах (или даже в разных облаках). Протокол A2A создает стандартный способ для них обнаруживать друг друга и взаимодействовать по протоколу HTTP. RemoteA2aAgent — это ADK-клиент для этого протокола.
- Откройте файл
agents/orchestrator/agent.py. - Найдите комментарий
# TODO: Define Remote Agentsили раздел с определениями удаленных агентов. - Добавьте следующий код для определения подключений. Убедитесь, что он размещен после импортов и перед любыми другими определениями агентов.
# ... existing code ... # Connect to the Researcher (Localhost port 8001) researcher_url = os.environ.get("RESEARCHER_AGENT_CARD_URL", "http://localhost:8001/a2a/agent/.well-known/agent-card.json") researcher = RemoteA2aAgent( name="researcher", agent_card=researcher_url, description="Gathers information using Google Search.", # IMPORTANT: Save the output to state for the Judge to see after_agent_callback=create_save_output_callback("research_findings"), # IMPORTANT: Use authenticated client for communication httpx_client=create_authenticated_client(researcher_url) ) # Connect to the Judge (Localhost port 8002) judge_url = os.environ.get("JUDGE_AGENT_CARD_URL", "http://localhost:8002/a2a/agent/.well-known/agent-card.json") judge = RemoteA2aAgent( name="judge", agent_card=judge_url, description="Evaluates research.", after_agent_callback=create_save_output_callback("judge_feedback"), httpx_client=create_authenticated_client(judge_url) ) # Content Builder (Localhost port 8003) content_builder_url = os.environ.get("CONTENT_BUILDER_AGENT_CARD_URL", "http://localhost:8003/a2a/agent/.well-known/agent-card.json") content_builder = RemoteA2aAgent( name="content_builder", agent_card=content_builder_url, description="Builds the course.", httpx_client=create_authenticated_client(content_builder_url) )
8. 🛑 Проверка эскалации
Для завершения цикла необходим способ его остановки. Если судья выдает «Пройдено», мы хотим немедленно выйти из цикла и перейти к конструктору контента.
Пользовательская логика с использованием BaseAgent
Подробный анализ: Не все агенты используют LLM. Иногда требуется простая логика на Python. BaseAgent позволяет определить агента, который просто выполняет код. В этом случае мы проверяем состояние сессии и используем EventActions(escalate=True) чтобы дать сигнал LoopAgent остановиться.
- Все еще в файле
agents/orchestrator/agent.py. - Найдите заполнитель TODO
EscalationChecker. - Замените его следующей реализацией:
class EscalationChecker(BaseAgent): """Checks the judge's feedback and escalates (breaks the loop) if it passed.""" async def _run_async_impl( self, ctx: InvocationContext ) -> AsyncGenerator[Event, None]: # Retrieve the feedback saved by the Judge feedback = ctx.session.state.get("judge_feedback") print(f"[EscalationChecker] Feedback: {feedback}") # Check for 'pass' status is_pass = False if isinstance(feedback, dict) and feedback.get("status") == "pass": is_pass = True # Handle string fallback if JSON parsing failed elif isinstance(feedback, str) and '"status": "pass"' in feedback: is_pass = True if is_pass: # 'escalate=True' tells the parent LoopAgent to stop looping yield Event(author=self.name, actions=EventActions(escalate=True)) else: # Continue the loop yield Event(author=self.name) escalation_checker = EscalationChecker(name="escalation_checker")
Ключевая концепция: управление потоком выполнения посредством событий.
Агенты общаются не только с помощью текста, но и с помощью событий . Отправив событие с escalate=True , этот агент посылает сигнал своему родительскому объекту ( LoopAgent ). LoopAgent запрограммирован на перехват этого сигнала и завершение цикла.
9. 🔁 Исследовательский цикл

Нам необходима обратная связь: Исследование -> Оценка -> (Неудача) -> Исследование -> ...
- Все еще в файле
agents/orchestrator/agent.py. - Добавьте определение
research_loop. Разместите его после классаEscalationCheckerи экземпляраescalation_checker.research_loop = LoopAgent( name="research_loop", description="Iteratively researches and judges until quality standards are met.", sub_agents=[researcher, judge, escalation_checker], max_iterations=3, )
Ключевая концепция: LoopAgent
LoopAgent последовательно перебирает своих sub_agents .
-
researcher: Находит данные. -
judge: Оценивает данные. -
escalation_checker: Определяет, следует лиyield Event(escalate=True). Если выполняетсяescalate=True, цикл прерывается досрочно. В противном случае он перезапускается с исследователя (доmax_iterations).
10. 🔗 Финальный трубопровод

Наконец, сшейте все вместе.
- Все еще в файле
agents/orchestrator/agent.py. - Определите параметр
root_agentв конце файла. Убедитесь, что он заменяет любой существующий заполнительroot_agent = None.root_agent = SequentialAgent( name="course_creation_pipeline", description="A pipeline that researches a topic and then builds a course from it.", sub_agents=[research_loop, content_builder], )
Ключевая концепция: Иерархическая композиция
Обратите внимание, что research_loop сам по себе является агентом ( LoopAgent ). Мы рассматриваем его так же, как и любого другого подагента в SequentialAgent . Такая компонуемость позволяет строить сложную логику, вкладывая простые шаблоны (циклы внутри последовательностей, последовательности внутри маршрутизаторов и т. д.).
11. 💻 Занимайтесь на местном уровне
Прежде чем запускать все компоненты, давайте посмотрим, как ADK имитирует распределенную среду локально.
Подробный анализ: как работает местное развитие
В микросервисной архитектуре каждый агент работает как отдельный сервер. При развертывании у вас будет 4 различных сервиса Cloud Run. Имитация этого локально может быть проблематичной, если вам придется открывать 4 вкладки терминала и выполнять 4 команды.
Этот скрипт запускает процессы uvicorn для Researcher (порт 8001), Judge (8002) и Content Builder (8003). Он устанавливает переменные среды, такие как RESEARCHER_AGENT_CARD_URL , и передает их Orchestrator (порт 8004). Именно так мы будем настраивать это в облаке позже!

- Запустите скрипт оркестровки:
Это запускает 4 отдельных процесса../run_local.sh - Проверьте это:
- При использовании Cloud Shell: нажмите кнопку « Предварительный просмотр веб-страницы» (в правом верхнем углу терминала) -> Предварительный просмотр на порту 8080 -> Измените порт на
8000. - При локальном запуске: откройте
http://localhost:8000в браузере. - Задание: «Создайте курс об истории кофе».
- Обратите внимание: Организатор вызывает Исследователь. Результат передается Судье. Если Судья не проходит проверку, цикл продолжается!
- "Внутренняя ошибка сервера" / Ошибки аутентификации: Если вы видите ошибки аутентификации (например, связанные с
google-auth), убедитесь, что вы выполнилиgcloud auth application-default loginесли работаете на локальном компьютере. В Cloud Shell убедитесь, что переменная средыGOOGLE_CLOUD_PROJECTустановлена правильно. - Ошибки терминала: Если команда завершается с ошибкой в новом окне терминала, не забудьте повторно экспортировать переменные среды (
GOOGLE_CLOUD_PROJECTи т. д.).
- При использовании Cloud Shell: нажмите кнопку « Предварительный просмотр веб-страницы» (в правом верхнем углу терминала) -> Предварительный просмотр на порту 8080 -> Измените порт на
- Тестирование агентов в изолированном режиме: даже при работе всей системы можно тестировать отдельные агенты, напрямую обращаясь к их портам. Это полезно для отладки конкретного компонента без запуска всей цепочки.
- Только для исследователей (порт 8001):
http://localhost:8001 - Только для судей (порт 8002):
http://localhost:8002 - Только для Content Builder (порт 8003):
http://localhost:8003 - Orchestrator (порт 8004):
http://localhost:8004(Прямой доступ к логике Orchestrator)
- Только для исследователей (порт 8001):
12. 🚀 Развертывание в Cloud Run
Окончательная проверка будет проведена в облаке. Мы развернем каждый агент как отдельную службу.
Понимание конфигурации развертывания
При развертывании агентов в Cloud Run мы передаем несколько переменных среды для настройки их поведения и подключения:
-
GOOGLE_CLOUD_PROJECT: Гарантирует, что агент использует правильный проект Google Cloud для логирования и вызовов Vertex AI. -
GOOGLE_GENAI_USE_VERTEXAI: Указывает фреймворку агентов (ADK) использовать Vertex AI для вывода модели вместо прямого вызова API Gemini. -
[AGENT]_AGENT_CARD_URL: Этот параметр крайне важен для оркестратора. Он указывает оркестратору, где найти удаленных агентов. Установив его на развернутый URL-адрес Cloud Run (в частности, путь к карточке агента), мы позволяем оркестратору обнаруживать и взаимодействовать с исследователем, судьей и создателем контента через Интернет.
- Задействуйте исследователя:
Скопируйте URL-адрес:gcloud run deploy researcher \ --source agents/researcher/ \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT \ --set-env-vars GOOGLE_GENAI_USE_VERTEXAI="true"RESEARCHER_URL=$(gcloud run services describe researcher --region us-central1 --format='value(status.url)') echo $RESEARCHER_URL - Вывести на поле боя судью:
Скопируйте URL-адрес:gcloud run deploy judge \ --source agents/judge/ \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT \ --set-env-vars GOOGLE_GENAI_USE_VERTEXAI="true"JUDGE_URL=$(gcloud run services describe judge --region us-central1 --format='value(status.url)') echo $JUDGE_URL - Разверните конструктор контента:
Скопируйте URL-адрес:gcloud run deploy content-builder \ --source agents/content_builder/ \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT \ --set-env-vars GOOGLE_GENAI_USE_VERTEXAI="true"CONTENT_BUILDER_URL=$(gcloud run services describe content-builder --region us-central1 --format='value(status.url)') echo $CONTENT_BUILDER_URL - Разверните Orchestrator: используйте полученные переменные среды для настройки Orchestrator.
Скопируйте URL-адрес:gcloud run deploy orchestrator \ --source agents/orchestrator/ \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars RESEARCHER_AGENT_CARD_URL=$RESEARCHER_URL/a2a/agent/.well-known/agent-card.json \ --set-env-vars JUDGE_AGENT_CARD_URL=$JUDGE_URL/a2a/agent/.well-known/agent-card.json \ --set-env-vars CONTENT_BUILDER_AGENT_CARD_URL=$CONTENT_BUILDER_URL/a2a/agent/.well-known/agent-card.json \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT \ --set-env-vars GOOGLE_GENAI_USE_VERTEXAI="true"ORCHESTRATOR_URL=$(gcloud run services describe orchestrator --region us-central1 --format='value(status.url)') echo $ORCHESTRATOR_URL - Разверните интерфейсную часть:
gcloud run deploy course-creator \ --source app \ --region us-central1 \ --allow-unauthenticated \ --labels dev-tutorial=prod-ready-1 \ --set-env-vars AGENT_SERVER_URL=$ORCHESTRATOR_URL \ --set-env-vars GOOGLE_CLOUD_PROJECT=$GOOGLE_CLOUD_PROJECT - Тестирование удаленного развертывания: откройте URL-адрес развернутого Orchestrator. Теперь он полностью работает в облаке, используя бессерверную инфраструктуру Google для масштабирования ваших агентов! Подсказка : все микросервисы и их URL-адреса вы найдете в интерфейсе Cloud Run.
13. Резюме
Поздравляем! Вы успешно создали и развернули готовую к использованию распределенную многоагентную систему.
Что мы достигли
- Разложили сложную задачу на составляющие : вместо одного большого задания мы разделили работу на специализированные роли (исследователь, судья, разработчик контента).
- Внедрен контроль качества : Мы использовали
LoopAgentи структурированныйJudge, чтобы гарантировать, что на заключительный этап поступает только высококачественная информация. - Создано для продакшена : Используя протокол Agent-to-Agent (A2A) и Cloud Run , мы создали систему, в которой каждый агент представляет собой независимый, масштабируемый микросервис. Это гораздо надежнее, чем запускать все в одном скрипте Python.
- Оркестрация : Мы использовали
SequentialAgentиLoopAgentдля определения четких шаблонов управления потоком выполнения.
Следующие шаги
Теперь, когда у вас есть основа, вы можете расширить эту систему:
- Добавьте больше инструментов : предоставьте исследователю доступ к внутренним документам или API.
- Улучшите работу судьи : добавьте более конкретные критерии или даже этап с участием человека.
- Замена моделей : Попробуйте использовать разные модели для разных агентов (например, более быструю модель для Судьи, более мощную модель для Автора контента).
Теперь вы готовы создавать сложные и надежные агентные рабочие процессы в Google Cloud!