7. 스트리밍 & 다중 턴 상호작용

7. 스트리밍 & 다중 턴 상호작용 (LangGraph 예제)

Hello World 예제는 A2A의 기본 메커니즘을 보여줘요. 견고한 스트리밍, 태스크 상태 관리, LLM 기반 다중 턴 대화 같은 더 고급 기능을 보려면 a2a-samples/samples/python/agents/langgraph/에 있는 LangGraph 예제를 살펴볼게요. 이 예제는 LangChain과 LangGraph를 통해 Gemini 모델을 사용해 환전 질문에 답하는 "Currency Agent"(환율 에이전트)를 제공해요.

출처: 문서

본문

LangGraph 예제 설정하기

아직 Gemini API 키가 없다면 하나 만들어요.

환경 변수: a2a-samples/samples/python/agents/langgraph/ 디렉터리에 .env 파일을 만들어요.

echo "GOOGLE_API_KEY=YOUR_API_KEY_HERE" > .env

YOUR_API_KEY_HERE를 실제 Gemini API 키로 바꿔요.

의존성 설치(아직 안 했다면): langgraph 예제는 langchain-google-genai, langgraph 같은 의존성을 포함한 자체 pyproject.toml을 가져요. a2a-samples 루트에서 pip install -e .[dev]로 SDK를 설치했다면 워크스페이스 예제의 의존성(langgraph-example 포함)도 함께 설치됐을 거예요. 임포트 오류가 발생하면 루트 디렉터리에서의 기본 SDK 설치가 성공했는지 확인해요.

LangGraph 서버 실행하기

터미널에서 a2a-samples/samples/python/agents/langgraph/app 디렉터리로 이동하고 (SDK 루트에서 만든) 가상 환경이 활성화되어 있는지 확인해요. LangGraph 에이전트 서버를 시작해요:

python __main__.py

이렇게 하면 보통 http://localhost:10000에서 서버가 시작돼요.

LangGraph 에이전트와 상호작용하기

새 터미널 창을 열고 가상 환경을 활성화한 뒤 a2a-samples/samples/python/agents/langgraph/app으로 이동해요. 테스트 클라이언트를 실행해요:

python test_client.py

이제 __main__.py가 실행 중인 터미널 창에서 Ctrl+C를 입력해 서버를 종료할 수 있어요.

시연되는 핵심 기능

langgraph 예제는 몇 가지 중요한 A2A 개념을 보여줘요.

LLM 통합:

  • agent.py는 CurrencyAgent를 정의해요. ChatGoogleGenerativeAI와 LangGraph의 create_react_agent를 사용해 사용자 질의를 처리해요.
  • 이는 실제 LLM이 에이전트 로직에 어떻게 힘을 실어줄 수 있는지 보여줘요.

태스크 상태 관리:

  • samples/langgraph/__main__.py는 InMemoryTaskStore로 DefaultRequestHandler를 초기화해요.
httpx_client = httpx.AsyncClient()
push_config_store = InMemoryPushNotificationConfigStore()
push_sender = BasePushNotificationSender(httpx_client=httpx_client,
                config_store=push_config_store)
request_handler = DefaultRequestHandler(
    agent_executor=CurrencyAgentExecutor(),
    task_store=InMemoryTaskStore(),
    push_config_store=push_config_store,
    push_sender= push_sender
)
server = A2AStarletteApplication(
    agent_card=agent_card, http_handler=request_handler
)

uvicorn.run(server.build(), host=host, port=port)
  • CurrencyAgentExecutor(samples/langgraph/agent_executor.py에 있음)의 execute 메서드는 DefaultRequestHandler가 호출할 때, 현재 태스크(있다면)를 담은 RequestContext와 상호작용해요.
  • Send Message의 경우 DefaultRequestHandler는 TaskStore를 사용해 상호작용 간 태스크 상태를 유지·검색해요. 에이전트 실행 흐름이 여러 단계를 거치거나 지속적인 태스크를 만들면 Send Message에 대한 응답은 전체 Task 객체가 돼요.
  • test_client.py의 run_single_turn_test는 Task 객체를 받아 get_task로 질의하는 것을 보여줘요.

TaskStatusUpdateEvent와 TaskArtifactUpdateEvent로 스트리밍:

  • CurrencyAgentExecutor의 execute 메서드는 비스트리밍·스트리밍 요청을 모두 처리하며 DefaultRequestHandler가 조율해요.
  • LangGraph 에이전트가 요청을 처리하면서(get_exchange_rate 같은 도구 호출을 포함할 수 있음) CurrencyAgentExecutor는 EventQueue에 다양한 유형의 이벤트를 큐에 넣어요.
    • TaskStatusUpdateEvent: 중간 업데이트용(예: "Looking up exchange rates...", "Processing the exchange rates...").
    • TaskArtifactUpdateEvent: 최종 답변이 준비되면 아티팩트로 큐에 넣어요. lastChunk 플래그는 True예요.
    • state=TaskState.completed로 최종 TaskStatusUpdateEvent를 보내 태스크가 끝났음을 알리고 스트림을 닫아요.
  • test_client.py의 run_streaming_test 함수는 서버에서 받는 개별 이벤트 청크들을 출력해요.

다중 턴 대화(TaskState.input_required):

  • CurrencyAgent는 질의가 모호할 때 명확화를 요청할 수 있어요(예: 사용자가 "100 USD는 얼마인가요?"라고 물을 때).
  • 이런 경우 CurrencyAgentExecutor는 status.state가 TaskState.input_required이고 status.message에 에이전트의 질문(예: "어느 통화로 변환할까요?")이 담긴 TaskStatusUpdateEvent를 큐에 넣어요. 이 이벤트 후 스트림이 닫혀요.
  • test_client.py의 run_multi_turn_test 함수는 이를 보여줘요.
    • 초기 모호한 질의를 보내요.
    • 에이전트가 (큐에 넣은 이벤트를 처리하는 DefaultRequestHandler를 통해) 상태가 input_required인 Task로 응답해요.
    • 클라이언트는 첫 턴의 Task 응답에서 taskId와 contextId를 포함한 두 번째 메시지를 보내 누락된 정보("in GBP")를 제공해요. 이는 같은 태스크를 계속하는 것이에요.

코드 탐색하기

이 파일들을 잠시 살펴보세요.

  • __main__.py: A2AStarletteApplication과 DefaultRequestHandler를 사용한 서버 설정. AgentCard 정의에 capabilities.streaming=True가 포함된 점을 주목하세요.
  • agent.py: LangGraph, LLM 모델, 도구 정의를 가진 CurrencyAgent.
  • agent_executor.py: execute(와 cancel) 메서드를 구현하는 CurrencyAgentExecutor. RequestContext로 진행 중인 태스크를 이해하고 EventQueue로 다양한 이벤트(TaskStatusUpdateEvent, TaskArtifactUpdateEvent, 태스크가 없으면 첫 이벤트로 암묵적으로 새 Task 객체)를 다시 보내요.
  • test_client.py: 태스크 ID를 가져오고 다중 턴 대화에 사용하는 것 포함, 다양한 상호작용 패턴을 보여줘요.

이 예제는 A2A가 에이전트 간 복잡하고 상태를 가진 비동기 상호작용을 어떻게 촉진하는지 훨씬 풍부하게 보여줘요.

더 알아보기 (Learn more)