基于图的智能体工作流¶
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):图路由节点暂未直接支持实时的音频/视频流交互。
- 特定第三方集成:部分早期的第三方集成可能无法在图节点上下文中正常运行。