이 Codelab 정보
1. 학습할 내용
환영합니다. 오늘은 아주 멋진 여정을 시작할 예정입니다. 먼저 인기 있는 소셜 이벤트 플랫폼인 InstaVibe를 생각해 보겠습니다. 그룹 활동을 계획하는 것은 좋은 방법이지만 일부 사용자에게는 번거로운 일로 느껴질 수 있습니다. 모든 친구가 관심을 가질 만한 주제를 파악한 다음 끝없는 이벤트 또는 장소 옵션을 살펴보고 모든 것을 조정하는 작업을 생각해 보세요. 엄청난 수의 버그죠. 바로 여기에서 AI, 특히 지능형 상담사를 도입하여 실질적인 변화를 가져올 수 있습니다.
이러한 에이전트가 사용자와 친구의 환경설정을 이해하기 위해 영리하게 '리슨'하고 사전에 맞춤화된 멋진 활동을 제안하는 등 어려운 작업을 처리할 수 있는 시스템을 구축하는 것이 목표입니다. Google의 목표는 InstaVibe의 소셜 계획을 원활하고 만족스러운 것으로 전환하는 것입니다. 이러한 스마트 어시스턴트를 빌드하려면 적절한 도구로 탄탄한 기반을 다져야 합니다.
다음은 표시되는 개념입니다.
Google ADK 기반: Google의 에이전트 개발 키트 (ADK)를 사용하여 첫 번째 지능형 에이전트를 빌드하는 기본사항을 익힙니다. 필수 구성요소, 에이전트 수명 주기, 프레임워크의 내장 도구를 효과적으로 활용하는 방법을 이해합니다.
Model Context Protocol (MCP)으로 에이전트 기능 확장: 에이전트가 특수한 작업을 실행하고 특정 정보에 액세스할 수 있도록 에이전트에 맞춤 도구와 컨텍스트를 제공하는 방법을 알아봅니다. Model Context Protocol (MCP) 개념을 소개합니다. 이 컨텍스트를 제공하도록 MCP 서버를 설정하는 방법을 알아봅니다.
상담사 상호작용 및 조정 설계: 단일 상담사를 넘어 상담사 조정을 이해합니다. 간단한 순차 워크플로에서 루프, 조건부 로직, 병렬 처리가 포함된 복잡한 시나리오에 이르기까지 다양한 상호작용 패턴을 설계합니다. 모듈식 작업을 관리하기 위해 ADK 프레임워크 내에 하위 에이전트 개념을 도입합니다.
공동작업 멀티 에이전트 시스템 빌드: 여러 에이전트가 협력하여 복잡한 목표를 달성하는 시스템을 설계하는 방법을 알아봅니다. 에이전트 간 (A2A) 통신 프로토콜을 학습하고 구현하여 분산 에이전트 (다른 머신이나 서비스에서 실행될 수 있음)가 안정적으로 상호작용할 수 있는 표준화된 방법을 수립합니다.
Google Cloud에서 에이전트 프로덕션화: 에이전트 애플리케이션을 개발 환경에서 클라우드로 전환합니다. Google Cloud Platform (GCP)에서 확장 가능하고 강력한 멀티 에이전트 시스템을 설계하고 배포하기 위한 권장사항을 알아봅니다. Cloud Run과 같은 GCP 서비스를 활용하는 방법을 알아보고 에이전트를 호스팅하고 관리하기 위한 최신 Google Agent Engine의 기능을 살펴보세요.
2. 아키텍처
InstaVibe를 통한 AI 기반 소셜 계획
소셜 동향 분석이란 무엇인가요?
소셜 동향 분석은 소셜 미디어, 포럼, 뉴스 사이트와 같은 플랫폼에서 디지털 대화를 모니터링하여 사람들이 특정 주제, 브랜드 또는 업계에 관해 어떻게 생각하고 있는지 파악하는 과정입니다. 이는 대중의 감정, 트렌드, 사용자 요구사항에 대한 유용한 정보를 제공합니다. 이 워크숍에서는 에이전트 기반 시스템 내에서 이 개념을 활용합니다.
InstaVibe팀에 속해 있습니다
젊은 성인을 대상으로 한 인기 소셜 이벤트 플랫폼을 운영하는 성공적인 스타트업인 'InstaVibe'에서 근무하고 있다고 가정해 보겠습니다. 상황이 순조롭지만 많은 기술 회사와 마찬가지로 팀에 AI를 사용한 혁신을 요구하는 투자자의 압력이 있습니다. 내부적으로도 다른 사용자만큼 참여도가 높지 않은 사용자 세그먼트를 발견했습니다. 이러한 사용자는 그룹 활동을 시작하는 데 소극적이거나 계획 과정이 어렵다고 생각하는 것일 수 있습니다. 이는 회사의 입장에서 이 중요한 사용자 그룹의 플랫폼 참여도가 낮아진다는 의미입니다.
팀의 연구에 따르면 AI 기반 지원이 이러한 사용자의 환경을 크게 개선할 수 있습니다. 사용자와 친구의 관심분야를 기반으로 관련 활동을 사전에 제안하여 외출 계획을 간소화하는 것이 목표입니다. 광고주와 광고 대행사가 직면한 문제는 다음과 같습니다. AI 에이전트가 관심분야 파악, 활동 조사, 잠재적인 초기 조정 등 시간이 많이 걸리는 작업을 자동화할 수 있는 방법은 무엇일까요?
에이전트 기반 솔루션 (프로토타입 개념)
멀티 에이전트 시스템을 기반으로 하는 프로토타입 기능을 개발할 것을 제안합니다. 개념적 분류는 다음과 같습니다.
- 소셜 프로파일링 에이전트: 이 에이전트는 소셜 동향 분석 기법을 사용하여 사용자 연결, 상호작용, 사용자의 선호와 관련된 잠재적으로 더 광범위한 대중 동향을 분석합니다. 이 목적은 공통 관심분야와 적절한 활동 특성 (예: 조용한 모임에 대한 선호도, 특정 취미)을 파악하는 것입니다.
- 이벤트 계획 에이전트: 소셜 프로파일링 에이전트의 통계를 사용하여 온라인 리소스에서 확인된 기준 (예: 위치, 관심분야)에 부합하는 특정 이벤트, 장소 또는 아이디어를 검색합니다.
- 플랫폼 상호작용 에이전트 (MCP 사용): 이 에이전트는 활동 계획 에이전트로부터 최종 계획을 가져옵니다. 주요 기능은 사전 정의된 MCP (Model Context Protocol) 도구를 활용하여 InstaVibe 플랫폼과 직접 상호작용하는 것입니다. 이 도구는 상담사에게 이벤트 제안 초안을 작성하고 계획을 요약하는 게시물을 작성하는 기능을 제공합니다.
- Orchestrator Agent: 이 에이전트는 중앙 조정자 역할을 합니다. InstaVibe 플랫폼에서 초기 사용자 요청을 수신하고 전반적인 목표 (예: '나와 친구를 위한 이벤트 계획')을 파악한 다음 논리적인 순서로 특정 작업을 적절한 전문 상담사에게 위임합니다. 상담사 간의 정보 흐름을 관리하고 최종 결과가 사용자에게 다시 전송되도록 합니다.
주요 아키텍처 요소 및 기술
Google Cloud Platform(GCP)
- Vertex AI:
- Gemini 모델: Google 에이전트의 추론 및 의사결정 기능을 지원하는 Gemini와 같은 Google의 최신 대규모 언어 모델 (LLM)에 대한 액세스를 제공합니다.
- Vertex AI Agent Engine: 오케스트레이터 에이전트를 배포, 호스팅, 확장하는 데 사용되는 관리형 서비스로, 프로덕션화를 간소화하고 인프라 복잡성을 추상화합니다.
- Cloud Run: 컨테이너화된 애플리케이션을 배포하기 위한 서버리스 플랫폼입니다. Google에서는 다음과 같은 목적으로 쿠키를 사용합니다.
- 기본 InstaVibe 웹 애플리케이션을 호스팅합니다.
- 개별 A2A 지원 에이전트 (Planner, 소셜 프로파일링, 플랫폼 상호작용)를 독립적인 마이크로서비스로 배포합니다.
- MCP 도구 서버를 실행하여 상담사가 InstaVibe의 내부 API를 사용할 수 있도록 합니다.
- Spanner: 전 세계적으로 분산된 완전 관리형의 strong consistency 관계형 데이터베이스입니다. 이 워크샵에서는 GRAPH DDL 및 쿼리 기능을 사용하여 그래프 데이터베이스로서의 기능을 활용하여 다음 작업을 수행합니다.
- 복잡한 소셜 관계 (사용자, 친구 관계, 이벤트 참석, 게시물)를 모델링하고 저장합니다.
- 소셜 프로파일링 상담사를 위해 이러한 관계를 효율적으로 쿼리할 수 있도록 합니다.
- Artifact Registry: 컨테이너 이미지를 저장, 관리, 보호하기 위한 완전 관리형 서비스입니다.
- Cloud Build: Google Cloud에서 빌드를 실행하는 서비스입니다. 이 도구를 사용하여 상담사 및 애플리케이션 소스 코드에서 Docker 컨테이너 이미지를 자동으로 빌드합니다.
- Cloud Storage: Cloud Build와 같은 서비스에서 빌드 아티팩트를 저장하는 데 사용하고 Agent Engine에서 운영 요구사항을 위해 사용합니다.
- 핵심 상담사 프레임워크 및 프로토콜:
- Google의 에이전트 개발 키트 (ADK): 다음을 위한 기본 프레임워크입니다.
- 개별 지능형 에이전트의 핵심 로직, 동작, 명령 집합을 정의합니다.
- 에이전트 수명 주기, 상태, 메모리 (단기 세션 상태 및 잠재적으로 장기 지식) 관리
- 상담사가 외부와 상호작용하는 데 사용할 수 있는 도구 (예: Google 검색 또는 맞춤 제작 도구) 통합
- 하위 에이전트의 순차, 루프, 병렬 실행을 포함한 멀티 에이전트 워크플로 조정
- Agent-to-Agent (A2A) 통신 프로토콜: 다음을 지원하는 개방형 표준입니다.
- 별도의 서비스로 실행되거나 다른 머신에서 실행되더라도 다양한 AI 상담사 간의 직접적이고 표준화된 커뮤니케이션 및 공동작업
- 상담사가 상담사 카드를 통해 서로의 기능을 탐색하고 할 일을 위임할 수 있습니다. 이는 Orchestrator 에이전트가 전문 Planner, 소셜, 플랫폼 에이전트와 상호작용하는 데 매우 중요합니다.
- A2A Python 라이브러리 (a2a-python): ADK 에이전트가 A2A 프로토콜을 말하도록 하는 데 사용되는 구체적인 라이브러리입니다. 다음에 필요한 서버 측 구성요소를 제공합니다.
- 상담사를 A2A 규정을 준수하는 서버로 노출합니다.
- 검색을 위한 '상담사 카드' 게재를 자동으로 처리합니다.
- 다른 상담사 (예: Orchestrator)로부터 들어오는 작업 요청을 수신하고 관리합니다.
- Model Context Protocol (MCP): 에이전트가 다음을 수행할 수 있는 개방형 표준입니다.
- 표준화된 방식으로 외부 도구, 데이터 소스, 시스템에 연결하고 활용합니다.
- Google의 플랫폼 상호작용 에이전트는 MCP 클라이언트를 사용하여 MCP 서버와 통신하며, MCP 서버는 InstaVibe 플랫폼의 기존 API와 상호작용하는 도구를 노출합니다.
- Google의 에이전트 개발 키트 (ADK): 다음을 위한 기본 프레임워크입니다.
- 디버깅 도구:
- A2A 검사기: A2A 검사기는 이 워크샵 전반에서 A2A 지원 상담사와 연결하고, 검사하고, 상호작용하는 데 사용되는 웹 기반 디버깅 도구입니다. 최종 프로덕션 아키텍처의 일부는 아니지만 개발 워크플로에서 중요한 부분입니다. 다음과 같은 이점을 제공합니다.
- 상담사 카드 뷰어: 상담사의 공개 기능을 가져오고 유효성을 검사합니다.
- 실시간 채팅 인터페이스: 즉시 테스트할 수 있도록 배포된 상담사에게 메시지를 직접 보낼 수 있습니다.
- 디버그 콘솔: 검사기와 에이전트 간에 교환되는 원시 JSON-RPC 메시지를 확인합니다.
- A2A 검사기: A2A 검사기는 이 워크샵 전반에서 A2A 지원 상담사와 연결하고, 검사하고, 상호작용하는 데 사용되는 웹 기반 디버깅 도구입니다. 최종 프로덕션 아키텍처의 일부는 아니지만 개발 워크플로에서 중요한 부분입니다. 다음과 같은 이점을 제공합니다.
- 언어 모델 (LLM): 시스템의 '두뇌':
- Google의 Gemini 모델: 특히 gemini-2.0-flash와 같은 버전을 사용합니다. 이러한 모델은 다음과 같은 이유로 선택됩니다.
- 고급 추론 및 안내 따르기: 복잡한 프롬프트를 이해하고, 상세한 안내를 따르고, 작업에 관해 추론하는 능력이 있어 상담사 의사결정을 지원하는 데 적합합니다.
- 도구 사용 (함수 호출): Gemini 모델은 ADK를 통해 제공된 도구를 사용해야 하는 시점과 방법을 결정하는 데 탁월하여 상담사가 정보를 수집하거나 작업을 실행할 수 있도록 지원합니다.
- 효율성 (플래시 모델): '플래시' 변형은 성능과 비용 효율성의 균형이 잘 잡혀 있어 빠른 응답이 필요한 여러 대화형 상담사 작업에 적합합니다.
- Google의 Gemini 모델: 특히 gemini-2.0-flash와 같은 버전을 사용합니다. 이러한 모델은 다음과 같은 이유로 선택됩니다.
Google Cloud 크레딧이 필요하신가요?
3. 시작하기 전에
👉Google Cloud 콘솔 상단에서 Cloud Shell 활성화 (Cloud Shell 창 상단에 있는 터미널 모양 아이콘)를 클릭합니다.
👉'편집기 열기' 버튼 (연필이 있는 열려 있는 폴더 모양)을 클릭합니다. 이렇게 하면 창에서 Cloud Shell 코드 편집기가 열립니다. 왼쪽에 파일 탐색기가 표시됩니다.
👉그림과 같이 하단 상태 표시줄에서 Cloud Code 로그인 버튼을 클릭합니다. 안내에 따라 플러그인을 승인합니다. 상태 표시줄에 Cloud Code - 프로젝트 없음이 표시되면 이를 선택한 다음 드롭다운에서 'Google Cloud 프로젝트 선택'을 선택하고 내가 만든 프로젝트 목록에서 특정 Google Cloud 프로젝트를 선택합니다.
👉 Google Cloud 프로젝트 ID를 찾습니다.
- Google Cloud 콘솔(https://console.cloud.google.com)을 엽니다.
- 페이지 상단의 프로젝트 드롭다운에서 이 워크샵에 사용할 프로젝트를 선택합니다.
- 프로젝트 ID는 대시보드의 프로젝트 정보 카드에 표시됩니다.
👉클라우드 IDE에서 터미널을 엽니다.
👉💻 터미널에서 다음 명령어를 사용하여 이미 인증되었는지, 프로젝트가 프로젝트 ID로 설정되어 있는지 확인합니다.
gcloud auth list
👉💻 GitHub에서 instavibe-bootstrap
프로젝트를 클론합니다.
git clone -b adk-1.2.1-a2a-0.2.7 https://github.com/weimeilin79/instavibe-bootstrap.git
chmod +x ~/instavibe-bootstrap/init.sh
chmod +x ~/instavibe-bootstrap/set_env.sh
프로젝트 구조 이해
빌드를 시작하기 전에 방금 클론한 instavibe-bootstrap
프로젝트의 레이아웃을 잠시 살펴보겠습니다. 이렇게 하면 워크숍 전반에서 파일을 찾고 수정할 위치를 알 수 있습니다.
instavibe-bootstrap/
├── agents/
│ ├── orchestrate/
│ ├── planner/
│ ├── platform_mcp_client/
│ └── social/
├── instavibe/
│ ├── static/
│ └── templates/
├── tools/
│ └── instavibe/
├── utils/
├── init.sh
└── set_env.sh
주요 디렉터리는 다음과 같습니다.
agents/
: AI 시스템의 핵심입니다. 각 하위 디렉터리 (planner/, social/ 등)에는 특정 지능형 에이전트의 소스 코드가 포함되어 있습니다.agent.py
: 각 상담사 폴더 내에서 상담사 로직이 있는 기본 파일입니다.a2a_server.py
: 이 파일은 ADK 에이전트를 에이전트 간 (A2A) 서버로 래핑합니다.Dockerfile
: Cloud Run 또는 Agent Engine에 에이전트를 배포하기 위한 컨테이너 이미지를 빌드하는 방법을 정의합니다.
instavibe/
: 이 디렉터리에는 InstaVibe 웹 애플리케이션의 전체 소스 코드가 포함되어 있습니다.tools/
: 이 디렉터리는 상담사가 사용할 수 있는 외부 도구를 빌드하기 위한 것입니다.instavibe/
에는 Model Context Protocol (MCP) 서버가 포함되어 있습니다.
이 모듈식 구조는 웹 애플리케이션을 다양한 AI 구성요소와 분리하여 전체 시스템을 더 쉽게 관리, 테스트, 배포할 수 있도록 합니다.
👉💻 초기화 스크립트를 실행합니다.
이 스크립트에서 Google Cloud 프로젝트 ID를 입력하라는 메시지가 표시됩니다.
init.sh
스크립트에서 메시지가 표시되면 마지막 단계에서 찾은 Google Cloud 프로젝트 ID를 입력합니다.
cd ~/instavibe-bootstrap
./init.sh
👉💻 필요한 프로젝트 ID를 설정합니다.
gcloud config set project $(cat ~/project_id.txt) --quiet
👉💻 다음 명령어를 실행하여 필요한 Google Cloud API를 사용 설정합니다.
gcloud services enable run.googleapis.com \
cloudfunctions.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
spanner.googleapis.com \
apikeys.googleapis.com \
iam.googleapis.com \
compute.googleapis.com \
aiplatform.googleapis.com \
cloudresourcemanager.googleapis.com \
maps-backend.googleapis.com
👉💻 필요한 모든 환경 변수를 설정합니다.
export PROJECT_ID=$(gcloud config get project)
export PROJECT_NUMBER=$(gcloud projects describe ${PROJECT_ID} --format="value(projectNumber)")
export SERVICE_ACCOUNT_NAME=$(gcloud compute project-info describe --format="value(defaultServiceAccount)")
export SPANNER_INSTANCE_ID="instavibe-graph-instance"
export SPANNER_DATABASE_ID="graphdb"
export GOOGLE_CLOUD_PROJECT=$(gcloud config get project)
export GOOGLE_GENAI_USE_VERTEXAI=TRUE
export GOOGLE_CLOUD_LOCATION="us-central1"
권한 설정
👉💻 권한을 부여합니다. 터미널에서 다음을 실행합니다.
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/spanner.admin"
# Spanner Database User
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/spanner.databaseUser"
# Artifact Registry Admin
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/artifactregistry.admin"
# Cloud Build Editor
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/cloudbuild.builds.editor"
# Cloud Run Admin
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/run.admin"
# IAM Service Account User
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/iam.serviceAccountUser"
# Vertex AI User
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/aiplatform.user"
# Logging Writer (to allow writing logs)
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/logging.logWriter"
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/logging.viewer"
👉 IAM 콘솔에서 결과를 확인합니다.
👉💻 터미널에서 다음 명령어를 실행하여 Artifact Registry 저장소를 만듭니다. Cloud Run 또는 Agent Engine에 배포하기 전에 에이전트, MCP 서버, InstaVibe 애플리케이션의 모든 Docker 이미지가 여기에 저장됩니다.
export REPO_NAME="introveally-repo"
gcloud artifacts repositories create $REPO_NAME \
--repository-format=docker \
--location=us-central1 \
--description="Docker repository for InstaVibe workshop"
API 키용 지도 플랫폼 설정
InstaVibe 애플리케이션에서 Google 지도 서비스를 사용하려면 API 키를 만들고 적절하게 제한해야 합니다.
👉 새 탭에서 API 및 서비스 > 사용자 인증 정보로 이동합니다. '사용자 인증 정보' 페이지에서 상단의 + 사용자 인증 정보 만들기 버튼을 클릭합니다. 드롭다운 메뉴에서 API 키를 선택합니다.
👉 새로 생성된 API 키가 표시된 대화상자가 나타납니다. 나중에 애플리케이션 구성에 필요합니다.
👉 'API 키 생성 완료' 대화상자에서 닫기를 클릭합니다.
👉 새 API 키가 표시됩니다 (예: 'API 키 1')을 입력합니다. 오른쪽의 햄버거 아이콘을 클릭하고 API 키 수정을 선택하여 'API 키 제한 및 이름 바꾸기' 페이지를 엽니다.
👉 상단의 이름 입력란에서 기본 이름을 Maps Platform API 키로 변경합니다. (🚨🚨중요🚨🚨 이 이름을 사용하세요.)
Maps Platform API Key
👉 '애플리케이션 제한사항' 섹션에서 없음이 선택되어 있는지 확인합니다.
👉 'API 제한사항' 섹션에서 키 제한 라디오 버튼을 선택합니다.
👉 'API 선택' 드롭다운 메뉴를 클릭합니다. 표시되는 검색창에 Maps JavaScript API
를 입력하고 목록에서 선택합니다.
👉 '확인'을 클릭합니다.
👉 페이지 하단의 저장 버튼을 클릭합니다.
이제 'Maps Platform API 키'라는 API 키를 만들고 'Maps JavaScript API'만 사용할 수 있도록 제한했으며 프로젝트에 API가 사용 설정되었는지 확인했습니다.
4. 그래프 데이터베이스 설정
지능형 상담사를 빌드하기 전에 InstaVibe 소셜 네트워크 내의 풍부한 연결을 저장하고 이해할 방법이 필요합니다. 이때 그래프 데이터베이스가 필요합니다. 행과 열의 테이블에 데이터를 저장하는 기존의 관계형 데이터베이스와 달리 그래프 데이터베이스는 노드(예: 사람, 이벤트, 게시물)와 노드를 연결하는 관계(예: 친구 관계, 이벤트 참석, 멘션) 측면에서 데이터를 나타내고 쿼리하도록 설계되었습니다. 이 구조는 실제 소셜 네트워크의 구조를 반영하므로 소셜 미디어 애플리케이션에 매우 강력합니다. 다양한 항목이 어떻게 상호 연결되는지 직관적으로 살펴볼 수 있기 때문입니다.
Google은 Google Cloud Spanner를 사용하여 이 그래프 데이터베이스를 구현하고 있습니다. Spanner는 전 세계적으로 분산된 strong consistency를 갖는 관계형 데이터베이스로 알려져 있지만, 관계형 테이블 위에 직접 그래프 구조를 정의하고 쿼리할 수도 있습니다.
이를 통해 Spanner의 확장성, 트랜잭션 일관성, 익숙한 SQL 인터페이스의 이점과 AI 기반 기능에 중요한 복잡한 소셜 동력을 분석하기 위한 그래프 쿼리의 표현력을 모두 활용할 수 있습니다.
👉💻 Cloud Shell IDE 터미널에서 Google Cloud에서 필요한 인프라를 프로비저닝합니다. 먼저 데이터베이스의 전용 컨테이너 역할을 하는 Spanner 인스턴스를 만듭니다. 인스턴스가 준비되면 InstaVibe의 모든 테이블과 그래프 데이터가 저장될 실제 Spanner 데이터베이스를 인스턴스 내에 만듭니다.
. ~/instavibe-bootstrap/set_env.sh
gcloud spanner instances create $SPANNER_INSTANCE_ID \
--config=regional-us-central1 \
--description="GraphDB Instance InstaVibe" \
--processing-units=100 \
--edition=ENTERPRISE
gcloud spanner databases create $SPANNER_DATABASE_ID \
--instance=$SPANNER_INSTANCE_ID \
--database-dialect=GOOGLE_STANDARD_SQL
👉💻 기본 서비스 계정에 Spanner 읽기/쓰기 액세스 권한 부여
echo "Granting Spanner read/write access to ${SERVICE_ACCOUNT_NAME} for database ${SPANNER_DATABASE_ID}..."
gcloud spanner databases add-iam-policy-binding ${SPANNER_DATABASE_ID} \
--instance=${SPANNER_INSTANCE_ID} \
--member="serviceAccount:${SERVICE_ACCOUNT_NAME}" \
--role="roles/spanner.databaseUser" \
--project=${PROJECT_ID}
👉💻 지금 Python 가상 환경을 설정하고 필요한 Python 패키지를 설치한 다음 Spanner 내에서 그래프 데이터베이스 스키마를 설정하고 초기 데이터로 로드하고 setup.py
스크립트를 실행합니다.
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap
python -m venv env
source env/bin/activate
pip install -r requirements.txt
cd instavibe
python setup.py
👉 새 브라우저 탭에서 Google Cloud 콘솔로 이동하여 Spanner로 이동하면 Spanner 인스턴스 목록이 표시됩니다. instavibe-graph-instance
를 클릭합니다. 👉 인스턴스 개요 페이지에는 해당 인스턴스의 데이터베이스 목록이 표시됩니다.
graphdb
를 클릭합니다.
👉 데이터베이스의 왼쪽 탐색 창에서 Spanner 스튜디오 를 클릭합니다.
👉 쿼리 편집기 (제목 없는 쿼리 탭)에 다음 그래프 SQL 쿼리를 붙여넣습니다. 이 쿼리는 모든 사람 노드와 다른 사람 노드와의 직접적인 친구 관계를 찾습니다. 실행을 클릭하여 결과를 확인합니다.
Graph SocialGraph
MATCH result_paths = ((p:Person)-[f:Friendship]-(friend:Person))
RETURN SAFE_TO_JSON(result_paths) AS result_paths
👉 동일한 쿼리 편집기에서 이전 DDL을 대체하여 동일한 이벤트에 참석한 사용자를 찾습니다. 이는 공유 활동을 통한 간접적인 연결을 의미합니다.
Graph SocialGraph
MATCH result_paths = (p1:Person)-[:Attended]->(e:Event)<-[:Attended]-(p2:Person)
WHERE p1.person_id < p2.person_id
RETURN SAFE_TO_JSON(result_paths) AS result_paths
👉 이 쿼리는 특정 사용자의 친구가 작성한 게시물에 언급된 사용자를 살펴보는 다른 유형의 연결을 탐색합니다. 쿼리 편집기에서 다음 쿼리를 실행합니다.
Graph SocialGraph
MATCH result_paths = (user:Person {name: "Alice"})-[:Friendship]-(friend:Person)-[:Wrote]->(post:Post)-[:Mentioned]->(mentioned_person:Person)
WHERE user <> mentioned_person AND friend <> mentioned_person -- Avoid self-mentions or friend mentioning themselves in their own post if not intended
RETURN SAFE_TO_JSON(result_paths) AS result_paths
이러한 쿼리는 Spanner를 InstaVibe 애플리케이션의 그래프 데이터베이스로 사용하는 이점을 단적으로 보여줍니다. 소셜 데이터를 상호 연결된 그래프로 모델링하면 관계와 활동을 정교하게 분석할 수 있습니다. 이는 AI 상담사가 사용자 맥락을 이해하고 관심분야를 파악하며 궁극적으로 지능형 소셜 계획 지원을 제공하는 데 기본이 됩니다.
이제 기본 데이터 구조를 설정하고 테스트했으므로 상담사가 상호작용할 기존 InstaVibe 애플리케이션으로 이동하겠습니다.
5. InstaVibe의 현재 상태
AI 상담사가 어디에 적합한지 이해하려면 먼저 기존 InstaVibe 웹 애플리케이션을 배포하고 실행해야 합니다. 이 애플리케이션은 이미 설정된 Spanner 그래프 데이터베이스에 연결하는 사용자 인터페이스와 기본 기능을 제공합니다.
InstaVibe 애플리케이션은 Google 지도를 사용하여 이벤트 세부정보 페이지에 이벤트 위치를 시각적으로 표시합니다. 이 기능을 사용 설정하려면 애플리케이션에 앞에서 만든 API 키가 필요합니다. 다음 스크립트는 할당된 표시 이름 ('지도 플랫폼 API 키')을 사용하여 실제 키 문자열을 검색합니다.
👉💻 Cloud Shell IDE로 돌아갑니다. 아래 스크립트를 실행합니다. 그런 다음 출력을 주의 깊게 확인하여 표시된 GOOGLE_MAPS_API_KEY가 이전에 Google Cloud 콘솔에서 만들고 복사한 키와 일치하는지 확인합니다.
. ~/instavibe-bootstrap/set_env.sh
export KEY_DISPLAY_NAME="Maps Platform API Key"
GOOGLE_MAPS_KEY_ID=$(gcloud services api-keys list \
--project="${PROJECT_ID}" \
--filter="displayName='${KEY_DISPLAY_NAME}'" \
--format="value(uid)" \
--limit=1)
GOOGLE_MAPS_API_KEY=$(gcloud services api-keys get-key-string "${GOOGLE_MAPS_KEY_ID}" \
--project="${PROJECT_ID}" \
--format="value(keyString)")
echo "${GOOGLE_MAPS_API_KEY}" > ~/mapkey.txt
echo "Retrieved GOOGLE_MAPS_API_KEY: ${GOOGLE_MAPS_API_KEY}"
👉💻 이제 InstaVibe 웹 애플리케이션의 컨테이너 이미지를 빌드하고 Artifact Registry 저장소에 푸시하겠습니다.
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/instavibe/
export IMAGE_TAG="latest"
export APP_FOLDER_NAME="instavibe"
export IMAGE_NAME="instavibe-webapp"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="instavibe"
gcloud builds submit . \
--tag=${IMAGE_PATH} \
--project=${PROJECT_ID}
👉💻 새 빌드 InstaVibe 웹 앱 이미지를 Cloud Run에 배포합니다.
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/instavibe/
export IMAGE_TAG="latest"
export APP_FOLDER_NAME="instavibe"
export IMAGE_NAME="instavibe-webapp"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="instavibe"
gcloud run deploy ${SERVICE_NAME} \
--image=${IMAGE_PATH} \
--platform=managed \
--region=${REGION} \
--allow-unauthenticated \
--set-env-vars="SPANNER_INSTANCE_ID=${SPANNER_INSTANCE_ID}" \
--set-env-vars="SPANNER_DATABASE_ID=${SPANNER_DATABASE_ID}" \
--set-env-vars="APP_HOST=0.0.0.0" \
--set-env-vars="APP_PORT=8080" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}" \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--set-env-vars="GOOGLE_MAPS_API_KEY=${GOOGLE_MAPS_API_KEY}" \
--project=${PROJECT_ID} \
--min-instances=1
배포가 완료되면 Cloud Run 로그에 실행 중인 InstaVibe 애플리케이션의 공개 URL이 표시됩니다.
Google Cloud 콘솔의 Cloud Run 섹션으로 이동하여 instavibe 서비스를 선택해도 이 URL을 찾을 수 있습니다.
이제 웹브라우저에서 해당 URL을 열어 기본 InstaVibe 플랫폼을 살펴보세요. 설정된 그래프 데이터베이스를 기반으로 하는 게시물, 이벤트, 사용자 연결을 확인합니다.
이제 타겟 애플리케이션이 실행되고 있으므로 기능을 향상할 첫 번째 지능형 에이전트를 빌드해 보겠습니다.
6. ADK를 사용한 기본 상담사,이벤트 플래너
ADK 프레임워크
Google ADK 프레임워크 소개 이제 기반 (InstaVibe 앱 및 데이터베이스)이 마련되었으므로 Google의 에이전트 개발 키트 (ADK)를 사용하여 첫 번째 지능형 에이전트를 빌드할 수 있습니다.
에이전트 개발 키트 (ADK)는 AI 에이전트를 개발하고 배포하기 위해 특별히 설계된 유연한 모듈식 프레임워크입니다. 설계 원칙은 에이전트 개발을 기존 소프트웨어 개발과 더 유사하게 느끼게 하는 것입니다. 이를 통해 개발자가 단순한 단일 목적 작업에서 복잡한 멀티 에이전트 워크플로에 이르기까지 모든 작업을 처리할 수 있는 에이전트 아키텍처를 더 쉽게 만들고, 배포하고, 조정할 수 있도록 하는 것을 목표로 합니다.
ADK의 핵심은 안내, 구성(예: 선택한 언어 모델(예:Agent
Gemini) 및 작업을 실행하거나 정보를 수집하는 데 사용할 수 있는 Tools
집합입니다.
첫 번째 상담사는 '이벤트 플래너'입니다. 이 모델의 핵심 목적은 사용자의 외출 요청 (위치, 날짜, 관심분야 지정)을 받아 창의적이고 맞춤화된 추천을 생성하는 것입니다. 추천이 관련성 있고 최신 정보 (예: 주말에 열리는 특정 이벤트)를 기반으로 하기 위해 ADK의 기본 제공 도구인 Google 검색을 활용합니다. 이를 통해 상담사는 실시간 웹 결과를 기반으로 응답을 제공하고 사용자의 기준에 맞는 장소, 이벤트, 활동에 관한 최신 세부정보를 가져올 수 있습니다.
👉📝 Cloud Shell IDE로 돌아가 ~/instavibe-bootstrap/agents/planner/agent.py
에 다음 프롬프트와 안내를 추가하여 에이전트를 만듭니다.
from google.adk.agents import Agent
from google.adk.tools import google_search
root_agent = Agent(
name="planner_agent",
model="gemini-2.0-flash",
description="Agent tasked with generating creative and fun dating plan suggestions",
instruction="""
You are a specialized AI assistant tasked with generating creative and fun plan suggestions.
Request:
For the upcoming weekend, specifically from **[START_DATE_YYYY-MM-DD]** to **[END_DATE_YYYY-MM-DD]**, in the location specified as **[TARGET_LOCATION_NAME_OR_CITY_STATE]** (if latitude/longitude are provided, use these: Lat: **[TARGET_LATITUDE]**, Lon: **[TARGET_LONGITUDE]**), please generate a distinct dating plan suggestions.
Constraints and Guidelines for Suggestions:
1. Creativity & Fun: Plans should be engaging, memorable, and offer a good experience for a date.
2. Budget: All generated plans should aim for a moderate budget (conceptually "$$"), meaning they should be affordable yet offer good value, without being overly cheap or extravagant. This budget level should be *reflected in the choice of activities and venues*, but **do not** explicitly state "Budget: $$" in the `plan_description`.
3. Interest Alignment:
Consider the following user interests: **[COMMA_SEPARATED_LIST_OF_INTERESTS, e.g., outdoors, arts & culture, foodie, nightlife, unique local events, live music, active/sports]**. Tailor suggestions specifically to these where possible. The plan should *embody* these interests.
Fallback: If specific events or venues perfectly matching all listed user interests cannot be found for the specified weekend, you should create a creative and fun generic dating plan that is still appealing, suitable for the location, and adheres to the moderate budget. This plan should still sound exciting and fun, even if it's more general.
4. Current & Specific: Prioritize finding specific, current events, festivals, pop-ups, or unique local venues operating or happening during the specified weekend dates. If exact current events cannot be found, suggest appealing evergreen options or implement the fallback generic plan.
5. Location Details: For each place or event mentioned within a plan, you MUST provide its name, precise latitude, precise longitude, and a brief, helpful description.
6. Maximum Activities: The plan must contain a maximum of 3 distinct activities.
RETURN PLAN in MARKDOWN FORMAT
""",
tools=[google_search]
)
이제 첫 번째 에이전트가 정의되었습니다. ADK의 장점 중 하나는 직관적인 특성과 제공하는 편리한 도구입니다. 특히 유용한 도구는 ADK Dev UI로, 상담사를 대화식으로 테스트하고 실시간으로 응답을 확인할 수 있습니다.
👉💻 시작해 보겠습니다. 다음 명령어를 실행하면 ADK DEV UI가 실행됩니다.
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents
sed -i "s|^\(O\?GOOGLE_CLOUD_PROJECT\)=.*|GOOGLE_CLOUD_PROJECT=${PROJECT_ID}|" ~/instavibe-bootstrap/agents/planner/.env
adk web
명령어를 실행하면 터미널에 ADK 웹 서버가 시작되었음을 나타내는 다음과 유사한 출력이 표시됩니다.
+-----------------------------------------------------------------------------+
| ADK Web Server started |
| |
| For local testing, access at http://localhost:8000. |
+-----------------------------------------------------------------------------+
INFO: Application startup complete.
INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
👉 다음으로 브라우저에서 ADK Dev UI에 액세스합니다.
Cloud Shell 툴바 (일반적으로 오른쪽 상단)의 웹 미리보기 아이콘 (눈 또는 화살표가 있는 정사각형처럼 보임)에서 포트 변경을 선택합니다. 팝업 창에서 포트를 8000으로 설정하고 '변경 및 미리보기'를 클릭합니다. 그러면 Cloud Shell에서 ADK Dev UI가 표시된 새 브라우저 탭 또는 창이 열립니다.
브라우저에서 ADK Dev UI가 열리면 UI의 오른쪽 상단 드롭다운 메뉴에서 상호작용할 상담사로 planner를 선택합니다. 이제 오른쪽의 채팅 대화상자에서 상담사에게 할 일을 지정하세요. 예를 들어 다음과 같이 상담사와 대화해 보세요.
Search and plan something in Seattle for me this weekend
This weekend and I enjoy food and anime
날짜 제안 (선호사항)
July 12 2025
상담사가 요청을 처리하고 Google 검색 결과를 기반으로 계획을 제공하는 것을 확인할 수 있습니다.
에이전트와 상호작용하는 것은 좋지만, 특히 변경사항이 적용될 때 에이전트가 예상대로 일관되게 작동하는지 어떻게 알 수 있을까요?
기존의 소프트웨어 테스트 방법은 생성형이며 비결정론적 특성 때문에 AI 에이전트에 적합하지 않은 경우가 많습니다. 멋진 데모에서 안정적인 프로덕션 상담사로의 전환을 위해서는 확실한 평가 전략이 중요합니다. 생성형 모델의 최종 출력을 단순히 확인하는 것과 달리 에이전트를 평가하려면 종종 에이전트의 의사결정 프로세스와 다양한 시나리오에서 도구를 올바르게 사용하거나 안내를 따르는 능력을 평가해야 합니다. ADK는 이를 지원하는 기능을 제공합니다.
👉 ADK Dev UI의 왼쪽 탐색 메뉴에서 'Eval'(평가) 탭을 클릭합니다. 미리 로드된 plan_eval
라는 테스트 파일이 표시됩니다. 이 파일에는 계획자 에이전트를 테스트하기 위한 사전 정의된 입력과 기준이 포함되어 있습니다.
👉 'boston'과 같은 시나리오를 선택하고 평가 실행 버튼을 클릭합니다. 팝업 창이 표시되면 일치 점수를 0.3으로 낮추고 '시작'을 클릭합니다.
이렇게 하면 테스트 입력으로 에이전트가 실행되고 출력이 정의된 기대치를 충족하는지 확인됩니다. 이를 통해 상담사의 실적을 체계적으로 테스트할 수 있습니다.
👉 이제 더 엄격한 기준을 적용하면 어떻게 되는지 살펴보겠습니다. 'nyc' 시나리오를 선택하고 Run Evaluation을 다시 클릭합니다. 이번에는 일치 점수를 기본값 (응답 일치 점수: 0.7)으로 두고 '시작'을 클릭합니다. 결과가 실패로 표시됩니다. 이는 상담사의 광고 소재 출력이 사전 정의된 '골드' 답변과 완전히 일치하지 않기 때문에 예상된 결과입니다.
👉 실패 이유를 알아보려면 'nyc' 행에서 실패 아이콘을 클릭합니다. 이제 UI에 상담사의 실제 응답과 테스트 사례의 예상 응답이 나란히 표시됩니다. 이 뷰는 디버깅에 필수적이며, 이를 통해 에이전트의 출력이 어디에서 달라졌는지 정확하게 확인하고 그에 따라 안내를 미세 조정할 수 있습니다.
UI 및 평가를 살펴본 후 Cloud Shell 편집기 터미널로 돌아가 Ctrl+C
키를 눌러 ADK Dev UI를 중지합니다.
자유 형식 텍스트 출력은 시작하기에는 좋지만 InstaVibe와 같은 애플리케이션에서 상담사의 추천을 쉽게 사용하려면 구조화된 데이터 (예: JSON)가 훨씬 더 실용적입니다. 일관된 JSON 형식으로 계획을 반환하도록 상담사를 수정하겠습니다.
👉📝 ~/instavibe-bootstrap/agents/planner/agent.py
에서 상담사의 안내 문자열 내에 현재 RETURN PLAN in MARKDOWN FORMAT
라고 표시된 줄을 찾습니다. 이 줄을 다음과 같은 상세 JSON 구조로 바꿉니다.
Return your response *exclusively* as a single JSON object. This object should contain a top-level key, "fun_plans", which holds a plan objects. Each plan object in the list must strictly adhere to the following structure:
--json--
{
"plan_description": "A summary of the overall plan, consisting of **exactly three sentences**. Craft these sentences in a friendly, enthusiastic, and conversational tone, as if you're suggesting this awesome idea to a close friend. Make it sound exciting and personal, highlighting the positive aspects and appeal of the plan without explicitly mentioning budget or listing interest categories.",
"locations_and_activities": [
{
"name": "Name of the specific place or event",
"latitude": 0.000000, // Replace with actual latitude
"longitude": 0.000000, // Replace with actual longitude
"description": "A brief description of this place/event, why it's suitable for the date, and any specific details for the weekend (e.g., opening hours, event time)."
}
// Add more location/activity objects here if the plan involves multiple stops/parts
]
}
이제 JSON 출력을 구체적으로 요청하도록 상담사의 안내를 업데이트했으므로 변경사항을 확인해 보겠습니다.
👉💻 이전과 동일한 명령어를 사용하여 ADK Dev UI를 다시 실행합니다.
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents
adk web
탭이 이미 열려 있는 경우 탭을 새로고침합니다. 또는 이전과 동일한 단계에 따라 브라우저에서 ADK Dev UI를 엽니다 (포트 8000의 Cloud Shell 웹 미리보기 사용). UI가 로드되면 계획자 에이전트가 선택되어 있는지 확인합니다.
👉 이번에는 다른 요청을 해 보겠습니다. 채팅 대화상자에 다음을 입력합니다.
Plan an event Boston this weekend with art and coffee
상담사의 응답을 주의 깊게 살펴봅니다. 이제 순수한 대화형 텍스트 답장 대신 안내에 정의된 구조 (fun_plans, plan_description, locations_and_activities 등 포함)와 일치하는 JSON 객체 형식의 응답이 표시됩니다. 이제 에이전트가 InstaVibe 애플리케이션의 프로그래매틱 사용에 적합한 구조화된 출력을 생성할 수 있습니다.
JSON 출력을 확인한 후 Cloud Shell 터미널로 돌아가 Ctrl+C
키를 눌러 ADK Dev UI를 중지합니다.
ADK 구성요소
ADK Dev UI는 대화형 테스트에 적합하지만, 더 큰 애플리케이션이나 백엔드 서비스의 일부로 프로그래매틱 방식으로 에이전트를 실행해야 하는 경우가 많습니다. 작동 방식을 이해하려면 런타임 및 컨텍스트 관리와 관련된 몇 가지 핵심 ADK 개념을 살펴보겠습니다.
의미 있는 멀티턴 대화를 하려면 상담사가 컨텍스트를 이해해야 합니다. 즉, 연속성을 유지하기 위해 이전에 말한 내용과 한 일을 기억해야 합니다. ADK는 세션, 상태, 메모리를 통해 이 컨텍스트를 관리하는 구조화된 방법을 제공합니다.
- 세션: 사용자가 상담사와 상호작용을 시작하면 세션이 생성됩니다. 특정 단일 채팅 대화목록의 컨테이너라고 생각하면 됩니다. 고유 ID, 상호작용 기록 (이벤트), 현재 작업 데이터 (상태), 마지막 업데이트 시간과 같은 메타데이터를 보유합니다.
- 상태: 단일 세션 내에서 상담사의 단기 작업 메모리입니다. 에이전트가 현재 태스크를 완료하는 데 필요한 임시 정보를 저장할 수 있는 변경 가능한 사전입니다 (예: 지금까지 수집된 사용자 환경설정, 도구 호출의 중간 결과).
- 메모리: 이는 상담사가 여러 세션에서 장기적으로 기억하거나 외부 지식 베이스에 액세스할 수 있는 잠재력을 나타냅니다. Session과 State는 즉각적인 대화를 처리하는 반면, Memory (대부분 MemoryService에서 관리)를 사용하면 에이전트가 이전 상호작용이나 구조화된 데이터 소스에서 정보를 검색하여 더 광범위한 지식 맥락을 제공할 수 있습니다. 참고: 간단한 클라이언트는 편의상 메모리 서비스를 사용합니다. 즉, 메모리/상태는 스크립트가 실행되는 동안만 유지됩니다.
- 이벤트: 세션 내의 모든 상호작용 (사용자 메시지, 상담사 응답, 도구 사용 요청, 도구 결과, 상태 변경, 오류)은 변경 불가능한 이벤트로 기록됩니다. 이렇게 하면 대화의 스크립트 및 작업 기록을 중심으로 시간순으로 정리된 로그가 생성됩니다.
그러면 에이전트가 실행될 때 이러한 작업은 어떻게 관리되나요? 이는 실행기의 역할입니다.
- Runner: Runner는 ADK에서 제공하는 핵심 실행 엔진입니다. 개발자가 에이전트와 에이전트에서 사용하는 도구를 정의하면 Runner가 사용자의 요청을 처리하는 프로세스를 조정합니다. 세션을 관리하고, 이벤트 흐름을 처리하고, 상태를 업데이트하고, 기본 언어 모델을 호출하고, 도구 호출을 조정하며, MemoryService와 상호작용할 수 있습니다. 마치 모든 부분이 올바르게 작동하도록 하는 지휘자와 같다고 생각하면 됩니다.
Runner를 사용하여 에이전트를 Dev UI와 완전히 독립된 독립형 Python 애플리케이션으로 실행할 수 있습니다.
간단한 클라이언트 스크립트를 만들어 계획자 에이전트를 프로그래매틱 방식으로 호출해 보겠습니다.
👉📝 ~/instavibe-bootstrap/agents/planner/planner_client.py
파일에서 기존 가져오기 아래에 다음 Python 코드를 추가합니다. planner_client.py
의 가져오기 아래에 다음을 추가합니다.
async def async_main():
session_service = InMemorySessionService()
session = await session_service.create_session(
state={}, app_name='planner_app', user_id='user_dc'
)
query = "Plan Something for me in San Francisco this weekend on wine and fashion "
print(f"User Query: '{query}'")
content = types.Content(role='user', parts=[types.Part(text=query)])
root_agent = agent.root_agent
runner = Runner(
app_name='planner_app',
agent=root_agent,
session_service=session_service,
)
print("Running agent...")
events_async = runner.run_async(
session_id=session.id, user_id=session.user_id, new_message=content
)
async for event in events_async:
print(f"Event received: {event}")
if __name__ == '__main__':
try:
asyncio.run(async_main())
except Exception as e:
print(f"An error occurred: {e}")
이 코드는 세션 및 아티팩트 관리를 위한 메모리 내 서비스를 설정하고 (이 예에서는 간단하게 유지), 세션을 만들고, 사용자 쿼리를 정의하고, 에이전트로 실행자를 구성한 다음 에이전트를 비동기식으로 실행하여 실행 중에 생성된 각 이벤트를 출력합니다.
👉💻 이제 터미널에서 다음 클라이언트 스크립트를 실행합니다.
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents
python -m planner.planner_client
👀 출력을 확인합니다. 최종 JSON 계획만 표시되는 것이 아니라 에이전트의 실행 흐름 중에 생성된 각 이벤트 객체의 세부 구조가 표시됩니다. 여기에는 초기 사용자 메시지 이벤트, 도구 호출과 관련된 잠재적 이벤트 (예: Google 검색), 마지막으로 JSON 계획이 포함된 모델의 응답 이벤트가 포함됩니다. 이 상세 이벤트 스트림은 ADK 런타임 내에서 진행되는 단계별 처리를 디버그하고 이해하는 데 매우 유용합니다.
Running agent...
Event received: content=Content(parts=[Part(video_metadata=None, thought=None, code_execution_result=None, executable_code=None, file_data=None, function_call=None, function_response=None, inline_data=None, text='```json\n{\n "fun_plans": [\n {\n "plan_description": "Embark on a stylish adventure through Hayes Valley,
...(turncated)
, offering a variety of fashion styles to browse and enjoy."\n }\n ]\n }\n ]\n}\n```')], role='model') grounding_metadata=GroundingMetadata(grounding_chunks=[GroundingChunk(retrieved_context=None, web=GroundingChunkWeb(domain='islands.com', title='islands.com', uri='http
...(turncated)
QyTpPV7jS6wUt-Ix7GuP2mC9J4eY_8Km6Vv44liF9cb2VSs='))], grounding_supports=[GroundingSupport(confide
...(turncated)
>\n', sdk_blob=None), web_search_queries=['..e']) partial=None turn_complete=None error_code=None error_message=None interrupted=None custom_metadata=None invocation_id='e-04d97b8b-9021-47a5-ab41-17b5cbb4bf03' author='location_search_agent' actions=EventActions(skip_summarization=None, state_delta={}, artifact_delta={}, transfer_to_agent=None, escalate=None, requested_auth_configs={}) long_running_tool_ids=None branch=None id='CInHdkKw' timestamp=1746978846.232674
스크립트가 계속 실행되거나 중단되는 경우 Ctrl+C
를 눌러 수동으로 중지해야 할 수 있습니다.
7. 플랫폼 상호작용 에이전트 - MCP 서버와 상호작용
ADK는 상담사를 구성하는 데 도움이 되지만, 실제 작업을 실행하려면 외부 시스템 또는 API와 상호작용해야 하는 경우가 많습니다.
Model Context Protocol (MCP)
Model Context Protocol (MCP)은 에이전트와 같은 AI 애플리케이션이 외부 데이터 소스, 도구, 시스템에 연결되는 방식을 표준화하도록 설계된 개방형 표준입니다. 이 도구는 범용 인터페이스를 제공하여 모든 AI 애플리케이션과 데이터 소스 조합에 맞게 맞춤 통합을 해야 하는 문제를 해결하는 것을 목표로 합니다. MCP는 AI 애플리케이션 (호스트) 내에 있는 MCP 클라이언트가 MCP 서버에 대한 연결을 관리하는 클라이언트-서버 아키텍처를 활용합니다. 이러한 서버는 로컬 데이터에 액세스하거나, API를 통해 원격 서비스와 상호작용하거나, 사전 정의된 프롬프트를 제공하는 등 특정 기능을 노출하는 외부 프로그램으로, AI 모델이 현재 정보에 액세스하고 초기 학습 외의 작업을 실행할 수 있도록 합니다. 이 구조를 사용하면 AI 모델이 표준화된 방식으로 외부 기능을 검색하고 상호작용할 수 있으므로 통합이 더 간단해지고 확장성이 향상됩니다.
InstaVibe MCP 서버 빌드 및 배포
상담사는 결국 InstaVibe 플랫폼 자체와 상호작용해야 합니다.특히 플랫폼의 기존 API를 사용하여 게시물을 만들고 이벤트를 등록해야 합니다. InstaVibe 애플리케이션은 이미 표준 HTTP 엔드포인트를 통해 다음과 같은 기능을 노출합니다.
Enpoint | URL | HTTP 메서드 | 설명 |
게시물 만들기 | api/posts | POST | 새 게시물을 추가하는 API 엔드포인트 JSON 본문이 예상됩니다. |
일정 만들기 | api/events | POST | 새 이벤트 및 참석자를 추가하는 API 엔드포인트 (간소화된 스키마) |
MCP를 통해 상담사가 이러한 기능을 사용할 수 있도록 하려면 먼저 이러한 API 호출을 둘러싸는 래퍼 역할을 하는 간단한 Python 함수를 만들어야 합니다. 이러한 함수는 HTTP 요청 로직을 처리합니다.
👉 먼저 게시물을 만드는 래퍼 함수를 구현해 보겠습니다. ~/instavibe-bootstrap/tools/instavibe/instavibe.py
파일을 열고 #REPLACE ME CREATE POST
주석을 다음 Python 코드로 바꿉니다.
def create_post(author_name: str, text: str, sentiment: str, base_url: str = BASE_URL):
"""
Sends a POST request to the /posts endpoint to create a new post.
Args:
author_name (str): The name of the post's author.
text (str): The content of the post.
sentiment (str): The sentiment associated with the post (e.g., 'positive', 'negative', 'neutral').
base_url (str, optional): The base URL of the API. Defaults to BASE_URL.
Returns:
dict: The JSON response from the API if the request is successful.
Returns None if an error occurs.
Raises:
requests.exceptions.RequestException: If there's an issue with the network request (e.g., connection error, timeout).
"""
url = f"{base_url}/posts"
headers = {"Content-Type": "application/json"}
payload = {
"author_name": author_name,
"text": text,
"sentiment": sentiment
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
print(f"Successfully created post. Status Code: {response.status_code}")
return response.json()
except requests.exceptions.RequestException as e:
print(f"Error creating post: {e}")
# Optionally re-raise the exception if the caller needs to handle it
# raise e
return None
except json.JSONDecodeError:
print(f"Error decoding JSON response from {url}. Response text: {response.text}")
return None
👉📝 다음으로 이벤트 생성 API의 래퍼 함수를 만듭니다. 동일한 ~/instavibe-bootstrap/tools/instavibe/instavibe.py
파일에서 #REPLACE ME CREATE EVENTS
주석을 다음 코드로 바꿉니다.
def create_event(event_name: str, description: str, event_date: str, locations: list, attendee_names: list[str], base_url: str = BASE_URL):
"""
Sends a POST request to the /events endpoint to create a new event registration.
Args:
event_name (str): The name of the event.
description (str): The detailed description of the event.
event_date (str): The date and time of the event (ISO 8601 format recommended, e.g., "2025-06-10T09:00:00Z").
locations (list): A list of location dictionaries. Each dictionary should contain:
'name' (str), 'description' (str, optional),
'latitude' (float), 'longitude' (float),
'address' (str, optional).
attendee_names (list[str]): A list of names of the people attending the event.
base_url (str, optional): The base URL of the API. Defaults to BASE_URL.
Returns:
dict: The JSON response from the API if the request is successful.
Returns None if an error occurs.
Raises:
requests.exceptions.RequestException: If there's an issue with the network request (e.g., connection error, timeout).
"""
url = f"{base_url}/events"
headers = {"Content-Type": "application/json"}
payload = {
"event_name": event_name,
"description": description,
"event_date": event_date,
"locations": locations,
"attendee_names": attendee_names,
}
try:
response = requests.post(url, headers=headers, json=payload)
response.raise_for_status() # Raise an exception for bad status codes (4xx or 5xx)
print(f"Successfully created event registration. Status Code: {response.status_code}")
return response.json()
except requests.exceptions.RequestException as e:
print(f"Error creating event registration: {e}")
# Optionally re-raise the exception if the caller needs to handle it
# raise e
return None
except json.JSONDecodeError:
print(f"Error decoding JSON response from {url}. Response text: {response.text}")
return None
보시다시피 이러한 함수는 기존 InstaVibe API를 둘러싼 간단한 래퍼입니다. 이 패턴은 서비스용 API가 이미 있는 경우 유용합니다. 이러한 래퍼를 만들어 에이전트용 도구로 기능을 쉽게 노출할 수 있습니다.
MCP 서버 구현
이제 작업을 실행하는 Python 함수 (InstaVibe API 호출)가 있으므로 MCP 서버 구성요소를 빌드해야 합니다. 이 서버는 MCP 표준에 따라 이러한 함수를 '도구'로 노출하므로 MCP 클라이언트 (예: 상담사)가 이를 검색하고 호출할 수 있습니다.
MCP 서버는 일반적으로 다음 두 가지 주요 기능을 구현합니다.
- list_tools: 클라이언트가 서버에서 사용 가능한 도구를 검색할 수 있도록 허용하고 이름, 설명, 필수 매개변수와 같은 메타데이터를 제공합니다(JSON 스키마를 사용하여 정의되는 경우가 많음).
- call_tool: 클라이언트가 요청한 특정 도구의 실행을 처리하여 도구의 이름과 인수를 수신하고 상응하는 작업을 실행합니다(예: 이 경우 API와 상호작용).
MCP 서버는 AI 모델에 실제 데이터 및 작업에 대한 액세스 권한을 제공하는 데 사용되므로 이메일 전송, 프로젝트 관리 시스템에서 작업 만들기, 데이터베이스 검색, 다양한 소프트웨어 및 웹 서비스와의 상호작용과 같은 작업을 할 수 있습니다. 초기 구현은 특히 개발 또는 '스튜디오' 환경에서 간편성을 위해 표준 입력/출력 (stdio)을 통해 통신하는 로컬 서버에 중점을 두었지만, 서버 전송 이벤트 (SSE)가 포함된 HTTP와 같은 프로토콜을 활용하는 원격 서버로 전환하는 것이 더 광범위한 채택과 엔터프라이즈 사용 사례에 더 적합합니다.
원격 아키텍처는 추가된 네트워크 통신 레이어에도 불구하고 상당한 이점을 제공합니다. 여러 AI 클라이언트가 단일 서버에 대한 액세스를 공유할 수 있고, 도구의 관리 및 업데이트를 중앙 집중화할 수 있으며, 민감한 데이터와 API 키를 여러 클라이언트 머신에 분산하지 않고 서버 측에 보관하여 보안을 강화할 수 있습니다. 또한 AI 모델을 외부 시스템 통합의 세부정보에서 분리하여 모든 AI 인스턴스가 자체 직접 통합을 관리해야 하는 것보다 전체 생태계를 더 확장 가능하고 안전하며 유지보수 가능하게 만듭니다.
Google은 HTTP 및 서버 전송 이벤트 (SSE)를 사용하여 MCP 서버를 구현할 예정입니다. 이는 장기 실행 도구 실행 및 엔터프라이즈 시나리오에 적합합니다.
👉📝 먼저 list_tools 엔드포인트를 구현해 보겠습니다. ~/instavibe-bootstrap/tools/instavibe/mcp_server.py
파일을 열고 #REPLACE ME - LIST TOOLS
주석을 다음 코드로 바꿉니다. :
@app.list_tools()
async def list_tools() -> list[mcp_types.Tool]:
"""MCP handler to list available tools."""
# Convert the ADK tool's definition to MCP format
mcp_tool_schema_event = adk_to_mcp_tool_type(event_tool)
mcp_tool_schema_post = adk_to_mcp_tool_type(post_tool)
print(f"MCP Server: Received list_tools request. \n MCP Server: Advertising tool: {mcp_tool_schema_event.name} and {mcp_tool_schema_post}")
return [mcp_tool_schema_event,mcp_tool_schema_post]
이 함수는 도구 (create_event, create_post)를 정의하고 연결되는 클라이언트에게 이를 알립니다.
👉📝 그런 다음 클라이언트의 실제 실행 요청을 처리하는 call_tool
엔드포인트를 구현합니다. 동일한 ~/instavibe-bootstrap/tools/instavibe/mcp_server.py
파일에서 #REPLACE ME - CALL TOOLS
주석을 다음 코드로 바꿉니다.
@app.call_tool()
async def call_tool(
name: str, arguments: dict
) -> list[mcp_types.TextContent | mcp_types.ImageContent | mcp_types.EmbeddedResource]:
"""MCP handler to execute a tool call."""
print(f"MCP Server: Received call_tool request for '{name}' with args: {arguments}")
# Look up the tool by name in our dictionary
tool_to_call = available_tools.get(name)
if tool_to_call:
try:
adk_response = await tool_to_call.run_async(
args=arguments,
tool_context=None, # No ADK context available here
)
print(f"MCP Server: ADK tool '{name}' executed successfully.")
response_text = json.dumps(adk_response, indent=2)
return [mcp_types.TextContent(type="text", text=response_text)]
except Exception as e:
print(f"MCP Server: Error executing ADK tool '{name}': {e}")
# Creating a proper MCP error response might be more robust
error_text = json.dumps({"error": f"Failed to execute tool '{name}': {str(e)}"})
return [mcp_types.TextContent(type="text", text=error_text)]
else:
# Handle calls to unknown tools
print(f"MCP Server: Tool '{name}' not found.")
error_text = json.dumps({"error": f"Tool '{name}' not implemented."})
return [mcp_types.TextContent(type="text", text=error_text)]
이 함수는 도구 이름과 인수를 수신하고, 앞에서 정의한 상응하는 Python 래퍼 함수를 찾아 실행하고 결과를 반환합니다.
👉💻 MCP 서버 로직이 정의되었으므로 이제 컨테이너로 패키징해야 합니다. 터미널에서 다음 스크립트를 실행하여 Cloud Build를 사용하여 Docker 이미지를 빌드합니다.
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/tools/instavibe
export IMAGE_TAG="latest"
export MCP_IMAGE_NAME="mcp-tool-server"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${MCP_IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="mcp-tool-server"
export INSTAVIBE_BASE_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep instavibe)/api
gcloud builds submit . \
--tag=${IMAGE_PATH} \
--project=${PROJECT_ID}
👉💻 그리고 Google Cloud Run에 이미지를 서비스로 배포합니다.
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/tools/instavibe
export IMAGE_TAG="latest"
export MCP_IMAGE_NAME="mcp-tool-server"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${MCP_IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="mcp-tool-server"
export INSTAVIBE_BASE_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep instavibe)/api
gcloud run deploy ${SERVICE_NAME} \
--image=${IMAGE_PATH} \
--platform=managed \
--region=${REGION} \
--allow-unauthenticated \
--set-env-vars="INSTAVIBE_BASE_URL=${INSTAVIBE_BASE_URL}" \
--set-env-vars="APP_HOST=0.0.0.0" \
--set-env-vars="APP_PORT=8080" \
--set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=TRUE" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}" \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--project=${PROJECT_ID} \
--min-instances=1
👉💻 배포가 완료되면 MCP 서버가 실행되고 공개 URL을 통해 액세스할 수 있습니다. MCP 클라이언트 역할을 하는 상담사가 연결할 위치를 알 수 있도록 이 URL을 캡처해야 합니다.
export MCP_SERVER_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep mcp-tool-server)/sse
이제 Google Cloud 콘솔의 Cloud Run 섹션에 mcp-tool-server 서비스가 '실행 중'으로 표시됩니다.
MCP 서버가 배포되고 URL이 캡처되었으므로 이제 MCP 클라이언트 역할을 하고 이 서버에서 노출된 도구를 활용할 에이전트를 구현할 수 있습니다.
8. 플랫폼 상호작용 에이전트 (MCP 사용)
MCP 클라이언트 MCP 클라이언트는 AI 애플리케이션 또는 상담사 내에 있는 구성요소로, AI 모델과 하나 이상의 MCP 서버 간의 인터페이스 역할을 합니다. Google의 구현에서는 이 클라이언트가 상담사 내에 직접 통합됩니다. 이 클라이언트의 기본 기능은 MCP 서버와 통신하여 list_tools
함수를 통해 사용 가능한 도구를 찾고, 이후 call_tool
함수를 사용하여 특정 도구의 실행을 요청하여 AI 모델 또는 호출을 조정하는 상담사가 제공한 필수 인수를 전달하는 것입니다.
이제 MCP 클라이언트 역할을 하는 에이전트를 빌드합니다. ADK 프레임워크 내에서 실행되는 이 에이전트는 방금 배포한 mcp-tool-server
와 통신합니다.
👉 먼저 실행 중인 MCP 서버에서 도구를 동적으로 가져오도록 에이전트 정의를 수정해야 합니다. agents/platform_mcp_client/agent.py
에서 #REPLACE ME - FETCH TOOLS
를 다음으로 바꿉니다.
"""Gets tools from the File System MCP Server."""
tools = MCPToolset(
connection_params=SseServerParams(url=MCP_SERVER_URL, headers={})
)
이 코드는 MCPToolset.from_server 메서드를 사용하여 MCP_SERVER_URL (이전에 환경 변수로 설정)에 연결하고 사용 가능한 도구 목록을 가져옵니다.
그런 다음 ADK 에이전트 정의에 이러한 동적으로 가져온 도구를 실제로 사용하도록 지시해야 합니다.
👉 agents/platform_mcp_client/agent.py
에서 #REPLACE ME - SET TOOLs
를 다음으로 바꿉니다.
tools=[tools],
👉💻 이제 ADK Dev UI를 사용하여 이 에이전트를 로컬에서 테스트하여 MCP 서버에 올바르게 연결되고 도구를 사용하여 실행 중인 InstaVibe 애플리케이션과 상호작용할 수 있는지 확인해 보겠습니다.
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
export MCP_SERVER_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep mcp-tool-server)/sse
cd ~/instavibe-bootstrap/agents
sed -i "s|^\(O\?GOOGLE_CLOUD_PROJECT\)=.*|GOOGLE_CLOUD_PROJECT=${PROJECT_ID}|" ~/instavibe-bootstrap/agents/platform_mcp_client/.env
sed -i "s|^\(O\?MCP_SERVER_URL\)=.*|MCP_SERVER_URL=${MCP_SERVER_URL}|" ~/instavibe-bootstrap/agents/platform_mcp_client/.env
adk web
브라우저에서 ADK Dev UI를 다시 엽니다 (포트 8000에서 Cloud Shell의 웹 미리보기 사용). 이번에는 오른쪽 상단 드롭다운에서 platform_mcp_client
상담사를 선택합니다.
create_post 도구를 테스트해 보겠습니다. 채팅 대화상자에 다음 요청을 입력합니다.
Create a post saying "Y'all I just got the cutest lil void baby 😭✨ Naming him Abyss bc he's deep, mysterious, and lowkey chaotic 🔥🖤 #VoidCat #NewRoomie" I'm Julia
상담사는 이를 처리하고 create_post 도구를 사용해야 하는지 확인한 후 MCP 서버와 통신하여 InstaVibe API를 호출해야 합니다.
👉 확인 단계: 상담사가 작업을 확인한 후 InstaVibe 애플리케이션이 실행 중인 탭을 열거나 새로고침합니다. '줄리아'의 새 게시물이 기본 피드에 표시됩니다.
👉💻 필요한 경우 별도의 터미널에서 이 스크립트를 실행하여 Instavibe 링크를 가져옵니다.
gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep instavibe
👉📝 이제 create_event 도구를 테스트해 보겠습니다. 채팅 대화상자에 다음과 같은 여러 줄 요청을 입력합니다.
Hey, can you set up an event for Hannah and George and me, and I'm Julia? Let's call it 'Mexico City Culinary & Art Day'.
here are more info
{"event_name": "Mexico City Culinary & Art Day",
"description": "A vibrant day in Mexico City for Hannah and George, starting with lunch at one of the city's best taco spots in the hip Condesa neighborhood, followed by an inspiring afternoon exploring the Museo Soumaya's stunning art collection.",
"event_date": "2025-10-17T12:00:00-06:00",
"locations": [
{
"name": "El Tizoncito",
"description": "Considered one of the original creators of tacos al pastor, El Tizoncito offers a legendary taco experience in the heart of Condesa. Their flavorful meats, house salsas, and casual vibe make it a must-visit for foodies.",
"latitude": 19.412179,
"longitude": -99.171308,
"address": "Av. Tamaulipas 122, Hipódromo, Cuauhtémoc, 06100 Ciudad de México, CDMX, Mexico"
},
{
"name": "Museo Soumaya",
"description": "An architectural icon in Mexico City, Museo Soumaya houses over 66,000 works of art, including pieces by Rodin, Dalí, and Rivera. The striking silver structure is a cultural landmark and a visual feast inside and out.",
"latitude": 19.440056,
"longitude": -99.204281,
"address": "Plaza Carso, Blvd. Miguel de Cervantes Saavedra 303, Granada, Miguel Hidalgo, 11529 Ciudad de México, CDMX, Mexico"
}
],
"attendee_names": ["Hannah", "George", Julia],
}
다시 한번 강조하지만 상담사는 MCP 서버를 통해 적절한 도구를 사용해야 합니다. 이벤트 탭에서 개별 이벤트를 클릭하면 실행의 세부적인 단계별 트레이스가 표시됩니다.
👉 인증 단계: 실행 중인 InstaVibe 애플리케이션으로 돌아가 '이벤트' 섹션 (또는 이에 상응하는 섹션)으로 이동합니다. 이제 새로 만든 '멕시코시티 요리 및 예술의 날' 이벤트가 표시됩니다.
이 예는 MCP를 통해 상담사가 외부 도구 (이 경우 InstaVibe의 API)를 표준화된 방식으로 활용할 수 있는 방법을 잘 보여줍니다.
두 작업을 모두 확인한 후 Cloud Shell 터미널로 돌아가서 Ctrl+C
를 눌러 ADK Dev UI를 중지합니다.
9. ADK의 워크플로 에이전트 및 멀티 에이전트
지금까지 Google 상담사는 외출을 계획하고 플랫폼과 상호작용할 수 있었습니다. 하지만 진정으로 맞춤설정된 계획을 위해서는 사용자의 사회적 관계를 이해해야 합니다. 친구의 활동을 면밀히 추적하지 않는 바쁜 사용자의 경우 이러한 맥락을 수동으로 수집하기는 어렵습니다. 이를 해결하기 위해 Spanner 그래프 데이터베이스를 활용하여 친구 활동과 관심분야를 분석하고 더 맞춤설정된 추천을 제공하는 소셜 프로파일링 에이전트를 빌드할 예정입니다.
먼저 이 에이전트가 그래프 데이터에 액세스할 수 있는 도구가 필요합니다.
👉📝 다음 Python 함수를 ~/instavibe-bootstrap/agents/social/instavibe.py
파일 끝에 추가합니다.
def get_person_attended_events(person_id: str)-> list[dict]:
"""
Fetches events attended by a specific person using Graph Query.
Args:
person_id (str): The ID of the person whose posts to fetch.
Returns: list[dict] or None.
"""
if not db_instance: return None
graph_sql = """
Graph SocialGraph
MATCH (p:Person)-[att:Attended]->(e:Event)
WHERE p.person_id = @person_id
RETURN e.event_id, e.name, e.event_date, att.attendance_time
ORDER BY e.event_date DESC
"""
params = {"person_id": person_id}
param_types_map = {"person_id": param_types.STRING}
fields = ["event_id", "name", "event_date", "attendance_time"]
results = run_graph_query( graph_sql, params=params, param_types=param_types_map, expected_fields=fields)
if results is None: return None
for event in results:
if isinstance(event.get('event_date'), datetime):
event['event_date'] = event['event_date'].isoformat()
if isinstance(event.get('attendance_time'), datetime):
event['attendance_time'] = event['attendance_time'].isoformat()
return results
def get_person_id_by_name( name: str) -> str:
"""
Fetches the person_id for a given name using SQL.
Args:
name (str): The name of the person to search for.
Returns:
str or None: The person_id if found, otherwise None.
Returns the ID of the *first* match if names are duplicated.
"""
if not db_instance: return None
sql = """
SELECT person_id
FROM Person
WHERE name = @name
LIMIT 1 -- Return only the first match in case of duplicate names
"""
params = {"name": name}
param_types_map = {"name": param_types.STRING}
fields = ["person_id"]
# Use the standard SQL query helper
results = run_sql_query( sql, params=params, param_types=param_types_map, expected_fields=fields)
if results: # Check if the list is not empty
return results[0].get('person_id') # Return the ID from the first dictionary
else:
return None # Name not found
def get_person_posts( person_id: str)-> list[dict]:
"""
Fetches posts written by a specific person using Graph Query.
Args:
person_id (str): The ID of the person whose posts to fetch.
Returns:
list[dict] or None: List of post dictionaries with ISO date strings,
or None if an error occurs.
"""
if not db_instance: return None
# Graph Query: Find the specific Person node, follow 'Wrote' edge to Post nodes
graph_sql = """
Graph SocialGraph
MATCH (author:Person)-[w:Wrote]->(post:Post)
WHERE author.person_id = @person_id
RETURN post.post_id, post.author_id, post.text, post.sentiment, post.post_timestamp, author.name AS author_name
ORDER BY post.post_timestamp DESC
"""
# Parameters now include person_id and limit
params = {
"person_id": person_id
}
param_types_map = {
"person_id": param_types.STRING
}
# Fields returned remain the same
fields = ["post_id", "author_id", "text", "sentiment", "post_timestamp", "author_name"]
results = run_graph_query(graph_sql, params=params, param_types=param_types_map, expected_fields=fields)
if results is None:
return None
# Convert datetime objects to ISO format strings
for post in results:
if isinstance(post.get('post_timestamp'), datetime):
post['post_timestamp'] = post['post_timestamp'].isoformat()
return results
def get_person_friends( person_id: str)-> list[dict]:
"""
Fetches friends for a specific person using Graph Query.
Args:
person_id (str): The ID of the person whose posts to fetch.
Returns: list[dict] or None.
"""
if not db_instance: return None
graph_sql = """
Graph SocialGraph
MATCH (p:Person {person_id: @person_id})-[f:Friendship]-(friend:Person)
RETURN DISTINCT friend.person_id, friend.name
ORDER BY friend.name
"""
params = {"person_id": person_id}
param_types_map = {"person_id": param_types.STRING}
fields = ["person_id", "name"]
results = run_graph_query( graph_sql, params=params, param_types=param_types_map, expected_fields=fields)
return results
이제 에이전트의 구조를 구성하는 방법을 알아보겠습니다. 여러 친구의 프로필을 분석한 후 결과를 요약하는 방법에는 여러 단계가 포함됩니다. 이는 ADK의 멀티 에이전트 기능, 특히 워크플로 에이전트를 사용하는 데 적합한 시나리오입니다.
Google의 ADK에서 워크플로 에이전트는 태스크를 직접 실행하지 않고 하위 에이전트라고 하는 다른 에이전트를 조정합니다. 이를 통해 모듈식 설계를 적용하여 복잡한 문제를 전문 구성요소로 분류할 수 있습니다. ADK는 다음과 같은 기본 제공 워크플로 유형을 제공합니다.
- 순차 (단계별)
- 병렬 (동시 실행)
- 및 루프 (반복 실행)
소셜 프로파일링 작업의 경우 디자인에서 루프 에이전트를 사용하여 반복 워크플로를 만듭니다. 한 번에 한 사람씩 처리하는 것이 목적입니다. profile_agent
는 데이터를 수집하고, summary_agent
는 분석을 업데이트하며, check_agent
는 다시 루프해야 하는지 결정합니다.
이 워크플로에 필요한 하위 에이전트를 정의해 보겠습니다.
👉📝 ~/instavibe-bootstrap/agents/social/agent.py
에서 #REPLACE FOR profile_agent
를 다음으로 바꿉니다.
profile_agent = LlmAgent(
name="profile_agent",
model="gemini-2.5-flash",
description=(
"Agent to answer questions about the this person social profile. Provide the person's profile using their name, make sure to fetch the id before getting other data."
),
instruction=(
"You are a helpful agent to answer questions about the this person social profile. You'll be given a list of names, provide the person's profile using their name, make sure to fetch the id before getting other data. Get one person at a time, start with the first one on the list, and skip if already provided. return this person's result"
),
tools=[get_person_posts,get_person_friends,get_person_id_by_name,get_person_attended_events],
)
그런 다음 에이전트는 수집된 프로필 정보 (루프 반복을 통해 누적됨)를 가져와 최종 요약을 생성하여 여러 사용자가 분석된 경우 공통점을 파악합니다.
👉📝 동일한 ~/instavibe-bootstrap/agents/social/agent.py
에서 #REPLACE FOR summary_agent
를 다음으로 바꿉니다.
summary_agent = LlmAgent(
name="summary_agent",
model="gemini-2.5-flash",
description=(
"Generate a comprehensive social summary as a single, cohesive paragraph. This summary should cover the activities, posts, friend networks, and event participation of one or more individuals. If multiple profiles are analyzed, the paragraph must also identify and integrate any common ground found between them."
),
instruction=(
"""
Your primary task is to synthesize social profile information into a single, comprehensive paragraph.
**Input Scope & Default Behavior:**
* If specific individuals are named by the user, focus your analysis on them.
* **If no individuals are specified, or if the request is general, assume the user wants an analysis of *all relevant profiles available in the current dataset/context*.**
**For each profile (whether specified or determined by default), you must analyze:**
1. **Post Analysis:**
* Systematically review their posts (e.g., content, topics, frequency, engagement).
* Identify recurring themes, primary interests, and expressed sentiments.
2. **Friendship Relationship Analysis:**
* Examine their connections/friends list.
* Identify key relationships, mutual friends (especially if comparing multiple profiles), and the general structure of their social network.
3. **Event Participation Analysis:**
* Investigate their past (and if available, upcoming) event participation.
* Note the types of events, frequency of attendance, and any notable roles (e.g., organizer, speaker).
**Output Generation (Single Paragraph):**
* **Your entire output must be a single, cohesive summary paragraph.**
* **If analyzing a single profile:** This paragraph will detail their activities, interests, and social connections based on the post, friend, and event analysis.
* **If analyzing multiple profiles:** This paragraph will synthesize the key findings regarding posts, friends, and events for each individual. Crucially, it must then seamlessly integrate or conclude with an identification and description of the common ground found between them (e.g., shared interests from posts, overlapping event attendance, mutual friends). The aim is a unified narrative within this single paragraph.
**Key Considerations:**
* Base your summary strictly on the available data.
* If data for a specific category (posts, friends, events) is missing or sparse for a profile, you may briefly acknowledge this within the narrative if relevant.
"""
),
output_key="summary"
)
루프를 중지해야 하는 시점 (요청된 모든 프로필이 요약된 시점)을 결정하는 방법이 필요합니다.
👉📝 동일한 ~/instavibe-bootstrap/agents/social/agent.py
에서 #REPLACE FOR check_agent
를 다음으로 바꿉니다.
check_agent = LlmAgent(
name="check_agent",
model="gemini-2.5-flash",
description=(
"Check if everyone's social profile are summarized and has been generated. Output 'completed' or 'pending'."
),
output_key="summary_status"
)
check_agent
에서 반환되고 State에 저장된 summary_status
를 명시적으로 확인하고 루프 에이전트에게 계속 (escalate=False)할지 또는 중지 (escalate=True)할지 알려주는 간단한 프로그래매틱 검사 (CheckCondition)를 추가합니다.
👉📝 동일한 ~/instavibe-bootstrap/agents/social/agent.py
에서 파일 상단에 있는 #REPLACE FOR CheckCondition
를 다음으로 바꿉니다.
class CheckCondition(BaseAgent):
async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
#log.info(f"Checking status: {ctx.session.state.get("summary_status", "fail")}")
log.info(f"Summary: {ctx.session.state.get("summary")}")
status = ctx.session.state.get("summary_status", "fail").strip()
is_done = (status == "completed")
yield Event(author=self.name, actions=EventActions(escalate=is_done))
루프 결과의 상태 및 콜백
Google ADK에서 상태는 실행 중인 에이전트의 메모리 또는 작업 데이터를 나타내는 중요한 개념입니다. 에이전트가 여러 단계, 도구 호출 또는 상호작용에서 유지해야 하는 정보를 보유하는 영구 컨텍스트입니다. 이 상태는 중간 결과, 사용자 정보, 후속 작업의 매개변수 또는 에이전트가 태스크를 진행하는 동안 기억해야 하는 기타 데이터를 저장할 수 있습니다.
이 시나리오에서 Loop Agent가 반복될 때 summary_agent
및 check_agent
는 출력 (summary 및 summary_status)을 상담사의 상태에 저장합니다. 이렇게 하면 반복 간에 정보가 유지됩니다. 하지만 Loop 에이전트 자체는 완료 시 상태에서 최종 요약을 자동으로 반환하지 않습니다.
ADK의 콜백을 사용하면 특정 이벤트(예: 도구 호출 완료 또는 에이전트 실행 완료 전)에 대한 응답으로 또는 에이전트 수명 주기의 특정 지점에서 실행할 맞춤 로직을 삽입할 수 있습니다. 에이전트의 동작을 맞춤설정하고 결과를 동적으로 처리하는 방법을 제공합니다.
CheckCondition이 에스컬레이션되었으므로 루프가 완료될 때 실행되는 after_agent_callback
를 사용합니다. 이 콜백 modify_output_after_agent
은 상태에서 최종 요약을 가져와 상담사의 최종 출력 메시지로 형식을 지정합니다.
👉📝 동일한 ~/instavibe-bootstrap/agents/social/agent.py
에서 #REPLACE FOR modify_output_after_agent
를 다음으로 바꿉니다.
def modify_output_after_agent(callback_context: CallbackContext) -> Optional[types.Content]:
agent_name = callback_context.agent_name
invocation_id = callback_context.invocation_id
current_state = callback_context.state.to_dict()
current_user_content = callback_context.user_content
print(f"[Callback] Exiting agent: {agent_name} (Inv: {invocation_id})")
print(f"[Callback] Current summary_status: {current_state.get("summary_status")}")
print(f"[Callback] Current Content: {current_user_content}")
status = current_state.get("summary_status").strip()
is_done = (status == "completed")
# Retrieve the final summary from the state
final_summary = current_state.get("summary")
print(f"[Callback] final_summary: {final_summary}")
if final_summary and is_done and isinstance(final_summary, str):
log.info(f"[Callback] Found final summary, constructing output Content.")
# Construct the final output Content object to be sent back
return types.Content(role="model", parts=[types.Part(text=final_summary.strip())])
else:
log.warning("[Callback] No final summary found in state or it's not a string.")
# Optionally return a default message or None if no summary was generated
return None
루트 루프 에이전트 정의
마지막으로 기본 LoopAgent를 정의합니다. 각 루프 반복 내에서 하위 에이전트를 순차적으로 조정합니다 (profile_agent -> summary_agent -> check_agent -> CheckCondition). 이 시퀀스는 max_iterations 횟수까지 또는 CheckCondition이 완료를 신호할 때까지 반복됩니다. after_agent_callback은 최종 요약이 반환되도록 합니다.
👉📝 동일한 ~/instavibe-bootstrap/agents/social/agent.py
에서 #REPLACE FOR root_agent
를 다음으로 바꿉니다.
root_agent = LoopAgent(
name="InteractivePipeline",
sub_agents=[
profile_agent,
summary_agent,
check_agent,
CheckCondition(name="Checker")
],
description="Find everyone's social profile on events, post and friends",
max_iterations=10,
after_agent_callback=modify_output_after_agent
)
ADK Dev UI를 사용하여 이 멀티 에이전트 워크플로를 테스트해 보겠습니다.
👉💻 ADK 웹 서버를 실행합니다.
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents
sed -i "s|^\(O\?GOOGLE_CLOUD_PROJECT\)=.*|GOOGLE_CLOUD_PROJECT=${PROJECT_ID}|" ~/instavibe-bootstrap/agents/social/.env
adk web
ADK Dev UI (웹 미리보기 경유 포트 8000)를 엽니다. 상담사 드롭다운 메뉴 (오른쪽 상단)에서 소셜 상담사를 선택합니다.
👉 이제 여러 사용자의 프로필을 작성하는 작업을 맡깁니다. 채팅 대화상자에 다음을 입력합니다.
Tell me about Mike and Bob
상담사가 응답한 후 (루프 및 여러 LLM 호출로 인해 다소 시간이 걸릴 수 있음) 최종 채팅 출력만 확인하지 마세요. ADK Dev UI의 왼쪽 창에서 '이벤트' 탭으로 이동합니다.
👉 확인 단계: 이벤트 탭에 실행의 단계별 세부 트레이스가 표시됩니다.
에이전트가 각 하위 에이전트를 호출하는 방식을 관찰한 후, 각 반복 내에 흐름이 profile_agent -> summary_agent -> check_agent, Checker로 이동할 것으로 예상됩니다. 하지만 실제로는 상담사의 강력한'자체 최적화'가 작동하는 것을 확인할 수 있습니다.
기본 모델은 전체 요청 (예: '마이크와 밥 프로파일링')을 실행할 때는 여러 번 반복하는 대신 단일 통합 턴에서 필요한 모든 데이터를 수집하는 등 가장 효율적인 경로를 선택하는 경우가 많습니다. profile_agent에서 실행한 도구 호출을 비롯하여 각 단계의 입력, 출력, 상태를 확인할 수 있습니다.
check_agent 및 CheckCondition의 상태 업데이트도 확인합니다.
이 시각적 트레이스는 최종 요약이 콜백에서 생성되고 반환될 때까지 멀티 에이전트 워크플로가 작동하는 방식을 이해하고 디버그하는 데 매우 유용합니다.
채팅 응답과 이벤트 트레이스를 살펴본 후 Cloud Shell 터미널로 돌아가 Ctrl+C
키를 눌러 ADK Dev UI를 중지합니다.
10. 상담사 간 (A2A) 커뮤니케이션
지금까지는 특수화된 상담사를 빌드했지만 이러한 상담사는 격리된 상태로 작동하거나 동일한 머신에서 사전 정의된 워크플로 내에서 작동합니다. 진정으로 분산되고 공동작업이 가능한 멀티 에이전트 시스템을 빌드하려면 별도의 서비스로 실행될 수 있는 에이전트가 서로를 검색하고 효과적으로 통신할 수 있는 방법이 필요합니다. 이때 Agent-to-Agent (A2A) 프로토콜이 사용됩니다.
A2A 프로토콜은 AI 에이전트 간의 상호 운용성 있는 통신을 위해 특별히 설계된 개방형 표준입니다. MCP는 상담사와 도구 간의 상호작용에 중점을 두고 A2A는 상담사와 상담사 간의 상호작용에 중점을 둡니다. 상담사는 다음을 수행할 수 있습니다.
- 탐색: 표준화된 상담사 카드를 통해 다른 상담사를 찾고 그들의 기능을 알아봅니다.
- 커뮤니케이션: 메시지와 데이터를 안전하게 교환합니다.
- 공동작업: 작업을 위임하고 작업을 조정하여 복잡한 목표를 달성합니다.
A2A 프로토콜은 에이전트가 기능과 연결 정보를 광고하는 데 사용할 수 있는 '에이전트 카드'와 같은 메커니즘을 통해 이러한 커뮤니케이션을 용이하게 합니다.
A2A는 익숙한 웹 표준 (HTTP, SSE, JSON-RPC)을 활용하며, 한 에이전트 (클라이언트)가 다른 에이전트 (원격 에이전트/서버)에 작업을 전송하는 클라이언트-서버 모델을 사용하는 경우가 많습니다. 이러한 표준화는 독립적으로 개발된 에이전트가 함께 작동할 수 있는 확장 가능한 모듈식 시스템을 구축하는 데 중요합니다.
InstaVibe 상담사를 위한 A2A 사용 설정
기존 Planner, 플랫폼 상호작용, 소셜 상담사가 A2A를 통해 다른 상담사에게 액세스할 수 있도록 하려면 각 상담사를 A2A 서버 구성요소로 래핑해야 합니다. 이 서버는 다음을 실행합니다.
- 상담사 카드 노출: HTTP 엔드포인트를 통해 상담사의 기능에 관한 표준 설명을 제공합니다.
- Tasks(Request Messages) 리슨: A2A 프로토콜에 따라 다른 상담사 (A2A 클라이언트)의 수신 태스크 요청을 수락합니다.
- 작업(요청 메시지) 실행 관리: 수신된 작업을 처리를 위해 기본 ADK 에이전트 로직에 전달합니다.
Planner Agent (A2A 사용 설정됨)
먼저 Planner 에이전트에 A2A 서버 레이어를 추가해 보겠습니다.
A2A 서버 시작 로직을 정의합니다. 이 코드는 AgentCard (에이전트의 공개 설명)를 정의하고, A2AServer를 구성하고, 이를 시작하여 PlatformAgentExecutor에 연결합니다.
👉📝 다음 코드를 ~/instavibe-bootstrap/agents/planner/a2a_server.py
끝에 추가합니다.
class PlannerAgent:
"""An agent to help user planning a event with its desire location."""
SUPPORTED_CONTENT_TYPES = ["text", "text/plain"]
def __init__(self):
self._agent = self._build_agent()
self.runner = Runner(
app_name=self._agent.name,
agent=self._agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="event_planner",
name="Event planner",
description="""
This agent generates multiple fun plan suggestions tailored to your specified location, dates, and interests,
all designed for a moderate budget. It delivers detailed itineraries,
including precise venue information (name, latitude, longitude, and description), in a structured JSON format.
""",
tags=["instavibe"],
examples=["What about Bostona MA this weekend?"],
)
self.agent_card = AgentCard(
name="Event Planner Agent",
description="""
This agent generates multiple fun plan suggestions tailored to your specified location, dates, and interests,
all designed for a moderate budget. It delivers detailed itineraries,
including precise venue information (name, latitude, longitude, and description), in a structured JSON format.
""",
url=f"{PUBLIC_URL}",
version="1.0.0",
defaultInputModes=PlannerAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=PlannerAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
def get_processing_message(self) -> str:
return "Processing the planning request..."
def _build_agent(self) -> LlmAgent:
"""Builds the LLM agent for the night out planning agent."""
return agent.root_agent
if __name__ == '__main__':
try:
plannerAgent = PlannerAgent()
request_handler = DefaultRequestHandler(
agent_executor=PlannerAgentExecutor(plannerAgent.runner,plannerAgent.agent_card),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=plannerAgent.agent_card,
http_handler=request_handler,
)
logger.info(f"Attempting to start server with Agent Card: {plannerAgent.agent_card.name}")
logger.info(f"Server object created: {server}")
uvicorn.run(server.build(), host='0.0.0.0', port=port)
except Exception as e:
logger.error(f"An error occurred during server startup: {e}")
exit(1)
👉💻 A2A 서버가 로컬에서 올바르게 시작되고 상담사 카드를 제공하는지 빠르게 테스트해 보겠습니다. 첫 번째 터미널에서 다음 명령어를 실행합니다.
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
cd ~/instavibe-bootstrap/agents/
python -m planner.a2a_server
👉 이제 다른 터미널 창을 엽니다. (터미널 패널에서 + 기호를 클릭합니다.)
👉💻 curl을 사용하여 로컬에서 실행 중인 서버에서 상담사 카드를 요청합니다.
curl http://localhost:10003/.well-known/agent.json | jq
정의한 AgentCard의 JSON 표현이 표시되어 서버가 실행 중이고 Planner 에이전트를 광고하고 있음을 확인할 수 있습니다.
서버가 실행 중인 첫 번째 터미널로 돌아가 Ctrl+C
를 눌러 중지합니다.
👉💻 A2A 서버 로직을 추가했으므로 이제 컨테이너 이미지를 빌드할 수 있습니다.
Planner 에이전트 빌드 및 배포
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/agents
# Set variables specific to the PLANNER agent
export IMAGE_TAG="latest"
export AGENT_NAME="planner"
export IMAGE_NAME="planner-agent"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="planner-agent"
export PUBLIC_URL="https://planner-agent-${PROJECT_NUMBER}.${REGION}.run.app"
echo "Building ${AGENT_NAME} agent..."
gcloud builds submit . \
--config=cloudbuild-build.yaml \
--project=${PROJECT_ID} \
--region=${REGION} \
--substitutions=_AGENT_NAME=${AGENT_NAME},_IMAGE_PATH=${IMAGE_PATH}
echo "Image built and pushed to: ${IMAGE_PATH}"
👉💻 Cloud Run에 Planner 에이전트를 배포합니다.
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/agents
# Set variables specific to the PLANNER agent
export IMAGE_TAG="latest"
export AGENT_NAME="planner"
export IMAGE_NAME="planner-agent"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="planner-agent"
export PUBLIC_URL="https://planner-agent-${PROJECT_NUMBER}.${REGION}.run.app"
gcloud run deploy ${SERVICE_NAME} \
--image=${IMAGE_PATH} \
--platform=managed \
--region=${REGION} \
--set-env-vars="A2A_HOST=0.0.0.0" \
--set-env-vars="A2A_PORT=8080" \
--set-env-vars="GOOGLE_GENAI_USE_VERTEXAI=TRUE" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}" \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--set-env-vars="PUBLIC_URL=${PUBLIC_URL}" \
--allow-unauthenticated \
--project=${PROJECT_ID} \
--min-instances=1
A2A Inspector를 사용하여 배포된 서비스가 실행 중이고 클라우드에서 상담사 카드를 올바르게 게재하는지 확인해 보겠습니다.
👉 Cloud Shell 툴바의 웹 미리보기 아이콘에서 포트 변경을 선택합니다. 포트를 8081로 설정하고 '변경 및 미리보기'를 클릭합니다. A2A Inspector 인터페이스가 포함된 새 브라우저 탭이 열립니다.
👉💻 터미널에서 배포된 계획자 에이전트의 URL을 가져옵니다.
export PLANNER_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep planner-agent)
echo ${PLANNER_AGENT_URL}
👉💻 출력 URL을 복사합니다.
👉 A2A Inspector UI에서 에이전트 URL 필드에 URL을 붙여넣고 '연결'을 클릭합니다.
👀 상담사의 카드 세부정보와 JSON이 상담사 카드 탭에 표시되면 연결이 완료된 것입니다.
👉 A2A 검사기에서 채팅 탭을 클릭합니다. 여기에서 배포된 에이전트와 직접 상호작용할 수 있습니다. 메시지를 보내 계획 기능을 테스트하세요. 예를 들면 다음과 같습니다.
Plan something for me in Boston MA this weekend, and I enjoy classical music
👀 원시 커뮤니케이션을 검사하려면 채팅 창에서 내 메시지 풍선을 클릭한 다음 상담사의 응답 풍선을 클릭합니다. 각 항목을 클릭하면 전송 또는 수신된 전체 JSON-RPC 2.0 메시지가 표시되며, 이는 디버깅에 매우 유용합니다.
A2A Inspector 탭을 가까이에 두세요. 닫지 마세요. 잠시 후 다른 두 에이전트를 테스트하기 위해 다시 사용합니다.
플랫폼 상호작용 에이전트 (A2A 사용 설정됨)
다음으로 플랫폼 상호작용 에이전트 (MCP를 사용하는 에이전트)에 대해 프로세스를 반복합니다.
👉📝 ~/instavibe-bootstrap/agents/platform_mcp_client/a2a_server.py
끝에 고유한 AgentCard를 포함하여 A2A 서버 설정을 정의합니다.
class PlatformAgent:
"""An agent that post event and post to instavibe."""
SUPPORTED_CONTENT_TYPES = ["text", "text/plain"]
def __init__(self):
self._agent = self._build_agent()
self.runner = Runner(
app_name=self._agent.name,
agent=self._agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="instavibe_posting",
name="Post social post and events on instavibe",
description="""
This "Instavibe" agent helps you create posts (identifying author, text, and sentiment – inferred if unspecified) and register
for events (gathering name, date, attendee). It efficiently collects required information and utilizes dedicated tools
to perform these actions on your behalf, ensuring a smooth sharing experience.
""",
tags=["instavibe"],
examples=["Create a post for me, the post is about my cute cat and make it positive, and I'm Alice"],
)
self.agent_card = AgentCard(
name="Instavibe Posting Agent",
description="""
This "Instavibe" agent helps you create posts (identifying author, text, and sentiment – inferred if unspecified) and register
for events (gathering name, date, attendee). It efficiently collects required information and utilizes dedicated tools
to perform these actions on your behalf, ensuring a smooth sharing experience.
""",
url=f"{PUBLIC_URL}",
version="1.0.0",
defaultInputModes=PlatformAgent.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=PlatformAgent.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
def get_processing_message(self) -> str:
return "Processing the social post and event request..."
def _build_agent(self) -> LlmAgent:
"""Builds the LLM agent for the Processing the social post and event request."""
return agent.root_agent
if __name__ == '__main__':
try:
platformAgent = PlatformAgent()
request_handler = DefaultRequestHandler(
agent_executor=PlatformAgentExecutor(platformAgent.runner,platformAgent.agent_card),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=platformAgent.agent_card,
http_handler=request_handler,
)
uvicorn.run(server.build(), host='0.0.0.0', port=port)
except Exception as e:
logger.error(f"An error occurred during server startup: {e}")
exit(1)
소셜 상담사 (A2A 사용 설정됨)
마지막으로 소셜 프로파일링 에이전트에 A2A를 사용 설정합니다.
👉📝 ~/instavibe-bootstrap/agents/social/a2a_server.py
끝에 A2A 서버 설정 및 AgentCard를 정의합니다.
class SocialAgent:
"""An agent that handles social profile analysis."""
SUPPORTED_CONTENT_TYPES = ["text", "text/plain"]
def __init__(self):
self._agent = self._build_agent()
self.runner = Runner(
app_name=self._agent.name,
agent=self._agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
capabilities = AgentCapabilities(streaming=True)
skill = AgentSkill(
id="social_profile_analysis",
name="Analyze Instavibe social profile",
description="""
Using a provided list of names, this agent synthesizes Instavibe social profile information by analyzing posts, friends, and events.
It delivers a comprehensive single-paragraph summary for individuals, and for groups, identifies commonalities in their social activities
and connections based on profile data.
""",
tags=["instavibe"],
examples=["Can you tell me about Bob and Alice?"],
)
self.agent_card = AgentCard(
name="Social Profile Agent",
description="""
Using a provided list of names, this agent synthesizes Instavibe social profile information by analyzing posts, friends, and events.
It delivers a comprehensive single-paragraph summary for individuals, and for groups, identifies commonalities in their social activities
and connections based on profile data.
""",
url=f"{PUBLIC_URL}",
version="1.0.0",
defaultInputModes=self.SUPPORTED_CONTENT_TYPES,
defaultOutputModes=self.SUPPORTED_CONTENT_TYPES,
capabilities=capabilities,
skills=[skill],
)
def get_processing_message(self) -> str:
return "Processing the social profile analysis request..."
def _build_agent(self) -> LoopAgent:
"""Builds the LLM agent for the social profile analysis agent."""
return agent.root_agent
if __name__ == '__main__':
try:
socialAgent = SocialAgent()
request_handler = DefaultRequestHandler(
agent_executor=SocialAgentExecutor(socialAgent.runner,socialAgent.agent_card),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
agent_card=socialAgent.agent_card,
http_handler=request_handler,
)
uvicorn.run(server.build(), host='0.0.0.0', port=port)
except Exception as e:
logger.error(f"An error occurred during server startup: {e}")
exit(1)
플랫폼 상호작용 및 소셜 에이전트 빌드 및 배포
이러한 에이전트는 Spanner에 액세스해야 하므로 배포 중에 SPANNER_INSTANCE_ID
, SPANNER_DATABASE_ID
, MCP_SERVER_URL
환경 변수가 올바르게 전달되는지 확인합니다.
👉💻 Cloud Build를 사용하여 빌드하고 Cloud Run에 배포합니다.
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/agents
export MCP_SERVER_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep mcp-tool-server)/sse
gcloud builds submit . \
--config=cloudbuild.yaml \
--project="${PROJECT_ID}" \
--region="${REGION}" \
--substitutions=\
_PROJECT_ID="${PROJECT_ID}",\
_PROJECT_NUMBER="${PROJECT_NUMBER}",\
_REGION="${REGION}",\
_REPO_NAME="${REPO_NAME}",\
_SPANNER_INSTANCE_ID="${SPANNER_INSTANCE_ID}",\
_SPANNER_DATABASE_ID="${SPANNER_DATABASE_ID}",\
_MCP_SERVER_URL="${MCP_SERVER_URL}"
👉💻 터미널에서 배포된 플랫폼 에이전트의 URL을 가져옵니다.
export PLATFORM_MPC_CLIENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep platform-mcp-client)
echo $PLATFORM_MPC_CLIENT_URL
👉💻 출력 URL을 복사합니다.
👉 A2A Inspector UI에서 에이전트 URL 필드에 URL을 붙여넣고 '연결'을 클릭합니다.
👀 상담사의 카드 세부정보와 JSON이 상담사 카드 탭에 표시되면 연결이 완료된 것입니다.
👉 A2A 검사기에서 채팅 탭을 클릭합니다. 여기에서 배포된 에이전트와 직접 상호작용할 수 있습니다. 메시지를 보내 에이전트의 게시물 작성 기능을 테스트해 보세요.
Create a post for me, the post says 'Paws, purrs, and ocean views 🐾☕🌊. Spent my morning at the Morning Seaside Cat Café, where every sip comes with a side of snuggles and sea breeze.' and make it positive, and I'm Oscar.
👀 원시 커뮤니케이션을 검사하려면 채팅 창에서 내 메시지 풍선을 클릭한 다음 상담사의 응답 풍선을 클릭합니다. 각 항목을 클릭하면 전송 또는 수신된 전체 JSON-RPC 2.0 메시지가 표시되며, 이는 디버깅에 매우 유용합니다.
👉💻 터미널에서 배포된 소셜 상담사의 URL을 가져옵니다.
export SOCIAL_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep social-agent)
echo $SOCIAL_AGENT_URL
👉💻 출력 URL을 복사합니다.
👉 A2A Inspector UI에서 에이전트 URL 필드에 URL을 붙여넣고 '연결'을 클릭합니다.
👀 상담사의 카드 세부정보와 JSON이 상담사 카드 탭에 표시되면 연결이 완료된 것입니다.
👉 A2A 검사기에서 채팅 탭을 클릭합니다. 여기에서 배포된 에이전트와 직접 상호작용하고 메시지를 보내 데이터베이스의 사용자 프로필을 분석할 수 있습니다.
Can you tell me about both Ian and Kevin's profile, what are their common interests?
👀 원시 커뮤니케이션을 검사하려면 채팅 창에서 내 메시지 풍선을 클릭한 다음 상담사의 응답 풍선을 클릭합니다. 각 항목을 클릭하면 전송 또는 수신된 전체 JSON-RPC 2.0 메시지가 표시되며, 이는 디버깅에 매우 유용합니다.
👉 좋습니다. 모든 상담사의 검사가 완료되었습니다. 이제 A2A 검사기 탭을 닫을 수 있습니다.
11. Orchestrator Agent (A2A 클라이언트)
이제 Cloud Run에서 독립적인 A2A 지원 서비스로 실행되는 세 가지 전문 상담사 (Planner, Platform, Social)가 있습니다. 마지막 부분은 Orchestrator Agent입니다. 이 에이전트는 중앙 조정자 또는 A2A 클라이언트 역할을 합니다. 사용자 요청을 수신하고 요청을 처리하는 데 필요한 원격 에이전트를 파악한 후(가능한 경우 순차적으로) A2A 프로토콜을 사용하여 이러한 원격 에이전트에게 태스크를 위임합니다. 이 워크숍에서는 ADK Dev UI를 사용하여 Orchestrator 에이전트를 로컬에서 실행합니다.
먼저 Orchestrator의 로직을 개선하여 Orchestrator가 감지하는 원격 에이전트의 등록을 처리하도록 합니다. 초기화 중에 가져온 상담사 카드의 연결 세부정보를 저장합니다.
👉📝 ~/instavibe-bootstrap/agents/orchestrate/agent.py
에서 #REPLACE ME REG AGENT CARD
를 다음으로 바꿉니다.
async with httpx.AsyncClient(timeout=30) as client:
for i, address in enumerate(REMOTE_AGENT_ADDRESSES):
log.info(f"--- STEP 3.{i}: Attempting connection to: {address} ---")
try:
card_resolver = A2ACardResolver(client, address)
card = await card_resolver.get_agent_card()
remote_connection = RemoteAgentConnections(agent_card=card, agent_url=address)
self.remote_agent_connections[card.name] = remote_connection
self.cards[card.name] = card
log.info(f"--- STEP 5.{i}: Successfully stored connection for {card.name} ---")
except Exception as e:
log.error(f"--- CRITICAL FAILURE at STEP 4.{i} for address: {address} ---")
log.error(f"--- The hidden exception type is: {type(e).__name__} ---")
log.error(f"--- Full exception details and traceback: ---", exc_info=True)
다음으로 ADK 내에서 Orchestrator 에이전트 자체의 도구를 정의합니다.
send_message
(작업을 위임하는 A2A 함수)
👉📝 ~/instavibe-bootstrap/agents/orchestrate/agent.py
의 #REPLACE ME CREATE AGENT
를 다음으로 바꿉니다.
def create_agent(self) -> Agent:
"""Synchronously creates the ADK Agent object."""
return Agent(
model="gemini-2.5-flash",
name="orchestrate_agent",
instruction=self.root_instruction,
before_agent_callback=self.before_agent_callback,
description=("Orchestrates tasks for child agents."),
tools=[self.send_message],
)
Orchestrator의 핵심 로직은 A2A를 사용하는 방법을 알려주는 안내에 있습니다.
👉📝 ~/instavibe-bootstrap/agents/orchestrate/agent.py
의 #REPLACE ME INSTRUCTIONS
를 다음과 같은 명령어 생성 메서드로 바꿉니다.
def root_instruction(self, context: ReadonlyContext) -> str:
current_agent = self.check_active_agent(context)
return f"""
You are an expert AI Orchestrator. Your primary responsibility is to intelligently interpret user requests, break them down into a logical plan of discrete actions, and delegate each action to the most appropriate specialized remote agent using the send_message function. You do not perform the tasks yourself but manage their assignment, sequence, and critically, their outcomes.
**Core Directives & Decision Making:**
* **Understand User Intent & Complexity:**
* Carefully analyze the user's request to determine the core task(s) they want to achieve. Pay close attention to keywords and the overall goal.
* Identify if the request requires a single agent or a sequence of actions from multiple agents. For example, "Analyze John Doe's profile and then create a positive post about his recent event attendance" would require two agents in sequence.
* **Task Planning & Sequencing (for Multi-Step Requests):**
* Before delegating, outline the clear sequence of agent tasks.
* Identify dependencies. If Task B requires output from Task A, execute them sequentially. If tasks are independent (like creating a post and then creating an event), execute them one after the other as separate delegations.
* Agent Reusability: An agent's completion of one task does not make it unavailable. If a user's plan involves multiple, distinct actions that fall under the same agent's expertise (e.g., create a post, then create an event), you must call that same agent again for the subsequent task.
* **Task Delegation & Management (using `send_message`):**
* **Delegation:** Use `send_message` to assign actionable tasks to the selected remote agent. Your `send_message` call MUST include:
* The `remote_agent_name` you've selected.
* The `user_request` or all necessary parameters extracted from the user's input, formatted in a way the target agent will understand.
* **Contextual Awareness for Remote Agents:** If a remote agent repeatedly requests user confirmation or seems to lack context, assume it lacks access to the full conversation history. In such cases, enrich your `send_message` with all necessary contextual information relevant to that specific agent from the conversation history.
* **Sequential Task Execution:**
* After a preceding task completes (indicated by the agent's response or a success signal), gather any necessary output from it.
* Then, use `send_message` for the next agent in the sequence, providing it with the user's original relevant intent and any necessary data obtained from the previous agent's task.
* **Active Agent Prioritization:** If an active agent is already engaged and the user's request is related to its current task, route subsequent related requests directly to that agent by providing updated context via `send_message`.
**Critical Success Verification:**
* You **MUST** wait for the tool_output after every send_message call before taking any further action.
* Your decision to proceed to the next task in a sequence **MUST** be based entirely on a confirmation of success from the tool_output of the previous task.
* If a tool call fails, returns an error, or the tool_output is ambiguous, you MUST STOP the sequence. Your next action is to report the exact failure or ambiguity to the user.
* DO NOT assume a task was successful. Do not invent success messages like "The event has been created." Only state that a task is complete if the tool's response explicitly says so.
**Communication with User:**
* **Transparent Communication:** Always present the complete and detailed response from the remote agent to the user. Do not summarize or filter unless explicitly instructed.
* When you delegate a task (or the first task in a sequence), clearly inform the user which remote agent is handling it.
* For multi-step requests, you can optionally inform the user of the planned sequence (e.g., "Okay, first I'll ask the 'Social Profile Agent' to analyze the profile, and then I'll have the 'Instavibe Posting Agent' create the post.").
* If waiting for a task in a sequence to complete, you can inform the user (e.g., "The 'Social Profile Agent' is currently processing. I'll proceed with the post once that's done.").
* **User Confirmation Relay:** If a remote agent asks for confirmation, and the user has not already provided it, just make up something.
* If the user's request is ambiguous, if necessary information is missing for any agent in the sequence, or if you are unsure about the plan, just make up something.
**Important Reminders:**
* **Autonomous Agent Engagement:** Never seek user permission before engaging with remote agents. If multiple agents are required to fulfill a request, connect with them directly without requesting user preference or confirmation.
* **Focused Information Sharing:** Provide remote agents with only relevant contextual information. Avoid extraneous details that are not directly pertinent to their task.
* **No Redundant Confirmations:** Do not ask remote agents for confirmation of information or actions they have already processed or committed to.
* **Tool Reliance:** Strictly rely on your available tools, primarily `send_message`, to address user requests. Do not generate responses based on assumptions. If information is insufficient, request clarification from the user.
* **Prioritize Recent Interaction:** Focus primarily on the most recent parts of the conversation when processing requests, while maintaining awareness of the overall goal for multi-step tasks.
* Always prioritize selecting the correct agent(s) based on their documented purpose.
* Ensure all information required by the chosen remote agent is included in the `send_message` call, including outputs from previous agents if it's a sequential task.
Agents:
{self.agents}
Current agent: {current_agent['active_agent']}`
"""
Orchestrator 및 전체 A2A 시스템 테스트
이제 전체 시스템을 테스트해 보겠습니다. ADK Dev UI를 사용하여 Orchestrator를 로컬에서 실행하면 Cloud Run에서 원격으로 실행되는 Planner, Platform, 소셜 상담사와 통신합니다.
👉💻 먼저 환경 변수 REMOTE_AGENT_ADDRESSES
에 배포된 A2A 지원 상담사의 URL이 쉼표로 구분되어 포함되어 있는지 확인합니다. 그런 다음 Orchestrator 에이전트에 필요한 환경 변수를 설정하고 ADK Dev UI를 실행합니다.
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
export PLATFORM_MPC_CLIENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep platform-mcp-client)
export PLANNER_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep planner-agent)
export SOCIAL_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep social-agent)
export REMOTE_AGENT_ADDRESSES=${PLANNER_AGENT_URL},${PLATFORM_MPC_CLIENT_URL},${SOCIAL_AGENT_URL}
cd ~/instavibe-bootstrap/agents
sed -i "s|^\(O\?REMOTE_AGENT_ADDRESSES\)=.*|REMOTE_AGENT_ADDRESSES=${REMOTE_AGENT_ADDRESSES}|" ~/instavibe-bootstrap/agents/orchestrate/.env
adk web
👉 ADK Dev UI를 엽니다 (웹 미리보기를 통해 포트를 8000으로 다시 변경).
👉 에이전트 드롭다운에서 orchestrate 에이전트를 선택합니다.
👉 이제 여러 원격 에이전트를 조정해야 하는 복잡한 작업을 할당합니다. 소셜 상담사와 플래너 상담사가 모두 포함된 첫 번째 예시를 시도해 보세요.
You are an expert event planner for a user named Diana.
Your task is to design a fun and personalized event.
Here are the details for the plan:
- Friends to invite: Ian, Nora
- Desired date: "2025-10-15"
- Location idea or general preference: "Chicago"
Your process should be:
1. Analyze the provided friend names. If you have access to a tool to get their InstaVibe profiles or summarized interests, please use it.
2. Based on their potential interests (or general good taste if profiles are unavailable), create a tailored plan for the outing, check if you have access to any event planner tools.
3. Ensure the plan includes the original `planned_date`.
The user wants a comprehensive plan that includes:
- The list of invited friends.
- A catchy and descriptive name for the event.
- The exact planned date for the event.
- A summary of what the group will do.
- Specific recommended spots (e.g., restaurants, bars, activity venues) with their names, (if possible, approximate latitude/longitude for mapping, and address), and a brief description of why it fits the plan.
- A short, exciting message that {Diana} can send to {Ian, Nora} to get them excited about the event.
ADK Dev UI 채팅 창에서 상호작용을 관찰합니다. Orchestrator의 응답에 주의를 기울이세요. 작업을 위임할 원격 상담사가 표시됩니다 (예: "좋습니다. 먼저 소셜 프로필 상담사에게 이안과 노라에 관해 문의하겠습니다.")
또한 UI의 '이벤트' 탭에서 원격 에이전트 URL에 대한 기본 도구 호출 (send_message)이 이루어지고 있는지 확인합니다.
👉 이제 플랫폼 통합 에이전트가 직접 포함된 두 번째 예시를 시도해 보세요.
Hey, can you register an event on Instavibe for Laura and Charlie? Let's call it 'Vienna Concert & Castles Day'.
here are more info
"event_name": "Vienna Concert & Castles Day",
"description": "A refined and unforgettable day in Vienna with Laura and Charlie. The day begins with a guided tour of the magnificent Schönbrunn Palace, showcasing imperial architecture and history. In the evening, enjoy a classical music concert in one of Vienna's most iconic concert halls.",
"event_date": "2025-10-14T10:00:00+02:00",
"locations": [
{
"name": "Schönbrunn Palace",
"description": "A UNESCO World Heritage Site and former imperial summer residence, Schönbrunn Palace offers opulent rooms, beautiful baroque gardens, and a glimpse into the life of the Habsburg monarchy. Visitors can stroll the grounds or take a guided historical tour.",
"latitude": 48.184516,
"longitude": 16.312222,
"address": "Schönbrunner Schloßstraße 47, 1130 Wien, Austria"
},
{
"name": "Musikverein Vienna",
"description": "Home to the world-renowned Vienna Philharmonic, the Musikverein is one of the finest concert halls in the world. Its 'Golden Hall' is famous for its acoustics and ornate design. Attendees can enjoy a powerful classical concert in an unforgettable setting.",
"latitude": 48.200132,
"longitude": 16.373777,
"address": "Musikvereinsplatz 1, 1010 Wien, Austria"
}
],
"attendee_names": ["Laura", "Charlie", "Oscar"] And I am Oscar
채팅과 이벤트 탭을 다시 모니터링합니다. 오케스트레이터는 이벤트를 만들어야 하는 필요성을 파악하고 제공된 모든 세부정보와 함께 '플랫폼 통합 에이전트'에 태스크를 위임해야 합니다. Trace 버튼을 클릭하여 트레이스를 확인하고 쿼리 응답 시간과 실행된 작업을 분석할 수도 있습니다.
그런 다음 이벤트가 InstaVibe 웹 애플리케이션에 표시되는지 확인할 수 있습니다.
이는 중앙 오케스트레이터가 전문 원격 에이전트에게 작업을 위임하는 ADK 및 A2A 프로토콜을 사용하는 멀티 에이전트 시스템을 성공적으로 구현한 예입니다.
테스트가 완료되면 ADK Dev UI (터미널의 Ctrl+C
)를 중지해야 합니다.
12. InstaVibe의 상담사 엔진 및 원격 호출
지금까지는 Cloud Run에서 특수 에이전트를 실행하고 ADK Dev UI를 사용하여 로컬에서 Orchestrator를 테스트했습니다. 프로덕션 시나리오에서는 상담사를 호스팅할 수 있는 강력하고 확장 가능하며 관리되는 환경이 필요합니다. 이때 Google Vertex AI Agent Engine이 유용합니다.
Agent Engine은 AI 에이전트 배포 및 확장을 위해 특별히 설계된 Vertex AI의 완전 관리형 서비스입니다. 인프라 관리, 보안, 운영 오버헤드를 추상화하므로 개발자 (특히 복잡한 클라우드 환경에 익숙하지 않은 개발자)가 서버를 관리하는 대신 상담사의 로직과 기능에 집중할 수 있습니다. 에이전트 워크로드에 최적화된 전용 런타임을 제공합니다.
이제 Orchestrator 에이전트를 Agent Engine에 배포합니다. 참고: 아래에 표시된 배포 메커니즘은 워크샵 자료에 제공된 맞춤 스크립트(agent_engine_app.py)를 사용합니다. 공식 ADK-to-Agent-Engine 직접 배포 도구가 아직 개발 중일 수 있기 때문입니다. 이 스크립트는 필요한 원격 에이전트 주소로 구성된 Orchestrator 에이전트의 패키징 및 배포를 처리합니다.
다음 명령어를 실행하여 Orchestrator 에이전트를 Agent Engine에 배포합니다. REMOTE_AGENT_ADDRESSES 환경 변수 (Cloud Run의 Planner, 플랫폼, 소셜 상담사의 URL 포함)가 이전 섹션에서 여전히 올바르게 설정되어 있는지 확인합니다.
👉💻 Orchestrate 에이전트를 Agent Engine에 배포합니다. 참고: 이는 배포에 대한 자체 구현이며 ADK에는 배포를 지원하는 CLI가 있습니다. BYO-SA가 구현된 후 이를 업데이트하겠습니다.
cd ~/instavibe-bootstrap/agents/
. ~/instavibe-bootstrap/set_env.sh
source ~/instavibe-bootstrap/env/bin/activate
export PLATFORM_MPC_CLIENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep platform-mcp-client)
export PLANNER_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep planner-agent)
export SOCIAL_AGENT_URL=$(gcloud run services list --platform=managed --region=us-central1 --format='value(URL)' | grep social-agent)
export REMOTE_AGENT_ADDRESSES=${PLANNER_AGENT_URL},${PLATFORM_MPC_CLIENT_URL},${SOCIAL_AGENT_URL}
sed -i "s|^\(O\?REMOTE_AGENT_ADDRESSES\)=.*|REMOTE_AGENT_ADDRESSES=${REMOTE_AGENT_ADDRESSES}|" ~/instavibe-bootstrap/agents/orchestrate/.env
adk deploy agent_engine \
--display_name "orchestrate-agent" \
--project $GOOGLE_CLOUD_PROJECT \
--region $GOOGLE_CLOUD_LOCATION \
--staging_bucket gs://$GOOGLE_CLOUD_PROJECT-agent-engine \
--trace_to_cloud \
--requirements_file orchestrate/requirements.txt \
orchestrate
이제 Orchestrator가 관리형 Agent Engine 플랫폼에 호스팅되므로 InstaVibe 웹 애플리케이션이 Orchestrator와 통신해야 합니다. 웹 앱은 ADK Dev UI를 통해 상호작용하는 대신 Agent Engine 엔드포인트를 원격으로 호출합니다.
먼저 배포된 Orchestrator 에이전트의 고유 ID를 사용하여 Agent Engine 클라이언트를 초기화하도록 InstaVibe 애플리케이션 코드를 수정해야 합니다. 이 ID는 플랫폼에서 올바른 상담사 인스턴스를 타겟팅하는 데 필요합니다.
👉📝 ~/instavibe-bootstrap/instavibe/introvertally.py
를 열고 #REPLACE ME initiate agent_engine
를 다음 코드로 바꿉니다. 이렇게 하면 환경 변수 (곧 설정)에서 상담사 엔진 ID를 가져오고 클라이언트 객체를 가져옵니다.
ORCHESTRATE_AGENT_ID = os.environ.get('ORCHESTRATE_AGENT_ID')
agent_engine = agent_engines.get(ORCHESTRATE_AGENT_ID)
InstaVibe의 계획된 사용자 흐름에는 상담사와의 두 가지 상호작용이 포함됩니다. 첫 번째는 추천 계획을 생성하는 것이고 두 번째는 상담사가 실제로 이벤트를 플랫폼에 게시하기 전에 사용자에게 확인을 요청하는 것입니다.
이제 InstaVibe 웹 애플리케이션 (Cloud Run에서 실행됨)과 Orchestrator 에이전트 (Agent Engine에서 실행됨)가 별도의 서비스이므로 웹 앱은 에이전트와 상호작용하기 위해 Agent Engine 엔드포인트를 원격으로 호출해야 합니다.
👉📝 계획 추천을 생성하기 위해 초기 호출을 실행하는 코드를 업데이트해 보겠습니다. 동일한 introvertally.py
파일에서 #REPLACE ME Query remote agent get plan
를 agent_engine 클라이언트를 사용하여 사용자의 요청을 전송하는 다음 스니펫으로 바꿉니다.
agent_engine.stream_query(
user_id=user_id,
message=prompt_message,
)
👉📝 그런 다음 사용자의 확인을 처리하는 코드 (예: 사용자가 '계획 확인'을 클릭할 때)를 업데이트합니다. 그러면 Agent Engine의 동일한 대화에 후속 메시지가 전송되어 Orchestrator에게 이벤트 게시를 진행하도록 지시합니다 (이벤트 게시는 플랫폼 통합 상담사에게 위임됨). introvertally.py
에서 확인을 위한 #REPLACE ME Query remote agent for confirmation
를 다음으로 바꿉니다.
agent_engine.stream_query(
user_id=agent_session_user_id,
message=prompt_message,
)
웹 애플리케이션의 경로에 이러한 함수에 대한 액세스 권한이 필요합니다. introvertally.py의 필수 함수가 Flask 경로 파일에 가져와졌는지 확인합니다.
👉📝 cd ~/instavibe-bootstrap/instavibe/ally_routes.py
에서 먼저 인스턴스를 가리키고 # REPLACE ME TO ADD IMPORT
를 다음으로 바꿉니다.
from introvertally import call_agent_for_plan, post_plan_event
👉📝 InstaVibe에 프로토타입 기능을 추가합니다. ~/instavibe-bootstrap/instavibe/templates/base.html
에서 <!–REPLACE_ME_LINK_TO_INTROVERT_ALLY–>를 다음으로 바꿉니다.
<li class="nav-item">
<a class="nav-link" href="{{ url_for('ally.introvert_ally_page') }}">Introvert Ally</a>
</li>
InstaVibe 앱을 다시 배포하려면 Agent Engine에 배포한 Orchestrator 에이전트의 특정 Resource ID
가 필요합니다.
현재 gcloud
를 통해 프로그래매틱 방식으로 이를 검색하는 것이 제한적일 수 있으므로 도우미 Python 스크립트 (워크샵에 제공된 temp-endpoint.py
)를 사용하여 ID를 가져와 환경 변수에 저장합니다.
👉💻 다음 명령어를 실행하여 스크립트를 실행합니다. 이 스크립트는 Agent Engine Endpoint ID를 캡처하고 에이전트 엔진의 기본 서비스 계정에 필요한 권한을 부여합니다. 참고: 현재 사용자가 수정할 수 없으므로 기본 서비스 계정을 사용하도록 스크립트가 구성되어 있습니다.
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/instavibe/
source ~/instavibe-bootstrap/env/bin/activate
python temp-endpoint.py
export ORCHESTRATE_AGENT_ID=$(cat temp_endpoint.txt)
echo "ORCHESTRATE_AGENT_ID set to: ${ORCHESTRATE_AGENT_ID}"
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:service-$PROJECT_NUMBER@gcp-sa-aiplatform-re.iam.gserviceaccount.com" \
--role="roles/viewer"
마지막으로, Agent Engine에서 실행 중인 에이전트에 연결하는 방법을 알 수 있도록 업데이트된 코드와 새 ORCHESTRATE_AGENT_ID
환경 변수를 사용하여 InstaVibe 웹 애플리케이션을 다시 배포해야 합니다.
👉💻 다음 명령어는 InstaVibe 애플리케이션 이미지를 다시 빌드하고 새 버전을 Cloud Run에 배포합니다.
. ~/instavibe-bootstrap/set_env.sh
cd ~/instavibe-bootstrap/instavibe/
export IMAGE_TAG="latest"
export APP_FOLDER_NAME="instavibe"
export IMAGE_NAME="instavibe-webapp"
export IMAGE_PATH="${REGION}-docker.pkg.dev/${PROJECT_ID}/${REPO_NAME}/${IMAGE_NAME}:${IMAGE_TAG}"
export SERVICE_NAME="instavibe"
echo "Building ${APP_FOLDER_NAME} webapp image..."
gcloud builds submit . \
--tag=${IMAGE_PATH} \
--project=${PROJECT_ID}
echo "Deploying ${SERVICE_NAME} to Cloud Run..."
gcloud run deploy ${SERVICE_NAME} \
--image=${IMAGE_PATH} \
--platform=managed \
--region=${REGION} \
--allow-unauthenticated \
--set-env-vars="SPANNER_INSTANCE_ID=${SPANNER_INSTANCE_ID}" \
--set-env-vars="SPANNER_DATABASE_ID=${SPANNER_DATABASE_ID}" \
--set-env-vars="APP_HOST=0.0.0.0" \
--set-env-vars="APP_PORT=8080" \
--set-env-vars="GOOGLE_CLOUD_LOCATION=${REGION}" \
--set-env-vars="GOOGLE_CLOUD_PROJECT=${PROJECT_ID}" \
--set-env-vars="GOOGLE_MAPS_API_KEY=${GOOGLE_MAPS_API_KEY}" \
--set-env-vars="ORCHESTRATE_AGENT_ID=${ORCHESTRATE_AGENT_ID}" \
--project=${PROJECT_ID} \
--min-instances=1 \
--cpu=2 \
--memory=2Gi
최종 배포가 완료되면 다른 브라우저 탭에서 InstaVibe 애플리케이션 URL로 이동합니다.
AI 기반 InstaVibe 환경 전체 테스트
'InstaVibe Ally' 기능이 Vertex AI Agent Engine을 통해 조정되고 A2A를 통해 통신하는 멀티 에이전트 시스템을 기반으로 출시되었습니다.
'InstaVibe Ally'를 클릭하고 이벤트를 계획해 달라고 요청합니다.
상담사가 작업하는 동안 오른쪽의 활동 로그를 확인합니다 (90~120초가 걸릴 수 있음). 계획이 표시되면 검토한 후 '계획 확인'을 클릭하여 게시를 진행합니다.
이제 오케스트레이터가 플랫폼 에이전트에게 InstaVibe 내에서 게시물과 이벤트를 만들도록 지시합니다.
InstaVibe 홈페이지에서 새 게시물과 이벤트를 확인합니다.
이벤트 페이지에는 상담사가 생성한 세부정보가 반영됩니다.
Cloud Trace로 성능 분석
이 과정은 다소 시간이 걸릴 수 있습니다. Vertex AI Agent Engine은 Cloud Trace와 통합되어 멀티 에이전트 시스템의 지연 시간을 분석할 수 있습니다.
Google Cloud 콘솔에서 Traces로 이동하여 Span에서 agent_run[orchestrate_agent]
를 선택합니다. 몇 개의 Span이 표시되면 클릭합니다.
트레이스 세부정보에서 시간이 더 오래 걸린 부분을 식별할 수 있습니다. 예를 들어 Planner 에이전트 호출은 검색 접지 및 복잡한 생성으로 인해 지연 시간이 더 길어질 수 있습니다.
마찬가지로 게시물과 이벤트를 만들 때 Orchestrator가 데이터를 처리하고 Platform 에이전트의 도구 호출을 준비하는 데 소비한 시간이 표시될 수 있습니다.
이러한 트레이스를 살펴보면 상담사 시스템의 성능을 파악하고 최적화하는 데 도움이 됩니다.
축하합니다. Google의 ADK, A2A, MCP, Google Cloud 서비스를 사용하여 정교한 멀티 에이전트 AI 시스템을 빌드, 배포, 테스트했습니다. 상담사 조정, 도구 사용, 상태 관리, 클라우드 배포를 처리하여 InstaVibe용으로 작동하는 AI 기반 기능을 만들었습니다. 워크샵을 완료해 주셔서 감사합니다.
13. 삭제
Google Cloud 계정에 계속 요금이 청구되지 않도록 하려면 이 워크샵에서 만든 리소스를 삭제하는 것이 중요합니다. 다음 명령어를 사용하면 Spanner 인스턴스, Cloud Run 서비스, 아티팩트 레지스트리 저장소, API 키, Vertex AI Agent Engine, 연결된 IAM 권한을 삭제할 수 있습니다.
중요:
- 워크숍에 사용된 것과 동일한 Google Cloud 프로젝트에서 이러한 명령어를 실행해야 합니다.
- Cloud Shell 터미널을 닫은 경우 $PROJECT_ID, $SPANNER_INSTANCE_ID와 같은 일부 환경 변수가 설정되지 않을 수 있습니다. 워크숍 설정 단계에서와 같이 다시 내보내거나 아래 명령어의 변수를 실제 값으로 바꿔야 합니다.
- 이 명령어는 리소스를 영구적으로 삭제합니다. 이 프로젝트에 다른 중요한 데이터가 있는지 실행하기 전에 다시 한번 확인하세요.
👉💻 다음 스크립트를 실행하여 정리합니다.
환경 변수 재설정
. ~/instavibe-bootstrap/set_env.sh
에이전트 엔진 삭제:
cd ~/instavibe-bootstrap/utils
source ~/instavibe-bootstrap/env/bin/activate
export ORCHESTRATE_AGENT_ID=$(cat ~/instavibe-bootstrap/instavibe/temp_endpoint.txt)
echo "ORCHESTRATE_AGENT_ID set to: ${ORCHESTRATE_AGENT_ID}"
python remote_delete.py
deactivate
echo "Vertex AI Agent Engine deletion initiated."
Cloud Run 서비스를 삭제합니다.
# InstaVibe Web Application
gcloud run services delete instavibe --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
# MCP Tool Server
gcloud run services delete mcp-tool-server --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
# Planner Agent (A2A Server)
gcloud run services delete planner-agent --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
# Platform MCP Client Agent (A2A Server)
gcloud run services delete platform-mcp-client --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
# Social Agent (A2A Server)
gcloud run services delete social-agent --platform=managed --region=${REGION} --project=${PROJECT_ID} --quiet
echo "Cloud Run services deletion initiated."
A2A Inspector Docker 컨테이너 중지 및 삭제
docker rm --force a2a-inspector
Spanner 인스턴스 삭제:
echo "Deleting Spanner instance: ${SPANNER_INSTANCE_ID}..."
gcloud spanner instances delete ${SPANNER_INSTANCE_ID} --project=${PROJECT_ID} --quiet
echo "Spanner instance deletion initiated."
Artifact Registry 저장소 삭제:
echo "Deleting Artifact Registry repository: ${REPO_NAME}..."
gcloud artifacts repositories delete ${REPO_NAME} --location=${REGION} --project=${PROJECT_ID} --quiet
echo "Artifact Registry repository deletion initiated."
서비스 계정에서 역할 삭제:
echo "Removing roles from service account: $SERVICE_ACCOUNT_NAME in project $PROJECT_ID"
# Remove Project-level roles for default service account
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/spanner.admin"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/spanner.databaseUser"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/artifactregistry.admin"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/cloudbuild.builds.editor"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/run.admin"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/iam.serviceAccountUser"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/aiplatform.user"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/logging.logWriter"
gcloud projects remove-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$SERVICE_ACCOUNT_NAME" \
--role="roles/logging.viewer"
echo "All specified roles have been removed."
로컬 워크숍 파일 삭제:
echo "Removing local workshop directory ~/instavibe-bootstrap..."
rm -rf ~/instavibe-bootstrap
rm -rf ~/a2a-inspector
rm -f ~/mapkey.txt
rm -f ~/project_id.txt
echo "Local directory removed."