How can I escalate in SequentialAgent like LoopAgent does?

[mrcaidev]

I have a SequentialAgent that looks like:

before_agent -> review_agent -> after_agent

And if review_agent outputs False, I want to immediately break the chain of SequentialAgent, so that after_agent will not be executed.

In LoopAgent, I can set EventActions(escalate=True) to break the loop. Is there an equivalent of that for SequentialAgent?

[adk-bot]

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

TLDR: SequentialAgent does not have an escalate parameter like LoopAgent. To stop the sequence, an agent in the chain should set ctx.end_invocation = True based on your condition.

---

Hello! That's a great question. You are correct that SequentialAgent is designed for a fixed, linear execution of sub-agents and does not have the same built-in escalation mechanism as LoopAgent [1].

However, you can achieve the behavior you're looking for by having an agent in the sequence decide when to terminate the entire workflow. Any agent can signal the framework to stop processing by setting ctx.end_invocation = True in the InvocationContext [2].

Here is how you can apply this to your use case:

1. review_agent: Ensure this agent saves its boolean output to the session state. For example:

   from adk.core import Event, InProcAgent, InvocationContext
   
   class ReviewAgent(InProcAgent):
       async def _run_async_impl(self, ctx: InvocationContext) -> Event:
           # Your review logic here...
           review_passed = False # or True
           ctx.session.state["review_passed"] = review_passed
           return Event()

2. Create a new "Gate" Agent: Create a small agent whose only job is to check the state and stop the invocation if the condition is met.

   from adk.core import Event, InProcAgent, InvocationContext
   
   class StopOnReviewFailedAgent(InProcAgent):
       async def _run_async_impl(self, ctx: InvocationContext) -> Event:
           if ctx.session.state.get("review_passed") is False:
               print("Review failed, stopping sequence.")
               ctx.end_invocation = True
           return Event()

3. Update your SequentialAgent: Insert this new agent into the sequence immediately after review_agent.

   # before_agent -> review_agent -> stop_agent -> after_agent

This way, after_agent will only run if review_agent's result was True.

For more complex branching or conditional logic, the recommended approach is to create a custom agent that can implement that logic using standard Python if/else statements to call different sub-agents [3]. However, for simply stopping a sequence, the end_invocation flag is the most direct method.

---

[1] Sequential Agents Documentation
[2] Invocation Context Documentation
[3] Custom Agents Documentation

[mrcaidev]

This does not work. I guess I have to write a custom agent for that.