Skip to content

顺序智能体

SequentialAgent

SequentialAgent 是一种工作流智能体,它按照列表中指定的顺序执行其子智能体。

当你希望执行按固定、严格的顺序进行时,请使用 SequentialAgent

示例

  • 你想构建一个可以总结任何网页的智能体,使用两个工具:get_page_contentssummarize_page。因为智能体必须先调用 get_page_contents 然后才能调用 summarize_page(你不能凭空总结!),所以你应该使用 SequentialAgent 构建你的智能体。

与其他工作流智能体一样,SequentialAgent 不是由 LLM 驱动的,因此它的执行方式是确定性的。也就是说,工作流智能体只关注它们的执行方式(即按顺序),而不是它们的内部逻辑;工作流智能体的工具或子智能体可能使用也可能不使用 LLM。

工作原理

当调用 SequentialAgentrun_async() 方法时,它执行以下操作:

  1. 迭代: 它按照提供的顺序遍历 sub_agents 列表。
  2. 子智能体执行: 对于列表中的每个子智能体,它调用该子智能体的 run_async() 方法。

顺序智能体

完整示例:代码开发流水线

考虑一个简化的代码开发流水线:

  • 代码编写智能体: 一个基于规范生成初始代码的 LlmAgent
  • 代码审查智能体: 一个检查生成代码中的错误、风格问题和是否遵守最佳实践的 LlmAgent。它接收代码编写智能体的输出。
  • 代码重构智能体: 一个接收已审查的代码(和审查者的评论)并重构它以提高质量和解决问题的 LlmAgent

SequentialAgent 非常适合这种情况:

SequentialAgent(sub_agents=[CodeWriterAgent, CodeReviewerAgent, CodeRefactorerAgent])

这确保了代码先被编写,然后被审查,最后被重构,按照严格、可靠的顺序进行。每个子智能体的输出通过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")