Google ADK와 LangGraph 비교 (Node,Edge 기반 워크플로우 정의기능)

LangGraph가 노드와 엣지 구성을 통해서 워크플로우를 정교하게 관리할 수 있는데, 그러면 Google ADK도 가능할까 해서, 예제 코드를 Gemini를 이용해서 구성해봤다. 

 

일단 결론 부터 이야기하자면, Google ADK는 노드와 엣지의 개념은 없지만, 노드대신 에이전트로 정의한후, 이를 워크플로우로 연결이 가능하고, Loop와 Branch역시 가능하다. 

아직 Node와 ADK의 Agent차이에 대해서는 좀 더 고민해봐야겠지만, 개념상으로 비슷한것은 만들수있다는 이야기.

from google.adk.agents import LlmAgent, SequentialAgent, BaseAgent
from google.adk.events import Event, EventActions
from google.adk.runners import Runner
from google.adk.agents.invocation_context import InvocationContext
from typing import AsyncGenerator

# 1. 콘텐츠 생성 에이전트 정의
story_writer = LlmAgent(
    name="StoryWriter",
    instruction="""당신은 창의적인 작가입니다.
    사용자가 제공한 주제에 대해 짧은 이야기를 작성해주세요."""
)

# 2. 콘텐츠 검토 에이전트 정의
story_critic = LlmAgent(
    name="StoryCritic",
    instruction="""당신은 비평가입니다.
    주어진 이야기가 주제에 맞고 흥미로운지 평가하고,
    '승인' 또는 '수정'으로만 답변해주세요."""
)

# 3. 조건부 흐름을 위한 사용자 정의 에이전트
class ConditionalRouter(BaseAgent):
    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        # 이전 단계(StoryCritic)의 결과 확인
        critic_decision = ctx.session.state.get("critic_decision")

        # '승인'일 경우 워크플로우 종료
        if critic_decision == "승인":
            yield Event(author=self.name, actions=EventActions(escalate=True))
        else:
            # 그 외의 경우(수정) 다음 단계로 진행
            yield Event(author=self.name, actions=EventActions(escalate=False))

# 4. 워크플로우 구성
# SequentialAgent를 사용하여 에이전트들을 순차적으로 연결합니다.
# LoopAgent와 같은 다른 워크플로우 에이전트를 사용하여 반복적인 작업을 처리할 수도 있습니다. [8, 9]
generative_flow = SequentialAgent(
    name="GenerativeWorkflow",
    sub_agents=[
        story_writer,
        # LlmAgent의 출력을 세션 상태에 저장하여 다음 에이전트가 사용할 수 있도록 합니다.
        LlmAgent(
            name="SaveCriticInput",
            output_key="critic_input",  # story_writer의 출력을 critic_input 키로 저장
            instruction="""다음 에이전트가 사용할 수 있도록 이전 단계의 출력을 저장합니다."""
        ),
        story_critic,
        # story_critic의 출력을 critic_decision 키로 저장
        LlmAgent(
            name="SaveCriticDecision",
            output_key="critic_decision",
            instruction="""다음 에이전트가 사용할 수 있도록 이전 단계의 출력을 저장합니다."""
        ),
        ConditionalRouter(name="Router")
    ]
)

# 5. 워크플로우 실행
async def main():
    runner = Runner()
    # '우주 탐험'이라는 주제로 워크플로우 시작
    final_event = await runner.run_async(
        agent=generative_flow,
        input_text="우주 탐험에 대한 짧은 이야기를 써줘."
    )
    print("최종 결과:", final_event)

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())