顺序型智能体
SequentialAgent
SequentialAgent是一种工作流智能体，它会按照列表中指定的顺序依次执行其子智能体。
当您需要以固定且严格的顺序执行任务时，请使用SequentialAgent。
示例场景
- 假设您需要构建一个能总结任意网页内容的智能体，该智能体需要使用两个工具：
get_page_contents和summarize_page。由于该智能体必须始终先调用get_page_contents再调用summarize_page（没有内容就无法生成摘要！），此时您应该使用SequentialAgent来构建这个智能体。
与其他工作流智能体相同，SequentialAgent并非由大模型驱动，因此其执行过程是完全确定性的。需要注意的是，工作流智能体仅关注执行顺序（即顺序执行），而不干涉内部逻辑——工作流智能体包含的工具或子智能体可能会使用大模型，也可能不会。
工作原理
当调用SequentialAgent的run_async()方法时，会执行以下操作：
- 顺序遍历：按照预设顺序遍历
sub_agents列表 - 子智能体执行：为列表中的每个子智能体调用其
run_async()方法
完整示例：代码开发流水线
来看一个简化的代码开发流程：
- 代码编写智能体：基于需求文档生成初始代码的
LlmAgent - 代码审查智能体：检查生成代码是否存在错误、风格问题及是否符合最佳实践的
LlmAgent，它会接收代码编写智能体的输出 - 代码重构智能体：根据审查意见对代码进行质量优化的
LlmAgent
这种情况非常适合使用SequentialAgent：
这样可以严格确保代码先被编写，再被审查，最后进行重构，执行顺序完全可靠。每个子智能体的输出会通过output_key机制存储在状态中传递给下一个智能体。
代码实现
from google.adk.agents.sequential_agent import SequentialAgent
from google.adk.agents.llm_agent import LlmAgent
from google.genai import types
from google.adk.sessions import InMemorySessionService
from google.adk.runners import Runner
# --- Constants ---
APP_NAME = "code_pipeline_app"
USER_ID = "dev_user_01"
SESSION_ID = "pipeline_session_01"
GEMINI_MODEL = "gemini-2.0-flash"
# --- 1. Define Sub-Agents for Each Pipeline Stage ---
# Code Writer Agent
# Takes the initial specification (from user query) and writes code.
code_writer_agent = LlmAgent(
name="CodeWriterAgent",
model=GEMINI_MODEL,
instruction="""You are a Code Writer AI.
Based on the user's request, write the initial Python code.
Output *only* the raw code block.
""",
description="Writes initial code based on a specification.",
# Stores its output (the generated code) into the session state
# under the key 'generated_code'.
output_key="generated_code"
)
# Code Reviewer Agent
# Takes the code generated by the previous agent (read from state) and provides feedback.
code_reviewer_agent = LlmAgent(
name="CodeReviewerAgent",
model=GEMINI_MODEL,
instruction="""You are a Code Reviewer AI.
Review the Python code provided in the session state under the key 'generated_code'.
Provide constructive feedback on potential errors, style issues, or improvements.
Focus on clarity and correctness.
Output only the review comments.
""",
description="Reviews code and provides feedback.",
# Stores its output (the review comments) into the session state
# under the key 'review_comments'.
output_key="review_comments"
)
# Code Refactorer Agent
# Takes the original code and the review comments (read from state) and refactors the code.
code_refactorer_agent = LlmAgent(
name="CodeRefactorerAgent",
model=GEMINI_MODEL,
instruction="""You are a Code Refactorer AI.
Take the original Python code provided in the session state key 'generated_code'
and the review comments found in the session state key 'review_comments'.
Refactor the original code to address the feedback and improve its quality.
Output *only* the final, refactored code block.
""",
description="Refactors code based on review comments.",
# Stores its output (the refactored code) into the session state
# under the key 'refactored_code'.
output_key="refactored_code"
)
# --- 2. Create the SequentialAgent ---
# This agent orchestrates the pipeline by running the sub_agents in order.
code_pipeline_agent = SequentialAgent(
name="CodePipelineAgent",
sub_agents=[code_writer_agent, code_reviewer_agent, code_refactorer_agent]
# The agents will run in the order provided: Writer -> Reviewer -> Refactorer
)
# Session and Runner
session_service = InMemorySessionService()
session = session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
runner = Runner(agent=code_pipeline_agent, app_name=APP_NAME, session_service=session_service)
# Agent Interaction
def call_agent(query):
content = types.Content(role='user', parts=[types.Part(text=query)])
events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
for event in events:
if event.is_final_response():
final_response = event.content.parts[0].text
print("Agent Response: ", final_response)
call_agent("perform math addition")