事件¶

事件是智能体开发工具包（ADK）中信息流动的基本单位。它们代表智能体交互生命周期中的每一个重要事件，从初始用户输入到最终响应，以及中间的所有步骤。理解事件至关重要，因为它们是组件通信、状态管理和控制流方向的主要方式。

什么是事件以及为什么它们很重要¶

ADK 中的Event是一个不可变记录，代表智能体执行过程中的特定点。它捕获用户消息、智能体回复、请求使用工具（函数调用）、工具结果、状态变化、控制信号和错误。技术上，它是google.adk.events.Event类的一个实例，该类在基本LlmResponse结构的基础上增加了必要的 ADK 特定元数据和actions载荷。

# Event 的概念结构
# from google.adk.events import Event, EventActions
# from google.genai import types

# class Event(LlmResponse): # 简化视图
#     # --- LlmResponse 字段 ---
#     content: Optional[types.Content]
#     partial: Optional[bool]
#     # ... 其他响应字段 ...

#     # --- ADK 特定添加项 ---
#     author: str          # 'user'或智能体名称
#     invocation_id: str   # 整个交互运行的 ID
#     id: str              # 此特定事件的唯一 ID
#     timestamp: float     # 创建时间
#     actions: EventActions # 重要的副作用和控制
#     branch: Optional[str] # 层次路径
#     # ...

事件对 ADK 操作至关重要，主要有几个原因：

通信： 它们作为用户界面、Runner、智能体、LLM 和工具之间的标准消息格式。所有内容都以Event形式流动。
信号状态和制品（Artifacts）变化： 事件通过event.actions.state_delta携带状态修改指令，并通过event.actions.artifact_delta跟踪制品（Artifacts）更新。SessionService使用这些信号确保持久性。
控制流： 特定字段如event.actions.transfer_to_agent或event.actions.escalate作为信号引导框架，决定下一步执行哪个智能体或是否应终止循环。
历史和可观察性： 在session.events中记录的事件序列提供了交互的完整、按时间顺序排列的历史记录，对于调试、审计和逐步理解智能体行为非常宝贵。

本质上，从用户查询到智能体最终答案的整个过程，都是通过Event对象的生成、解释和处理来编排的。

理解和使用事件¶

作为开发者，你主要会与Runner产生的事件流进行交互。以下是如何理解和从中提取信息：

识别事件来源和类型¶

通过检查以下内容快速确定事件代表什么：

谁发送了它？（event.author）
- 'user'：表示直接来自最终用户的输入。
- 'AgentName'：表示来自特定智能体的输出或动作（例如，'WeatherAgent'，'SummarizerAgent'）。
主要载荷是什么？（event.content和event.content.parts）
- 文本： 如果存在event.content.parts[0].text，这可能是一条会话消息。
- 工具调用请求： 检查event.get_function_calls()。如果不为空，表示 LLM 正在请求执行一个或多个工具。列表中的每项都有.name和.args。
- 工具结果： 检查event.get_function_responses()。如果不为空，此事件携带来自工具执行的结果。每项都有.name和.response（工具返回的字典）。注意： 对于历史结构，content内部的role通常是'user'，但事件的author通常是请求工具调用的智能体。
是否为流式输出？（event.partial）
- True：这是来自 LLM 的不完整文本块；后面还会有更多。
- False或None：此部分内容已完成（尽管如果turn_complete也为 false，整体回合可能尚未完成）。

# 伪代码：基本事件识别
# async for event in runner.run_async(...):
#     print(f"事件来源：{event.author}")
#
#     if event.content and event.content.parts:
#         if event.get_function_calls():
#             print("  类型：工具调用请求")
#         elif event.get_function_responses():
#             print("  类型：工具结果")
#         elif event.content.parts[0].text:
#             if event.partial:
#                 print("  类型：流式文本块")
#             else:
#                 print("  类型：完整文本消息")
#         else:
#             print("  类型：其他内容（例如，代码结果）")
#     elif event.actions and (event.actions.state_delta or event.actions.artifact_delta):
#         print("  类型：状态/制品（Artifacts）更新")
#     else:
#         print("  类型：控制信号或其他")

