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.py。它定义了 HelloWorldAgentExecutor。

代理 (HelloWorldAgent): 这是一个简单的辅助类，封装了实际的“业务逻辑”。
```
class HelloWorldAgent:
    """Hello World Agent."""

    async def invoke(self) -> str:
        return 'Hello World'
```
它有一个简单的 invoke 方法，返回字符串“Hello World”。
执行器 (HelloWorldAgentExecutor): 此类实现了 AgentExecutor 接口。
- __init__:
```
class HelloWorldAgentExecutor(AgentExecutor):
    """Test AgentProxy Implementation."""

    def __init__(self):
        self.agent = HelloWorldAgent()
```
  它实例化了 HelloWorldAgent。
- execute:
```
async def execute(
    self,
    context: RequestContext,
    event_queue: EventQueue,
) -> None:
    result = await self.agent.invoke()
    await event_queue.enqueue_event(new_agent_text_message(result))
```
  当 message/send 或 message/stream 请求传入时（两者在此简化执行器中都由 execute 处理）
  1. 它调用 self.agent.invoke() 获取“Hello World”字符串。
  2. 它使用 new_agent_text_message 实用函数创建一个 A2A Message 对象。
  3. 它将此消息入队到 event_queue。底层的 DefaultRequestHandler 将处理此队列以将响应发送到客户端。对于像这样的单个消息，它将导致 message/send 的单个响应或 message/stream 的单个事件，然后流关闭。
- cancel: Hello World 示例的 cancel 方法只是引发一个异常，表明此基本代理不支持取消。
```
async def cancel(
    self, context: RequestContext, event_queue: EventQueue
) -> None:
    raise Exception('cancel not supported')
```

AgentExecutor 充当 A2A 协议（由请求处理程序和服务器应用程序管理）与您的代理特定逻辑之间的桥梁。它接收有关请求的上下文，并使用事件队列将结果或更新传回。

4. Agent Executor¶

AgentExecutor 接口¶

Helloworld 代理执行器¶

`AgentExecutor` 接口¶