Class SequentialAgent

src/google/adk/agents/sequential_agent.py:53–173

A shell agent that runs its sub-agents in sequence. .. deprecated:: SequentialAgent is deprecated and will be removed in future versions. Please use Workflow instead.

    ' Please use Workflow instead.'
)
class SequentialAgent(BaseAgent):
  """A shell agent that runs its sub-agents in sequence.

  .. deprecated::
    SequentialAgent is deprecated and will be removed in future versions.
    Please use Workflow instead.
  """

  config_type: ClassVar[Type[BaseAgentConfig]] = SequentialAgentConfig
  """The config type for this agent.

  DEPRECATED: This attribute is deprecated and will be removed in a future
  version, along with the AgentConfig YAML loader.
  """

  @override
  async def _run_async_impl(
      self, ctx: InvocationContext
  ) -> AsyncGenerator[Event, None]:
    if not self.sub_agents:
      return

    # Initialize or resume the execution state from the agent state.
    agent_state = self._load_agent_state(ctx, SequentialAgentState)
    start_index = self._get_start_index(agent_state)

    pause_invocation = False
    resuming_sub_agent = agent_state is not None
    for i in range(start_index, len(self.sub_agents)):
      sub_agent = self.sub_agents[i]
      if not resuming_sub_agent:
        # If we are resuming from the current event, it means the same event has
        # already been logged, so we should avoid yielding it again.
        if ctx.is_resumable:
          agent_state = SequentialAgentState(current_sub_agent=sub_agent.name)
          ctx.set_agent_state(self.name, agent_state=agent_state)
          yield self._create_agent_state_event(ctx)

      async with Aclosing(sub_agent.run_async(ctx)) as agen:
        async for event in agen:
          yield event
          if ctx.should_pause_invocation(event):
            pause_invocation = True

      # Skip the rest of the sub-agents if the invocation is paused.
      if pause_invocation:
        return

      # Reset the flag for the next sub-agent.
      resuming_sub_agent = False

    if ctx.is_resumable:
      ctx.set_agent_state(self.name, end_of_agent=True)
      yield self._create_agent_state_event(ctx)

  def _get_start_index(
      self,
      agent_state: SequentialAgentState,