提取关键信息¶

一旦知道了事件类型，就可以访问相关数据：

文本内容： text = event.content.parts[0].text（始终先检查event.content和event.content.parts）。

函数调用详情：

calls = event.get_function_calls()
if calls:
    for call in calls:
        tool_name = call.name
        arguments = call.args # 这通常是一个字典
        print(f"  工具：{tool_name}，参数：{arguments}")
        # 应用程序可能根据此分派执行

函数响应详情：

responses = event.get_function_responses()
if responses:
    for response in responses:
        tool_name = response.name
        result_dict = response.response # 工具返回的字典
        print(f"  工具结果：{tool_name} -> {result_dict}")

标识符：
- event.id：此特定事件实例的唯一 ID。
- event.invocation_id：该事件所属的整个用户请求到最终响应周期的 ID。对于日志记录和跟踪很有用。

检测动作和副作用¶

event.actions对象表示发生或应该发生的变化。在访问其字段之前，始终检查event.actions是否存在。

状态变化： delta = event.actions.state_delta提供了一个{key: value}对字典，表示在产生此事件的步骤中会话状态发生的修改。

if event.actions and event.actions.state_delta:
    print(f"  状态变化：{event.actions.state_delta}")
    # 如有必要，更新本地 UI 或应用程序状态

制品（Artifacts）保存： artifact_changes = event.actions.artifact_delta提供了一个{filename: version}字典，指示保存了哪些制品（Artifacts）及其新的版本号。

if event.actions and event.actions.artifact_delta:
    print(f"  已保存制品（Artifacts）：{event.actions.artifact_delta}")
    # UI 可能刷新制品（Artifacts）列表

控制流信号： 检查布尔标志或字符串值：

event.actions.transfer_to_agent（字符串）：控制应传递给指定的智能体。
event.actions.escalate（布尔值）：循环应终止。

event.actions.skip_summarization（布尔值）：工具结果不应由 LLM 进行总结。

if event.actions:
    if event.actions.transfer_to_agent:
        print(f"  信号：转移到{event.actions.transfer_to_agent}")
    if event.actions.escalate:
        print("  信号：升级（终止循环）")
    if event.actions.skip_summarization:
        print("  信号：跳过工具结果总结")

确定事件是否为"最终"响应¶

使用内置辅助方法event.is_final_response()来识别适合作为智能体一轮完整输出显示的事件。

目的： 过滤掉中间步骤（如工具调用、部分流式文本、内部状态更新），只保留最终面向用户的消息。
何时为True？
1. 事件包含工具结果（function_response）且skip_summarization为True。
2. 事件包含针对标记为is_long_running=True的工具的工具调用（function_call）。
3. 或者，以下全部条件满足：
  - 没有函数调用（get_function_calls()为空）。
  - 没有函数响应（get_function_responses()为空）。
  - 不是部分流块（partial不是True）。
  - 不以可能需要进一步处理/显示的代码执行结果结束。

用法： 在应用程序逻辑中过滤事件流。

# 伪代码：在应用程序中处理最终响应
# full_response_text = ""
# async for event in runner.run_async(...):
#     # 如果需要，累积流式文本...
#     if event.partial and event.content and event.content.parts and event.content.parts[0].text:
#         full_response_text += event.content.parts[0].text
#
#     # 检查是否为最终的、可显示的事件
#     if event.is_final_response():
#         print("\n--- 检测到最终输出 ---")
#         if event.content and event.content.parts and event.content.parts[0].text:
#              # 如果是流的最终部分，使用累积的文本
#              final_text = full_response_text + (event.content.parts[0].text if not event.partial else "")
#              print(f"向用户显示：{final_text.strip()}")
#              full_response_text = "" # 重置累加器
#         elif event.actions.skip_summarization:
#              # 如果需要，处理显示原始工具结果
#              response_data = event.get_function_responses()[0].response
#              print(f"显示原始工具结果：{response_data}")
#         elif event.long_running_tool_ids:
#              print("显示消息：工具正在后台运行...")
#         else:
#              # 如果适用，处理其他类型的最终响应
#              print("显示：非文本最终响应或信号。")

