How to force a Sequential Agent to stay at first Sub Agent until a condition is met?

[JahangirJadi]

I have a sequential workflow with two sub-agents:

Information Gatherer: Tasks with collecting specific details (Name, Date, Event Type) from the user.

Processor Agent: Takes that data and classifies it into a template.

The Problem: When I start the session, the Information Gatherer runs once, often with incomplete data, and immediately hands off the "empty" results to the Processor Agent. The sequence finishes before I can provide the missing details.

The Goal: I want to "lock" the workflow in the first stage. The first agent should gather all the info from the user until the Information Gatherer confirms that all required fields are present. Only then should the Processor Agent be allowed to start.

What I've tried: I tried a Loop Agent with 5 iterations, but it just cycles through the sub-agents internally and finishes in seconds without waiting for my typing.

Question: How do I implement a "Wait for User" or "Breakpoint" inside a sequential chain so the second agent doesn't execute prematurely?

[adk-bot]

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

TLDR: You can achieve this by creating a custom LongRunningFunctionTool for your "Information Gatherer" agent. This tool will handle the user interaction and will not complete until all the required information has been collected, effectively pausing the sequential workflow.

---

Hello! That's a great question about controlling the flow of a SequentialAgent. Here’s a way you can implement a "wait for user" mechanism in your workflow.

The recommended approach is to encapsulate the user interaction logic within a custom tool. This tool will be responsible for gathering the information from the user and will only finish once all the necessary data has been provided. By doing this, you ensure that the SequentialAgent does not proceed to the "Processor Agent" prematurely.

A LongRunningFunctionTool is well-suited for this scenario because it's designed for operations that are not instantaneous, such as waiting for human input.

Here's a conceptual example of how you can set this up:

from adk.tools import LongRunningFunctionTool
from adk.agents import LlmAgent, SequentialAgent

# 1. Define a tool to gather information from the user.
def get_user_info() -> str:
    """Gets Name, Date, and Event Type from the user."""
    # In a real application, you would have a more sophisticated way
    # to prompt the user, perhaps through a web interface or a command-line prompt.
    name = input("Please enter your Name: ")
    date = input("Please enter the Date: ")
    event_type = input("Please enter the Event Type: ")

    # Loop until all information is provided.
    while not all([name, date, event_type]):
        print("Missing information. Please provide all details.")
        if not name:
            name = input("Please enter your Name: ")
        if not date:
            date = input("Please enter the Date: ")
        if not event_type:
            event_type = input("Please enter the Event Type: ")

    return f"Name: {name}, Date: {date}, Event Type: {event_type}"

# 2. Create a LongRunningFunctionTool from this function.
information_gathering_tool = LongRunningFunctionTool(
    name="InformationGatherer",
    description="Gathers Name, Date, and Event Type from the user.",
    func=get_user_info,
)

# 3. Create the "Information Gatherer" agent with this tool.
information_gatherer_agent = LlmAgent(
    tools=[information_gathering_tool],
    instructions="You are an information gatherer. Use the InformationGatherer tool to collect details from the user."
)

# 4. Define the "Processor Agent".
processor_agent = LlmAgent(
    instructions="You are a processor. Classify the gathered information into a template."
)

# 5. Set up the SequentialAgent.
sequential_workflow = SequentialAgent(
    sub_agents=[
        information_gatherer_agent,
        processor_agent,
    ]
)

# When you run this workflow, the `get_user_info` function will be invoked,
# and it will wait for the user to provide all the inputs before returning.
# The "Processor Agent" will only start after the tool has finished.
# result = sequential_workflow.run("Start the information gathering process.")
# print(result)

This approach aligns with the Human-in-the-Loop pattern in ADK, where you can integrate human intervention points within your agent workflows. [1] The LongRunningFunctionTool is a key component for implementing this pattern. [2]

By using a tool to manage the interaction, you gain fine-grained control over the execution flow without needing to modify the core behavior of the SequentialAgent.

I hope this helps you build your workflow!

---

[1] ADK Documentation: Multi-agent Architectures
[2] ADK Documentation: Long-running function tools

[JahangirJadi]

The AI agent is supposed to collect the information dynamically. I do not want to make it static using function