Module 21.5: Custom Agents - Building Complex Orchestrators

Theory

Beyond Pre-Built Workflows

The ADK provides powerful pre-built workflow agents like SequentialAgent, ParallelAgent, and LoopAgent. These are fantastic for structured, predictable pipelines.

However, real-world AI applications often demand dynamic, complex routing logic that doesn't fit neatly into a straight line or a simple loop.

• What if you want to inspect a user's session state and decide on the fly which agent to call next?
• What if you need to run a sentiment analysis tool first, and only invoke an LLM if the user is angry?

For these scenarios, the ADK gives you the ultimate flexibility: Custom Agents.

Inheriting from BaseAgent

At the core of the ADK is the BaseAgent class. Every agent (including LlmAgent and the workflow agents) inherits from it. By creating your own class that inherits from BaseAgent, you gain complete control over the execution loop.

To build a custom agent, you must do two things:

1. Inherit from BaseAgent: Define your new class.
2. Override _run_async_impl: This is the engine of your agent. It is an asynchronous generator that yields Event objects.

The _run_async_impl Method

This method is where your custom logic lives. It receives an InvocationContext (ctx), which gives you access to the current session.state and the events history.

from google.adk.agents import BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncGenerator

class MyCustomRouter(BaseAgent):
    
    # You can define an __init__ to accept other agents as components
    def __init__(self, name: str, agent_a: BaseAgent, agent_b: BaseAgent, **kwargs):
        super().__init__(name=name, **kwargs)
        self.agent_a = agent_a
        self.agent_b = agent_b

    async def _run_async_impl(self, ctx: InvocationContext) -> AsyncGenerator[Event, None]:
        # 1. Inspect the state
        user_tier = ctx.session.state.get("user_tier", "standard")
        
        # 2. Make a decision
        if user_tier == "premium":
            chosen_agent = self.agent_a
        else:
            chosen_agent = self.agent_b
            
        # 3. Delegate execution to the chosen agent
        # We MUST yield the events generated by the sub-agent back up to the runner
        async for event in chosen_agent.run_async(ctx):
            yield event

Delegation vs. Tools

It's important to understand the difference between delegating to a sub-agent (like in the example above) and calling an agent as a tool (using AgentTool).

• Delegation (Custom/Workflow Agents): The parent agent hands over control completely. The sub-agent's events flow directly to the user. This is structural and deterministic.
• AgentTool (LlmAgent): The parent LLM agent decides if and when to call the sub-agent based on its prompt. The sub-agent runs, returns a result to the parent LLM, and the parent LLM synthesizes the final response.

Custom agents are used for structural delegation when you, the developer, want to hardcode the business logic rather than relying on an LLM to figure it out.

Key Takeaways

• While pre-built workflow agents are great for standard patterns, Custom Agents provide ultimate flexibility for dynamic routing and complex business logic.
• You build a custom agent by inheriting from BaseAgent and overriding the _run_async_impl asynchronous generator method.
• Inside _run_async_impl, you have full access to the InvocationContext (state and history) to make routing decisions.
• When delegating to a sub-agent inside _run_async_impl, you must yield the events from its run_async method back to the caller.