通过仔细检查事件的这些方面，你可以构建强大的应用程序，适当地响应 ADK 系统中流动的丰富信息。

事件如何流动：生成和处理¶

事件在不同点创建，并由框架系统地处理。理解这个流程有助于阐明行动和历史是如何管理的。

生成源：
- 用户输入： Runner通常将初始用户消息或对话中途输入包装成author='user'的Event。
- 智能体逻辑： 智能体（BaseAgent、LlmAgent）明确yield Event(...)对象（设置author=self.name）以传达响应或信号动作。
- LLM 响应： ADK 模型集成层（例如，google_llm.py）将原始 LLM 输出（文本、函数调用、错误）转换为Event对象，由调用智能体创作。
- 工具结果： 工具执行后，框架生成包含function_response的Event。author通常是请求工具的智能体，而content内的role设置为'user'作为 LLM 历史。
处理流程：
1. 产出： 由源生成并产出事件。
2. Runner 接收： 执行智能体的主Runner接收事件。
3. SessionService 处理（append_event）： Runner将事件发送到配置的SessionService。这是关键步骤：
  - 应用增量： 服务将event.actions.state_delta合并到session.state中，并根据event.actions.artifact_delta更新内部记录。（注：实际的制品（Artifacts）保存通常在context.save_artifact被调用时更早完成）。
  - 完成元数据： 如果不存在，分配唯一的event.id，可能更新event.timestamp。
  - 持久化到历史： 将处理后的事件追加到session.events列表。
4. 外部产出： Runner将处理后的事件向外产出到调用应用程序（例如，调用runner.run_async的代码）。

此流程确保状态变更和历史与每个事件的通信内容一致地记录。

常见事件示例（说明性模式）¶

以下是你可能在流中看到的典型事件的简明示例：

用户输入：

{
  "author": "user",
  "invocation_id": "e-xyz...",
  "content": {"parts": [{"text": "预订下周二去伦敦的航班"}]}
  // actions 通常为空
}

智能体最终文本响应： (is_final_response() == True)

{
  "author": "TravelAgent",
  "invocation_id": "e-xyz...",
  "content": {"parts": [{"text": "好的，我可以帮您。您能确认出发城市吗？"}]},
  "partial": false,
  "turn_complete": true
  // actions 可能包含 state delta 等
}

智能体流式文本响应： (is_final_response() == False)

{
  "author": "SummaryAgent",
  "invocation_id": "e-abc...",
  "content": {"parts": [{"text": "该文档讨论了三个主要点："}]},
  "partial": true,
  "turn_complete": false
}
// ... 后续有更多 partial=True 的事件 ...

工具调用请求（由 LLM）： (is_final_response() == False)

{
  "author": "TravelAgent",
  "invocation_id": "e-xyz...",
  "content": {"parts": [{"function_call": {"name": "find_airports", "args": {"city": "London"}}}]}
  // actions 通常为空
}

提供工具结果（给 LLM）： (is_final_response()取决于skip_summarization)

{
  "author": "TravelAgent",
  "invocation_id": "e-xyz...",
  "content": {
    "role": "user", // LLM 历史的角色
    "parts": [{"function_response": {"name": "find_airports", "response": {"result": ["LHR", "LGW", "STN"]}}}]
  }
  // actions 可能具有 skip_summarization=True 属性
}

状态更新事件： （is_final_response() == False)

{
  "author": "InternalUpdater",
  "invocation_id": "e-def...",
  "content": null,
  "actions": {
    "state_delta": {"user_status": "verified"},
    "artifact_delta": {"verification_doc.pdf": 2}
  }
}

智能体转移（切换）信号： （is_final_response() == False）

