1. 소개
개요
이 Codelab에서는 에이전트 개발 키트 (ADK)를 사용하여 확장 가능한 비동기 에이전트 시스템을 빌드하는 방법을 보여줍니다. PubSub 풀 구독의 작업을 처리하는 ADK 빠른 시작 날씨 에이전트를 호스팅하는 Cloud Run 작업자 풀을 만듭니다.
학습할 내용
- 에이전트 개발 키트 (ADK)로 단일 턴 에이전트를 만드는 방법
- PubSub 구독에서 가져오는 Cloud Run 작업자 풀을 배포하는 방법
2. 시작하기 전에
API 사용 설정
이 Codelab을 시작하기 전에 다음을 실행하여 다음 API를 사용 설정하세요.
gcloud services enable \
run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
pubsub.googleapis.com \
aiplatform.googleapis.com
3. 설정 및 요구사항
필수 리소스를 설정하려면 다음 단계를 따르세요.
- 이 Codelab의 환경 변수를 설정합니다.
export PROJECT_ID=<YOUR_PROJECT_ID>
export REGION=europe-west1
# AR repo
export AR_REPO="codelab-agent-wp"
# Application Names
export WORKER_APP_NAME="multi-tool-agent-worker"
# Pub/Sub Resources
export MY_TOPIC="pull-pubsub-topic-agent"
export MY_SUBSCRIPTION="agent-wp-sub"
# Service Accounts
export WORKER_SA_NAME="agent-worker-sa"
export WORKER_SA_ADDRESS="${WORKER_SA_NAME}@${PROJECT_ID}.iam.gserviceaccount.com"
서비스 계정 만들기
보안을 위해 작업자에게 필요한 권한만 부여되도록 전용 서비스 계정을 만듭니다.
작업자의 서비스 계정을 만듭니다.
gcloud iam service-accounts create ${WORKER_SA_NAME} \
--display-name="Service Account for ADK Agent Worker"
서비스 계정에 필요한 역할을 부여합니다. Pub/Sub에서 메시지를 가져오고 ADK에서 사용하는 Vertex AI 모델을 호출해야 합니다.
# Role for subscribing to Pub/Sub
gcloud projects add-iam-policy-binding ${PROJECT_ID} \
--member="serviceAccount:${WORKER_SA_ADDRESS}" \
--role="roles/pubsub.admin"
# Role for invoking Vertex AI
gcloud projects add-iam-policy-binding ${PROJECT_ID} \
--member="serviceAccount:${WORKER_SA_ADDRESS}" \
--role="roles/aiplatform.user"
Pub/Sub 리소스 만들기
작업 대기열 역할을 할 Pub/Sub 주제를 만듭니다.
gcloud pubsub topics create $MY_TOPIC
작업자가 메시지를 가져올 Pub/Sub 구독을 만듭니다.
gcloud pubsub subscriptions create $MY_SUBSCRIPTION --topic=$MY_TOPIC
4. Cloud Run 작업자 풀 만들기
agents-wp라는 프로젝트 디렉터리를 만듭니다.
mkdir agents-wp && cd agents-wp
Dockerfile를 만드는 방법
touch Dockerfile
Dockerfile에 다음 콘텐츠를 추가합니다.
FROM python:3.11-slim
WORKDIR /app
# Create a non-root user
RUN adduser --disabled-password --gecos "" myuser
# Switch to the non-root user
USER myuser
# Set up environment variables
ENV PATH="/home/myuser/.local/bin:$PATH"
# Copy agent files
COPY --chown=myuser:myuser multi_tool_agent/ /app/multi_tool_agent/
# Install dependencies from requirements.txt
RUN pip install -r /app/multi_tool_agent/requirements.txt
# Set the entrypoint to run the agent as a worker
CMD ["python3", "/app/multi_tool_agent/main.py"]
내부에 multi_tool_agent라는 하위 디렉터리를 만듭니다. 폴더 이름 multi_tool_agent에 밑줄이 있습니다. 이 폴더는 나중에 배포할 ADK 에이전트의 이름과 일치해야 합니다.
mkdir multi_tool_agent && cd multi_tool_agent
__init__.py 파일 만들기
touch __init__.py
__init__.py 파일에 다음을 추가합니다.
from . import agent
agent.py 파일 만들기
touch agent.py
agent.py 파일에 다음 내용을 추가합니다.
import datetime
from zoneinfo import ZoneInfo
from google.adk.agents.llm_agent import Agent
def get_weather(city: str) -> dict:
"""Retrieves the current weather report for a specified city.
Args:
city (str): The name of the city for which to retrieve the weather report.
Returns:
dict: status and result or error msg.
"""
print(f"--- Entering get_weather function for city: {city} ---")
if city.lower() == "new york":
result = {
"status": "success",
"report": (
"The weather in New York is sunny with a temperature of 25 degrees"
" Celsius (77 degrees Fahrenheit)."
),
}
else:
result = {
"status": "error",
"error_message": f"Weather information for '{city}' is not available.",
}
print(f"--- Exiting get_weather function with result: {result} ---")
return result
def get_current_time(city: str) -> dict:
"""Returns the current time in a specified city.
Args:
city (str): The name of the city for which to retrieve the current time.
Returns:
dict: status and result or error msg.
"""
print(f"--- Entering get_current_time function for city: {city} ---")
if city.lower() == "new york":
tz_identifier = "America/New_York"
else:
result = {
"status": "error",
"error_message": (
f"Sorry, I don't have timezone information for {city}."
),
}
print(f"--- Exiting get_current_time function with result: {result} ---")
return result
tz = ZoneInfo(tz_identifier)
now = datetime.datetime.now(tz)
report = (
f'The current time in {city} is {now.strftime("%Y-%m-%d %H:%M:%S %Z%z")}'
)
result = {"status": "success", "report": report}
print(f"--- Exiting get_current_time function with result: {result} ---")
return result
print("--- Creating root_agent ---")
root_agent = Agent(
name="weather_time_agent",
model="gemini-2.5-flash",
description=(
"Agent to answer questions about the time and weather in a city."
),
instruction=(
"You are a helpful agent who can answer user questions about the time and weather in a city."
),
tools=[get_weather, get_current_time],
)
print("--- root_agent created ---")
main.py 파일 만들기
touch main.py
main.py 파일에 다음을 추가합니다.
import asyncio
import os
from google.adk.runners import InMemoryRunner, Runner
from google.genai import types
from google.cloud import pubsub_v1
from agent import root_agent
# --- Runner-based Invocation with Proper Async Handling ---
APP_NAME = "multi_tool_agent_worker"
USER_ID = "pubsub_user"
async def process_message(runner: Runner, message_data: bytes):
"""Processes a single message using the agent runner."""
print(f"Processing message: {message_data}")
try:
prompt = message_data.decode("utf-8")
session = await runner.session_service.create_session(
app_name=APP_NAME,
user_id=USER_ID
)
final_response_text = ""
async for event in runner.run_async(
user_id=USER_ID,
session_id=session.id,
new_message=types.Content(
role="user", parts=[types.Part.from_text(text=prompt)]
),
):
if event.content and event.content.parts:
if event.author != "user":
# Filter out thought parts to get only the final response text
final_response_text += "".join(
part.text or "" for part in event.content.parts if not part.thought
)
print(f"Agent response: {final_response_text}")
except Exception as e:
print(f"Error processing message: {e}")
async def async_worker(queue: asyncio.Queue, runner: Runner):
"""Continuously gets messages from the queue and processes them."""
while True:
message = await queue.get()
if message is None: # Sentinel for stopping
break
await process_message(runner, message.data)
message.ack()
queue.task_done()
async def main():
"""Sets up the Pub/Sub subscriber and the async worker."""
project_id = os.environ.get("GOOGLE_CLOUD_PROJECT")
subscription_id = os.environ.get("SUBSCRIPTION_ID")
if not project_id or not subscription_id:
print("GOOGLE_CLOUD_PROJECT and SUBSCRIPTION_ID environment variables must be set.")
return
runner = InMemoryRunner(agent=root_agent, app_name=APP_NAME)
message_queue = asyncio.Queue()
subscriber = pubsub_v1.SubscriberClient()
subscription_path = subscriber.subscription_path(project_id, subscription_id)
loop = asyncio.get_running_loop()
callback = lambda message: loop.call_soon_threadsafe(
message_queue.put_nowait, message
)
print(f"Listening for messages on {subscription_path}...\n")
streaming_pull_future = subscriber.subscribe(subscription_path, callback=callback)
worker_task = asyncio.create_task(async_worker(message_queue, runner))
try:
# This will block until the subscription is cancelled or an error occurs.
await loop.run_in_executor(None, streaming_pull_future.result)
except KeyboardInterrupt:
print("Shutting down...")
finally:
streaming_pull_future.cancel()
await message_queue.put(None) # Stop the worker
await worker_task # Wait for the worker to finish
await runner.close()
subscriber.close()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
print("Exiting.")
requirements.txt 파일 만들기
touch requirements.txt
requirements.txt 파일에 다음을 추가합니다.
google-adk
google-cloud-pubsub
google-cloud-aiplatform
다음과 같은 폴더 구조가 있어야 합니다.
agents-wp
- multi_tool_agent
- __init__.py
- agent.py
- main.py
- requirements.txt
- Dockerfile
5. 빌드 및 배포
Artifact Registry 저장소 만들기
컨테이너 이미지를 저장할 위치가 필요합니다.
gcloud artifacts repositories create codelab-agent-wp \
--repository-format=docker \
--location=${REGION} \
--description="Repo for Cloud Run source deployments"
컨테이너 이미지 빌드
Dockerfile이 있는 루트 agents-wp 디렉터리로 이동합니다.
cd ..
다음 빌드 명령어를 실행합니다.
gcloud builds submit . --tag \
${REGION}-docker.pkg.dev/${PROJECT_ID}/${AR_REPO}/${WORKER_APP_NAME}:latest
Cloud Run에 배포
에이전트 작업자 이미지를 배포합니다.
gcloud beta run worker-pools deploy ${WORKER_APP_NAME} \
--image=${REGION}-docker.pkg.dev/${PROJECT_ID}/${AR_REPO}/${WORKER_APP_NAME}:latest \
--service-account=${WORKER_SA_ADDRESS} \
--region=${REGION} \
--set-env-vars="SUBSCRIPTION_ID=${MY_SUBSCRIPTION}" \
--set-env-vars="PYTHONUNBUFFERED=1" \
--set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=1" \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}"
6. 에이전트 테스트
Pub/Sub 주제에 직접 메시지를 게시하여 작업자를 테스트할 수 있습니다.
gcloud pubsub topics publish ${MY_TOPIC} --message="What is the weather in New York?"
이 명령어를 실행하여 Google Cloud 콘솔에서 multi-tool-agent-worker 서비스의 로그를 확인할 수 있습니다.
gcloud logging read 'resource.type="cloud_run_worker_pool" AND resource.labels.worker_pool_name="'$WORKER_APP_NAME'" AND resource.labels.location="'$REGION'"' --limit 10 --format="value(textPayload)"
메시지가 수신되어 처리되었음을 나타내는 출력과 함께 상담사의 응답이 표시됩니다.
Agent response: The weather in New York is sunny with a temperature of 25 degrees Celsius (77 degrees Fahrenheit).
7. 축하합니다.
축하합니다. Codelab을 완료했습니다.
작업자 풀 및 호스트 에이전트에 관한 Cloud Run 문서를 검토하는 것이 좋습니다.
학습한 내용
- 에이전트 개발 키트 (ADK)로 단일 턴 에이전트를 만드는 방법
- PubSub 구독에서 가져오는 Cloud Run 작업자 풀을 배포하는 방법
8. 삭제
요금이 청구되지 않도록 하려면 만든 리소스를 삭제하세요.
Cloud Run 작업자 풀 삭제
gcloud beta run worker-pools delete ${WORKER_APP_NAME} --region=${REGION}
Pub/Sub 리소스 삭제
gcloud pubsub subscriptions delete ${MY_SUBSCRIPTION}
gcloud pubsub topics delete ${MY_TOPIC}
Artifact Registry 저장소 삭제
gcloud artifacts repositories delete ${AR_REPO} --location=$REGION
서비스 계정 삭제
gcloud iam service-accounts delete ${WORKER_SA_ADDRESS}
전체 프로젝트를 삭제하려면 리소스 관리로 이동하여 2단계에서 만든 프로젝트를 선택하고 삭제를 선택합니다. 프로젝트를 삭제하면 Cloud SDK에서 프로젝트를 변경해야 합니다. gcloud projects list를 실행하여 사용 가능한 모든 프로젝트의 목록을 볼 수 있습니다.