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

[dcatonetih]

@adk-bot maybe this solution is better?

import random
import secrets
from typing import ClassVar, List
from google.adk.events import Event, EventActions
from google.adk.agents import BaseAgent, ParallelAgent, SequentialAgent
from google.genai import types


class Worker(BaseAgent):
    """Simple worker that calculates n²."""

    def __init__(self, *, name: str, run_id: str):
        super().__init__(name=name)
        self._run_id = run_id

    async def _run_async_impl(self, ctx):
        n = ctx.session.state.get(f"task:{self._run_id}:{self.name}", 0)
        result = n * n
        yield Event(
            author=self.name,
            content=types.Content(
                role=self.name, parts=[types.Part(text=f"{n}² = {result}")]
            ),
            actions=EventActions(
                state_delta={f"result:{self._run_id}:{self.name}": result}
            ),
        )


class PlannerAndRunner(BaseAgent):
    """Distributes tasks and dynamically creates a ParallelAgent."""

    POOL: ClassVar[List[str]] = ["w0", "w1", "w2"]

    async def _run_async_impl(self, ctx):
        run_id = secrets.token_hex(2)
        picked = random.sample(self.POOL, k=random.randint(1, len(self.POOL)))
        task_delta = {f"task:{run_id}:{name}": random.randint(1, 9) for name in picked}
        yield Event(
            author=self.name,
            content=types.Content(
                role=self.name,
                parts=[types.Part(text=f"Run {run_id} tasks {task_delta}")],
            ),
            actions=EventActions(state_delta={"current_run": run_id, **task_delta}),
        )
        parallel = ParallelAgent(
            name=f"block_{run_id}",
            sub_agents=[Worker(name=n, run_id=run_id) for n in picked],
        )
        async for ev in parallel.run_async(ctx):
            yield ev


class Aggregator(BaseAgent):
    """Aggregates results from workers."""

    async def _run_async_impl(self, ctx):
        run_id = ctx.session.state.get("current_run")
        vals = [
            v
            for k, v in ctx.session.state.items()
            if run_id and k.startswith(f"result:{run_id}:")
        ]
        yield Event(
            author=self.name,
            content=types.Content(
                role=self.name, parts=[types.Part(text=f"Sum = {sum(vals)}")]
            ),
            actions=EventActions(escalate=True),
        )


root_agent = SequentialAgent(
    name="root",
    sub_agents=[PlannerAndRunner(name="planner"), Aggregator(name="collector")],
)

[adk-bot]

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

TLDR: Your current "orphan agent" approach is discouraged because it breaks framework features like resumability and introspection. The recommended pattern is to define a single, reusable sub-agent for the parallel task and orchestrate the concurrent calls within your custom agent using asyncio.gather. This keeps the agent hierarchy static and visible to the framework.

---

Hi! This is an excellent question that touches on a common and powerful pattern: dynamic fan-out/gather workflows. Your current implementation is clever, but your intuition is correct—creating "orphan" agents at runtime is discouraged, and the documentation snippet you found explains exactly why.

Let's break down your questions.

Is the runtime ParallelAgent approach discouraged?

Yes. When you create an agent instance at runtime inside an execution method and don't declare it in the parent's sub_agents list, you are bypassing the mechanisms that make the ADK framework robust.

What are the consequences and misbehaviors?

By operating outside the declared agent hierarchy, you lose several key framework features:

1. Lifecycle Management & Resumability: This is a major one. If your workflow were to be interrupted, the ADK's resume feature for a ParallelAgent would be unable to function. It works by checking which of its known sub-agents have already completed [11]. Since your ParallelAgent is created from scratch on every run, the framework has no memory of the previous attempt and cannot resume gracefully; it would simply start over.

2. Introspection & Observability: The ADK developer tools, logging, and tracing systems rely on the declared agent hierarchy to provide a clear picture of the execution flow [2, 15]. Your dynamically created ParallelAgent and its sub-agents would be invisible to introspection tools. This makes debugging, tracing, and monitoring significantly more difficult, as you'd have a "black hole" in your execution graph.

3. Routing and Other Features: While you are invoking the agent directly, you lose the ability for the framework to ever route to these agents automatically [1, 5]. Furthermore, as you noted from the docs, future ADK features will likely lean on this declared hierarchy, and your implementation could break unexpectedly with future updates.

What is the recommended pattern for this task?

The goal is to keep your agent hierarchy static and declarative, while the execution logic within an agent remains dynamic. You don't need to create new Agent objects to run tasks in parallel.

Instead of instantiating a ParallelAgent and new LlmAgents in every run, you should:

1. Declare a single, reusable LlmAgent as a sub-agent of your GeneratorWorkflow. This agent will be responsible for processing one item.
2. Use asyncio.gather within your _run_async_impl to invoke this single agent multiple times with different inputs in parallel.

Here’s a conceptual rewrite of your GeneratorWorkflow:

import asyncio
from google.adk.agents import (
    BaseAgent,
    InvocationContext,
    LlmAgent,
)
from google.adk.events.event import Event

# ... (your Pydantic models)

class GeneratorWorkflow(BaseAgent):
    draft_item_list_generator: LlmAgent
    draft_item_detail_generator: LlmAgent  # Reusable agent for one item

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

        # 1. Generate the list of items
        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"))

        # 2. Create parallel invocation tasks for the reusable agent
        # The key is that we are NOT creating new agents here.
        # We are creating new, parallel tasks for an EXISTING sub-agent.
        parallel_tasks = []
        for item in draft_item_list.draft_items:
            # Create a separate branch in the session for isolated state
            branch_ctx = ctx.create_branch(f"detail_{item.id}")
            branch_ctx.session.state['draft_item_name'] = item.name
            
            task = self.draft_item_detail_generator.run_async(branch_ctx)
            parallel_tasks.append(task)

        # 3. Run all tasks concurrently using asyncio.gather
        # The ADK framework will correctly trace these parallel invocations
        # to the 'draft_item_detail_generator' agent.
        await asyncio.gather(*parallel_tasks)

        # Note: The results of each run will be in their respective branches.
        # You may need logic here to merge the results back into the main context if needed.

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

# ...

# The reusable agent is defined once
draft_item_detail_generator = LlmAgent(
    name="draft_item_detail_generator",
    model="gemini-2.5-flash",
    instruction="You are a helpful agent who can generate a detailed item based on the following description: {draft_item_name}",
    output_schema=DetailedDraftItem,
    output_key="detailed_item", # Output to its own state
)

root_agent = Agent(
    # ...
    sub_agents=[
        GeneratorWorkflow(
            name="generator_workflow",
            draft_item_list_generator=draft_item_list_generator,
            draft_item_detail_generator=draft_item_detail_generator, # Pass it in
        )
    ],
)

This revised pattern gives you the best of both worlds: the dynamic parallelism you need, combined with the stability, observability, and full feature set of the ADK framework.

Hope this helps!

[dcatonetih]

but will I still have observability and everything when running sub parallel agents like that?