{
  "author": "OrchestratorAgent",
  "invocation_id": "e-789...",
  "content": {"parts": [{"function_call": {"name": "transfer_to_agent", "args": {"agent_name": "BillingAgent"}}}]},
  "actions": {"transfer_to_agent": "BillingAgent"} // 由框架添加
}

循环升级信号： (is_final_response() == False)

{
  "author": "CheckerAgent",
  "invocation_id": "e-loop...",
  "content": {"parts": [{"text": "Maximum retries reached."}]}, // 可选内容
  "actions": {"escalate": true}
}

事件上下文和详细信息¶

除了核心概念外，以下是一些对特定用例重要的关于上下文和事件的具体细节：

ToolContext.function_call_id（链接工具动作）：
- 当 LLM 请求工具（FunctionCall）时，该请求有一个 ID。提供给你的工具函数的ToolContext包含这个function_call_id。
- 重要性： 这个 ID 对于将认证等操作（request_credential、get_auth_response）链接回发起它们的特定工具请求至关重要，特别是在一个回合中调用多个工具时。框架在内部使用这个 ID。
状态/制品（Artifacts）变更如何被记录：
- 当你使用CallbackContext或ToolContext修改状态（context.state['key'] = value）或保存制品（Artifacts）（context.save_artifact(...)）时，这些更改不会立即写入持久存储。
- 相反，它们会填充EventActions对象内的state_delta和artifact_delta字段。
- 这个EventActions对象被附加到更改后生成的下一个事件（例如，智能体的响应或工具结果事件）。
- SessionService.append_event方法从传入事件中读取这些增量，并将它们应用到会话的持久状态和制品（Artifacts）记录中。这确保更改在时间上与事件流相关联。
状态作用域前缀（app:、user:、temp:）：
- 当通过context.state管理状态时，你可以选择使用前缀：
  - app:my_setting：表示与整个应用程序相关的状态（需要持久化的SessionService）。
  - user:user_preference：表示在多个会话中与特定用户相关的状态（需要持久化的SessionService）。
  - temp:intermediate_result或无前缀：通常是会话特定的或当前调用的临时状态。
- 底层的SessionService决定如何处理这些前缀以实现持久性。
错误事件：
- Event可以表示错误。检查event.error_code和event.error_message字段（从LlmResponse继承）。
- 错误可能源自 LLM（例如，安全过滤器、资源限制）或如果工具严重失败，可能由框架打包。检查工具的FunctionResponse内容以获取典型的工具特定错误。
```
// 错误事件示例（概念性）
{
  "author": "LLMAgent",
  "invocation_id": "e-err...",
  "content": null,
  "error_code": "SAFETY_FILTER_TRIGGERED",
  "error_message": "由于安全设置，响应被阻止。",
  "actions": {}
}
```

这些细节为涉及工具认证、状态持久性范围和事件流中的错误处理的高级用例提供了更完整的画面。

使用事件的最佳实践¶

要在 ADK 应用程序中有效使用事件：

明确作者身份： 构建自定义智能体（BaseAgent）时，确保yield Event(author=self.name, ...)以正确地将智能体动作归属于历史记录。对于 LLM/工具事件，框架通常正确处理作者身份。
语义内容和动作： 使用event.content表示核心消息/数据（文本、函数调用/响应）。专门使用event.actions表示副作用（状态/制品（Artifacts）增量）或控制流（transfer、escalate、skip_summarization）。
幂等性意识： 理解SessionService负责应用event.actions中信号的状态/制品（Artifacts）更改。虽然 ADK 服务旨在保持一致性，但如果应用程序逻辑重新处理事件，请考虑潜在的下游影响。
使用is_final_response()： 在应用程序/UI 层依赖这个辅助方法来识别完整的、面向用户的文本响应。避免手动复制其逻辑。
利用历史记录： session.events列表是你的主要调试工具。检查作者、内容和动作的序列，以跟踪执行并诊断问题。
使用元数据： 使用invocation_id关联单个用户交互中的所有事件。使用event.id引用特定的、唯一的事件。

将事件视为具有明确内容和动作目的的结构化消息，是构建、调试和管理 ADK 中复杂智能体行为的关键。