基于图的智能体工作流¶

Supported in ADKPython v2.0.0

ADK 中的基于图的智能体工作流让你可以更精确地控制智能体的构建，创建结合代码逻辑与 AI 推理能力的确定性流程。基于图的工作流允许你将智能体逻辑定义为由执行节点和边组成的图，将 AI 驱动的智能体推理与确定性工具和代码结合在一起。

基于图的航班升级智能体

图 1. 一个基于图的航班升级智能体设计，结合了不同类型的工作流节点，包括函数、人工输入、工具以及大语言模型能力。

预构建的 ADK 模板工作流（例如顺序智能体）仅提供跨一组智能体的定义流程控制。你可以继续使用长提示词和工具构建标准 ADK 智能体，并将其用于基于图的工作流智能体。当你需要更精确的控制时，工作流智能体图为任务的路由和执行方式提供了更大的灵活性。基于图的工作流具有以下优势：

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

开始使用¶

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

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-flash-latest",
    instruction="""Return the name of a random city.
      Return only the name, nothing else.""",
    output_schema=str,
)

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

def lookup_time_function(node_input: str):
    """Simulate returning the current time in the specified city."""
    return CityTime(time_info="10:10 AM", city=node_input)

city_report_agent = Agent(
    name="city_report_agent",
    model="gemini-flash-latest",
    input_schema=CityTime,
    instruction="""Output following line:
    It is {CityTime.time_info} in {CityTime.city} right now.""",
    output_schema=str,
)

def completed_message_function(node_input: str):
    return Event(
        message=f"{node_input}\n WORKFLOW COMPLETED.",
    )

root_agent = Workflow(
    name="root_agent",
    edges=[
        ("START", city_generator_agent, lookup_time_function,
          city_report_agent, completed_message_function)
    ],
)

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

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

使用图构建流程¶

你可以使用基于提示词的智能体，通过 ADK 智能体的 instruction 字段中描述的任务和流程来定义多步骤流程。然而，随着指令和流程变得越来越长和复杂，确保智能体遵循每个步骤和指导方针就变得更加困难和不可靠。

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

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

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

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

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

process_message = Agent(
    name="process_message",
    model="gemini-flash-latest",
    instruction="""Classify user message into either "BUG", "CUSTOMER_SUPPORT",
      or "LOGISTICS". If you think a message applies to more than one category,
      reply with a comma separated list of categories.
   """,
    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="Handling bug...")

def response_2_support():
    return Event(message="Handling customer support...")

def response_3_logistics():
    return Event(message="Handling logistics...")

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 数组定义一个包含节点间路由的图，这些节点是离散任务，可以包括智能体、工具、你的代码，甚至其他 Workflows。关于构建高级工作流图的信息，请参阅为工作流智能体构建图路由。

已知限制¶

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

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