Running dynamically generated agents with ParallelAgent

[lucasbarzotto-axonify]

Hi!

I'm working on a project where we need to perform the following steps:

• Run an Agent to generate a list of draft items (structured output)
• For each generated draft item by the first agent, run another agent to generate a detailed item (also structured). These agents should run in parallel.

My current implementation is having a custom agent that:

• Runs the draft items generator
• For each run, it creates and runs a new ParallelAgent whose sub-agents are defined dynamically based on the output of the first agent.

Here's a simple example:

import logging
import uuid
from collections.abc import AsyncGenerator
from typing import Any

from google.adk.agents import (
    Agent,
    BaseAgent,
    InvocationContext,
    LlmAgent,
    ParallelAgent,
)
from google.adk.events.event import Event
from google.genai import types
from pydantic import BaseModel, Field
from pydantic.json_schema import SkipJsonSchema

logger = logging.getLogger(__name__)


class DraftItem(BaseModel):
    id: SkipJsonSchema[str] = Field(default_factory=lambda: str(uuid.uuid4()))
    name: str


class DraftItemList(BaseModel):
    draft_items: list[DraftItem]


class DetailedDraftItem(BaseModel):
    detailed_name: str
    detailed_description: str


class GeneratorWorkflow(BaseAgent):
    draft_item_list_generator: LlmAgent

    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event]:
        state: dict[str, Any] = ctx.session.state

        async for event in self.draft_item_list_generator.run_async(ctx):
            yield event

        draft_item_list = DraftItemList.model_validate(state.get("draft_item_list"))

        parallel_agent = (
            ParallelAgent(  # Orphan agent. It's not a sub-agent of GeneratorWorkflow.
                name="parallel_agent",
                sub_agents=[
                    self._create_sub_agent(item) for item in draft_item_list.draft_items
                ],
            )
        )
        async for event in parallel_agent.run_async(ctx):
            yield event

        yield Event(
            author=self.name,
            invocation_id=ctx.invocation_id,
            branch=ctx.branch,
            content=types.ModelContent(parts=[types.Part(text="Success!")]),
        )

    def _create_sub_agent(self, draft_item: DraftItem) -> BaseAgent:
        return LlmAgent(
            name=f"draft_item_generator_{draft_item.id}".replace("-", "_"),
            model="gemini-2.5-flash",
            instruction=f"You are a helpful agent who can generate a detailed item based on the following description: {draft_item.name}",
            output_schema=DetailedDraftItem,
            output_key=draft_item.id,
        )


draft_item_list_generator = LlmAgent(
    name="draft_item_list_generator",
    model="gemini-2.5-flash",
    instruction="You are a helpful agent who can generate a list of 4 draft items for the user.",
    output_schema=DraftItemList,
    output_key="draft_item_list",
)

root_agent = Agent(
    name="helpful_agent",
    model="gemini-2.5-flash",
    description="Agent to answer questions about the helpful agent.",
    instruction="You are a helpful agent who can answer user questions about the helpful agent."
    "If the user wants to generate items, transfer to the `generator_workflow` sub-agent.",
    sub_agents=[
        GeneratorWorkflow(
            name="generator_workflow",
            draft_item_list_generator=draft_item_list_generator,
        )
    ],
)

Note that in this case, the ParallelAgent is an orphan agent, since it needs to be recreated in every run.
The reason we recreate the ParallelAgent if every run, is that if we set it as a sub agent of the GeneratorWorkflow, it means it would be shared across different sessions, potentially resulting in race condition issues.

While this seems to work well, the documentation explicitly mentions that:

When initializing the BaseAgent using it's super() constructor, you should pass a sub agents list. This list tells the ADK framework about the agents that are part of this custom agent's immediate hierarchy. It's important for framework features like lifecycle management, introspection, and potentially future routing capabilities, even if your core execution logic (_run_async_impl) calls the agents directly via self.xxx_agent. Include the agents that your custom logic directly invokes at the top level.

My questions are:

• Is the runtime / orphan ParallelAgent approach (recreate-per-run) discouraged? If so, can you explain which ADK features would be affected — e.g., lifecycle management, introspection, routing, persistence, telemetry, or other behaviors — and what misbehaviors or limitations I should expect if I keep doing it this way?
• If it is discouraged, what is the recommended pattern for accomplishing this kind of task?

I'm happy to provide more details or examples if needed. Thanks!

[adk-bot]

Response from ADK Answering Agent (experimental, answer may be inaccurate)

TLDR: Your current approach of creating "orphan" agents at runtime is discouraged. It can interfere with essential ADK features like state management, introspection, and observability. The recommended pattern is to manage the parallel execution of your tasks within a single custom agent's _run_async_impl method, using a standard library like asyncio.gather to make concurrent LLM calls.

---

Hi! That's a great question about handling dynamic, parallel workloads with ADK.

Your concern is valid. The documentation snippet you highlighted is key: ADK's framework relies on knowing the hierarchy of agents at initialization time. 1 Creating agents dynamically at runtime, as you are doing with ParallelAgent, means the framework cannot properly manage them.

Why the "Orphan" Agent Approach Is Discouraged

By creating an unmanaged agent, you bypass several features that are critical for building robust and maintainable agent-based systems:

• State Management & Lifecycle: ADK manages the lifecycle and state of all registered sub-agents. Orphan agents are not part of this managed lifecycle, which can lead to unpredictable behavior in state persistence, session management, and resumability.
• Introspection & Observability: These dynamically created agents will not appear in the agent hierarchy. This makes debugging and tracing executions much more difficult, as they will be invisible to ADK's introspection and observability tools. 2
• Routing & Delegation: Automatic routing capabilities depend on the parent agent being aware of its sub-agents. Your orphan agents are invisible to the router, limiting the framework's ability to delegate tasks automatically.
• Telemetry & Other Framework Features: Logging, metrics, and other framework-level features that hook into the agent lifecycle may not function correctly for these orphan agents.

Recommended Pattern for Dynamic Parallel Tasks

Instead of dynamically creating ADK agents, the recommended approach is to contain the dynamic parallelism inside a single, well-defined custom agent. Your GeneratorWorkflow agent would be responsible for orchestrating these parallel tasks internally.

Here’s how you could structure it:

1. Define a Custom Agent: Keep your GeneratorWorkflow agent, which is properly registered within the ADK framework.
2. Generate Initial Items: Use your draft_item_list_generator to get the initial list of items, just as you are doing now.
3. Use asyncio for Parallelism: Within the _run_async_impl method of GeneratorWorkflow, create a list of coroutines. Instead of creating a new LlmAgent for each item, you can make direct, concurrent calls to the Gemini API using asyncio.gather.
4. Process and Yield Results: Once the parallel tasks are complete, process the results and yield the final Event objects. This keeps the ADK framework aware of the outcome while abstracting away the implementation details of the parallel execution.

This pattern allows you to leverage Python's native async capabilities for your dynamic workload while ensuring that your agent remains fully compliant with the ADK framework, benefiting from proper lifecycle management, state persistence, and observability.

---

This response was generated by an AI assistant. Please leave a 👍 if it was helpful.

[lucasbarzotto-axonify]

I believe using asyncio.gather to make direct, concurrent calls to the Gemini API is a bad option, since we'd loose ADK's features, such as:

• Observability and tracing for the agents
• Resumability control on the ParallelAgent and its sub agents
• Branch management of the ParallelAgent