다양한 에이전트 타입 제공
- 모든 기반은 BaseAgent를 상속하여 진행
- 3가지 에이전트
- LLM 에이전트
- 자연어 이해와 추론: 복잡한 언어적 맥락을 파악하고 논리적 추론을 수행
- 동적 의사결정: 상황에 따라 적절한 도구를 선택하고 실행 방향 결정
- 유연성: 예측하기 어려운 상황에서도 창의적이고 적응적인 해결책 제시
- 언어 중심 작업: 텍스트 생성, 번역, 요약 등 언어 관련 작업에 특
- 워크플로 에이전트
- SequentialAgent: 에이전트 순차적 실행
- ParallelAgent: 에이전트 동시에 병렬 실행
- LoopAgent: 특정 조건이 만족될 때까지 반복 실행
- 커스텀 에이전트
- 고유한 운영 로직 구현: 특별한 비즈니스 요구사항에 맞춘 독특한 동작 방식
- 전문화된 통합: 특정 시스템인 서비스와의 깊은 연동
- 유연한 결정론 수준: 필요에 따라 예측 가능하거나 적응적인 동작 선택
구글 ADK 적용
pip install google-adk==1.9.0 (최신버전 사용 시 버전 제거)
from dotenv import load_dotenv
from google.adk.agents import Agent
load_dotenv()
# 인사말 함수
def greet_user() -> str:
return "안녕하세요!"
root_agent = Agent(
name="hello_agent",
model="gemini-3.6-flash",
description="유저와 인사하는 에이전트입니다.",
instruction="사용자에게 반갑고 친절하게 인사해주세요.",
tools=[greet_user],
)
- adk web → agent.py 상위 프로젝트 폴더 경로 접근, adk web 명령어 터미널 실행
- adk run [프로젝트명]
- adk api_server → FastAPI 기반의 api 사이트 실행
멀티 에이전트
- 특이 사항 : 예시 코드 실행 시 api 사이트가 셧다운으로 인해 진행되지 않음
- 2가지의 에이전트를 구성
- 책 정보를 api를 통해 조회 하도록 하는 도구를 이용한 에이전트 구성
- 결과 값을 pydantic을 이용해 구조화된 json 결과로 도출시키는 에이전트 구성
- 2개의 에이전트를 하나로 합치는 root 에이전트 구성
- 루트 에이전트에서 instruction을 통해서 각 역할과 처리 순서를 정리하도록 구성함.
- 병렬 아닌 순차적 진행
- 힌트를 제공하여 책 정보를 찾도록 구성
워크플로우 에이전트
- 순차적이거나 병렬적인 에이전트를 동시에 실행하는 용도로 이용
- 예제코드 설명
- 뉴스, 날씨, 주식의 정보를 병렬로 조회
- 병렬처리 에이전트와 요약해주는 정보 실행 root 에이전트 구성
from dotenv import load_dotenv
from google.adk.agents import Agent, ParallelAgent, SequentialAgent
from google.adk.tools import google_search
# 병렬로 실행될 정보 수집 에이전트
# 날씨 정보 수집 에이전트 생성
weather_fetcher = Agent(
name="weather",
model="gemini-3.6-flash",
output_key="weather_info",
instruction="오늘의 날씨 정보를 제공하세요.",
tools=[google_search],
)
# 뉴스 정보 수집 에이전트 생성
news_fetcher = Agent(
name="news",
model="gemini-3.6-flash",
output_key="news_info",
instruction="오늘의 주요 뉴스를 요약하세요.",
tools=[google_search],
)
# 주식 정보 수집 에이전트 생성
stock_fetcher = Agent(
name="stocks",
model="gemini-3.6-flash",
output_key="stock_info",
instruction="주요 주식 시장 동향을 제공하세요.",
tools=[google_search],
)
parallel_fetcher = ParallelAgent(
name="multi_info_fetcher",
sub_agents=[weather_fetcher, news_fetcher, stock_fetcher],
description="여러 정보를 동시에 수집"
)
# 수집 결과를 요약하는 에이전트
summarizer = Agent(
name="daily_briefing",
model="gemini-3.6-flash",
instrunction="""
수집된 정보를 종합하여 일일 브리핑을 작성하세요:
- 날씨: {weather_info}
- 뉴스: {news_info}
- 주식: {stock_info}
간결하고 읽기 쉬운 형식으로 정리하세요.
""",
)
# 병렬 수집 후 종합하는 파이프라인
daily_briefing_pipeline = SequentialAgent(
name="daily_briefing_system",
sub_agents=[parallel_fetcher, summarizer],
description="정보를 병렬로 수집한 후 종합 브리핑 생성",
)
root_agent = daily_briefing_pipeline
스팸 체크 에이전트
- 워크플로 에이전트 중 반복해서 메시지 생성 혹은 확인할 수 있는 LoopAgent 있음
- LLM 결과 퀄리티가 일정하지 않고 랜덤한 경우가 종종 있어, 반복 실행하여 퀄리티 높이는 방식
- 이번 예시는 BaseAgent를 통해 커스텀 에이전트 생성, LlmAgent 통해 데이터 생성 및 스팸 여부 확인 에이전트 구성
from dotenv import load_dotenv
from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.events import Event, EventActions
from google.adk.agents.invocation_context import InvocationContext
from typing import AsyncGenerator
# 랜덤한 메시지 생성하는 에이전트
random_generator = LlmAgent(
name="RandomGenerator",
model="gemini-3.6-flash",
description="랜덤한 메시지를 생성하는 에이전트입니다. 스팸 메시지와 정상적인 메시지를 60:40 비율로 생성합니다.",
output_key="random_message",
instruction="스팸 메시지와 정상적인 메시지를 60:40 확률로 생성합니다. 스팸 메시지는 '[웹발신]'을 앞에 넣고, 정상적인 메시지는 따로 표시하지 않아도 됩니다. 반드시 하나의 메시지만 출력해야 합니다.",
)
# 스팸 메시지를 확인하는 에이전트
spam_checker = LlmAgent(
name="SpamChecker",
model="gemini-3.6-flash",
instruction="사용자의 입력이 스팸인지 확인하세요. 스팸이면 'fail', 아니면 'pass'를 반환하세요.",
output_key="spam_status",
)
# 상태를 확인하고, 정상 메시지이면 루프를 중단하도록 요청하는 커스텀 에이전트
class CheckStatusAndEscalate(BaseAgent):
async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
status = ctx.session.state.get("spam_status", "fail")
should_stop = status == "pass"
yield Event(author=self.name, actions=EventActions(escalate=should_stop))
root_agent = LoopAgent(
name="SpamCheckLoop",
max_iterations=10,
sub_agents=[
random_generator,
spam_checker,
CheckStatusAndEscalate(name="StopChecker"),
]
)