Class LoopAgent

src/google/adk/agents/loop_agent.py:57–180

A shell agent that run its sub-agents in a loop. When sub-agent generates an event with escalate or max_iterations are reached, the loop agent will stop. .. deprecated:: LoopAgent is deprecated and will be removed in future versions. Please use Workflow instead.

    ' Please use Workflow instead.'
)
class LoopAgent(BaseAgent):
  """A shell agent that run its sub-agents in a loop.

  When sub-agent generates an event with escalate or max_iterations are
  reached, the loop agent will stop.

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

  config_type: ClassVar[type[BaseAgentConfig]] = LoopAgentConfig
  """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.
  """

  max_iterations: Optional[int] = None
  """The maximum number of iterations to run the loop agent.

  If not set, the loop agent will run indefinitely until a sub-agent
  escalates.
  """

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

    agent_state = self._load_agent_state(ctx, LoopAgentState)
    is_resuming_at_current_agent = agent_state is not None
    times_looped, start_index = self._get_start_state(agent_state)

    should_exit = False
    pause_invocation = False
    while (
        not self.max_iterations or times_looped < self.max_iterations
    ) and not (should_exit or pause_invocation):
      for i in range(start_index, len(self.sub_agents)):
        sub_agent = self.sub_agents[i]

        if ctx.is_resumable and not is_resuming_at_current_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.
          agent_state = LoopAgentState(
              current_sub_agent=sub_agent.name,
              times_looped=times_looped,
          )
          ctx.set_agent_state(self.name, agent_state=agent_state)
          yield self._create_agent_state_event(ctx)

        is_resuming_at_current_agent = False

        async with Aclosing(sub_agent.run_async(ctx)) as agen:
          async for event in agen: