A2A + CrewAI + OpenRouter Chart Generation Agent Tutorial

Foco do Tutorial

Este tutorial irá guiá-lo através da prática das seguintes habilidades fundamentais:

• Integrando OpenRouter + CrewAI + A2A: Desenvolvimento completo end-to-end de Agente usando OpenRouter como provedor LLM, CrewAI como framework de Agente e protocolo A2A como interface padronizada
• Praticando Retorno de Dados de Imagem do Agente A2A: Aprenda como fazer Agentes gerarem e retornarem dados de imagem, não apenas respostas de texto
• Usando A2A Inspector para Depurar Aplicações A2A: Domine ferramentas de depuração profissionais para testar e validar suas aplicações de Agente

Início Rápido

1. Clonar o Código

git clone [email protected]:sing1ee/a2a-crewai-charts-agent.git
cd a2a-crewai-charts-agent

2. Criar Configuração de Ambiente

Crie o arquivo .env:

OPENROUTER_API_KEY=sk-or-v1-your-api-key-here
OPENAI_MODEL_NAME=openrouter/anthropic/claude-3.7-sonnet

3. Configuração de Ambiente e Execução

# Criar ambiente virtual
uv venv

# Ativar ambiente virtual
source .venv/bin/activate

# Executar aplicação
uv run .

A aplicação iniciará em http://localhost:10011.

Depurar com A2A Inspector

A2A Inspector é uma ferramenta poderosa especificamente projetada para depuração de aplicações A2A.

Passos de Depuração:

1. Acessar A2A Inspector: Abra https://inspector.a2aprotocol.ai

2. Conectar ao Seu Agente:

   • Digite o endereço do seu Agente no Inspector: http://localhost:10011
   • Clique em "Connect" para estabelecer conexão

3. Testar Funcionalidade do Agente:

   • Envie mensagem de teste: "Generate a chart of revenue: Jan,1000 Feb,2000 Mar,1500"
   • Observe o processo completo de interação do protocolo A2A
   • Visualize os dados de imagem retornados

4. Depurar e Monitorar:

   • Verifique capacidades e habilidades do Agente
   • Monitore o fluxo completo de requisição e resposta
   • Verifique transmissão correta de dados de imagem

Consulte a Documentação do A2A Inspector para guias de depuração mais detalhados.

Processo Principal e Introdução ao Código

Diagrama de Sequência da Arquitetura do Sistema

sequenceDiagram
    participant U as User
    participant A2A as A2A Inspector
    participant S as A2A Server
    participant H as DefaultRequestHandler
    participant E as ChartGenerationAgentExecutor
    participant CA as ChartGenerationAgent
    participant Crew as CrewAI Crew
    participant Tool as ChartGenerationTool
    participant MP as Matplotlib
    participant Cache as InMemoryCache

    U->>A2A: Send prompt "Generate chart: A,100 B,200"
    A2A->>S: HTTP POST /tasks with A2A message
    S->>H: Handle request with RequestContext
    H->>E: execute(context, event_queue)
    E->>CA: invoke(query, session_id)
    CA->>Crew: kickoff with inputs
    Crew->>Tool: generate_chart_tool(prompt, session_id)
    Tool->>Tool: Parse CSV data
    Tool->>MP: Create bar chart with matplotlib
    MP-->>Tool: Return PNG bytes
    Tool->>Cache: Store image with ID
    Cache-->>Tool: Confirm storage
    Tool-->>Crew: Return image ID
    Crew-->>CA: Return image ID
    CA-->>E: Return image ID
    E->>CA: get_image_data(session_id, image_key)
    CA->>Cache: Retrieve image data
    Cache-->>CA: Return Imagedata
    CA-->>E: Return Imagedata
    E->>H: Create FilePart with image bytes
    H->>S: enqueue completed_task event
    S-->>A2A: Return A2A response with image
    A2A-->>U: Display generated chart

Detalhes dos Componentes Principais

1. Inicialização do Servidor A2A (__main__.py)

