Skip to content

基于图的智能体工作流

Supported in ADKPython v2.0.0Alpha

ADK 中的基于图的工作流允许你以更高的精度构建智能体,创建将 AI 推理能力与确定性代码逻辑完美融合的可靠流程。通过这种模式,你可以将智能体逻辑定义为由执行节点 (Nodes)边 (Edges) 构成的图结构,从而实现 AI 智能体与工具、自定义代码的无缝协作。

基于图的航班升舱智能体设计

图 1. 结合了多种工作流节点(函数、人机协同、工具及 LLM 推理)的航班升舱智能体设计方案。

虽然 ADK 预置的工作流智能体(如顺序执行智能体)已经提供了跨智能体的流程控制,但它们通常仅限于简单的线性逻辑。如果你需要更精细、更灵活的任务调度与执行控制,图工作流将是你的不二之选。

基于图的工作流的核心优势:

  • 精确定义逻辑语义:通过明确勾勒路由逻辑,你可以微操不同节点间的状态转换。
  • 实现超复杂拓扑结构:支持多分支并行、条件循环以及复杂的全局状态管理。
  • 低成本函数链运行:在无需动用昂贵 LLM 推理的情况下,直接调用智能体工具或你的原生 Python 代码。
  • 极高的系统可靠性:通过依赖硬编码的节点路由而非变数极大的提示词 (Prompt),显著提升智能体的可预测性。

Alpha 发布版说明

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

请先查阅 ADK 2.0 安装指南,然后按照下文步骤开启你的图工作流之旅。

快速入门

本节将演示如何构建一个基础的顺序图工作流:该流程首先由 AI 智能体生成一个城市名称,接着通过函数节点查询该城市的时间,最后由另一个智能体播报结果。

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

# 1. 定义名称生成智能体
city_generator_agent = Agent(
    name="city_generator_agent",
    model="gemini-2.5-flash",
    instruction="""随机返回一个世界知名城市的名称。
      仅输出城市名,不要包含标点符号或前缀。""",
    output_schema=str,
)

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

# 3. 定义查询时间的函数节点
def lookup_time_function(node_input: str):
    """模拟获取指定城市的实时时间。"""
    # 此处你可以调用真实的外部 API
    return CityTime(time_info="上午 10:10", city=node_input)

# 4. 定义结果播报智能体
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,
)

# 5. 使用 Workflow 类编排工作流
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)
    ],
    # 节点执行顺序:START -> 生成 -> 查询 -> 播报
)

为什么不直接用提示词?

这段示例代码展示了如何使用 Workflow 类来组装一个简单的顺序工作流,并在 AI 智能体处理和代码执行之间交替进行。

虽然你也可以通过长提示词配合工具调用来实现上述功能,但图方法让你能够对每一步的执行顺序、数据契约(Schema)以及中间状态拥有绝对的控制权。

如果你想了解更多关于图工作流数据流转的细节,请参阅智能体工作流的数据处理

使用图构建复杂的业务流程

传统的提示词驱动智能体(Prompt-based Agents)通过文本指令来描述任务。然而,随着流程复杂度的增加,确保 LLM 严格遵守每一个微小的分支规则会变得极度困难且不可靠。

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

通过图工作流,你可以在代码中将每一个原子步骤明确定义为节点 (Node)。节点间的拓扑关系决定了任务的走向。下图清晰地展示了如何将一段复杂的文字指令重构为清晰的执行图:

提示词到图的转换示意图

图 2. 将复杂的基于提示词的指令结构重构为高可靠的基于图的工作流。

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

这种转变允许你显式地拆分任务链路,并根据需要在“非确定性的 AI 决策”与“确定性的硬业务逻辑”之间自如切换。

以下是实现 图 2 逻辑的代码示例:

# 获取分类智能体
classifier_agent = Agent(
    name="classifier",
    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 数组,你可以轻松定义包含智能体、工具、代码片段甚至是嵌套工作流的庞大复杂系统。欲深入了解高级路由技巧,请参阅为智能体工作流构建图路由

已知限制

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

  • 实时流 (Live Streaming):图路由节点暂未直接支持实时的音频/视频流交互。
  • 特定第三方集成:部分早期的第三方集成可能无法在图节点上下文中正常运行。