ADK Python Workflow 状态管理实战四种读写共享 State 的方式与源码级原理【免费下载链接】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本篇基于 ADKAgent Development Kit for Python官方示例workflows/state完整讲解Workflow 中共享状态state的四种读法与写法直接字典变更、通过Event增量更新、通过ctx.state读取、通过函数参数自动注入。文中结合 示例源码、事件快照 以及 ADK 工作流引擎的 State 类、FunctionNode 参数绑定逻辑 逐层拆解底层机制读完你可以理解 state 在事件流中如何以stateDelta形式持久化并能在自己项目中正确选用四种方式。一、为什么 Workflow 需要共享 State在 ADK Workflow 中数据在节点间流动有两条路径节点输出传递node output上游节点return或yield的值经边edge直接作为下游节点的node_input共享状态state一个在整个工作流执行期间共享的字典任何节点都可以写入和读取。当流程需要“在多个步骤中逐步收集信息”而不是简单地把上游输出原样传给下游时state 就是更合适的载体。示例 README 对它的定位是State is a dictionary shared across all nodes in the workflow execution, useful for gathering information across multiple steps without passing everything directly from one nodes output to anothers input.该示例演示了四种技术通过直接字典变更更新状态ctx.state[key] value通过 yield 一个事件更新状态yield Event(state{key: value})通过直接字典访问读取状态ctx.state[key]通过自动参数注入读取状态def func(key: str): ...二、示例工程结构示例位于 contributing/samples/workflows/state/ 目录文件作用agent.py定义 4 个节点函数和Workflow根代理tests/go.json一次完整运行的事件快照session events 最终 state可作为行为验证依据README.md本文的核心参考文档README 中给出了该示例的执行图Mermaid测试输入为Hello ADK!或Testing state management.README 中标注的 Sample Inputs。三、完整可运行代码一个四节点线性工作流下面是 agent.py 的核心代码可原样复制运行去掉许可证头后from google.adk import Event from google.adk import Workflow def process_initial_input(ctx, node_input: str): Takes initial input and sets it in state via direct dict modification. ctx.state[original_text] node_input return node_input def update_state_via_event(node_input: str): Returns an Event that implicitly updates the shared workflow state. yield Event(state{uppercased_text: node_input.upper()}) def read_state_via_ctx(ctx): Reads a state variable via direct dictionary access and appends to it. original ctx.state[original_text] uppercased ctx.state[uppercased_text] result f{uppercased} (Original was: {original}) ctx.state[appended_text] result return result def read_state_via_param(appended_text: str): Reads a state variable via automatic parameter injection. return fFinal Result: {appended_text}! root_agent Workflow( namestate_sample, edges[ ( START, process_initial_input, update_state_via_event, read_state_via_ctx, read_state_via_param, ), ], )两个要点边的写法edges接收EdgeItem可以是一个显式Edge对象也可以是“一个元组表示链式节点”。从 Graph 模块的类型定义 看ChainElement: TypeAlias NodeLike | tuple[NodeLike, ...] | RoutingMap EdgeItem: TypeAlias Edge | tuple[ChainElement, ...]因此(START, fn1, fn2, fn3, fn4)等价于START → fn1 → fn2 → fn3 → fn4的四条边。Workflow在构造时model_post_init会通过Graph.from_edge_items(self.edges)编译出图见 _workflow.py 的_build_graph。函数即节点每个普通 Python 函数都会被框架包装成FunctionNode。函数签名中名为ctx的参数会收到当前的Context名为node_input的参数会收到上游节点的输出其余参数则按 state 绑定规则自动注入下文第四节详述。四、四种 State 操作方式逐一解析4.1 写入方式一直接字典变更ctx.state[key] valuedef process_initial_input(ctx, node_input: str): ctx.state[original_text] node_input return node_inputctx.state返回的不是普通字典而是 State 类 的实例——一个“delta-aware”感知增量的状态容器。它的__setitem__实现同时写入当前值和待提交增量def __setitem__(self, key: str, value: Any) - None: Sets the value of the state dict for the given key. if self._schema is not None and isinstance(self._schema, type): _validate_state_entry(self._schema, key, value) self._value[key] value self._delta[key] value这带来两个关键行为写入即记录 delta所有改动都会进入_delta随后被挂到节点产出的事件上持久化见 4.3 节可选 schema 校验如果节点声明了state_schemaPydantic 模型每次写入都会经_validate_state_entry校验 key 是否存在于 schema、值是否符合字段类型违反时抛出StateSchemaError。此外任何包含:的 key如app:前缀会跳过校验这为app:、user:、temp:等前缀键保留了空间。4.2 写入方式二yield 带 state 的 Eventdef update_state_via_event(node_input: str): yield Event(state{uppercased_text: node_input.upper()})Event的state参数会被映射到actions.state_delta。在 Event 的序列化逻辑 中可以确认这一对应关系state: dict - actions.state_delta即在事件 JSON 里表现为actions.state_delta。这条路径的意义在于即使函数不接收ctx也能更新共享状态。FunctionNode._to_event在转换函数返回值/yield 值时会把ctx.actions.state_delta一并附着到事件上保证改动随事件持久化。注意该节点没有return只 yield 了一个无output的事件——从 tests/go.json 的事件e-3可见它只产生了stateDelta: {uppercased_text: GO}没有节点输出但下游节点仍通过边正常推进。4.3 两种写入方式如何进入事件流源码佐证运行一次输入为go的会话tests/go.json 记录了完整事件序列非常适合作为“state 如何落盘”的证据事件作者节点路径stateDelta输出e-2state_sampleprocess_initial_input1{original_text: go}goe-3state_sampleupdate_state_via_event1{uppercased_text: GO}无e-4state_sampleread_state_via_ctx1{appended_text: GO (Original was: go)}GO (Original was: go)e-5state_sampleread_state_via_param1无Final Result: GO (Original was: go)!可以看到无论是ctx.state直接赋值还是Event(state...)最终都以事件actions.state_delta的形式进入 session 事件流由会话服务统一持久化。会话最终 state 为state: { appended_text: GO (Original was: go), original_text: go, uppercased_text: GO }read_state_via_param作为终端节点无出边的输出会作为整个 Workflow 的输出由_finalize写入ctx.output见 _workflow.py 的_finalizeterminal node 没有出边的节点。4.4 读取方式一ctx.state[key]直接访问def read_state_via_ctx(ctx): original ctx.state[original_text] uppercased ctx.state[uppercased_text] result f{uppercased} (Original was: {original}) ctx.state[appended_text] result return result由于State.__getitem__优先查_delta再查_value同一个工作流执行中上游节点刚刚写入的值对下游立即可见无需等待任何提交过程。State还提供了get(key, default)、setdefault、update(delta)和has_delta()等字典式接口方便节点批量操作。4.5 读取方式二参数自动注入parameter bindingdef read_state_via_param(appended_text: str): return fFinal Result: {appended_text}!这是最“声明式”的读法只要函数参数名与 state 中的 key 同名ADK 就会自动注入该值。其实现位于 FunctionNode 的_bind_parametersFunctionNode的parameter_binding默认为state即非ctx/node_input参数一律从ctx.state中查找找到同名 key 后若参数有类型注解还会经 PydanticTypeAdapter做强制类型转换_coerce_param支持dict → BaseModel、Content → str等若 state 中不存在该 key 且参数没有默认值框架抛出WorkflowDataError提示 Missing value for parameter ...而不是静默传None——这为 state 键的拼写错误提供了显式失败。一个配套的保护机制在 Workflow 的_validate_state_schema当 Workflow 声明了state_schema时构造期就会检查所有FunctionNode的每个参数ctx、node_input、self除外是否都在 schema 字段中否则抛出StateSchemaError并列出已声明字段。换言之注入机制 schema 校验组合起来让 state 键在编译期/构造期就可被静态检查。4.6 四种方式对比与选用建议方式读/写是否需要ctx参数典型场景ctx.state[k] v写是顺手写入、与返回值一起产出yield Event(state{...})写否函数不关心 ctx、纯生成器风格ctx.state[k]/State.get读是需要读多个键、或键名需运行时计算参数注入def fn(k: T)读否依赖固定键想要类型转换与缺失即报错五、State 的生命周期与持久化边界结合 State 类 与 Context.state 属性 的文档字符串可以确认State 属于会话session而非单次调用。ctx.state的 docstring 明确写道The delta-aware state of the current session. For any state change, you can mutate this object directly, e.g. ctx.state[foo] bar。同一个 session 跨多轮对话时state 会持续存在改动以 delta 形式提交。State维护_value当前值与_delta未提交增量两层节点执行中产生的_delta会被挂到事件actions.state_delta由会话服务统一写入存储。go.json 中每个事件只携带本节点产生的增量正体现了这一点命名空间前缀State定义了APP_PREFIX app:、USER_PREFIX user:、TEMP_PREFIX temp:三个前缀常量。从源码结构看这些前缀键用于区分会话 state 的不同作用域例如temp:键通常只保留在会话内而不做跨用户持久化示例中使用的都是无前缀的普通键且校验逻辑对含:的键直接放行。一个需要注意的细节State.__setitem__中有一段 TODO 注释make new change only store in delta, so that self._value is only updated at the storage commit time说明当前实现下_value在写入时即被同步更新——即同一执行内读到的永远是最新值这是 4.4 节中“上游写入、下游立即可见”的直接原因。六、如何运行与验证以仓库内的示例为蓝本标准运行方式是将其组织为 ADK 应用目录agent 目录下暴露root_agent然后用 ADK 的命令行工具跑adk run app或adk web调试本文仓库为只读参考实际运行请将contributing/samples/workflows/state/的 agent.py 复制到你的项目结构中。验证行为时对照 tests/go.json 是最直接的方式输入go最终输出应为Final Result: GO (Original was: go)!事件流中e-2、e-3、e-4三个节点各贡献一次stateDeltae-5只产出最终输出、不再改动 state最终 sessionstate应包含original_text、uppercased_text、appended_text三个键。若把read_state_via_param的参数名改错例如appended_text写成appened_text运行时会立即得到WorkflowDataError而不是错误的最终结果——这正是参数注入方式的价值所在。七、小结Workflow 的 state 是贯穿全部节点的共享字典与节点输出传递互补前者适合跨步骤“收集/沉淀”中间结果后者适合严格的流水线传值写入有两条路径ctx.state赋值 /yield Event(state...)二者最终统一收敛为事件的actions.state_delta并由会话服务持久化读取也有两条路径ctx.state访问 / 同名参数注入参数注入由FunctionNode在parameter_bindingstate模式下自动完成并附带类型转换与缺失键报错若需要静态约束 state 键集合可为 Workflow 配置state_schema构造期即校验所有FunctionNode的注入参数是否合法StateSchemaError。参考文件README、agent.py、tests/go.json、Workflow 实现、FunctionNode、Graph 边定义、State 类、Context.state、Event。【免费下载链接】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），仅供参考 ADK Python Workflow 状态管理实战：四种读写共享 State 的方式与源码级原理
ADK Python Workflow 状态管理实战四种读写共享 State 的方式与源码级原理【免费下载链接】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本篇基于 ADKAgent Development Kit for Python官方示例workflows/state完整讲解Workflow 中共享状态state的四种读法与写法直接字典变更、通过Event增量更新、通过ctx.state读取、通过函数参数自动注入。文中结合 示例源码、事件快照 以及 ADK 工作流引擎的 State 类、FunctionNode 参数绑定逻辑 逐层拆解底层机制读完你可以理解 state 在事件流中如何以stateDelta形式持久化并能在自己项目中正确选用四种方式。一、为什么 Workflow 需要共享 State在 ADK Workflow 中数据在节点间流动有两条路径节点输出传递node output上游节点return或yield的值经边edge直接作为下游节点的node_input共享状态state一个在整个工作流执行期间共享的字典任何节点都可以写入和读取。当流程需要“在多个步骤中逐步收集信息”而不是简单地把上游输出原样传给下游时state 就是更合适的载体。示例 README 对它的定位是State is a dictionary shared across all nodes in the workflow execution, useful for gathering information across multiple steps without passing everything directly from one nodes output to anothers input.该示例演示了四种技术通过直接字典变更更新状态ctx.state[key] value通过 yield 一个事件更新状态yield Event(state{key: value})通过直接字典访问读取状态ctx.state[key]通过自动参数注入读取状态def func(key: str): ...二、示例工程结构示例位于 contributing/samples/workflows/state/ 目录文件作用agent.py定义 4 个节点函数和Workflow根代理tests/go.json一次完整运行的事件快照session events 最终 state可作为行为验证依据README.md本文的核心参考文档README 中给出了该示例的执行图Mermaid测试输入为Hello ADK!或Testing state management.README 中标注的 Sample Inputs。三、完整可运行代码一个四节点线性工作流下面是 agent.py 的核心代码可原样复制运行去掉许可证头后from google.adk import Event from google.adk import Workflow def process_initial_input(ctx, node_input: str): Takes initial input and sets it in state via direct dict modification. ctx.state[original_text] node_input return node_input def update_state_via_event(node_input: str): Returns an Event that implicitly updates the shared workflow state. yield Event(state{uppercased_text: node_input.upper()}) def read_state_via_ctx(ctx): Reads a state variable via direct dictionary access and appends to it. original ctx.state[original_text] uppercased ctx.state[uppercased_text] result f{uppercased} (Original was: {original}) ctx.state[appended_text] result return result def read_state_via_param(appended_text: str): Reads a state variable via automatic parameter injection. return fFinal Result: {appended_text}! root_agent Workflow( namestate_sample, edges[ ( START, process_initial_input, update_state_via_event, read_state_via_ctx, read_state_via_param, ), ], )两个要点边的写法edges接收EdgeItem可以是一个显式Edge对象也可以是“一个元组表示链式节点”。从 Graph 模块的类型定义 看ChainElement: TypeAlias NodeLike | tuple[NodeLike, ...] | RoutingMap EdgeItem: TypeAlias Edge | tuple[ChainElement, ...]因此(START, fn1, fn2, fn3, fn4)等价于START → fn1 → fn2 → fn3 → fn4的四条边。Workflow在构造时model_post_init会通过Graph.from_edge_items(self.edges)编译出图见 _workflow.py 的_build_graph。函数即节点每个普通 Python 函数都会被框架包装成FunctionNode。函数签名中名为ctx的参数会收到当前的Context名为node_input的参数会收到上游节点的输出其余参数则按 state 绑定规则自动注入下文第四节详述。四、四种 State 操作方式逐一解析4.1 写入方式一直接字典变更ctx.state[key] valuedef process_initial_input(ctx, node_input: str): ctx.state[original_text] node_input return node_inputctx.state返回的不是普通字典而是 State 类 的实例——一个“delta-aware”感知增量的状态容器。它的__setitem__实现同时写入当前值和待提交增量def __setitem__(self, key: str, value: Any) - None: Sets the value of the state dict for the given key. if self._schema is not None and isinstance(self._schema, type): _validate_state_entry(self._schema, key, value) self._value[key] value self._delta[key] value这带来两个关键行为写入即记录 delta所有改动都会进入_delta随后被挂到节点产出的事件上持久化见 4.3 节可选 schema 校验如果节点声明了state_schemaPydantic 模型每次写入都会经_validate_state_entry校验 key 是否存在于 schema、值是否符合字段类型违反时抛出StateSchemaError。此外任何包含:的 key如app:前缀会跳过校验这为app:、user:、temp:等前缀键保留了空间。4.2 写入方式二yield 带 state 的 Eventdef update_state_via_event(node_input: str): yield Event(state{uppercased_text: node_input.upper()})Event的state参数会被映射到actions.state_delta。在 Event 的序列化逻辑 中可以确认这一对应关系state: dict - actions.state_delta即在事件 JSON 里表现为actions.state_delta。这条路径的意义在于即使函数不接收ctx也能更新共享状态。FunctionNode._to_event在转换函数返回值/yield 值时会把ctx.actions.state_delta一并附着到事件上保证改动随事件持久化。注意该节点没有return只 yield 了一个无output的事件——从 tests/go.json 的事件e-3可见它只产生了stateDelta: {uppercased_text: GO}没有节点输出但下游节点仍通过边正常推进。4.3 两种写入方式如何进入事件流源码佐证运行一次输入为go的会话tests/go.json 记录了完整事件序列非常适合作为“state 如何落盘”的证据事件作者节点路径stateDelta输出e-2state_sampleprocess_initial_input1{original_text: go}goe-3state_sampleupdate_state_via_event1{uppercased_text: GO}无e-4state_sampleread_state_via_ctx1{appended_text: GO (Original was: go)}GO (Original was: go)e-5state_sampleread_state_via_param1无Final Result: GO (Original was: go)!可以看到无论是ctx.state直接赋值还是Event(state...)最终都以事件actions.state_delta的形式进入 session 事件流由会话服务统一持久化。会话最终 state 为state: { appended_text: GO (Original was: go), original_text: go, uppercased_text: GO }read_state_via_param作为终端节点无出边的输出会作为整个 Workflow 的输出由_finalize写入ctx.output见 _workflow.py 的_finalizeterminal node 没有出边的节点。4.4 读取方式一ctx.state[key]直接访问def read_state_via_ctx(ctx): original ctx.state[original_text] uppercased ctx.state[uppercased_text] result f{uppercased} (Original was: {original}) ctx.state[appended_text] result return result由于State.__getitem__优先查_delta再查_value同一个工作流执行中上游节点刚刚写入的值对下游立即可见无需等待任何提交过程。State还提供了get(key, default)、setdefault、update(delta)和has_delta()等字典式接口方便节点批量操作。4.5 读取方式二参数自动注入parameter bindingdef read_state_via_param(appended_text: str): return fFinal Result: {appended_text}!这是最“声明式”的读法只要函数参数名与 state 中的 key 同名ADK 就会自动注入该值。其实现位于 FunctionNode 的_bind_parametersFunctionNode的parameter_binding默认为state即非ctx/node_input参数一律从ctx.state中查找找到同名 key 后若参数有类型注解还会经 PydanticTypeAdapter做强制类型转换_coerce_param支持dict → BaseModel、Content → str等若 state 中不存在该 key 且参数没有默认值框架抛出WorkflowDataError提示 Missing value for parameter ...而不是静默传None——这为 state 键的拼写错误提供了显式失败。一个配套的保护机制在 Workflow 的_validate_state_schema当 Workflow 声明了state_schema时构造期就会检查所有FunctionNode的每个参数ctx、node_input、self除外是否都在 schema 字段中否则抛出StateSchemaError并列出已声明字段。换言之注入机制 schema 校验组合起来让 state 键在编译期/构造期就可被静态检查。4.6 四种方式对比与选用建议方式读/写是否需要ctx参数典型场景ctx.state[k] v写是顺手写入、与返回值一起产出yield Event(state{...})写否函数不关心 ctx、纯生成器风格ctx.state[k]/State.get读是需要读多个键、或键名需运行时计算参数注入def fn(k: T)读否依赖固定键想要类型转换与缺失即报错五、State 的生命周期与持久化边界结合 State 类 与 Context.state 属性 的文档字符串可以确认State 属于会话session而非单次调用。ctx.state的 docstring 明确写道The delta-aware state of the current session. For any state change, you can mutate this object directly, e.g. ctx.state[foo] bar。同一个 session 跨多轮对话时state 会持续存在改动以 delta 形式提交。State维护_value当前值与_delta未提交增量两层节点执行中产生的_delta会被挂到事件actions.state_delta由会话服务统一写入存储。go.json 中每个事件只携带本节点产生的增量正体现了这一点命名空间前缀State定义了APP_PREFIX app:、USER_PREFIX user:、TEMP_PREFIX temp:三个前缀常量。从源码结构看这些前缀键用于区分会话 state 的不同作用域例如temp:键通常只保留在会话内而不做跨用户持久化示例中使用的都是无前缀的普通键且校验逻辑对含:的键直接放行。一个需要注意的细节State.__setitem__中有一段 TODO 注释make new change only store in delta, so that self._value is only updated at the storage commit time说明当前实现下_value在写入时即被同步更新——即同一执行内读到的永远是最新值这是 4.4 节中“上游写入、下游立即可见”的直接原因。六、如何运行与验证以仓库内的示例为蓝本标准运行方式是将其组织为 ADK 应用目录agent 目录下暴露root_agent然后用 ADK 的命令行工具跑adk run app或adk web调试本文仓库为只读参考实际运行请将contributing/samples/workflows/state/的 agent.py 复制到你的项目结构中。验证行为时对照 tests/go.json 是最直接的方式输入go最终输出应为Final Result: GO (Original was: go)!事件流中e-2、e-3、e-4三个节点各贡献一次stateDeltae-5只产出最终输出、不再改动 state最终 sessionstate应包含original_text、uppercased_text、appended_text三个键。若把read_state_via_param的参数名改错例如appended_text写成appened_text运行时会立即得到WorkflowDataError而不是错误的最终结果——这正是参数注入方式的价值所在。七、小结Workflow 的 state 是贯穿全部节点的共享字典与节点输出传递互补前者适合跨步骤“收集/沉淀”中间结果后者适合严格的流水线传值写入有两条路径ctx.state赋值 /yield Event(state...)二者最终统一收敛为事件的actions.state_delta并由会话服务持久化读取也有两条路径ctx.state访问 / 同名参数注入参数注入由FunctionNode在parameter_bindingstate模式下自动完成并附带类型转换与缺失键报错若需要静态约束 state 键集合可为 Workflow 配置state_schema构造期即校验所有FunctionNode的注入参数是否合法StateSchemaError。参考文件README、agent.py、tests/go.json、Workflow 实现、FunctionNode、Graph 边定义、State 类、Context.state、Event。【免费下载链接】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），仅供参考