ADK 事件模型Event 与 NodeInfo深度指南会话、动作与工作流路由的底层数据结构【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python在 Google Agent Development KitADK中一次对话或工作流执行本质上是一串**事件Event**的有序流动用户输入、模型回复、函数调用与返回、状态更新、路由决策……这些信息都封装在Event对象中并最终持久化到会话Session里。本文以 docs/guides/events/event/index.md 为骨架结合 event.py 及其配套源码与单测系统讲解Event与NodeInfo的设计、字段、序列化规则和高级用法。读完你将能正确构造各类事件、理解便捷参数的路由机制、掌握 camelCase 序列化约定并能在工作流路由与上下文隔离场景中直接使用事件能力。Event 是什么一次对话/工作流的最小单元在 ADK 中会话与工作流执行被建模为一串事件的序列。Event类定义于 src/google/adk/events/event.py代表这个序列中的单个单元它从三个维度刻画一次交互Content内容用户与 Agent 之间交换的消息包括纯文本、函数调用function call、函数返回function response等存储于content字段类型为google.genai.types.Content。Actions动作事件携带的副作用或指令例如状态更新state_delta、路由决策route、Agent 转移transfer_to_agent、UI 渲染请求render_ui_widgets等封装在EventActions对象中见 event_actions.py。Metadata元数据关于谁生成了事件、何时生成、来自工作流的哪个部分的信息例如author、invocation_id、timestamp与node_info。NodeInfo专门承载生成该事件的工作流节点元数据节点路径、run ID 等用于追踪执行路径与运行标识。从源码依赖关系看Event是框架的血液Session通过events: list[Event]保存有序的完整事件历史见 session.py而Workflow/NodeRunner则依赖事件驱动执行流与状态管理。理解Event是理解 ADK 会话持久化、多 Agent 协作与工作流编排的起点。快速上手创建与使用 Event1. 基础文本事件最简单的用法是传入author与一条文本消息。message是content的便捷别名构造时会自动转换为types.Contentfrom google.adk.events.event import Event # 创建一个用户消息事件 user_event Event(authoruser, messageHello, agent!) # message 是 content 的便捷别名 print(user_event.message.parts[0].text) # 输出: Hello, agent!在 test_event.py 的TestMessageConstructor中单测验证了message参数可接受字符串、types.Content、types.Part以及 Part 列表全部会被转换为content字段。2. 携带状态增量State Delta的事件事件可以携带需要应用到会话状态上的更新。传入state字典后它会被路由到actions.state_deltafrom google.adk.events.event import Event # 创建一个更新状态的 Agent 事件 state_event Event( authormy_agent, messageIve updated the user preference., state{user_theme: dark} ) print(state_event.actions.state_delta) # 输出: {user_theme: dark}注意state_delta的语义是增量更新运行时会将该字典合并进会话状态而不是整体替换。在源码中state_delta: dict[str, Any] Field(default_factorydict)event_actions.py默认值为空字典Event()不带任何参数也能正常构造对应单测test_event_constructor_without_state。3. 携带节点元数据NodeInfo的事件当事件在工作流内部产生时通常都会带上节点信息。框架会自动填充NodeInfo[!NOTE]NodeInfo由 ADK 框架自动填充。你可以在应用逻辑中读取这些字段但不应手动构造或修改node_info。from google.adk.events.event import Event, NodeInfo node_event Event( authoragent_node, node_pathparent_workflow/child_noderun-123, outputsome_result ) print(node_event.node_info.path) # 输出: parent_workflow/child_noderun-123 print(node_event.node_info.name) # 输出: child_node print(node_event.node_info.run_id) # 输出: run-123node_path的格式为节点名run_id的层级路径name返回去掉run_id后的干净节点名run_id通过 _node_path_builder.py 中的_NodePathBuilder从路径字符串解析得出。工作原理Event 的内部机制继承自 LlmResponse直接包裹模型响应Event直接继承自LlmResponse定义于 llm_response.py这意味着它天然具备模型响应的一切能力content、grounding_metadata地面引用元数据、usage_metadataToken 用量、finish_reason、error_code/error_message、partial流式分片标记等。此外还继承了get_function_calls()与get_function_responses()两个方法用于按顺序提取事件内容中的函数调用与函数返回llm_response.py。这一设计让 Event 可以直接包裹 Gemini 等模型的生成结果无需二次转换。便捷参数路由_accept_convenience_kwargsEvent构造器接受若干便捷参数它们会被自动路由到嵌套的 Pydantic 模型中。路由逻辑由model_validator(modebefore)方法_accept_convenience_kwargs完成event.py便捷参数路由目标说明messagecontent经google.genai._transformers.t_content转换为types.Contentstateactions.state_delta字典形式的状态增量routeactions.route工作流图边的路由值node_pathnode_info.path节点路径字符串关于该路由有两个值得注意的细节互斥校验如果同时传入message与content会抛出ValueError提示message and content are mutually exclusive. Use one or the other.对应单测test_message_and_content_raises。非破坏性路由过程基于传入字典的拷贝进行改写Event.model_validate(data)不会修改调用方的原始 dict单测test_model_validate_does_not_mutate_input_dict专门验证了这一点。此外message还是一个可读写的 property读取时别名到content赋值时同样会经过类型转换若子类把message声明为真实字段则该 property 会优先路由到子类字段详见 event.py 与单测TestMessageSubclassField。序列化统一 camelCaseEvent与NodeInfo的 Pydantic 配置均使用alias_generatoralias_generators.to_camelpopulate_by_nameTrue即字段名在 Python 中是 snake_case序列化输出统一为 camelCaseevent Event( invocation_idi-1, node_infoNodeInfo(patha/b, output_for[c], message_as_outputTrue), ) dumped event.model_dump(by_aliasTrue) print(dumped[invocationId]) # i-1 print(dumped[nodeInfo][outputFor]) # [c] print(dumped[nodeInfo][messageAsOutput]) # True因此当通过网络发送事件或持久化保存时务必使用model_dump(by_aliasTrue)以保证与 ADK API 的兼容。单测test_event_serialization_always_camel_case会递归校验序列化结果中不存在任何 snake_case 键test_event.py。另一个序列化细节是long_running_tool_ids长运行函数调用 ID 集合因为set的迭代顺序随进程随机化默认序列化会导致同一事件每次输出顺序不同、破坏调试器或缓存的 diff 逻辑因此源码通过field_serializer将其稳定输出为排序后的列表event.py对应单测TestLongRunningToolIdsSerialization验证了排序与往返一致性。生命周期唯一 ID 与时间戳每个事件在初始化时都会被分配唯一的 UUIDid与timestamp除非显式传入。id的生成发生在model_post_init钩子中event.py通过Event.new_id()调用平台抽象层platform_uuid.new_uuid()生成timestamp默认值为platform_time.get_time()event.py。单测验证了ID 缺省时自动分配、两个事件 ID 互不相同、显式传入的idfixed-id会被保留test_event_id_*系列。深入核心字段Event 主要字段一览字段类型说明authorstruser或产生该事件的 Agent 名称invocation_idstr调用 ID追加进会话前应非空contenttypes.Content对话内容文本/函数调用/函数返回actionsEventActions动作集默认空对象outputAny工作流节点的通用数据输出node_infoNodeInfo工作流节点元数据path、run_id 等long_running_tool_idsset[str]长运行函数调用 ID 集合仅函数调用事件有效branchstr事件分支格式如agent_1.agent_2.agent_3用于隔离同级子 Agent 的会话历史isolation_scopestr逻辑上下文隔离标签Task API 内部机制勿直接使用id/timestampstr/float唯一标识与时间戳初始化时自动生成EventActions事件能携带哪些动作EventActionsevent_actions.py是事件动作维度的完整定义主要包括skip_summarization为 True 时不对函数响应调用模型做摘要仅函数响应事件使用state_delta状态增量更新默认{}artifact_delta制品文件版本更新键为文件名、值为版本号transfer_to_agent/transfer_reason转移至指定 Agent 及其原因escalate向更高级 Agent 升级requested_auth_configs工具响应请求的认证配置键为函数调用 IDrequested_tool_confirmations请求的工具确认键为函数调用 IDcompaction事件压缩信息EventCompaction含起止时间戳与压缩内容end_of_agent当前 Agent 本次运行结束同一 Agent 在一次调用中因循环可能出现多个为 True 的事件仅由 ADK 工作流设置agent_state当前事件的 Agent 状态快照用于 checkpoint 与恢复仅由 ADK 工作流设置rewind_before_invocation_id回滚目标调用 ID仅回滚事件设置route工作流图边匹配的路由值类型为bool / int / str / 列表之一render_ui_widgets需要 UI 渲染的控件列表set_model_response模型响应的结构化输出。另外两个动作字段state_delta与agent_state在序列化时都采用了modewrap的兜底策略若值中存在 Pydantic 无法序列化的对象如 Python 可调用对象会用其字符串表示替换保证整个事件结构仍可持久化而不崩溃event_actions.py。NodeInfo 与节点路径解析NodeInfoevent.py包含三个字段与三个派生属性path工作流节点路径。若工作流 A 下直接挂节点 BB 发出事件则路径为A/BAgent 状态事件的路径为A。output_for该事件输出同时算作哪些节点路径的输出如[wf/A1/B1, wf/A1]。message_as_output为 True 时事件内容本身就是节点输出无需单独的 output 事件。派生属性run_id、parent_run_id、name均由_NodePathBuilder解析得到_node_path_builder.py。该工具类还提供static_path剥离所有 run ID、parent、append、is_descendant_of、is_direct_child_of等层次关系判定方法是工作流引擎追踪动态节点状态的基础设施。高级应用工作流路由Workflow Routing工作流使用Event传递路由决策节点通过设置route路由到actions.route向工作流引擎发出下一条边该走哪条的信号。事件本身与路由行为解耦只需一行代码routing_event Event(authorrouter_node, routesuccess_path)从工作流实现看route的值会在节点执行后被取出用于图边匹配例如 _workflow.py 中读取子上下文route并据此选择后续边。route支持bool / int / str / 列表多种类型便于映射到不同类型的边匹配条件。上下文隔离Context Isolationisolation_scope字段被 Task API 用于隔离被委派 Agent 的对话。带有特定isolation_scope如task:fc-987的事件只对运行在同一 scope内的 Agent 可见从而防止委派任务看到主对话的完整历史scoped_event Event(authortask_agent, isolation_scopetask:fc-987)源码注释明确指出其典型用法委派的任务 Agent 以发起函数调用 IDfc_id为 scope只看到自己任务的事件与聊天协调者的更广对话相互隔离event.py。相关内容在 docs/guides/events/request_input/index.md 所属的事件体系文档中也有呼应。判断最终响应is_final_response()应用层和 UI 层经常需要判断某个事件是否为 Agent 的最终、面向用户回复。Event提供了is_final_response()帮助方法event.py其判定规则为若actions.skip_summarization为 True 或存在long_running_tool_ids直接判定为最终响应若存在error_code且非partial且无函数调用判定为最终响应否则需同时满足无函数调用、无函数返回、非partial分片、无尾部代码执行结果。注意当一次调用中多个 Agent 参与时每个 Agent 都可能有一个is_final_response()为 True 的事件。test_event.py 对全部判定分支纯文本、空事件、函数调用/返回、partial、尾部代码执行结果、错误事件等都有完整单测覆盖可作为判断行为的权威参考。限制与注意事项NodeInfo 由框架管理node_info字段以及node_path构造参数由 ADK 框架在工作流执行期间负责分配。生产代码中不应手动设置或修改node_info。内部字段勿依赖isolation_scope是内部实现细节可能在未通知的情况下变更外部代码不应读取、写入或依赖其语义。message 与 content 互斥不能在Event构造器中同时指定message与content否则抛出ValueError。序列化注意别名跨进程/跨网络传输事件时必须使用model_dump(by_aliasTrue)以获得 camelCase 输出直接按字段名保存会产生与 ADK API 不兼容的数据。延伸阅读事件实现源码src/google/adk/events/event.py、src/google/adk/events/event_actions.py、src/google/adk/events/_node_path_builder.py事件基类src/google/adk/models/llm_response.py会话对事件的使用src/google/adk/sessions/session.py行为验证单测tests/unittests/events/test_event.py官方指南docs/guides/events/event/index.md、docs/guides/events/request_input/index.md【免费下载链接】adk-pythonAn open-source, code-first Python toolkit for building, evaluating, and deploying sophisticated AI agents with flexibility and control.项目地址: https://gitcode.com/GitHub_Trending/ad/adk-python创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考