智能体工作流的数据处理¶

Supported in ADKPython v2.0.0Alpha

在智能体与基于图的节点之间构建和管理数据结构对于使用 ADK 构建可靠的流程至关重要。本指南解释了基于图的工作流和协作智能体中的数据处理方式，包括信息如何使用事件 (Events) 在图节点之间传输和接收。它涵盖了事件、数据、内容和状态的关键参数，并解释了如何使用数据格式架构 (Schema) 和特定的指令语法为函数节点和智能体节点实现结构化数据传输。

Alpha 发布版

ADK 2.0 是一个 Alpha 发布版，与之前的 ADK 版本一起使用时可能会导致破坏性变更。如果你需要向后兼容性（例如在生产环境中），请不要使用 ADK 2.0。我们鼓励你测试此版本，并欢迎提供你的反馈！

警告：请勿混用 ADK 2.0 和 ADK 1.0 的数据存储系统

如果你在 ADK 2.0 项目中使用持久化存储，请勿让 ADK 2.0 项目与 ADK 1.0 项目共享存储，包括但不限于会话存储、记忆系统和评估数据。这样做可能会导致数据丢失或使数据在 ADK 1.0 项目中无法使用。

工作流图中的事件 (Events)¶

在基于图的工作流中，你使用事件 (Events) 传递数据。工作流图中的所有执行节点都会消费和发送事件。本节涵盖了在工作流节点之间传输和接收数据的基本知识。

事件具有用于在节点之间传输不同类型数据的特定参数。节点数据处理的关键参数如下：

output：用于在节点之间传递信息的参数。
message：旨在作为对用户响应的数据。
state：在整个 ADK 会话期间，通过事件 (Events) 在节点间自动持久化的数据。

事件还携带有关工作流的其他信息，包括事件的源节点。

节点的事件输入与输出¶

图中的每个节点都通过 Event 类接收和传输数据。使用 yield 语法将数据移交给下一个节点，如下列代码片段所示：

from google.adk import Event

# 定义一个简单的函数节点
def my_function_node(node_input: str):
    output_value = node_input.upper()
    return Event(output=output_value) # 返回结果

当输出不需要额外处理的 Event 数据时，请使用 return 语法。当发送需要额外处理的数据，或者如果你正在生成多个数据项时，可以使用多个 yield 命令。每次 yield 调用都会向 Event 上的数据对象列表添加内容，并将其传递给图的下一个节点。不带参数的 return 或 yield 命令会向下一个节点传递一个 None 值。

事件的 output 参数¶

Event 的 output 参数是在图节点间传递数据的标准方式。下一个节点会接收到一个包含该数据的节点输入 (node input) 对象，如下列代码示例所示：

# 节点 1：生成结果
def my_function_node_1():
    return Event(output="The Result")

# 节点 2：处理输入
def my_function_node_2(node_input: Content):
    output_value = node_input.parts[0].text.lower()
    return Event(output=output_value) # 结果为 "the result"

你可以以可序列化的格式传递更长的结构化数据，如本代码示例所示：

# 传递结构化数据
def my_function_node_3():
    yield Event(
        output={
            "city_name": "Paris",
            "city_time": "10:10 AM",
        },
    )

注意：Event.output 限制

节点在每次执行时仅允许发送单个 Event.output 数据负载。这一限制意味着虽然你可以在一个节点中使用多个 yield，但如果拥有两个或更多带有 Event.output 的 yield 命令，将导致运行时错误。

事件的 message 参数¶

Event 的 message 参数用于传递旨在作为用户响应的数据。通常，你不应在智能体代码中使用 message 参数，除非该参数专门用于向用户提供信息或向用户请求信息。以下代码示例展示了如何在工作流执行期间向用户提供信息：

# 向用户发送消息
async def user_message(node_input: str):
  """告知用户研究过程正在开始。"""
  yield Event(message="正在开始研究过程...")

事件的 state 参数¶

Event 的 state 参数用于在整个 ADK 会话期间维护一组小型数据值。state 参数中的值在节点之间自动持久化，旨在引导更复杂工作流的执行。节点可以修改 state 值，修改后的 state 值对下游节点可用。以下代码示例展示了 state 如何跨节点持久化：

