4. Agent Executor¶

A2A agent 处理请求并生成响应/事件的核心逻辑由 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) -> str:
        return 'Hello World'
```
它有一个简单的 invoke 方法，返回字符串 "Hello World"。
Executor (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()
    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： Helloworld 示例的 cancel 方法只是抛出一个异常，表明这个基本 agent 不支持取消操作。
```
async def cancel(
    self, context: RequestContext, event_queue: EventQueue
) -> None:
    raise Exception('cancel not supported')
```

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

4. Agent Executor¶

AgentExecutor 接口¶

Helloworld Agent Executor¶

`AgentExecutor` 接口¶