Class ParallelAgent

src/google/adk/agents/parallel_agent.py:156–230

A shell agent that runs its sub-agents in parallel in an isolated manner. This approach is beneficial for scenarios requiring multiple perspectives or attempts on a single task, such as: - Running different algorithms simultaneously. - Generating multiple responses for review by a subseque

    ' Please use Workflow instead.'
)
class ParallelAgent(BaseAgent):
  """A shell agent that runs its sub-agents in parallel in an isolated manner.

  This approach is beneficial for scenarios requiring multiple perspectives or
  attempts on a single task, such as:

  - Running different algorithms simultaneously.
  - Generating multiple responses for review by a subsequent evaluation agent.

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

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

    agent_state = self._load_agent_state(ctx, BaseAgentState)
    if ctx.is_resumable and agent_state is None:
      ctx.set_agent_state(self.name, agent_state=BaseAgentState())
      yield self._create_agent_state_event(ctx)

    agent_runs = []
    # Prepare and collect async generators for each sub-agent.
    for sub_agent in self.sub_agents:
      sub_agent_ctx = _create_branch_ctx_for_sub_agent(self, sub_agent, ctx)

      # Only include sub-agents that haven't finished in a previous run.
      if not sub_agent_ctx.end_of_agents.get(sub_agent.name):
        agent_runs.append(sub_agent.run_async(sub_agent_ctx))

    pause_invocation = False
    try:
      merge_func = (
          _merge_agent_run
          if sys.version_info >= (3, 11)
          else _merge_agent_run_pre_3_11
      )
      async with Aclosing(merge_func(agent_runs)) as agen:
        async for event in agen:
          yield event
          if ctx.should_pause_invocation(event):
            pause_invocation = True

      if pause_invocation:
        return