4. Agent Executor

4. Agent Executor

A2A 에이전트가 요청을 처리하고 응답/이벤트를 생성하는 핵심 로직은 Agent Executor가 담당해요. A2A Python SDK는 여러분이 구현하는 추상 기본 클래스 a2a.server.agent_execution.AgentExecutor를 제공해요.

출처: 문서

본문

AgentExecutor 인터페이스

AgentExecutor 클래스는 두 가지 주요 메서드를 정의해요.

  • async def execute(self, context: RequestContext, event_queue: EventQueue): 응답이나 이벤트 스트림을 기대하는 들어오는 요청을 처리해요. 사용자 입력(context로 접근 가능)을 처리하고 event_queue를 사용해 Message, Task, TaskStatusUpdateEvent, TaskArtifactUpdateEvent 객체를 다시 보내요.
  • async def cancel(self, context: RequestContext, event_queue: EventQueue): 진행 중인 태스크를 취소하는 요청을 처리해요.

RequestContext는 사용자 메시지와 기존 태스크 세부사항 같은 들어오는 요청에 대한 정보를 제공해요. EventQueue는 실행기가 이벤트를 클라이언트에 다시 보내는 데 사용해요.

Helloworld Agent Executor

agent_executor.py를 살펴볼게요. HelloWorldAgentExecutor를 정의해요.

Agent(HelloWorldAgent): 실제 "비즈니스 로직"을 캡슐화하는 간단한 헬퍼 클래스예요.

class HelloWorldAgent:
    """Hello World Agent."""

    async def invoke(self, user_request: str) -> str:
        """Invoke the Hello World agent to generate a response."""
        return f'Hello, World! I have received your request ({user_request})'

"Hello, World!" 문자열을 반환하는 간단한 invoke 메서드가 있어요.

Executor(HelloWorldAgentExecutor): AgentExecutor 인터페이스를 구현하는 클래스예요.

__init__:

class HelloWorldAgentExecutor(AgentExecutor):
    """Test AgentProxy Implementation."""

    def __init__(self) -> None:
        self.agent = HelloWorldAgent()

HelloWorldAgent를 인스턴스화해요.

execute:

async def execute(
    self,
    context: RequestContext,
    event_queue: EventQueue,
) -> None:
    """Process user request."""
    # 1. Collect a task from request context
    if context.current_task:
        task = context.current_task
    else:
        # 1.1 If there is no task, create one and add it event queue
        task = new_task_from_user_message(context.message)
        await event_queue.enqueue_event(task)

    # 2. Update task status in EventQueue using TaskUpdater class object
    task_updater = TaskUpdater(
        event_queue=event_queue, task_id=task.id, context_id=task.context_id
    )
    await task_updater.update_status(
        state=TaskState.TASK_STATE_WORKING,
        message=new_text_message('Processing request...'),
    )

    # 3. Collect user request from request content and invoke LLM agent to generate content
    query = get_message_text(context.message)
    if query:
        result = await self.agent.invoke(user_request=query)
    else:
        result = 'No text input is provided!'

    # 4. Add generated response as an artifact to EventQueue
    await task_updater.add_artifact(parts=[new_text_part(text=result, media_type='text/plain')])
    print('Result: ', result)

    # 5. Update task status to completed
    await task_updater.update_status(
        state=TaskState.TASK_STATE_COMPLETED,
        message=new_text_message('Request is completed!'),
    )

Send Message 또는 Send Streaming Message 요청이 들어오면(이 단순화된 실행기에서는 둘 다 execute가 처리), 다음 단계가 일어나요.

  • 1단계: A2A instance(서버)가 컨텍스트에서 현재 태스크를 가져와요. 컨텍스트에 태스크가 없으면 새 태스크를 만들고 EventQueue에 추가해요.
  • 2단계: 에이전트가 처리를 시작했음을 나타내기 위해 TASK_STATE_WORKING 상태의 TaskStatusUpdateEvent를 큐에 넣어요.
  • 3단계: self.agent.invoke()를 호출해 실제 비즈니스 로직을 실행해요(단순히 "Hello, World!"를 반환).
  • 4단계: 에이전트의 결과 텍스트를 담은 TaskArtifactUpdateEvent를 큐에 넣어요.
  • 5단계: 마지막으로 태스크를 마무리하기 위해 TASK_STATE_COMPLETED 상태의 TaskStatusUpdateEvent를 큐에 넣어요.

AgentExecutor는 A2A 프로토콜(요청 핸들러와 서버 애플리케이션이 관리)과 여러분 에이전트의 특정 로직 사이의 다리 역할을 해요. 요청에 대한 컨텍스트를 받고, 이벤트 큐를 사용해 결과나 업데이트를 다시 전달하죠.

더 알아보기 (Learn more)