A2A Exemplo: Travel Planner OpenRouter

Esta é uma implementação em Python que adere ao protocolo A2A (Agent2Agent). Uma demonstração de assistente de viagem implementada com base no SDK oficial a2a-python do Google e no OpenAI Python SDK. É um assistente de viagem em conformidade com as especificações do modelo OpenAI, capaz de fornecer serviços de planejamento de viagens.

Código Fonte

A2A Travel Planner OpenRouter

Arquitetura do Projeto

Este projeto demonstra como construir um agente de planejamento de viagens interoperável usando o protocolo A2A, incluindo os seguintes componentes principais:

Travel Planner Agent: Lógica principal do assistente de viagem baseada em interface compatível com OpenAI
Agent Executor: Adaptador do protocolo A2A que conecta a lógica do agente ao servidor A2A
A2A Server: Servidor compatível com protocolo A2A fornecendo interface padronizada de comunicação entre agentes
Loop Client: Cliente de teste para interagir com o servidor A2A

Diagrama de Sequência do Fluxo de Trabalho

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

    Client->>A2AServer: Solicitar Cartão do Agente
    A2AServer->>Client: Retornar Cartão do Agente (habilidades, capacidades)

    Note over Client,A2AServer: Usuário consulta planejamento de viagem

    Client->>A2AServer: message/sendStream (solicitação de streaming)
    A2AServer->>RequestHandler: Rotear solicitação de streaming
    RequestHandler->>Executor: execute(context, event_queue)
    
    Executor->>Agent: stream(query)
    Agent->>LLM: chat.completions.create(stream=True)
    
    loop Processamento de resposta de streaming
        LLM-->>Agent: Retornar fragmento de conteúdo de streaming
        Agent-->>Executor: yield {'content': chunk, 'done': False}
        Executor-->>RequestHandler: TaskArtifactUpdateEvent
        RequestHandler-->>A2AServer: Enviar evento SSE
        A2AServer-->>Client: Atualização de conteúdo de streaming
    end
    
    LLM-->>Agent: Resposta final concluída
    Agent-->>Executor: yield {'content': '', 'done': True}
    Executor-->>RequestHandler: TaskArtifactUpdateEvent final
    RequestHandler-->>A2AServer: Evento SSE final
    A2AServer-->>Client: Resposta de streaming concluída

Fluxo de Trabalho Principal

Recuperação do Cartão do Agente: Cliente primeiro recupera o cartão do agente do servidor A2A para entender as capacidades e habilidades do agente
Processamento de Solicitação de Streaming: Cliente envia solicitação de mensagem de streaming com consulta do usuário
Execução do Agente: Executor do agente processa a solicitação e chama a lógica principal do agente planejador de viagens
Interação com LLM: Agente conduz conversa de streaming com LLM compatível com OpenAI
Resposta em Tempo Real: Transmite respostas para o cliente em tempo real via Server-Sent Events (SSE)

Começando

Configure as variáveis de ambiente:

Copie o arquivo de exemplo e configure suas credenciais de API.

cp env.example .env

Edite o arquivo .env com seus valores reais:

# Obrigatório: Sua chave de API para o serviço de modelo de IA
API_KEY=your_actual_api_key_here

# Opcional: Nome do modelo (padrão: google/gemini-2.0-flash-001)
MODEL_NAME=google/gemini-2.0-flash-001

# Opcional: URL base para o serviço de API
BASE_URL=https://openrouter.ai/api/v1

Instale as dependências e inicie o servidor:

uv venv
source .venv/bin/activate

uv sync
uv run .

Execute o cliente de loop em um novo terminal:

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

Configuração

A aplicação usa variáveis de ambiente para configuração:

API_KEY (obrigatório): Sua chave de API para o serviço de modelo de IA
MODEL_NAME (opcional): O nome do modelo a ser usado (padrão: "google/gemini-2.0-flash-001")
BASE_URL (opcional): A URL base para o serviço de API (padrão: "https://openrouter.ai/api/v1")

Características Técnicas

Implementação Atual

✅ Conformidade com Protocolo A2A: Totalmente compatível com a especificação do protocolo Agent2Agent
✅ Resposta de Streaming: Suporta geração de conteúdo de streaming em tempo real
✅ Compatível com OpenAI: Suporta qualquer interface de API compatível com OpenAI
✅ Design Modular: Separação clara entre lógica do agente e adaptação do protocolo
✅ Configuração de Ambiente: Configuração flexível de variáveis de ambiente

Planos de Melhorias Futuras

Melhoria do Gerenciamento de Estado de Tarefas

Baseado no exemplo Google A2A LangGraph, planejando adicionar as seguintes funcionalidades:

🔄 Gerenciamento do Ciclo de Vida de Tarefas: Implementar rastreamento completo do estado de tarefas (enviado → trabalhando → concluído/falhado)
🔄 Suporte a Conversa Multi-turno: Adicionar estado input_required para suportar cenários complexos de planejamento de viagem que requerem esclarecimento do usuário
🔄 Persistência de Tarefas: Implementar persistência de estado de tarefas para tarefas de planejamento de longa duração
🔄 Tratamento de Erro Aprimorado: Estados de erro mais detalhados e mecanismos de recuperação
🔄 Cancelamento de Tarefas: Suportar cancelamento de tarefas em andamento

Exemplo de Gerenciamento de Estado

# Exemplo de implementação futura de gerenciamento de estado
class TravelPlannerTaskManager:
    async def handle_complex_query(self, query: str, context: RequestContext):
        # Detectar se mais informações são necessárias
        if self.needs_clarification(query):
            return TaskStatus(
                state=TaskState.input_required,
                message="Mais informações necessárias: Por favor, forneça destino específico, datas e faixa de orçamento"
            )
        
        # Executar planejamento complexo de múltiplas etapas
        task_id = await self.create_long_running_task(query)
        return TaskStatus(
            state=TaskState.working,
            taskId=task_id,
            message="Criando plano de viagem detalhado..."
        )

Adições de Funcionalidades Planejadas

📋 Suporte a Dados Estruturados: Adicionar suporte DataPart para coleta de preferências de viagem baseada em formulários
🖼️ Suporte Multimídia: Suportar FilePart para gerar e processar imagens de viagem, mapas, etc.
🔍 Integração de Ferramentas: Integrar APIs externas (clima, voos, hotéis, etc.) como chamadas de ferramentas
🌐 Suporte Multi-idioma: Estender capacidades de planejamento de viagem multi-idioma
📊 Métricas de Análise: Adicionar coleta de tempo de execução de tarefas, métricas de taxa de sucesso

Visite A2A