Is there a way to stop a sequential agent at the first subagent?

[AeonMind]

This question is about use SequentialAgent. for example, I create a SequentialAgent as root agent and with two sub agents, In some case, i want to the all workflow stop after the first sub agent, for exmaple, user input "Hello", the first sub agent response to user is engouh, no need to run the sencond sub agent.

[ac-machache]

@AeonMind, How would you determine if the second agent has to be invoked or not, based on that you have many options to achieve this, after first agent callback, before second agent callback, or create a custom sequential agent with your own logic.

[adk-bot]

Response from ADK Answering Agent

TLDR: To stop a SequentialAgent after the first sub-agent, you should create a custom agent. This will allow you to implement conditional logic to control the execution flow.

---

Hello! That's an excellent question. While SequentialAgent is designed to run all its sub-agents in order, you can achieve conditional execution by creating your own custom agent. This gives you precise control over the workflow. [1]

A custom agent inherits from BaseAgent and implements its logic in the _run_async_impl method. Inside this method, you can call your sub-agents and, based on the results or state, decide whether to proceed to the next one.

Here’s a conceptual example in Python of how you could structure your custom agent to solve your problem:

from google.adk.agents import BaseAgent, InvocationContext
from typing import AsyncGenerator

class ConditionalSequentialAgent(BaseAgent):
    async def _run_async_impl(
        self, ctx: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        # Run the first sub-agent
        async for event in self.sub_agents[0].run_async(ctx):
            yield event

        # Check a condition to decide if you should stop.
        # This condition could be based on the session state, the last event, etc.
        # For example, let's check if a specific key was set in the state by the first agent.
        if ctx.session().state.get("should_stop"):
            return  # Stop execution

        # If the condition is not met, run the second sub-agent
        if len(self.sub_agents) > 1:
            async for event in self.sub_agents[1].run_async(ctx):
                yield event

# You would then instantiate this agent with your sub-agents:
# my_conditional_agent = ConditionalSequentialAgent(
#     sub_agents=[first_sub_agent, second_sub_agent]
# )

In this example, the ConditionalSequentialAgent runs the first sub-agent and then checks for a should_stop key in the session state. The first sub-agent would be responsible for setting this key. If the key is present and True, the agent stops; otherwise, it proceeds to the second sub-agent. You can customize the condition to fit your specific needs.

For more details, you can refer to the documentation on custom agents.

---

[1] https://github.com/google/adk-docs/blob/main/docs/agents/custom-agents.md