support parallel agent

Author: jinnigu

core/src/main/java/com/google/adk/agents/ParallelAgent.java

@@ -20,6 +20,7 @@
 import com.google.adk.agents.ConfigAgentUtils.ConfigurationException;
 import com.google.adk.events.Event;
 import io.reactivex.rxjava3.core.Flowable;
+import io.reactivex.rxjava3.schedulers.Schedulers;
 import java.util.ArrayList;
 import java.util.List;
 import org.slf4j.Logger;
@@ -131,7 +132,7 @@ protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
 
     List<Flowable<Event>> agentFlowables = new ArrayList<>();
     for (BaseAgent subAgent : currentSubAgents) {
-      agentFlowables.add(subAgent.runAsync(invocationContext));
+      agentFlowables.add(subAgent.runAsync(invocationContext).subscribeOn(Schedulers.io()));
     }
     return Flowable.merge(agentFlowables);
   }

core/src/test/java/com/google/adk/agents/ParallelAgentTest.java

@@ -110,4 +110,52 @@ public void runAsync_noSubAgents_returnsEmptyFlowable() {
 
     assertThat(events).isEmpty();
   }
+
+  static class BlockingAgent extends BaseAgent {
+    private final long sleepMillis;
+
+    private BlockingAgent(String name, long sleepMillis) {
+      super(name, "Blocking Agent", ImmutableList.of(), null, null);
+      this.sleepMillis = sleepMillis;
+    }
+
+    @Override
+    protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
+      return Flowable.fromCallable(
+          () -> {
+            Thread.sleep(sleepMillis);
+            return Event.builder()
+                .author(name())
+                .branch(invocationContext.branch().orElse(null))
+                .invocationId(invocationContext.invocationId())
+                .content(Content.fromParts(Part.fromText("Done")))
+                .build();
+          });
+    }
+
+    @Override
+    protected Flowable<Event> runLiveImpl(InvocationContext invocationContext) {
+      throw new UnsupportedOperationException("Not implemented");
+    }
+  }
+
+  @Test
+  public void runAsync_blockingSubAgents_shouldExecuteInParallel() {
+    long sleepTime = 1000;
+    BlockingAgent agent1 = new BlockingAgent("agent1", sleepTime);
+    BlockingAgent agent2 = new BlockingAgent("agent2", sleepTime);
+
+    ParallelAgent parallelAgent =
+        ParallelAgent.builder().name("parallel_agent").subAgents(agent1, agent2).build();
+
+    InvocationContext invocationContext = createInvocationContext(parallelAgent);
+
+    long startTime = System.currentTimeMillis();
+    List<Event> events = parallelAgent.runAsync(invocationContext).toList().blockingGet();
+    long duration = System.currentTimeMillis() - startTime;
+
+    assertThat(events).hasSize(2);
+    // If parallel, duration should be less than 2 * sleepTime (2000ms).
+    assertThat(duration).isLessThan((long) (2.0 * sleepTime));
+  }
 }