Google ADK : Loop Agent · 평가 · API 서버 · SSE · Vertex AI 배포

# email-refiner/agent.py
from google.adk.agents import Agent, LoopAgent

email_refiner_agent = LoopAgent(
    name="Email Refiner Agent",
    description=EMAIL_OPTIMIZER_DESCRIPTION,
    sub_agents=[
        clarity_agent,
        tone_stylist_agent,
        persuation_agent,
        email_synthesizer_agent,
        literary_critic_agent,
    ],
)

root_agent = email_refiner_agent

# escalate라고 불리는 이유는 이게 더 상위 단계의 agent로 넘어갈 수 있기 때문.
# 정리하면 LoopAgent 내부에서 escalate를 쓰면 Loop가 종료. 만약 상위의 분류 에이전트가 있고 하위 에이전트가 많이 있는 계층적인 구조에서
# escalate를 사용하면 하위 에이전트에서 escalate를 사용해서 대화의 주도권을 상위 에이전트로 보냄.
async def escalate_email_complete(tool_context: ToolContext):
    """Use this tool only when the email is good to go."""
    tool_context.actions.escalate = True
    return "Email optimization complete."

# email-refiner/agent.py
email_refiner_agent = LoopAgent(
    name="EmailRefinerAgent",
    # 50 이상 loop를 돌 수 없음.
    max_iterations=50,
    description=EMAIL_OPTIMIZER_DESCRIPTION,
    sub_agents=[
        clarity_agent,
        tone_stylist_agent,
        persuation_agent,
        email_synthesizer_agent,
        literary_critic_agent,
    ],
)

import requests

BASE_URL = "http://127.0.0.1:8000"
APP_NAME = "travel_advisor_agent"
USER_ID = "u_123"
SESSOIN_ID = "25745dc0-e59d-487e-a53c-ba39814870bb"

# adk web 실행 시 session이 자동으로 생성되었었는데,
# 이를 API로 동일하게 구현하기 위해선 /apps/{app_name}/users/{user_id}/sessions 로 POST 요청
# -> session id 반환
response = requests.post(f"{BASE_URL}/apps/{APP_NAME}/users/{USER_ID}/sessions")
# {'id': '663fd92c-aced-4b8f-8cb7-32501ab0bf20', 'appName': 'travel_advisor_agent', 'userId': 'u_123', 'state': {}, 'events': [], 'lastUpdateTime': 1782737032.916954}
print(response.json())

# 다음으로는 run으로 agent 실행. (/run, /run_sse)
# 2가지 옵션이 있는데, 하나는 run이고 다른 하나는 run_sse임.

message = {
    "appName": APP_NAME,
    "userId": USER_ID,
    "sessionId": SESSOIN_ID,
    "newMessage": {
        "parts": [
            {"text": "I want to go to Tokyo"}
        ],
        "role": "user"
    }
}

response = requests.post(f"{BASE_URL}/run", json=message)

for event in response.json():
    content = event.get("content")
    parts = content.get("parts")
    for part in parts:
        function_call = part.get("functionCall", None)
        if function_call:
            print(function_call.get("name"))
        text = part.get("text", None)
        if text:
            print(text)

import sseclient
import json

response = requests.post(f"{BASE_URL}/run_sse", json=message, stream=True)

# SSEClient 생성
client = sseclient.SSEClient(response)

# 서버가 보내는 이벤트에 반복문
for event in client.events():
    data = json.loads(event.data)
    content = data.get("content")
    parts = data.get("parts")
    for part in parts:
        function_call = part.get("functionCall", None)
        if function_call:
            print(function_call.get("name"))
        text = part.get("text", None)
        if text:
            print(text)

# run으로 agent 실행 시 appName, userId, sessionId 등 아래와 같은 값이 필요.
message = {
    "appName": APP_NAME,
    "userId": USER_ID,
    "sessionId": SESSOIN_ID,
    "newMessage": {
        "parts": [
            {"text": "I want to go to Tokyo"}
        ],
        "role": "user"
    },
    "streaming": True
}

from google.adk.agents import Agent
from google.adk.models.lite_llm import LiteLlm
from google.adk.tools.tool_context import ToolContext

from google.adk.sessions import DatabaseSessionService  # OpenAI Agent SDK에 있는 SQLiteSession과 비슷한 역할.
from google.adk.runners import Runner
from google.genai import types

# SessionService 생성
session_service = DatabaseSessionService(db_url="sqlite+aiosqlite:///./session.db")

# session 생성
session = await session_service.create_session(
    app_name="weather_agent",
    user_id="u_123",
    state={
        "user_name": "toma"
    }
)

MODEL = LiteLlm(model="openai/gpt-4o")


async def get_weather(tool_context: ToolContext, location: str):
 ...

async def get_exchange_rate(
    tool_context: ToolContext, from_currency: str, to_currency: str, amount: float
):
 ...


travel_advisor_agent = Agent(
 ...
)

# Runner 생성 -> agent 전달
runner = Runner(
    agent=travel_advisor_agent,
    session_service=session_service,
    app_name="weather_agent"
)

# ADK의 메세지 형식(규칙)
message = types.Content(role="user", parts=[types.Part(text="I'm going to Vietnam, tell me all about it.")])

# Runner 실행
async for event in runner.run_async(user_id="u_123", session_id=session.id, new_message=message):
    if event.is_final_response():
        print(event.content)
    else:
        # Event 안에서 FunctionCall을 찾을 필요 없이 이 메서드들을 호출하면 됨.
        print(event.get_function_calls())
        print(event.get_function_responses())

# deploy.py
import dotenv

dotenv.load_dotenv()

import os

import vertexai
import vertexai.agent_engines
from vertexai.preview import reasoning_engines

from travel_advisor_agent.agent import travel_advisor_agent

PROJECT_ID = "프로젝트 ID"
LOCATION = "AI 엔진 지원 지역"
BUCKET = "gs://내 버킷이름"

# vertexai 초기화
vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=BUCKET,
)

app = reasoning_engines.AdkApp(
    agent=travel_advisor_agent,
    enable_tracing=True,
)

# 구글 클라우드 서비스인 agent engines 안에 agent 생성
remote_app = vertexai.agent_engines.create(
    display_name="Travel Advisor Agent",
    agent_engine=app,
    # 필요한 의존성
    requirements=[
        "google-cloud-aiplatform[adk,agent_engines]",
        "litellm",
    ],
    # 에이전트를 실행하는데 필요한 모든 폴더
    extra_packages=["travel_advisor_agent"],
    env_vars={
        "OPENAI_API_KEY": os.environ.get("OPENAI_API_KEY"),
    },
)

# remote.py
import vertexai
from vertexai import agent_engines

PROJECT_ID = "프로젝트 ID"
LOCATION = "지역"

vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
)

# 배포된 것을 가져옴
# deployments = agent_engines.list()

# for deployment in deployments:
#     print(deployment)

DEPLOYMENT_ID = "위 스크립트 실행 시 표시되는 ID"

SESSION_ID = "생성한 세션의 ID"

# 연결
remote_app = agent_engines.get(DEPLOYMENT_ID)

# 필수! 에이전트를 삭제하지 않으면 과금이 발생함!!!
remote_app.delete(force=True)


# 세션 생성
# # remote_session = remote_app.create_session(user_id="u_123")

# # print(remote_session["id"])

# 유저-중재자-에이전트 간 상호작용

# for event in remote_app.stream_query(
#     user_id="u_123",
#     session_id=SESSION_ID,
#     message="I'm going to Laos, any tips?",
# ):
#     print(event, "\n", "=" * 50)