Track 4. Custom Agent Solutions

import os from typing import Optional from google.adk.agents import Agent from google.adk.apps import App from google.adk.models import Gemini from google.cloud import bigquery from google.genai import types # 최신 Vertex AI 추론 모델 지정 (Gemini 3.6 Flash) MODEL_NAME = "gemini-3.6-flash" def get_table_schema(table_id: str) -> dict: """BigQuery 테이블의 컬럼 이름, 데이터 타입, 설명 등 스키마 정보를 동적으로 조회합니다.""" # [고객 환경 설정]: 환경 변수 GOOGLE_CLOUD_PROJECT 또는 실제 GCP 프로젝트 ID 주입 project_id = os.environ.get("GOOGLE_CLOUD_PROJECT", "YOUR_GCP_PROJECT_ID") try: client = bigquery.Client(project=project_id) table = client.get_table(table_id.strip().strip("`")) fields = [ {"name": f.name, "type": f.field_type, "mode": f.mode, "description": f.description or "설명 없음"} for f in table.schema ] return {"status": "SUCCESS", "table_id": table_id, "num_rows": table.num_rows, "fields": fields} except Exception as e: return {"status": "ERROR", "message": f"테이블 스키마 조회 실패 ({table_id}): {str(e)}"} def query_bigquery_table(sql_query: str) -> dict: """BigQuery 데이터셋에서 읽기 전용 SELECT 쿼리를 실행하고 실제 테이블 레코드를 반환합니다.""" # 1. 보안 가드레일: SELECT/WITH 읽기 전용 쿼리만 허용 clean_sql = sql_query.strip().upper() if not clean_sql.startswith("SELECT") and not clean_sql.startswith("WITH"): return {"error": "보안 정책상 SELECT 및 WITH 읽기 전용 쿼리만 실행할 수 있습니다."} # 2. BigQuery 클라이언트 초기화 및 쿼리 실행 project_id = os.environ.get("GOOGLE_CLOUD_PROJECT", "YOUR_GCP_PROJECT_ID") try: client = bigquery.Client(project=project_id) job = client.query(sql_query) rows = [dict(row) for row in job.result(max_results=50)] return {"status": "SUCCESS", "row_count": len(rows), "records": rows} except Exception as e: return {"status": "ERROR", "error_detail": str(e), "guidance": "BigQuery 테이블 경로를 정확히 지정하세요."} # 3. Google ADK Agent 정의 및 도구 바인딩 root_agent = Agent( name="bigquery_data_analyst", model=Gemini(model=MODEL_NAME, retry_options=types.HttpRetryOptions(attempts=3)), instruction="""당신은 Google Cloud BigQuery 기반의 엔터프라이즈 데이터 분석 전문가입니다. [동작 원칙] 1. 테이블 스키마 파악: 특정 테이블에 대한 분석 요청이 오면 먼저 `get_table_schema`를 호출하여 정확한 컬럼명과 타입을 확인하세요. 2. 표준 SQL 작성 및 실행: 스키마를 기반으로 표준 GoogleSQL SELECT 쿼리를 작성하여 `query_bigquery_table`을 호출하세요. - 공개 데이터셋: `bigquery-public-data.thelook_ecommerce.orders` - 사내 데이터셋: `YOUR_GCP_PROJECT_ID.retail_data.sales_records` 3. 분석 및 시각화 브리핑: 반환된 실제 쿼리 레코드를 바탕으로 마크다운 표와 3줄 핵심 인사이트(Key Takeaways)를 작성하세요. 4. 보안 가드레일: 읽기 전용(SELECT/WITH) 쿼리만 생성하며 DDL/DML은 절대 실행하지 않습니다.""", tools=[get_table_schema, query_bigquery_table], ) app = App(root_agent=root_agent, name="bigquery_analytics_app")

