1. 개요
이 Codelab에서는 공급망 오케스트레이터 에이전트를 빌드합니다. 이 애플리케이션을 사용하면 자연어를 사용하여 인벤토리를 분석하고, 물류를 추적하고, 공급망 위험을 관리할 수 있습니다.
Google의 에이전트 개발 키트 (ADK)를 활용하여 컨텍스트를 유지하고, Vertex AI 메모리 뱅크를 통해 사용자 환경설정을 기억하며, MCP Toolbox를 통해 AlloyDB에 저장된 대규모 데이터 세트와 상호작용하는 다중 에이전트 아키텍처를 빌드합니다.
빌드할 항목
다음으로 구성된 Python Flask 애플리케이션
글로벌 오케스트레이터 에이전트: 대화 흐름과 위임을 관리하는 루트 에이전트입니다.
전문가 에이전트: 도메인별 작업을 위한 'InventorySpecialist' 및 'LogisticsManager'입니다.
메모리 통합: Vertex AI 메모리 뱅크를 사용하는 단기 세션 메모리 및 장기 메모리
내러티브 UI: 에이전트의 추론 프로세스 (추적 컨텍스트)를 시각화하는 웹 인터페이스입니다.
학습할 내용
- Google ADK를 사용하여 전문 에이전트 및 하위 에이전트를 만드는 방법
- 장기 에이전트 메모리를 위해 Vertex AI 메모리 뱅크를 통합하는 방법
- MCP 도구 상자를 사용하여 에이전트를 AlloyDB 데이터 도구에 연결하는 방법
- ADK 콜백을 구현하여 에이전트 추론을 추적하고 시각화하는 방법
- Cloud Run을 사용하여 솔루션을 배포하거나 로컬로 실행하는 방법
아키텍처
기술 스택
- PostgreSQL용 AlloyDB: 50,000개 이상의 공급망 레코드를 보유하는 고성능 운영 데이터베이스 역할을 합니다. 벡터 검색 및 검색을 지원합니다.
- 데이터베이스용 MCP 도구 상자: '오케스트레이션 마에스트로' 역할을 하며, 에이전트가 호출할 수 있는 실행 가능한 도구로 AlloyDB 데이터를 노출합니다.
- 에이전트 개발 키트 (ADK): 에이전트, 요청 사항, 도구를 정의하는 데 사용되는 프레임워크입니다.
- Vertex AI 메모리 뱅크: 장기 메모리를 제공하여 에이전트가 세션 전반에서 사용자 환경설정과 이전 상호작용을 기억할 수 있습니다.
- Vertex AI 세션 서비스: 단기 대화 컨텍스트를 관리합니다.
흐름
- 사용자 질문: 사용자가 질문을 합니다 (예: '프리미엄 아이스크림 재고 확인').
- 메모리 확인: 오케스트레이터가 메모리 뱅크에서 관련 과거 정보 (예: '사용자는 EMEA 지역 관리자임')를 확인합니다.
- 위임: 조정자가 InventorySpecialist에게 작업을 위임합니다.
- 도구 실행: 전문가가 MCP 도구 상자에서 제공하는 도구를 사용하여 AlloyDB를 쿼리합니다.
- 대답: 에이전트가 데이터를 처리하고 마크다운 형식의 표를 반환합니다.
- 메모리 저장소: 중요한 상호작용은 메모리 뱅크에 다시 저장됩니다.
요구사항
2. 시작하기 전에
프로젝트 만들기
- Google Cloud 콘솔의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.
- Cloud 프로젝트에 결제가 사용 설정되어 있어야 하므로 프로젝트에 결제가 사용 설정되어 있는지 확인하는 방법을 알아보세요.
- Google Cloud에서 실행되는 명령줄 환경인 Cloud Shell을 사용합니다. Google Cloud 콘솔 상단에서 Cloud Shell 활성화를 클릭합니다.
- Cloud Shell에 연결되면 다음 명령어를 사용하여 이미 인증되었는지, 프로젝트가 프로젝트 ID로 설정되었는지 확인합니다.
gcloud auth list
- Cloud Shell에서 다음 명령어를 실행하여 gcloud 명령어가 프로젝트를 알고 있는지 확인합니다.
gcloud config list project
- 프로젝트가 설정되지 않은 경우 다음 명령어를 사용하여 설정합니다.
gcloud config set project <YOUR_PROJECT_ID>
- 필요한 API 사용 설정: 링크를 따라 API를 사용 설정합니다.
또는 gcloud 명령어를 사용할 수 있습니다. gcloud 명령어 및 사용법은 문서를 참조하세요.
주의사항 및 문제 해결
'유령 프로젝트' 증후군 |
|
결제 바리케이드 | 프로젝트를 사용 설정했지만 결제 계정을 잊었습니다. AlloyDB는 고성능 엔진이므로 '연료 탱크' (결제)가 비어 있으면 시작되지 않습니다. |
API 전파 지연 | 'API 사용 설정'을 클릭했지만 명령줄에 여전히 |
할당량 질문 | 새 체험판 계정을 사용하는 경우 AlloyDB 인스턴스의 리전별 할당량에 도달할 수 있습니다. |
'숨겨진' 서비스 에이전트 | AlloyDB 서비스 에이전트에는 |
3. 데이터베이스 설정
애플리케이션의 핵심은 PostgreSQL용 AlloyDB입니다. 강력한 벡터 기능을 활용하고 열 기반 엔진을 통합하여 50,000개 이상의 SCM 레코드의 삽입을 생성했습니다. 이를 통해 거의 실시간 벡터 분석이 가능해져 상담사가 대규모 데이터 세트에서 몇 밀리초 만에 인벤토리 이상치 또는 물류 위험을 식별할 수 있습니다.
이 실습에서는 AlloyDB를 테스트 데이터의 데이터베이스로 사용합니다. 클러스터를 사용하여 데이터베이스, 로그와 같은 모든 리소스를 보유합니다. 각 클러스터에는 데이터에 대한 액세스 포인트를 제공하는 기본 인스턴스가 있습니다. 테이블에는 실제 데이터가 저장됩니다.
테스트 데이터 세트가 로드될 AlloyDB 클러스터, 인스턴스, 테이블을 만들어 보겠습니다.
- 아래 버튼을 클릭하거나 Google Cloud 콘솔 사용자가 로그인한 브라우저에 링크를 복사합니다.
또는 결제 계정을 사용한 프로젝트에서 Cloud Shell 터미널로 이동하여 GitHub 저장소 를 클론하고 아래 명령어를 사용하여 프로젝트로 이동합니다.
git clone https://github.com/AbiramiSukumaran/easy-alloydb-setup
cd easy-alloydb-setup
- 이 단계를 완료하면 저장소가 로컬 Cloud Shell 편집기에 복제되며 프로젝트 폴더에서 아래 명령어를 실행할 수 있습니다 (프로젝트 디렉터리에 있는지 확인하는 것이 중요함).
sh run.sh
- 이제 UI를 사용하여 터미널에서 링크를 클릭하거나 터미널에서 '웹에서 미리보기' 링크를 클릭합니다.
- 시작하려면 프로젝트 ID, 클러스터, 인스턴스 이름을 입력하세요.
- 로그가 스크롤되는 동안 커피를 마시세요. 여기에서 백그라운드에서 이 작업이 어떻게 이루어지는지 확인할 수 있습니다.
주의사항 및 문제 해결
'인내심' 문제 | 데이터베이스 클러스터는 무거운 인프라입니다. 페이지를 새로고침하거나 '멈춘 것 같아서' Cloud Shell 세션을 종료하면 부분적으로 프로비저닝되어 수동 개입 없이는 삭제할 수 없는 '고스트' 인스턴스가 생성될 수 있습니다. |
지역 불일치 |
|
좀비 클러스터 | 이전에 클러스터에 동일한 이름을 사용했고 삭제하지 않은 경우 스크립트에서 클러스터 이름이 이미 있다고 표시할 수 있습니다. 클러스터 이름은 프로젝트 내에서 고유해야 합니다. |
Cloud Shell 시간 제한 | 커피를 마시는 데 30분이 걸리면 Cloud Shell이 절전 모드로 전환되어 |
4. 스키마 프로비저닝
AlloyDB 클러스터와 인스턴스가 실행되면 AlloyDB Studio SQL 편집기로 이동하여 AI 확장 프로그램을 사용 설정하고 스키마를 프로비저닝합니다.
인스턴스 생성이 완료될 때까지 기다려야 할 수 있습니다. 완료되면 클러스터를 만들 때 만든 사용자 인증 정보를 사용하여 AlloyDB에 로그인합니다. PostgreSQL에 인증하려면 다음 데이터를 사용하세요.
- 사용자 이름 : '
postgres' - 데이터베이스 : '
postgres' - 비밀번호 : '
alloydb' (또는 생성 시 설정한 비밀번호)
AlloyDB Studio에 인증되면 편집기에 SQL 명령어가 입력됩니다. 마지막 창 오른쪽에 있는 더하기 기호를 사용하여 여러 편집기 창을 추가할 수 있습니다.
필요에 따라 실행, 형식, 지우기 옵션을 사용하여 편집기 창에 AlloyDB 명령어를 입력합니다.
확장 프로그램 사용 설정
이 앱을 빌드하기 위해 확장 프로그램 pgvector 및 google_ml_integration를 사용합니다. pgvector 확장 프로그램을 사용하면 벡터 임베딩을 저장하고 검색할 수 있습니다. google_ml_integration 확장 프로그램은 SQL에서 예측을 수행하기 위해 Vertex AI 예측 엔드포인트에 액세스하는 데 사용하는 함수를 제공합니다. 다음 DDL을 실행하여 이러한 확장 프로그램을 사용 설정합니다.
CREATE EXTENSION IF NOT EXISTS google_ml_integration CASCADE;
CREATE EXTENSION IF NOT EXISTS vector;
테이블 만들기
AlloyDB Studio에서 아래 DDL 문을 사용하여 테이블을 만들 수 있습니다.
DROP TABLE IF EXISTS shipments;
DROP TABLE IF EXISTS products;
-- 1. Product Inventory Table
CREATE TABLE products (
id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
category VARCHAR(100),
stock_level INTEGER,
distribution_center VARCHAR(100),
region VARCHAR(50),
embedding vector(768),
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 2. Logistics & Shipments
CREATE TABLE shipments (
shipment_id SERIAL PRIMARY KEY,
product_id INTEGER REFERENCES products(id),
status VARCHAR(50), -- 'In Transit', 'Delayed', 'Delivered', 'Pending'
estimated_arrival TIMESTAMP,
route_efficiency_score DECIMAL(3, 2)
);
embedding 열은 일부 텍스트 필드의 벡터 값을 저장할 수 있습니다.
데이터 수집
아래 SQL 문 집합을 실행하여 제품 테이블에 50,000개의 레코드를 대량 삽입합니다.
-- We use a CROSS JOIN pattern with realistic naming segments to create meaningful variety
DO $$
DECLARE
brand_names TEXT[] := ARRAY['Artisan', 'Nature', 'Elite', 'Pure', 'Global', 'Eco', 'Velocity', 'Heritage', 'Aura', 'Summit'];
product_types TEXT[] := ARRAY['Ice Cream', 'Body Wash', 'Laundry Detergent', 'Shampoo', 'Mayonnaise', 'Deodorant', 'Tea', 'Soup', 'Face Cream', 'Soap'];
variants TEXT[] := ARRAY['Classic', 'Gold', 'Premium', 'Eco-Friendly', 'Organic', 'Night-Repair', 'Extra-Fresh', 'Zero-Sugar', 'Sensitive', 'Maximum-Strength'];
regions TEXT[] := ARRAY['EMEA', 'APAC', 'LATAM', 'NAMER'];
dcs TEXT[] := ARRAY['London-Hub', 'Mumbai-Central', 'Sao-Paulo-Logistics', 'Singapore-Port', 'Rotterdam-Gate', 'New-York-DC'];
BEGIN
INSERT INTO products (name, category, stock_level, distribution_center, region)
SELECT
b || ' ' || v || ' ' || t as name,
CASE
WHEN t IN ('Ice Cream', 'Mayonnaise', 'Tea', 'Soup') THEN 'Food & Refreshment'
WHEN t IN ('Body Wash', 'Shampoo', 'Deodorant', 'Face Cream', 'Soap') THEN 'Personal Care'
ELSE 'Home Care'
END as category,
floor(random() * 20000 + 100)::int as stock_level,
dcs[floor(random() * 6 + 1)] as distribution_center,
regions[floor(random() * 4 + 1)] as region
FROM
unnest(brand_names) b,
unnest(variants) v,
unnest(product_types) t,
generate_series(1, 50); -- 10 * 10 * 10 * 50 = 50,000 records
END $$;
데모 관련 레코드를 삽입하여 임원 스타일 질문에 대한 예측 가능한 답변을 보장합니다.
-- These ensure you have predictable answers for specific "Executive" questions
INSERT INTO products (name, category, stock_level, distribution_center, region) VALUES
('Magnum Ultra Gold Limited Edition', 'Food & Refreshment', 45, 'Rotterdam-Gate', 'EMEA'),
('Dove Pro-Health Deep Moisture', 'Personal Care', 12000, 'Mumbai-Central', 'APAC'),
('Hellmanns Real Organic Mayonnaise', 'Food & Refreshment', 8000, 'London-Hub', 'EMEA');
배송 데이터 삽입
-- Shipments Generation (More shipments than products)
INSERT INTO shipments (product_id, status, estimated_arrival, route_efficiency_score)
SELECT
id,
CASE
WHEN random() > 0.8 THEN 'Delayed'
WHEN random() > 0.4 THEN 'In Transit'
ELSE 'Delivered'
END,
NOW() + (random() * 10 || ' days')::interval,
(random() * 0.5 + 0.5)::decimal(3,2)
FROM products
WHERE random() > 0.3; -- Create shipments for ~70% of products
-- Add duplicate shipments for some products to show complex logistics
INSERT INTO shipments (product_id, status, estimated_arrival, route_efficiency_score)
SELECT id, 'In Transit', NOW() + INTERVAL '12 days', 0.88
FROM products
LIMIT 5000;
권한 부여
아래 문을 실행하여 'embedding' 함수에 대한 실행 권한을 부여합니다.
GRANT EXECUTE ON FUNCTION embedding TO postgres;
AlloyDB 서비스 계정에 Vertex AI 사용자 역할 부여
Google Cloud IAM 콘솔에서 AlloyDB 서비스 계정 (service-<<PROJECT_NUMBER>>@gcp-sa-alloydb.iam.gserviceaccount.com)에 'Vertex AI 사용자' 역할에 대한 액세스 권한을 부여합니다. PROJECT_NUMBER에는 프로젝트 번호가 표시됩니다.
또는 Cloud Shell 터미널에서 아래 명령어를 실행할 수 있습니다.
PROJECT_ID=$(gcloud config get-value project)
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:service-$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)")@gcp-sa-alloydb.iam.gserviceaccount.com" \
--role="roles/aiplatform.user"
임베딩 생성
다음으로 의미 있는 특정 텍스트 필드의 벡터 임베딩을 생성해 보겠습니다.
WITH
rows_to_update AS (
SELECT
id
FROM
products
WHERE
embedding IS NULL
LIMIT
5000 )
UPDATE
products
SET
embedding = ai.embedding('text-embedding-005', name || ' ' || category || ' ' || distribution_center || ' ' || region)::vector
FROM
rows_to_update
WHERE
products.id = rows_to_update.id
AND embedding IS null;
위의 문에서 한도를 5,000으로 설정했으므로 열 임베딩이 NULL인 행이 테이블에 없을 때까지 반복해서 실행하세요.
주의사항 및 문제 해결
'비밀번호 기억 상실증' 루프 | '원클릭' 설정을 사용했고 비밀번호가 기억나지 않는 경우 콘솔의 인스턴스 기본 정보 페이지로 이동하여 '수정'을 클릭하여 |
'확장 프로그램을 찾을 수 없음' 오류 |
|
IAM 전파 격차 |
|
벡터 차원 불일치 |
|
프로젝트 ID 오타 |
|
5. 도구 및 도구 상자 설정
데이터베이스용 MCP 도구 상자는 데이터베이스용 오픈소스 MCP 서버입니다. 연결 풀링, 인증, 기타와 같은 복잡한 작업을 처리하여 더 쉽고 빠르고 안전하게 도구를 개발할 수 있습니다. 도구 상자를 사용하면 에이전트가 데이터베이스의 데이터에 액세스할 수 있는 생성형 AI 도구를 빌드할 수 있습니다.
데이터베이스용 모델 컨텍스트 프로토콜 (MCP) 도구 상자를 '컨덕터'로 사용합니다. 에이전트와 AlloyDB 간의 표준화된 미들웨어 역할을 합니다. tools.yaml 구성을 정의하면 툴박스에서 복잡한 데이터베이스 작업을 search_products_by_context 또는 check_inventory_levels와 같은 깔끔한 실행 가능한 도구로 자동 노출합니다. 이렇게 하면 에이전트 로직 내에서 수동 연결 풀링이나 상용구 SQL이 필요하지 않습니다.
Toolbox 서버 설치
Cloud Shell 터미널에서 새 도구 yaml 파일과 툴박스 바이너리를 저장할 폴더를 만듭니다.
mkdir scm-agent-toolbox
cd scm-agent-toolbox
새 폴더 내에서 다음 명령어 집합을 실행합니다.
# see releases page for other versions
export VERSION=0.27.0
curl -L -o toolbox https://storage.googleapis.com/genai-toolbox/v$VERSION/linux/amd64/toolbox
chmod +x toolbox
그런 다음 Cloud Shell 편집기로 이동하여 새 폴더 내에 tools.yaml 파일을 만들고 이 저장소 파일의 콘텐츠를 tools.yaml 파일에 복사합니다.
sources:
supply_chain_db:
kind: "alloydb-postgres"
project: "YOUR_PROJECT_ID"
region: "us-central1"
cluster: "YOUR_CLUSTER"
instance: "YOUR_INSTANCE"
database: "postgres"
user: "postgres"
password: "YOUR_PASSWORD"
tools:
search_products_by_context:
kind: postgres-sql
source: supply_chain_db
description: Find products in the inventory using natural language search and vector embeddings.
parameters:
- name: search_text
type: string
description: Description of the product or category the user is looking for.
statement: |
SELECT name, category, stock_level, distribution_center, region
FROM products
ORDER BY embedding <=> ai.embedding('text-embedding-005', $1)::vector
LIMIT 5;
check_inventory_levels:
kind: postgres-sql
source: supply_chain_db
description: Get precise stock levels for a specific product name.
parameters:
- name: product_name
type: string
description: The exact or partial name of the product.
statement: |
SELECT name, stock_level, distribution_center, last_updated
FROM products
WHERE name ILIKE '%' || $1 || '%'
ORDER BY stock_level DESC;
track_shipment_status:
kind: postgres-sql
source: supply_chain_db
description: Retrieve real-time logistics and shipping status for a specific region or product.
parameters:
- name: region
type: string
description: The geographical region to filter shipments (e.g., EMEA, APAC).
statement: |
SELECT p.name, s.status, s.estimated_arrival, s.route_efficiency_score
FROM shipments s
JOIN products p ON s.product_id = p.id
WHERE p.region = $1
ORDER BY s.estimated_arrival ASC;
analyze_supply_chain_risk:
kind: postgres-sql
source: supply_chain_db
description: Rerank and filter shipments based on risk profiles and efficiency scores using Google ML reranker.
parameters:
- name: risk_context
type: string
description: The business context for risk analysis (e.g., 'heatwave impact' or 'port strike').
statement: |
WITH initial_ranking AS (
SELECT s.shipment_id, p.name, s.status, p.distribution_center,
ROW_NUMBER() OVER () AS ref_number
FROM shipments s
JOIN products p ON s.product_id = p.id
WHERE s.status != 'Delivered'
LIMIT 10
),
reranked_results AS (
SELECT index, score FROM
ai.rank(
model_id => 'semantic-ranker-default-003',
search_string => $1,
documents => (SELECT ARRAY_AGG(name || ' at ' || distribution_center ORDER BY ref_number) FROM initial_ranking)
)
)
SELECT i.name, i.status, i.distribution_center, r.score
FROM initial_ranking i, reranked_results r
WHERE i.ref_number = r.index
ORDER BY r.score DESC;
toolsets:
supply_chain_toolset:
- search_products_by_context
- check_inventory_levels
- track_shipment_status
- analyze_supply_chain_risk
이제 로컬 서버에서 tools.yaml 파일을 테스트합니다.
./toolbox --tools-file "tools.yaml"
UI에서 테스트할 수도 있습니다.
./toolbox --ui
완벽합니다. 이 모든 항목이 작동하는지 확인한 후 다음과 같이 Cloud Run에 배포합니다.
Cloud Run 배포
- PROJECT_ID 환경 변수를 설정합니다.
export PROJECT_ID="my-project-id"
- gcloud CLI를 초기화합니다.
gcloud init
gcloud config set project $PROJECT_ID
- 다음 API가 사용 설정되어 있어야 합니다.
gcloud services enable run.googleapis.com \
cloudbuild.googleapis.com \
artifactregistry.googleapis.com \
iam.googleapis.com \
secretmanager.googleapis.com
- 아직 백엔드 서비스 계정이 없는 경우 다음 단계를 따라 만듭니다.
gcloud iam service-accounts create toolbox-identity
- 보안 비밀 관리자 사용 권한을 부여합니다.
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member serviceAccount:toolbox-identity@$PROJECT_ID.iam.gserviceaccount.com \
--role roles/secretmanager.secretAccessor
- AlloyDB 소스 (roles/alloydb.client 및 roles/serviceusage.serviceUsageConsumer)에 특정한 서비스 계정에 추가 권한 부여
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member serviceAccount:toolbox-identity@$PROJECT_ID.iam.gserviceaccount.com \
--role roles/alloydb.client
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member serviceAccount:toolbox-identity@$PROJECT_ID.iam.gserviceaccount.com \
--role serviceusage.serviceUsageConsumer
- tools.yaml을 보안 비밀로 업로드합니다.
gcloud secrets create tools-scm-agent --data-file=tools.yaml
- 이미 보안 비밀이 있고 보안 비밀 버전을 업데이트하려면 다음을 실행하세요.
gcloud secrets versions add tools-scm-agent --data-file=tools.yaml
- Cloud Run에 사용할 컨테이너 이미지에 환경 변수를 설정합니다.
export IMAGE=us-central1-docker.pkg.dev/database-toolbox/toolbox/toolbox:latest
- 다음 명령어를 사용하여 Cloud Run에 도구 상자를 배포합니다.
AlloyDB에서 공개 액세스를 사용 설정한 경우 (권장하지 않음) Cloud Run에 배포하려면 아래 명령어를 따르세요.
gcloud run deploy toolbox-scm-agent \
--image $IMAGE \
--service-account toolbox-identity \
--region us-central1 \
--set-secrets "/app/tools.yaml=tools-scm-agent:latest" \
--args="--tools-file=/app/tools.yaml","--address=0.0.0.0","--port=8080" \
--allow-unauthenticated
VPC 네트워크를 사용하는 경우 아래 명령어를 사용합니다.
gcloud run deploy toolbox-scm-agent \
--image $IMAGE \
--service-account toolbox-identity \
--region us-central1 \
--set-secrets "/app/tools.yaml=tools-scm-agent:latest" \
--args="--tools-file=/app/tools.yaml","--address=0.0.0.0","--port=8080" \
# TODO(dev): update the following to match your VPC details
--network <<YOUR_NETWORK_NAME>> \
--subnet <<YOUR_SUBNET_NAME>> \
--allow-unauthenticated
6. 에이전트 설정
에이전트 개발 키트 (ADK)를 사용하여 모놀리식 프롬프트에서 전문화된 멀티 에이전트 아키텍처로 전환했습니다.
- InventorySpecialist: 제품 재고 및 창고 측정항목에 중점을 둡니다.
- LogisticsManager: 글로벌 배송 경로 및 위험 분석 전문가.
- GlobalOrchestrator: 추론을 사용하여 작업을 위임하고 결과를 종합하는 '브레인'입니다.
이 저장소를 프로젝트에 클론하고 살펴보겠습니다.
이를 클론하려면 Cloud Shell 터미널 (루트 디렉터리 또는 이 프로젝트를 만들려는 위치)에서 다음 명령어를 실행합니다.
git clone https://github.com/AbiramiSukumaran/scm-memory-agent
- 이렇게 하면 프로젝트가 생성되며 Cloud Shell 편집기에서 이를 확인할 수 있습니다.
- 프로젝트 및 인스턴스의 값으로 .env 파일을 업데이트해야 합니다.
코드 둘러보기
Orchestrator 에이전트 간단히 살펴보기
Go to app.py and you should be able to see the following snippet:
orchestrator = adk.Agent(
name="GlobalOrchestrator",
model="gemini-2.5-flash",
description="Global Supply Chain Orchestrator root agent.",
instruction="""
You are the Global Supply Chain Brain. You are responsible for products, inventory and logistics.
You also have access to the memory tool, remember to include all the information that the tool can provide you with about the user before you respond.
1. Understand intent and delegate to specialists. As the Global Orchestrator, you have access to the full conversation history with the user.
When you transfer a query to a specialist agent, sub agent or tool, share the important facts and information from your memory to them so they can operate with the full context.
2. Ensure the final response is professional and uses Markdown tables for data.
3. If a specialist provides a long list, ensure only the top 10 items are shown initially.
4. Conclude with a brief, high-level executive summary of what the data implies.
""",
tools=[adk.tools.preload_memory_tool.PreloadMemoryTool()],
sub_agents=[inventory_agent, logistics_agent],
#after_agent_callback=auto_save_session_to_memory_callback,
)
이 스니펫은 사용자의 대화나 요청을 수신하고 작업에 따라 해당 하위 에이전트나 사용자에게 해당 도구를 라우팅하는 오케스트레이터 에이전트인 루트의 정의입니다.
- 인벤토리 에이전트를 살펴보겠습니다.
inventory_agent = adk.Agent(
name="InventorySpecialist",
model="gemini-2.5-flash",
description="Specialist in product stock and warehouse data.",
instruction="""
Analyze inventory levels.
1. Use 'search_products_by_context' or 'check_inventory_levels'.
2. ALWAYS format results as a clean Markdown table.
3. If there are many results, display only the TOP 10 most relevant ones.
4. At the end, state: 'There are additional records available. Would you like to see more?'
""",
tools=tools
)
이 특정 하위 에이전트는 제품을 맥락에 맞게 검색하고 재고 수준을 확인하는 등의 인벤토리 활동에 특화되어 있습니다.
- 다음은 물류 하위 에이전트입니다.
logistics_agent = adk.Agent(
name="LogisticsManager",
model="gemini-2.5-flash",
description="Expert in global shipping routes and logistics tracking.",
instruction="""
Check shipment statuses.
1. Use 'track_shipment_status' or 'analyze_supply_chain_risk'.
2. ALWAYS format results as a clean Markdown table.
3. Limit initial output to the top 10 shipments.
4. Ask if the user needs the full manifest if more results exist.
""",
tools=tools
)
이 특정 하위 에이전트는 배송 추적, 공급망의 위험 분석과 같은 물류 활동을 전문으로 합니다.
- 지금까지 설명한 3가지 에이전트는 모두 도구를 사용하며, 도구는 이전 섹션에서 이미 배포한 Toolbox 서버를 통해 참조됩니다. 아래 스니펫을 참고하세요.
from toolbox_core import ToolboxSyncClient
TOOLBOX_SERVER = os.environ["TOOLBOX_SERVER"]
TOOLBOX_TOOLSET = os.environ["TOOLBOX_TOOLSET"]
# --- ADK TOOLBOX CONFIGURATION ---
toolbox = ToolboxSyncClient(TOOLBOX_SERVER)
tools = toolbox.load_toolset(TOOLBOX_TOOLSET)
이 특정 하위 에이전트는 배송 추적, 공급망의 위험 분석과 같은 물류 활동을 전문으로 합니다.
7. 에이전트 엔진
초기 실행에서 Agent Engine 만들기
import vertexai
GOOGLE_CLOUD_PROJECT = os.environ["GOOGLE_CLOUD_PROJECT"]
GOOGLE_CLOUD_LOCATION = os.environ["GOOGLE_CLOUD_LOCATION"]
client = vertexai.Client(
project=GOOGLE_CLOUD_PROJECT,
location=GOOGLE_CLOUD_LOCATION
)
agent_engine = client.agent_engines.create()
- 다음 실행을 위해 메모리 뱅크 구성으로 Agent Engine을 업데이트합니다.
agent_engine = client.agent_engines.update(
name=APP_NAME,
config={
"context_spec": {
"memory_bank_config": {
"generation_config": {
"model": f"projects/{PROJECT_ID}/locations/{GOOGLE_CLOUD_LOCATION}/publishers/google/models/gemini-2.5-flash"
}
}
}
})
8. 컨텍스트, 실행, 메모리
컨텍스트 관리는 상담사가 상태 비저장 봇이 아닌 지속적인 파트너처럼 느껴지도록 두 개의 별도 레이어로 나뉩니다.
단기 기억 (세션): VertexAiSessionService를 통해 관리되며 단일 상호작용 내에서 즉각적인 이벤트 기록 (사용자 메시지, 도구 응답)을 추적합니다.
장기 기억 (Memory Bank): adk.memorybankservice를 통해 Vertex AI Memory Bank로 구동됩니다. 이 레이어는 특정 운송업체에 대한 사용자의 선호도 또는 반복되는 창고 지연과 같은 '의미 있는' 정보를 추출하고 세션 전반에 걸쳐 유지합니다.
대화 범위 내에서 세션 메모리의 세션 초기화
현재 사용자의 현재 앱에 대한 세션을 만드는 스니펫의 부분입니다.
from google.adk.sessions import VertexAiSessionService
...
session_service = VertexAiSessionService(
project=PROJECT_ID,
location=GOOGLE_CLOUD_LOCATION,
)
...
# Initialize the session *outside* of the route handler to avoid repeated creation
session = None
session_lock = threading.Lock()
async def initialize_session():
global session
try:
session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID)
print(f"Session {session.id} created successfully.") # Add a log
except Exception as e:
print(f"Error creating session: {e}")
session = None # Ensure session is None in case of error
# Create the session on app startup
asyncio.run(initialize_session())
장기 메모리를 위해 Vertex AI 메모리 뱅크 초기화
이는 에이전트 엔진의 Vertex AI 메모리 뱅크 서비스 객체를 인스턴스화하는 스니펫의 부분입니다.
from google.adk.memory import InMemoryMemoryService
from google.adk.memory import VertexAiMemoryBankService
...
try:
memory_bank_service = adk.memory.VertexAiMemoryBankService(
agent_engine_id=AGENT_ENGINE_ID,
project=PROJECT_ID,
location=GOOGLE_CLOUD_LOCATION,
)
#in_memory_service = InMemoryMemoryService()
print("Memory Bank Service initialized successfully.")
except Exception as e:
print(f"Error initializing Memory Bank Service: {e}")
memory_bank_service = None
runner = adk.Runner(
agent=orchestrator,
app_name=APP_NAME,
session_service=session_service,
memory_service=memory_bank_service,
)
...
무엇이 구성되나요?
스니펫의 이 부분에서는 장기 기억을 위해 Vertex AI 메모리 뱅크 서비스를 구성합니다. 특정 사용자의 특정 앱에 대한 세션을 Vertex AI 메모리 뱅크 내에 메모리로 문맥에 맞게 저장합니다.
에이전트 실행의 일부로 실행되는 것은 무엇인가요?
async def run_and_collect():
final_text = ""
try:
async for event in runner.run_async(
new_message=content,
user_id=user_id,
session_id=session_id
):
if hasattr(event, 'author') and event.author:
if not any(log['agent'] == event.author for log in execution_logs):
execution_logs.append({
"agent": event.author,
"action": "Analyzing data requirements...",
"type": "orchestration_event"
})
if hasattr(event, 'text') and event.text:
final_text = event.text
elif hasattr(event, 'content') and hasattr(event.content, 'parts'):
for part in event.content.parts:
if hasattr(part, 'text') and part.text:
final_text = part.text
except Exception as e:
print(f"Error during runner.run_async: {e}")
raise # Re-raise the exception to signal failure
finally:
gc.collect()
return final_text
사용자의 입력 콘텐츠를 범위 내에 사용자 ID와 세션 ID가 있는 new_message 객체로 처리합니다. 그러면 에이전트가 인계받아 에이전트 응답이 처리되고 반환됩니다.
장기 메모리에 저장되는 내용은 무엇인가요?
앱 및 사용자 범위의 세션 세부정보는 세션 변수에 추출됩니다.
그런 다음 이 세션은 'add_session_to_memory' 메서드를 사용하여 Vertex AI 메모리 뱅크 객체의 현재 앱에 대한 현재 사용자의 메모리로 추가됩니다.
session = asyncio.run(session_service.get_session(app_name=APP_NAME, user_id=USER_ID, session_id=session.id))
if memory_bank_service and session: # Check memory service AND session
try:
#asyncio.run(in_memory_service.add_session_to_memory(session))
asyncio.run(memory_bank_service.add_session_to_memory(session))
'''
client.agent_engines.memories.generate(
scope={"app_name": APP_NAME, "user_id": USER_ID},
name=APP_NAME,
direct_contents_source={
"events": [
{"content": content}
]
},
config={"wait_for_completion": True},
)
'''
print("Successfully added session to memory.******")
print(session.id)
except Exception as e:
print(f"Error adding session to memory: {e}")
메모리 검색
컨텍스트의 일부로 오케스트레이터 및 기타 에이전트에 전달하려면 앱 이름과 사용자 이름을 범위로 사용하여 저장된 장기 기억을 검색해야 합니다 (기억을 저장한 범위이기 때문).
results = client.agent_engines.memories.retrieve(
name=APP_NAME,
scope={"app_name": APP_NAME, "user_id": USER_ID}
)
# RetrieveMemories returns a pager. You can use `list` to retrieve all pages' memories.
list(results)
print(list(results))
검색된 메모리는 컨텍스트의 일부로 어떻게 로드되나요?
루트 에이전트가 메모리 뱅크에서 컨텍스트를 미리 로드할 수 있도록 Orchestrator 에이전트의 정의에 다음 속성을 사용합니다. 이는 하위 상담사를 위해 도구 상자 서버에서 액세스하는 도구 외에 추가로 제공되는 도구입니다.
tools=[adk.tools.preload_memory_tool.PreloadMemoryTool()],
콜백 컨텍스트
엔터프라이즈 공급망에는 '블랙박스'가 있을 수 없습니다. ADK의 CallbackContext를 사용하여 Narrative Engine을 만듭니다. 에이전트의 실행에 연결하여 모든 사고 과정과 도구 호출을 캡처하고 UI 사이드바로 스트리밍합니다.
- 추적 이벤트: 'GlobalOrchestrator가 데이터 요구사항을 분석하는 중...'
- 추적 이벤트: '재고 수준을 위해 InventorySpecialist에 위임...'
- 추적 이벤트: '메모리 뱅크에서 이전 공급업체 지연 패턴 가져오는 중...'
이 감사 추적은 디버깅에 매우 유용하며 인간 작업자가 에이전트의 자율적인 결정을 신뢰할 수 있도록 합니다.
from google.adk.agents.callback_context import CallbackContext
...
# --- ADK CALLBACKS (Narrative Engine) ---
execution_logs = []
async def trace_callback(context: CallbackContext):
"""
Captures agent and tool invocation flow for the UI narrative.
"""
agent_name = context.agent.name
event = {
"agent": agent_name,
"action": "Processing request steps...",
"type": "orchestration_event"
}
execution_logs.append(event)
return None
...
이제 끝났습니다. 프로젝트를 성공적으로 클론하고 에이전트, 메모리, 컨텍스트의 세부정보를 살펴봤습니다.
클론된 저장소의 프로젝트 폴더로 이동하여 다음 명령어를 실행하여 테스트할 수 있습니다.
>> pip install -r requirements.txt
>> python app.py
이렇게 하면 에이전트가 로컬로 시작되고 테스트할 수 있습니다.
9. Cloud Run에 배포해 보겠습니다.
- 프로젝트가 클론되고 프로젝트의 루트 폴더 내에 있는지 확인한 Cloud Shell 터미널에서 다음 명령어를 실행하여 Cloud Run에 배포합니다.
Cloud Shell 터미널에서 다음을 실행합니다.
gcloud run deploy supply-chain-agent --source . --platform managed --region us-central1 --allow-unauthenticated --set-env-vars GOOGLE_CLOUD_PROJECT=<<YOUR_PROJECT>>,GOOGLE_CLOUD_LOCATION=us-central1,GOOGLE_GENAI_USE_VERTEXAI=TRUE,REASONING_ENGINE_APP_NAME=<<YOUR_APP_ENGINE_URL>>,TOOLBOX_SERVER=<<YOUR_TOOLBOX_SERVER>>,TOOLBOX_TOOLSET=supply_chain_toolset,AGENT_ENGINE_ID=<<YOUR_AGENT_ENGINE_ID>>
자리표시자 <<YOUR_PROJECT>>, <<YOUR_APP_ENGINE_URL>>, <<YOUR_TOOLBOX_SERVER>> 및 <<YOUR_AGENT_ENGINE_ID>>의 값을 바꿉니다.
명령어가 완료되면 서비스 URL이 출력됩니다. 복사합니다.
- Cloud Run 서비스 계정에 AlloyDB 클라이언트 역할을 부여합니다.이렇게 하면 서버리스 애플리케이션이 데이터베이스로 안전하게 터널링할 수 있습니다.
Cloud Shell 터미널에서 다음을 실행합니다.
# 1. Get your Project ID and Project Number
PROJECT_ID=$(gcloud config get-value project)
PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format="value(projectNumber)")
# 2. Grant the AlloyDB Client role
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member="serviceAccount:$PROJECT_NUMBER-compute@developer.gserviceaccount.com" \
--role="roles/alloydb.client"
이제 서비스 URL (앞서 복사한 Cloud Run 엔드포인트)을 사용하여 앱을 테스트합니다.
참고: 서비스 문제가 발생하고 메모리가 이유로 언급된 경우 할당된 메모리 한도를 1GiB로 늘려 테스트해 보세요.
10. 삭제
이 실습을 완료한 후에는 AlloyDB 클러스터와 인스턴스를 삭제해야 합니다.
인스턴스와 함께 클러스터를 정리해야 합니다.
11. 축하합니다
AlloyDB의 속도, MCP 도구 상자의 오케스트레이션 효율성, Vertex AI Memory Bank의 '기관 메모리'를 결합하여 진화하는 공급망 시스템을 구축했습니다. 질문에 답할 뿐만 아니라 싱가포르 창고에서 항상 몬순 관련 지연이 발생한다는 점을 기억하고, 사용자가 묻기도 전에 배송 경로를 변경하도록 선제적으로 제안합니다.