Google ADK SequentialAgent子代理未等待用户输入的解决方法问询

我正在使用Google Agent Development Kit构建一个简单工作流，要求每个子代理需提示用户输入，验证通过后再继续。但运行SequentialAgent时，它会立即依次执行所有子代理，不会等待我回复第一个提示。

最小可复现代码 #

from google.adk.agents import LlmAgent, SequentialAgent

# First agent: prompt for “5”
a1 = LlmAgent(
    name="CheckFive",
    model="gemini-2.0-flash",
    instruction="""
    Ask the user for an integer. 
    If it’s not 5, reply “Exiting” and stop.
    Otherwise reply “Got 5” and store it.
    """,
    output_key="value1"
)

# Second agent: prompt for “7”
a2 = LlmAgent(
    name="CheckSeven",
    model="gemini-2.0-flash",
    instruction="""
    I see the first number was {value1}. 
    Now ask for another integer. If it’s not 7, exit; otherwise store it.
    """,
    output_key="value2"
)

# Third agent: compute sum
a3 = LlmAgent(
    name="Summer",
    model="gemini-2.0-flash",
    instruction="""
    I have two numbers: {value1} and {value2}. 
    Calculate and reply with their sum.
    """,
    output_key="sum"
)

root_agent = SequentialAgent(
    name="CheckAndSum",
    sub_agents=[a1, a2, a3]
)

实际现象 #

• 调用root_agent后，立即得到所有三个提示的拼接内容或最终响应，完全没有机会输入“5”或“7”。

预期流程 #

1. CheckFive 询问：“Please enter an integer.”
2. 输入5，代理回复“Got 5”并存储value1=5。
3. CheckSeven 询问：“Please enter another integer.”
4. 输入7，代理回复“Got 7”并存储value2=7。
5. Summer 回复“The sum is 12.”

解决方案 #

核心原因 #

SequentialAgent默认是批量执行所有子代理，不会自动暂停等待用户输入。要实现交互式流程，需要手动控制每个子代理的执行时机，在每个步骤后收集用户输入，验证通过后再传递给下一个代理。

交互式实现代码 #

from google.adk.agents import LlmAgent
import sys

# 保留原代理定义
a1 = LlmAgent(
    name="CheckFive",
    model="gemini-2.0-flash",
    instruction="""
    Ask the user for an integer. 
    If it’s not 5, reply “Exiting” and stop.
    Otherwise reply “Got 5” and store it.
    """,
    output_key="value1"
)

a2 = LlmAgent(
    name="CheckSeven",
    model="gemini-2.0-flash",
    instruction="""
    I see the first number was {value1}. 
    Now ask for another integer. If it’s not 7, exit; otherwise store it.
    """,
    output_key="value2"
)

a3 = LlmAgent(
    name="Summer",
    model="gemini-2.0-flash",
    instruction="""
    I have two numbers: {value1} and {value2}. 
    Calculate and reply with their sum.
    """,
    output_key="sum"
)

def run_interactive_workflow():
    context = {}
    
    # 第一步：处理CheckFive
    # 获取初始提示
    prompt_response = a1.run(inputs={})
    print(prompt_response["response"])
    
    # 等待用户输入并验证
    user_input = input("> ")
    validation = a1.run(inputs={"user_input": user_input})
    print(validation["response"])
    
    if "Exiting" in validation["response"]:
        return
    context["value1"] = validation["value1"]
    
    # 第二步：处理CheckSeven
    prompt_response = a2.run(inputs=context)
    print(prompt_response["response"])
    
    user_input = input("> ")
    validation = a2.run(inputs={**context, "user_input": user_input})
    print(validation["response"])
    
    if "Exiting" in validation["response"]:
        return
    context["value2"] = validation["value2"]
    
    # 第三步：计算总和
    final_result = a3.run(inputs=context)
    print(final_result["response"])

if __name__ == "__main__":
    run_interactive_workflow()

关键说明 #

• 手动控制执行顺序：不再依赖SequentialAgent的自动批量执行，逐个调用子代理的run()方法。
• 用户输入交互：每个代理给出提示后，用input()获取用户输入，再将输入传给代理完成验证。
• 上下文传递：将前一个代理验证通过的结果存入上下文字典，传递给下一个代理使用。
• 终止逻辑：如果验证失败（回复包含"Exiting"），直接终止流程。

内容的提问来源于stack exchange，提问作者Dinesh Sonachalam