Skip to content

基于图的智能体工作流

Supported in ADKPython v2.0.0Alpha

ADK 中的基于图的工作流让你能以更精确的控制力构建智能体,创建结合了代码逻辑和 AI 推理能力的确定性流程。基于图的工作流允许你将智能体逻辑定义为执行节点和边 (edges) 的图,从而将 AI 驱动的智能体与确定性的工具及代码结合起来。

基于图的航班升舱智能体

图 1. 针对航班升舱的基于图的智能体设计,结合了不同类型的工作流节点,包括函数 (Functions)、人类输入、工具 (Tools) 和 LLM 能力。

预构建的 ADK 工作流智能体,例如 顺序智能体,仅在一组智能体之间提供定义好的流程控制。你可以继续构建带有长提示词、工具的标准 ADK 智能体,并在基于图的工作流智能体中使用它们。当你需要更精确的控制时,工作流智能体图在任务路由和执行方式上为你提供了更大的灵活性。基于图的工作流具有以下优势:

  • 定义精确逻辑:明确规划出路由逻辑,以管理不同节点之间的转换。
  • 实现复杂结构:构建支持分支和状态管理的智能体工作流。
  • 无需 AI 即可运行函数链:在不调用生成式 AI 模型的情况下调用智能体工具和你自己的代码。
  • 增强可靠性:通过依赖结构化节点定义而不是仅依赖提示词,提高智能体的可预测性。

Alpha 发布版

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

按照 安装 ADK 2.0 的说明操作,然后查看以下说明以开始使用基于图的工作流。

快速入门

本节介绍如何开始使用基于图的智能体。以下示例展示了如何创建一个顺序的基于图的智能体工作流,该工作流生成一个城市名称,通过代码函数查找该城市的当前时间,最后由智能体报告该信息。

from google.adk import Agent
from google.adk import Workflow
from google.adk import Event
from pydantic import BaseModel

# 定义生成城市名称的智能体
city_generator_agent = Agent(
    name="city_generator_agent",
    model="gemini-2.5-flash",
    instruction="""返回一个随机城市的名称。
      只返回名称,不要返回其他任何内容。""",
    output_schema=str,
)

# 定义城市时间数据模型
class CityTime(BaseModel):
    time_info: str  # 时间信息
    city: str       # 城市名称

# 定义查找时间的函数节点
def lookup_time_function(node_input: str):
    """模拟返回指定城市的当前时间。"""
    return CityTime(time_info="10:10 AM", city=node_input)

# 定义报告信息的智能体
city_report_agent = Agent(
    name="city_report_agent",
    model="gemini-2.5-flash",
    input_schema=CityTime,
    instruction="""输出以下行:
    现在 {CityTime.city} 的时间是 {CityTime.time_info}。""",
    output_schema=str,
)

# 定义完成消息函数
def completed_message_function(node_input: str):
    return Event(
        message=f"{node_input}\n 工作流执行完成。",
    )

# 创建并组合工作流
root_agent = Workflow(
    name="root_agent",
    edges=[
        ("START", city_generator_agent, lookup_time_function,
          city_report_agent, completed_message_function)
    ],
)

这段示例代码展示了如何使用 Workflow 类来组装一个简单的顺序工作流,并在 AI 智能体处理和代码执行之间交替进行。虽然你可以使用单个智能体配合较长的提示词和工具调用来执行这些步骤,但基于图的方法让你能对任务执行顺序和每一步的数据输出进行精确控制。

有关基于图的工作流中数据处理的更多信息,请参见 工作流节点与智能体的数据处理

使用图构建流程

你可以使用基于提示词的智能体,通过 ADK 智能体的 instructions 字段描述任务和步骤,来定义多步骤流程。然而,随着你的指令和流程变得越来越长、越来越复杂,确保智能体遵循每一个步骤和准则会变得更加复杂且不可靠。

基于图的工作流智能体相比于基于提示词的智能体具有显著优势,它允许你在代码中明确定义整体流程工作流。通过基于图的智能体工作流,流程的每一步都可以定义为图中的一个执行节点 (Node),每个节点可以是 AI 智能体、工具或你编写的代码。下图说明了一个简单的基于提示词的智能体如何转换为工作流智能体图:

从基于提示词的智能体到基于图的工作流

图 2. 将基于提示词的智能体指令结构转换为基于图的工作流。

从基于提示词的智能体转向基于图的工作流智能体,允许你显式地拆分流程中的任务,以定义特定的执行流。一旦定义完成,智能体应用程序就会按照图中的步骤流动,根据需要在非确定性的 AI 驱动智能体和确定性代码之间切换。

以下代码示例展示了图 2 中的工作流如何使用 Workflow 类转换为基于图的智能体:

# 定义分类智能体
process_message = Agent(
    name="process_message",
    model="gemini-2.5-flash",
    instruction="""将用户消息分类为 "BUG"、"CUSTOMER_SUPPORT" 
      或 "LOGISTICS"。如果你认为一条消息适用于多个类别,
      请以逗号分隔的类别列表形式回复。
   """,
    output_schema=str,
)

# 定义路由函数
def router(node_input: str):
    routes = node_input.split(",")
    routes = [route.strip() for route in routes]
    # 返回路由事件
    return Event(route=routes)

# 定义各分支的处理逻辑
def response_1_bug():
    return Event(message="正在处理 Bug...")

def response_2_support():
    return Event(message="正在处理客户支持...")

def response_3_logistics():
    return Event(message="正在处理物流...")

# 组合工作流
root_agent = Workflow(
   name="routing_workflow",
   edges=[
       ("START", process_message, router),
       ( router,
           {
               "BUG": response_1_bug,
               "CUSTOMER_SUPPORT": response_2_support,
               "LOGISTICS": response_3_logistics,
           }
       )
   ],
)

这段示例代码展示了如何使用 edges 数组来定义具有一组节点 (nodes) 之间路由的图,这些节点是离散的任务,可以包含智能体、工具、你的代码,甚至是额外的工作流 (Workflows)。有关为工作流构建高级图的信息,请参见 为工作流智能体构建图路由

已知限制

基于图的工作流存在一些已知限制。它们与以下 ADK 特性不兼容

  • 实时流 (Live Streaming) 功能与基于图的工作流不兼容。
  • 集成 (Integrations):某些第三方集成可能与基于图的工作流不兼容。