A2A Protocol

A2A + CrewAI + OpenRouter Chart Generation Agent Tutorial

deutschEnglishespañolfrançaisहिंदी日本語한국어portuguêsрусский中文
MILO•
Share
A2A + CrewAI + OpenRouter Chart Generation Agent Tutorial

튜토리얼 중점 사항

이 튜토리얼은 다음 핵심 기술들을 연습하는 과정을 안내합니다:

  • OpenRouter + CrewAI + A2A 통합: OpenRouter를 LLM 제공자로, CrewAI를 에이전트 프레임워크로, A2A 프로토콜을 표준화된 인터페이스로 사용하는 완전한 엔드투엔드 에이전트 개발
  • A2A 에이전트 이미지 데이터 반환 연습: 에이전트가 텍스트 응답뿐만 아니라 이미지 데이터를 생성하고 반환하는 방법 학습
  • A2A Inspector로 A2A 애플리케이션 디버깅: 에이전트 애플리케이션을 테스트하고 검증하는 전문 디버깅 도구 마스터

빠른 시작

1. 코드 클론

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

2. 환경 설정 생성

.env 파일 생성:

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

3. 환경 설정 및 실행

# 가상 환경 생성
uv venv

# 가상 환경 활성화
source .venv/bin/activate

# 애플리케이션 실행
uv run .

애플리케이션은 http://localhost:10011에서 시작됩니다.

A2A Inspector로 디버깅

A2A Inspector는 A2A 애플리케이션 디버깅을 위해 특별히 설계된 강력한 도구입니다.

디버깅 단계:

  1. A2A Inspector 접근: https://inspector.a2aprotocol.ai 열기

  2. 에이전트에 연결:

    • Inspector에 에이전트 주소 입력: http://localhost:10011
    • "Connect" 클릭하여 연결 설정

  3. 에이전트 기능 테스트:

    • 테스트 메시지 전송: "Generate a chart of revenue: Jan,1000 Feb,2000 Mar,1500"
    • 완전한 A2A 프로토콜 상호작용 과정 관찰
    • 반환된 이미지 데이터 확인

  4. 디버그 및 모니터링:

    • 에이전트의 능력과 스킬 확인
    • 완전한 요청 및 응답 흐름 모니터링
    • 올바른 이미지 데이터 전송 검증

더 자세한 디버깅 가이드는 A2A Inspector 문서를 참조하세요.

주요 프로세스 및 코드 소개

시스템 아키텍처 시퀀스 다이어그램

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

핵심 컴포넌트 세부사항

1. A2A 서버 초기화 (__main__.py)

# 에이전트 능력과 스킬 정의
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'],
)

# 에이전트 카드 생성
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],
)

핵심 포인트:

  • AgentCapabilities는 지원되는 에이전트 기능 정의 (여기서는 스트리밍 비활성화)
  • AgentSkill은 특정 에이전트 스킬과 사용 예제 설명
  • AgentCard는 A2A 프로토콜에서 에이전트의 신원

2. CrewAI 에이전트 구현 (agent.py)

class ChartGenerationAgent:
    def __init__(self):
        # 전문 차트 생성 에이전트 생성
        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],
        )

        # 작업 정의
        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,
        )

핵심 포인트:

  • CrewAI의 Agent 클래스는 AI 어시스턴트 역할과 능력 정의
  • Task 클래스는 특정 작업 실행 로직 설명
  • tools 매개변수를 통해 커스텀 도구가 에이전트에 통합

3. 차트 생성 도구

