4. 智能体执行器¶

A2A 智能体如何处理请求和生成响应/事件的核心逻辑由智能体执行器处理。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： Helloworld 示例的 cancel 方法只是抛出异常，表示此基本智能体不支持取消。
```
async def cancel(
    self, context: RequestContext, event_queue: EventQueue
) -> None:
    raise Exception('cancel not supported')
```

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

4. 智能体执行器¶

AgentExecutor 接口¶

Helloworld 智能体执行器¶

`AgentExecutor` 接口¶