顺序智能体¶
SequentialAgent
¶
SequentialAgent
是一种工作流智能体,它按照列表中指定的顺序执行其子智能体。
当你希望执行按固定、严格的顺序进行时,请使用 SequentialAgent
。
示例¶
- 你想构建一个可以总结任何网页的智能体,使用两个工具:
get_page_contents
和summarize_page
。因为智能体必须先调用get_page_contents
然后才能调用summarize_page
(你不能凭空总结!),所以你应该使用SequentialAgent
构建你的智能体。
与其他工作流智能体一样,SequentialAgent
不是由 LLM 驱动的,因此它的执行方式是确定性的。也就是说,工作流智能体只关注它们的执行方式(即按顺序),而不是它们的内部逻辑;工作流智能体的工具或子智能体可能使用也可能不使用 LLM。
工作原理¶
当调用 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")