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

本篇基于 ADK（Agent 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 node's output to another's input.

该示例演示了四种技术：

1. 通过直接字典变更更新状态：ctx.state["key"] = "value"
2. 通过 yield 一个事件更新状态：yield Event(state={"key": "value"})
3. 通过直接字典访问读取状态：ctx.state["key"]
4. 通过自动参数注入读取状态：def func(key: str): ...

二、示例工程结构

示例位于 contributing/samples/workflows/state/ 目录：

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 f"Final Result: {appended_text}!" root_agent = Workflow( name="state_sample", edges=[ ( "START", process_initial_input, update_state_via_event, read_state_via_ctx, read_state_via_param, ), ], )

两个要点：

1. 边的写法：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。

2. 函数即节点：每个普通 Python 函数都会被框架包装成FunctionNode。函数签名中名为ctx的参数会收到当前的Context，名为node_input的参数会收到上游节点的输出；其余参数则按 state 绑定规则自动注入（下文第四节详述）。

四、四种 State 操作方式逐一解析

4.1 写入方式一：直接字典变更ctx.state["key"] = "value"

def process_initial_input(ctx, node_input: str): ctx.state["original_text"] = node_input return node_input

ctx.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_schema（Pydantic 模型），每次写入都会经_validate_state_entry校验 key 是否存在于 schema、值是否符合字段类型，违反时抛出StateSchemaError。此外任何包含:的 key（如app:前缀）会跳过校验，这为app:、user:、temp:等前缀键保留了空间。

4.2 写入方式二：yield 带 state 的 Event

def 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 如何落盘”的证据：

可以看到：无论是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 的_finalize：terminal 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 binding）

def read_state_via_param(appended_text: str): return f"Final Result: {appended_text}!"

这是最“声明式”的读法：只要函数参数名与 state 中的 key 同名，ADK 就会自动注入该值。其实现位于 FunctionNode 的_bind_parameters：

• FunctionNode的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 四种方式对比与选用建议

五、State 的生命周期与持久化边界

结合 State 类 与 Context.state 属性 的文档字符串，可以确认：

1. 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 会持续存在；
2. 改动以 delta 形式提交。State维护_value（当前值）与_delta（未提交增量）两层，节点执行中产生的_delta会被挂到事件actions.state_delta，由会话服务统一写入存储。go.json 中每个事件只携带本节点产生的增量，正体现了这一点；
3. 命名空间前缀：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三个节点各贡献一次stateDelta，e-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_binding="state"模式下自动完成，并附带类型转换与缺失键报错；
• 若需要静态约束 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