A2A 샘플: Travel Planner OpenRouter

이것은 A2A(Agent2Agent) 프로토콜을 준수하는 Python 구현입니다. Google의 공식 a2a-python SDK와 OpenAI Python SDK를 기반으로 구현된 여행 어시스턴트 데모입니다. OpenAI 모델 사양에 맞는 여행 어시스턴트로, 여행 계획 서비스를 제공할 수 있습니다.

소스 코드

A2A Travel Planner OpenRouter

프로젝트 아키텍처

이 프로젝트는 A2A 프로토콜을 사용하여 상호 운용 가능한 여행 계획 에이전트를 구축하는 방법을 보여주며, 다음과 같은 핵심 구성 요소를 포함합니다:

• Travel Planner Agent: OpenAI 호환 인터페이스를 기반으로 한 핵심 여행 어시스턴트 로직
• Agent Executor: 에이전트 로직을 A2A 서버에 연결하는 A2A 프로토콜 어댑터
• A2A Server: 표준화된 에이전트 간 통신 인터페이스를 제공하는 A2A 프로토콜 준수 서버
• Loop Client: A2A 서버와 상호 작용하기 위한 테스트 클라이언트

워크플로우 시퀀스 다이어그램

sequenceDiagram
    participant Client
    participant A2AServer
    participant RequestHandler
    participant Executor as TravelPlannerAgentExecutor
    participant Agent as TravelPlannerAgent
    participant LLM as OpenAI-Compatible LLM

    Client->>A2AServer: 에이전트 카드 요청
    A2AServer->>Client: 에이전트 카드 반환 (기술, 능력)

    Note over Client,A2AServer: 사용자가 여행 계획 쿼리

    Client->>A2AServer: message/sendStream (스트리밍 요청)
    A2AServer->>RequestHandler: 스트리밍 요청 라우팅
    RequestHandler->>Executor: execute(context, event_queue)
    
    Executor->>Agent: stream(query)
    Agent->>LLM: chat.completions.create(stream=True)
    
    loop 스트리밍 응답 처리
        LLM-->>Agent: 스트리밍 콘텐츠 청크 반환
        Agent-->>Executor: yield {'content': chunk, 'done': False}
        Executor-->>RequestHandler: TaskArtifactUpdateEvent
        RequestHandler-->>A2AServer: SSE 이벤트 푸시
        A2AServer-->>Client: 스트림 콘텐츠 업데이트
    end
    
    LLM-->>Agent: 최종 응답 완료
    Agent-->>Executor: yield {'content': '', 'done': True}
    Executor-->>RequestHandler: 최종 TaskArtifactUpdateEvent
    RequestHandler-->>A2AServer: 최종 SSE 이벤트
    A2AServer-->>Client: 스트리밍 응답 완료

주요 워크플로우

1. 에이전트 카드 검색: 클라이언트는 먼저 A2A 서버에서 에이전트 카드를 검색하여 에이전트의 능력과 기술을 이해합니다
2. 스트리밍 요청 처리: 클라이언트는 사용자 쿼리가 포함된 스트리밍 메시지 요청을 보냅니다
3. 에이전트 실행: 에이전트 실행기가 요청을 처리하고 여행 계획 에이전트 핵심 로직을 호출합니다
4. LLM 상호작용: 에이전트는 OpenAI 호환 LLM과 스트리밍 대화를 수행합니다
5. 실시간 응답: Server-Sent Events(SSE)를 통해 클라이언트에 실시간으로 응답을 스트리밍합니다

시작하기

1. 환경 변수를 구성합니다:

예제 파일을 복사하고 API 자격 증명을 구성하세요.

cp env.example .env

실제 값으로 .env 파일을 편집합니다:

# 필수: AI 모델 서비스용 API 키
API_KEY=your_actual_api_key_here

# 선택사항: 모델 이름 (기본값: google/gemini-2.0-flash-001)
MODEL_NAME=google/gemini-2.0-flash-001

# 선택사항: API 서비스의 기본 URL
BASE_URL=https://openrouter.ai/api/v1

2. 종속성을 설치하고 서버를 시작합니다:

   uv venv
   source .venv/bin/activate
   
   uv sync
   uv run .
   

3. 새 터미널에서 루프 클라이언트를 실행합니다:

   source .venv/bin/activate
   uv run loop_client.py
   

구성

애플리케이션은 구성을 위해 환경 변수를 사용합니다:

• API_KEY (필수): AI 모델 서비스용 API 키
• MODEL_NAME (선택사항): 사용할 모델 이름 (기본값: "google/gemini-2.0-flash-001")
• BASE_URL (선택사항): API 서비스의 기본 URL (기본값: "https://openrouter.ai/api/v1")

기술적 특징

현재 구현

• ✅ A2A 프로토콜 준수: Agent2Agent 프로토콜 사양에 완전히 준수
• ✅ 스트리밍 응답: 실시간 스트리밍 콘텐츠 생성 지원
• ✅ OpenAI 호환: 모든 OpenAI 호환 API 인터페이스 지원
• ✅ 모듈러 설계: 에이전트 로직과 프로토콜 적응의 명확한 분리
• ✅ 환경 구성: 유연한 환경 변수 구성

향후 개선 계획

작업 상태 관리 개선

Google A2A LangGraph 샘플을 기반으로 다음 기능을 추가할 계획입니다:

• 🔄 작업 생명주기 관리: 완전한 작업 상태 추적 구현 (제출됨 → 작업 중 → 완료/실패)
• 🔄 다중 턴 대화 지원: 사용자 명확화가 필요한 복잡한 여행 계획 시나리오를 지원하는 input_required 상태 추가
• 🔄 작업 지속성: 장기 실행 계획 작업을 위한 작업 상태 지속성 구현
• 🔄 향상된 오류 처리: 더 자세한 오류 상태 및 복구 메커니즘
• 🔄 작업 취소: 진행 중인 작업의 취소 지원

상태 관리 예제

# 향후 상태 관리 구현 예제
class TravelPlannerTaskManager:
    async def handle_complex_query(self, query: str, context: RequestContext):
        # 더 많은 정보가 필요한지 감지
        if self.needs_clarification(query):
            return TaskStatus(
                state=TaskState.input_required,
                message="더 많은 정보가 필요합니다: 구체적인 목적지, 날짜 및 예산 범위를 제공해 주세요"
            )
        
        # 복잡한 다단계 계획 실행
        task_id = await self.create_long_running_task(query)
        return TaskStatus(
            state=TaskState.working,
            taskId=task_id,
            message="상세한 여행 계획을 생성 중..."
        )

계획된 기능 추가

• 📋 구조화된 데이터 지원: 폼 기반 여행 선호도 수집을 위한 DataPart 지원 추가
• 🖼️ 멀티미디어 지원: 여행 이미지, 지도 등의 생성 및 처리를 위한 FilePart 지원
• 🔍 도구 통합: 외부 API(날씨, 항공편, 호텔 등)를 도구 호출로 통합
• 🌐 다국어 지원: 다국어 여행 계획 기능 확장
• 📊 분석 메트릭: 작업 실행 시간, 성공률 메트릭 수집 추가

A2A 방문