# 初始化状态
async def init_state_node(attempts: int = 0):
  yield Event(
      state={
          "attempts": attempts,
      },
  )

# 尝试任务并更新状态
async def task_attempt_node(node_input: Content, attempts: int):
  yield Event(
      state={
          "attempts": attempts + 1,
      },
  )

# 读取状态
async def read_state_node(ctx: WorkflowContext):
  print(f"当前尝试次数状态: {ctx.state}") # 输出: attempts: 1

# 定义工作流
root_agent = Workflow(
    name="root_agent",
    edges=[("START", init_state_node, task_attempt_node, read_state_node)],
)

注意：state 属性的数据限制

state 参数不应被用于在节点间持久化大量数据。请使用 artifact 或其他数据持久化机制（如数据库工具）在工作流的生命周期内持久化大型数据资源。

使用架构 (Schema) 约束节点的输入与输出数据¶

你可以设置输入和输出数据架构，以约束任何节点（包括函数节点 (FunctionNodes) 和智能体 (Agents)）的输入和输出数据格式。以下参数是任何节点的可选设置。你可以根据智能体项目的需要，在任何工作流节点上设置这两个参数或其中之一。

input_schema：使用扩展自 BaseModel 的类设置预期的输入架构。
output_schema：使用扩展自 BaseModel 的类设置要求的输出架构。

下面的代码示例展示了如何为子智能体设置输入和输出架构。

from google.adk import Agent
from pydantic import BaseModel

# 定义输入架构
class FlightSearchInput(BaseModel):
    origin: str           # 机场代码，例如 "SFO"
    destination: str      # 机场代码，例如 "CDG"
    departure_date: date  # 日期，例如 date(2026, 3, 15)
    passengers: int = 1   # 乘客人数

# 定义输出架构
class FlightSearchOutput(BaseModel):
    flights: list[Flight]
    cheapest_price: float

# 实例化带有架构的智能体
flight_searcher = Agent(
    name="flight_searcher",
    instruction="搜索可用航班。",
    input_schema=FlightSearchInput,
    output_schema=FlightSearchOutput,
    tools=[search_flights_api],
    mode="single-turn",
    ...
)

# 协调者智能体
assistant = Agent(
    name="assistant",
    instruction="你帮助用户规划行程。",
    sub_agents=[flight_searcher],
    ...
)

在智能体中访问结构化数据¶

当你从子智能体或工作流节点（如函数节点）将结构化数据传递到智能体时，可以使用特定语法将该数据添加到智能体的指令中。具体而言，你可以使用花括号 { } 选择输入架构属性，或者使用 < > 指定输入架构属性、from 关键字以及提供数据的节点名称。以下代码片段展示了两种包含通过智能体输入架构 (input schema) 传递的数据的方法：

# 定义城市时间架构
class CityTime(BaseModel):
    time_info: str  # 时间信息
    city: str       # 城市名称

# 函数节点：获取时间
def lookup_time_function(city: str):
    """模拟返回指定城市的当前时间。"""
    return Event(output=CityTime(time_info='10:10 AM', city=city))

# 智能体：报告城市时间
city_report_agent = Agent(
    name="city_report_agent",
    model="gemini-2.5-flash",
    input_schema=CityTime,

    # 方式一：基于类和参数的数据选择
    # instruction="""
    #     请按以下格式返回一句话：
    #     现在 {CityTime.city} 的时间是 {CityTime.time_info}。
    # """,

    # 方式二：基于源节点名称的更严格数据选择
    instruction="""
        请按以下格式返回一句话：
        现在 <CityTime.city from lookup_time_function> 的时间是
        <CityTime.time_info from lookup_time_function>。
    """,
)

# 根级工作流
root_agent = Workflow(
    name="root_agent",
    edges=[
        (START, city_generator_agent, lookup_time_function, city_report_agent)
    ],
)

有关此工作流的完整但简化版本，请参见基于图的智能体工作流。