Advanced Concept

Building custom agents by directly implementing _run_async_impl provides powerful control but is more complex than using the predefined LlmAgent or standard WorkflowAgent types. We recommend understanding those foundational agent types first before tackling custom orchestration logic.

Custom agents¶

Custom agents provide the ultimate flexibility in ADK, allowing you to define arbitrary orchestration logic by inheriting directly from BaseAgent and implementing your own control flow. This goes beyond the predefined patterns of SequentialAgent, LoopAgent, and ParallelAgent, enabling you to build highly specific and complex agentic workflows.

Introduction: Beyond Predefined Workflows¶

What is a Custom Agent?¶

A Custom Agent is essentially any class you create that inherits from google.adk.agents.BaseAgent and implements its core execution logic within the _run_async_impl asynchronous method. You have complete control over how this method calls other agents (sub-agents), manages state, and handles events.

Why Use Them?¶

While the standard Workflow Agents (SequentialAgent, LoopAgent, ParallelAgent) cover common orchestration patterns, you'll need a Custom agent when your requirements include:

Conditional Logic: Executing different sub-agents or taking different paths based on runtime conditions or the results of previous steps.
Complex State Management: Implementing intricate logic for maintaining and updating state throughout the workflow beyond simple sequential passing.
External Integrations: Incorporating calls to external APIs, databases, or custom Python libraries directly within the orchestration flow control.
Dynamic Agent Selection: Choosing which sub-agent(s) to run next based on dynamic evaluation of the situation or input.
Unique Workflow Patterns: Implementing orchestration logic that doesn't fit the standard sequential, parallel, or loop structures.

Implementing Custom Logic:¶

The heart of any custom agent is the _run_async_impl method. This is where you define its unique behavior.

Signature: async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
Asynchronous Generator: It must be an async def function and return an AsyncGenerator. This allows it to yield events produced by sub-agents or its own logic back to the runner.
ctx (InvocationContext): Provides access to crucial runtime information, most importantly ctx.session.state, which is the primary way to share data between steps orchestrated by your custom agent.

Key Capabilities within _run_async_impl:

Calling Sub-Agents: You invoke sub-agents (which are typically stored as instance attributes like self.my_llm_agent) using their run_async method and yield their events:
```
async for event in self.some_sub_agent.run_async(ctx):
    # Optionally inspect or log the event
    yield event # Pass the event up
```

Managing State: Read from and write to the session state dictionary (ctx.session.state) to pass data between sub-agent calls or make decisions:

# Read data set by a previous agent
previous_result = ctx.session.state.get("some_key")

# Make a decision based on state
if previous_result == "some_value":
    # ... call a specific sub-agent ...
else:
    # ... call another sub-agent ...

# Store a result for a later step (often done via a sub-agent's output_key)
# ctx.session.state["my_custom_result"] = "calculated_value"

Implementing Control Flow: Use standard Python constructs (if/elif/else, for/while loops, try/except) to create sophisticated, conditional, or iterative workflows involving your sub-agents.

Managing Sub-Agents and State¶

Typically, a custom agent orchestrates other agents (like LlmAgent, LoopAgent, etc.).

Initialization: You usually pass instances of these sub-agents into your custom agent's __init__ method and store them as instance attributes (e.g., self.story_generator = story_generator_instance). This makes them accessible within _run_async_impl.
sub_agents List: When initializing the BaseAgent using super().__init__(...), 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 _run_async_impl calls the agents directly via self.xxx_agent. Include the agents that your custom logic directly invokes at the top level.
State: As mentioned, ctx.session.state is the standard way sub-agents (especially LlmAgents using output_key) communicate results back to the orchestrator and how the orchestrator passes necessary inputs down.

Design Pattern Example: `StoryFlowAgent`¶

Let's illustrate the power of custom agents with an example pattern: a multi-stage content generation workflow with conditional logic.

Goal: Create a system that generates a story, iteratively refines it through critique and revision, performs final checks, and crucially, regenerates the story if the final tone check fails.

Why Custom? The core requirement driving the need for a custom agent here is the conditional regeneration based on the tone check. Standard workflow agents don't have built-in conditional branching based on the outcome of a sub-agent's task. We need custom Python logic (if tone == "negative": ...) within the orchestrator.

Part 1: Simplified custom agent Initialization¶

We define the StoryFlowAgent inheriting from BaseAgent. In __init__, we store the necessary sub-agents (passed in) as instance attributes and tell the BaseAgent framework about the top-level agents this custom agent will directly orchestrate.

Part 2: Defining the Custom Execution Logic¶

This method orchestrates the sub-agents using standard Python async/await and control flow.

Explanation of Logic:

The initial story_generator runs. Its output is expected to be in ctx.session.state["current_story"].
The loop_agent runs, which internally calls the critic and reviser sequentially for max_iterations times. They read/write current_story and criticism from/to the state.
The sequential_agent runs, calling grammar_check then tone_check, reading current_story and writing grammar_suggestions and tone_check_result to the state.
Custom Part: The if statement checks the tone_check_result from the state. If it's "negative", the story_generator is called again, overwriting the current_story in the state. Otherwise, the flow ends.

Part 3: Defining the LLM Sub-Agents¶

These are standard LlmAgent definitions, responsible for specific tasks. Their output_key parameter is crucial for placing results into the session.state where other agents or the custom orchestrator can access them.

GEMINI_2_FLASH = "gemini-2.0-flash" # Define model constant

Part 4: Instantiating and Running the custom agent¶

Finally, you instantiate your StoryFlowAgent and use the Runner as usual.

(Note: The full runnable code, including imports and execution logic, can be found linked below.)

Full Code Example¶

Storyflow Agent

# Full runnable code for the StoryFlowAgent example