from google.adk.agents import Agent from google.adk.apps import App from google.adk.models import Gemini from google.genai import types # 최신 Vertex AI 추론 모델 지정 (Gemini 3.6 Flash) MODEL_NAME = "gemini-3.6-flash" def check_department_budget(department: str) -> dict: """부서별 당월 잔여 예산 한도를 조회합니다 (ERP 연동).""" # 실제 환경에서는 SAP, Oracle 등 사내 ERP/DB API와 연동됩니다. budgets = { "마케팅팀": {"total": 50000000, "spent": 32000000, "remaining": 18000000}, "개발1팀": {"total": 80000000, "spent": 45000000, "remaining": 35000000}, "영업본부": {"total": 60000000, "spent": 58000000, "remaining": 2000000}, } return budgets.get(department, {"total": 30000000, "spent": 10000000, "remaining": 20000000}) def evaluate_and_request_approval(department: str, amount: int, item_name: str, reason: str) -> dict: """지출 결재를 요청하고 전결 규정에 따라 자동 승인 또는 HITL 결재 티켓을 발급합니다.""" # 전결 규정 기준 금액: 3,000,000 KRW (300만 원) APPROVAL_THRESHOLD = 3000000 budget_info = check_department_budget(department) # 1. 예산 초과 검증 (Budget Guardrail) if amount > budget_info["remaining"]: return { "status": "REJECTED_BUDGET_EXCEEDED", "message": f"잔여 예산({budget_info['remaining']:,}원) 초과로 결재가 자동 반려되었습니다." } # 2. 전결 기준 초과 시 인간 관리자 승인 분기 (HITL) if amount > APPROVAL_THRESHOLD: ticket_id = f"APPR-2026-{abs(hash(item_name + reason)) % 10000:04d}" return { "status": "REQUIRES_HUMAN_APPROVAL", "ticket_id": ticket_id, "request_amount_krw": amount, "message": f"지출 금액({amount:,}원)이 전결 기준({APPROVAL_THRESHOLD:,}원)을 초과하여 부서장 결재 티켓이 발급되었습니다.", "approval_link": f"https://erp.enterprise-corp.com/approvals/{ticket_id}" } # 3. 전결 기준 이하 시 자동 승인 (Auto-Approval) return { "status": "AUTO_APPROVED", "ticket_id": f"AUTO-{abs(hash(item_name)) % 10000:04d}", "message": f"{department}의 {item_name} 지출({amount:,}원)이 전결 규정에 따라 자동 승인되었습니다." } root_agent = Agent( name="erp_approval_bot", model=Gemini(model=MODEL_NAME, retry_options=types.HttpRetryOptions(attempts=3)), instruction="""사내 전사적 자원 관리(ERP) 지출 결재 거버넌스 에이전트입니다. 1. 사용자의 요청에서 부서, 품목, 금액, 사유를 정확히 추출하세요. 2. `check_department_budget`으로 예산 잔액을 먼저 확인하세요. 3. `evaluate_and_request_approval` 도구를 호출하여 규정 준수 여부를 검토하세요. 4. 관리자 승인이 필요한 경우 승인 링크와 티켓 번호를 강조하여 안내하세요.""", tools=[check_department_budget, evaluate_and_request_approval], ) app = App(root_agent=root_agent, name="erp_approval_app")

from google.adk.agents import Agent from google.adk.apps import App from google.adk.models import Gemini from google.genai import types MODEL_NAME = "gemini-3.6-flash" def calculate_cagr(start_val: float, end_val: float, years: int) -> float: """초기값, 최종값, 경과 연수를 받아 연평균 복합 성장률(CAGR)을 계산합니다.""" if start_val <= 0 or years <= 0: return 0.0 return round((((end_val / start_val) ** (1.0 / years)) - 1.0) * 100.0, 2) # 1. 시장 조사 전담 서브 에이전트 (Market Researcher) researcher = Agent( name="market_researcher", model=Gemini(model=MODEL_NAME), instruction="글로벌 IT 및 클라우드 시장 동향, 경쟁 구도, 기회 요인을 구조화하여 요약하세요.", description="시장 조사 전담 서브 에이전트", ) # 2. 재무 지표 및 CAGR 계산 전담 서브 에이전트 (Financial Analyst) financial_analyst = Agent( name="financial_analyst", model=Gemini(model=MODEL_NAME), instruction="시장 데이터와 calculate_cagr 툴을 활용하여 연평균 성장률(CAGR)과 재무 지표를 계산하세요.", tools=[calculate_cagr], description="재무 및 성장성 계산 전담 서브 에이전트", ) # 3. 최상위 총괄 디렉터 (Supervisor Agent) strategy_director = Agent( name="strategy_director", model=Gemini(model=MODEL_NAME, retry_options=types.HttpRetryOptions(attempts=3)), instruction="""당신은 신사업 전략 기획 총괄 수석 디렉터입니다. 1. `market_researcher` 서브 에이전트로 시장 기회를 조사하세요. 2. `financial_analyst` 서브 에이전트로 성장률(CAGR)을 계산하세요. 3. 두 산출물을 종합하여 C-Level 보고용 '1-Page Strategy Blueprint'를 완성하세요.""", sub_agents=[researcher, financial_analyst], ) app = App(root_agent=strategy_director, name="strategy_orchestrator_app")

# 1. 환경 변수 설정 (고객 및 실습 프로젝트 환경에 맞게 입력) export GOOGLE_CLOUD_PROJECT="YOUR_GCP_PROJECT_ID" export REGION="us-central1" # Gemini Enterprise App 전체 리소스 ID (형식: projects/{PROJECT_NUM}/locations/global/collections/default_collection/engines/{APP_ID}) export GE_APP_ID="projects/YOUR_PROJECT_NUMBER/locations/global/collections/default_collection/engines/YOUR_APP_ID" # 2. Vertex AI Agent Runtime(Reasoning Engine) 컨테이너 배포 agents-cli deploy --project="${GOOGLE_CLOUD_PROJECT}" --region="${REGION}" --deployment-target=agent_runtime --service-name="track4-custom-agent" --no-confirm-project # 3. Gemini Enterprise 사내 앱에 ADK 네이티브 연동으로 커스텀 에이전트 등록 agents-cli publish gemini-enterprise --registration-type adk --agent-runtime-id "projects/YOUR_PROJECT_NUMBER/locations/us-central1/reasoningEngines/YOUR_REASONING_ENGINE_ID" --gemini-enterprise-app-id "${GE_APP_ID}" --display-name "Track 4 Custom Agent" --description "BigQuery 분석, HITL ERP 결재, Multi-Agent 협업 전담 커스텀 에이전트" --tool-description "BigQuery 데이터 분석, ERP 결재 승인, CAGR 성장률 계산"