@tool('ChartGenerationTool')
def generate_chart_tool(prompt: str, session_id: str) -> str:
    """Generates a bar chart image from CSV-like input using matplotlib."""
    
    # CSV 데이터 파싱
    df = pd.read_csv(StringIO(prompt))
    df.columns = ['Category', 'Value']
    df['Value'] = pd.to_numeric(df['Value'], errors='coerce')
    
    # 막대 차트 생성
    fig, ax = plt.subplots()
    ax.bar(df['Category'], df['Value'])
    ax.set_xlabel('Category')
    ax.set_ylabel('Value')
    ax.set_title('Bar Chart')
    
    # PNG 바이트로 저장
    buf = BytesIO()
    plt.savefig(buf, format='png')
    plt.close(fig)
    buf.seek(0)
    image_bytes = buf.read()
    
    # 이미지 인코딩 및 캐싱
    data = Imagedata(
        bytes=base64.b64encode(image_bytes).decode('utf-8'),
        mime_type='image/png',
        name='generated_chart.png',
        id=uuid4().hex,
    )
    
    # 캐시에 이미지 저장
    session_data = cache.get(session_id) or {}
    session_data[data.id] = data
    cache.set(session_id, session_data)
    
    return data.id

핵심 포인트:

  • @tool 데코레이터를 사용하여 함수를 CrewAI 도구로 변환
  • pandas로 CSV 데이터 파싱, matplotlib로 차트 생성
  • 네트워크 전송을 위해 이미지를 base64 인코딩으로 저장
  • 여러 사용자의 데이터 격리 관리를 위해 세션 ID 사용

4. A2A 실행자 (agent_executor.py)

class ChartGenerationAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # 사용자 입력 가져오기
        query = context.get_user_input()
        
        # CrewAI 에이전트 호출
        result = self.agent.invoke(query, context.context_id)
        
        # 생성된 이미지 데이터 가져오기
        data = self.agent.get_image_data(
            session_id=context.context_id, 
            image_key=result.raw
        )
        
        if data and not data.error:
            # 이미지 바이트를 포함하는 파일 부분 생성
            parts = [
                Part(
                    root=FilePart(
                        file=FileWithBytes(
                            bytes=data.bytes,
                            mimeType=data.mime_type,
                            name=data.name,
                        )
                    )
                )
            ]
        else:
            # 오류 시 텍스트 메시지 반환
            parts = [Part(root=TextPart(text=data.error or 'Failed to generate chart image.'))]
        
        # 완료된 작업을 이벤트 큐에 추가
        event_queue.enqueue_event(
            completed_task(
                context.task_id,
                context.context_id,
                [new_artifact(parts, f'chart_{context.task_id}')],
                [context.message],
            )
        )

핵심 포인트:

  • AgentExecutor는 A2A 프로토콜 실행 레이어
  • RequestContext를 통해 사용자 요청 획득
  • CrewAI 응답을 A2A 프로토콜 형식으로 변환
  • 파일 타입 데이터(이미지) 반환 지원

5. 캐시 시스템 (utils.py)

class InMemoryCache:
    """만료 없는 간단한 스레드 안전 인메모리 캐시."""
    
    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

핵심 포인트:

  • 스레드 안전 인메모리 캐시 구현
  • 생성된 이미지 데이터 저장에 사용
  • 사용자 데이터 혼동 방지를 위한 세션 격리 지원

기술 스택 요약

  • A2A 프로토콜: 표준화된 에이전트 통신 프로토콜
  • CrewAI: 멀티 에이전트 협업 프레임워크
  • OpenRouter: LLM API 집계 서비스
  • Matplotlib: Python 차트 생성 라이브러리
  • Pandas: 데이터 처리 라이브러리
  • UV: 모던 Python 패키지 관리자

확장 제안

  1. 더 많은 차트 타입 지원: 파이 차트, 라인 차트, 스캐터 플롯 등
  2. 데이터 검증 추가: 더 강력한 입력 데이터 검증 및 오류 처리
  3. 영구 캐시: Redis나 파일 시스템을 사용한 이미지 저장
  4. 스트리밍 지원: 실시간 차트 생성 진행상황 지원
  5. 멀티모달 입력: 텍스트 전용 입력 대신 CSV 파일 업로드 지원

이 튜토리얼을 통해 현대 AI 기술 스택을 사용한 실용적인 에이전트 구축 핵심 기술을 마스터했습니다. 이 차트 생성 에이전트는 더 복잡한 데이터 분석 애플리케이션의 기반으로 활용할 수 있습니다.

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.