# Definir capacidades e habilidades do Agente
capabilities = AgentCapabilities(streaming=False)
skill = AgentSkill(
    id='chart_generator',
    name='Chart Generator',
    description='Generate a chart based on CSV-like data passed in',
    tags=['generate image', 'edit image'],
    examples=['Generate a chart of revenue: Jan,$1000 Feb,$2000 Mar,$1500'],
)

# Criar cartão do Agente
agent_card = AgentCard(
    name='Chart Generator Agent',
    description='Generate charts from structured CSV-like data input.',
    url=f'http://{host}:{port}/',
    version='1.0.0',
    defaultInputModes=ChartGenerationAgent.SUPPORTED_CONTENT_TYPES,
    defaultOutputModes=ChartGenerationAgent.SUPPORTED_CONTENT_TYPES,
    capabilities=capabilities,
    skills=[skill],
)

Pontos Principais:

• AgentCapabilities define funções de Agente suportadas (streaming desabilitado aqui)
• AgentSkill descreve habilidades específicas do Agente e exemplos de uso
• AgentCard é a identidade do Agente no protocolo A2A

2. Implementação do Agente CrewAI (agent.py)

class ChartGenerationAgent:
    def __init__(self):
        # Criar Agente especializado de geração de gráficos
        self.chart_creator_agent = Agent(
            role='Chart Creation Expert',
            goal='Generate a bar chart image based on structured CSV input.',
            backstory='You are a data visualization expert who transforms structured data into visual charts.',
            verbose=False,
            allow_delegation=False,
            tools=[generate_chart_tool],
        )

        # Definir tarefa
        self.chart_creation_task = Task(
            description=(
                "You are given a prompt: '{user_prompt}'.\n"
                "If the prompt includes comma-separated key:value pairs (e.g. 'a:100, b:200'), "
                "reformat it into CSV with header 'Category,Value'.\n"
                "Ensure it becomes two-column CSV, then pass that to the 'ChartGenerationTool'.\n"
                "Use session ID: '{session_id}' when calling the tool."
            ),
            expected_output='The id of the generated chart image',
            agent=self.chart_creator_agent,
        )

Pontos Principais:

• A classe Agent do CrewAI define funções e capacidades do assistente de IA
• A classe Task descreve lógica específica de execução de tarefas
• Ferramentas personalizadas são integradas ao Agente através do parâmetro tools

3. Ferramenta de Geração de Gráficos

@tool('ChartGenerationTool')
def generate_chart_tool(prompt: str, session_id: str) -> str:
    """Generates a bar chart image from CSV-like input using matplotlib."""
    
    # Analisar dados CSV
    df = pd.read_csv(StringIO(prompt))
    df.columns = ['Category', 'Value']
    df['Value'] = pd.to_numeric(df['Value'], errors='coerce')
    
    # Gerar gráfico de barras
    fig, ax = plt.subplots()
    ax.bar(df['Category'], df['Value'])
    ax.set_xlabel('Category')
    ax.set_ylabel('Value')
    ax.set_title('Bar Chart')
    
    # Salvar como bytes PNG
    buf = BytesIO()
    plt.savefig(buf, format='png')
    plt.close(fig)
    buf.seek(0)
    image_bytes = buf.read()
    
    # Codificar e cachear imagem
    data = Imagedata(
        bytes=base64.b64encode(image_bytes).decode('utf-8'),
        mime_type='image/png',
        name='generated_chart.png',
        id=uuid4().hex,
    )
    
    # Armazenar imagem no cache
    session_data = cache.get(session_id) or {}
    session_data[data.id] = data
    cache.set(session_id, session_data)
    
    return data.id

Pontos Principais:

• Use o decorador @tool para converter função em ferramenta CrewAI
• Use pandas para analisar dados CSV, matplotlib para gerar gráficos
• Imagens armazenadas como codificação base64 para transmissão em rede
• Use IDs de sessão para gerenciar isolamento de dados para múltiplos usuários

4. Executor A2A (agent_executor.py)

class ChartGenerationAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # Obter entrada do usuário
        query = context.get_user_input()
        
        # Chamar Agente CrewAI
        result = self.agent.invoke(query, context.context_id)
        
        # Obter dados de imagem gerados
        data = self.agent.get_image_data(
            session_id=context.context_id, 
            image_key=result.raw
        )
        
        if data and not data.error:
            # Criar parte de arquivo contendo bytes de imagem
            parts = [
                Part(
                    root=FilePart(
                        file=FileWithBytes(
                            bytes=data.bytes,
                            mimeType=data.mime_type,
                            name=data.name,
                        )
                    )
                )
            ]
        else:
            # Retornar mensagem de texto em caso de erro
            parts = [Part(root=TextPart(text=data.error or 'Failed to generate chart image.'))]
        
        # Adicionar tarefa concluída à fila de eventos
        event_queue.enqueue_event(
            completed_task(
                context.task_id,
                context.context_id,
                [new_artifact(parts, f'chart_{context.task_id}')],
                [context.message],
            )
        )

Pontos Principais:

• AgentExecutor é a camada de execução do protocolo A2A
• Obter requisições do usuário através de RequestContext
• Converter respostas CrewAI para formato do protocolo A2A
• Suporte para retornar dados do tipo arquivo (imagens)

5. Sistema de Cache (utils.py)

class InMemoryCache:
    """Cache simples thread-safe em memória sem expiração."""
    
    def __init__(self):
        self._lock = threading.Lock()
        self._store: dict[str, Any] = {}
    
    def get(self, key: str) -> Any | None:
        with self._lock:
            return self._store.get(key)
    
    def set(self, key: str, value: Any) -> None:
        with self._lock:
            self._store[key] = value

Pontos Principais:

• Implementação de cache em memória thread-safe
• Usado para armazenar dados de imagem gerados
• Suporta isolamento de sessão para evitar confusão de dados do usuário

Resumo da Stack Tecnológica

• Protocolo A2A: Protocolo de comunicação de Agente padronizado
• CrewAI: Framework de colaboração multi-Agente
• OpenRouter: Serviço de agregação de API LLM
• Matplotlib: Biblioteca de geração de gráficos Python
• Pandas: Biblioteca de processamento de dados
• UV: Gerenciador de pacotes Python moderno

Sugestões de Extensão

1. Suportar Mais Tipos de Gráficos: Gráficos de pizza, gráficos de linha, gráficos de dispersão, etc.
2. Adicionar Validação de Dados: Validação de dados de entrada mais forte e tratamento de erros
3. Cache Persistente: Usar Redis ou sistema de arquivos para armazenar imagens
4. Suporte a Streaming: Suporte ao progresso de geração de gráficos em tempo real
5. Entrada Multimodal: Suporte para upload de arquivos CSV em vez de apenas entrada de texto

Através deste tutorial, você dominou as habilidades fundamentais de construir Agentes práticos usando stack de tecnologia de IA moderna. Este Agente de geração de gráficos pode servir como base para aplicações de análise de dados mais complexas.

System Prompt for AI

You are a professional assistant based on the CrewAI chart generation agent tutorial. This tutorial provides a comprehensive guide on building a complete AI chart generation application. I can help you with:

## What I Can Help You With:
- **Quick Start**: Guide you from scratch to build an agent application using OpenRouter + CrewAI + A2A protocol
- **Solve Technical Issues**: Answer specific questions about environment setup, code implementation, and debugging
- **Understand Core Concepts**: Explain how A2A protocol, CrewAI framework, and agent architecture work
- **Practical Guidance**: Provide best practices for image data handling, tool integration, and cache design
- **Debug Support**: Teach you to use A2A Inspector for professional debugging and troubleshooting

## You Can Ask Me:
- "How to configure OpenRouter API keys?"
- "How do CrewAI Agents and Tasks work?"
- "How to return image data after chart generation?"
- "How to connect and test my agent with A2A Inspector?"
- "How is the cache system implemented in the code?"
- "How to extend support for more chart types?"

I'll provide accurate, practical answers based on the tutorial content to help you quickly master modern AI agent development skills.

You can visit [A2AProtocol.ai](https://a2aprotocol.ai/) for more tutorials.