feat(core): opt memory concurrency and exception handling (#196)

Author: disaster1-tesk

agentscope-core/src/main/java/io/agentscope/core/exception/CompositeAgentException.java

@@ -0,0 +1,84 @@
+/*
+ * Copyright 2024-2025 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.agentscope.core.exception;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class CompositeAgentException extends RuntimeException {
+    private final List<AgentExceptionInfo> causes;
+
+    public CompositeAgentException(String message, List<AgentExceptionInfo> causes) {
+        super(message);
+        this.causes = new ArrayList<>(causes != null ? causes : Collections.emptyList());
+    }
+
+    @Override
+    public String getMessage() {
+        String baseMessage = super.getMessage();
+        if (causes == null || causes.isEmpty()) {
+            return baseMessage;
+        }
+
+        StringBuilder sb = new StringBuilder(baseMessage);
+        sb.append(" (caused by: ");
+
+        for (int i = 0; i < causes.size(); i++) {
+            if (i > 0) {
+                sb.append(", ");
+            }
+
+            AgentExceptionInfo cause = causes.get(i);
+
+            if (cause.agentId() != null) {
+                sb.append("agentId=").append(cause.agentId());
+                if (cause.agentName() != null) {
+                    sb.append("(").append(cause.agentName()).append(")");
+                }
+            } else if (cause.agentName() != null) {
+                sb.append("agentName=").append(cause.agentName());
+            }
+
+            Throwable throwable = cause.throwable();
+            if (throwable != null) {
+                if (cause.agentId() != null || cause.agentName() != null) {
+                    sb.append(": ");
+                }
+                sb.append(throwable.getClass().getSimpleName())
+                        .append(": ")
+                        .append(
+                                throwable.getMessage() != null
+                                        ? throwable.getMessage()
+                                        : "No message");
+            } else {
+                if (cause.agentId() != null || cause.agentName() != null) {
+                    sb.append(": ");
+                }
+                sb.append("Unknown error");
+            }
+        }
+        sb.append(")");
+
+        return sb.toString();
+    }
+
+    public List<AgentExceptionInfo> getCauses() {
+        return Collections.unmodifiableList(causes);
+    }
+
+    public record AgentExceptionInfo(String agentId, String agentName, Throwable throwable) {}
+}

agentscope-core/src/main/java/io/agentscope/core/memory/InMemoryMemory.java

@@ -22,7 +22,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.stream.Collectors;
 
 /**
@@ -36,7 +36,7 @@
  */
 public class InMemoryMemory extends StateModuleBase implements Memory {
 
-    private final List<Msg> messages = new CopyOnWriteArrayList<>();
+    private final ConcurrentLinkedQueue<Msg> messages = new ConcurrentLinkedQueue<>();
     private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
 
     /**

agentscope-core/src/main/java/io/agentscope/core/pipeline/FanoutPipeline.java

@@ -16,10 +16,11 @@
 package io.agentscope.core.pipeline;
 
 import io.agentscope.core.agent.AgentBase;
+import io.agentscope.core.exception.CompositeAgentException;
 import io.agentscope.core.message.Msg;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
-import java.util.concurrent.atomic.AtomicReference;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Schedulers;
@@ -87,7 +88,7 @@ public Mono<List<Msg>> execute(Msg input, Class<?> structuredOutputClass) {
 
     /**
      * Execute agents concurrently using reactive merge with true parallelism.
-     * All agents are executed even if some fail, but the first error is propagated.
+     * All agents are executed even if some fail, but the all error is propagated.
      *
      * <p>Implementation: Each agent's call is subscribed on a separate thread from the
      * {@link Schedulers#boundedElastic()} scheduler, enabling true concurrent execution
@@ -99,7 +100,9 @@ public Mono<List<Msg>> execute(Msg input, Class<?> structuredOutputClass) {
      */
     private Mono<List<Msg>> executeConcurrent(Msg input, Class<?> structuredOutputClass) {
         // Collect all agent results and errors
-        AtomicReference<Throwable> firstError = new AtomicReference<>();
+
+        List<CompositeAgentException.AgentExceptionInfo> errors =
+                Collections.synchronizedList(new ArrayList<>());
 
         List<Mono<Msg>> agentMonos =
                 agents.stream()
@@ -113,10 +116,13 @@ private Mono<List<Msg>> executeConcurrent(Msg input, Class<?> structuredOutputCl
 
                                     return mono.subscribeOn(Schedulers.boundedElastic())
                                             .doOnError(
-                                                    e -> {
-                                                        // Capture the first error encountered
-                                                        firstError.compareAndSet(null, e);
-                                                    })
+                                                    throwable ->
+                                                            errors.add(
+                                                                    new CompositeAgentException
+                                                                            .AgentExceptionInfo(
+                                                                            agent.getAgentId(),
+                                                                            agent.getName(),
+                                                                            throwable)))
                                             .onErrorResume(e -> Mono.empty());
                                 })
                         .toList();
@@ -126,9 +132,11 @@ private Mono<List<Msg>> executeConcurrent(Msg input, Class<?> structuredOutputCl
                 .flatMap(
                         results -> {
                             // If there was an error, propagate the first one
-                            Throwable error = firstError.get();
-                            if (error != null) {
-                                return Mono.error(error);
+                            if (!errors.isEmpty()) {
+                                return Mono.error(
+                                        new CompositeAgentException(
+                                                "Multiple agent execution failures occurred",
+                                                errors));
                             }
                             return Mono.just(results);
                         });

agentscope-core/src/test/java/io/agentscope/core/pipeline/FanoutPipelineTest.java

@@ -17,12 +17,15 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import io.agentscope.core.ReActAgent;
 import io.agentscope.core.agent.test.MockModel;
 import io.agentscope.core.agent.test.TestUtils;
+import io.agentscope.core.exception.CompositeAgentException;
 import io.agentscope.core.memory.InMemoryMemory;
 import io.agentscope.core.message.Msg;
 import io.agentscope.core.tool.Toolkit;
@@ -131,7 +134,11 @@ void shouldPropagatePartialFailure() {
                         () -> pipeline.execute(input).block(TIMEOUT),
                         "Fanout pipeline should surface the failure");
 
-        assertEquals("Simulated error", exception.getMessage(), "Unexpected exception message");
+        assertInstanceOf(CompositeAgentException.class, exception);
+        CompositeAgentException compositeException = (CompositeAgentException) exception;
+        assertEquals(1, compositeException.getCauses().size(), "Expected one cause");
+        assertTrue(
+                exception.getMessage().contains("Simulated error"), "Unexpected exception message");
         assertEquals(1, model1.getCallCount(), "Successful agent should still be invoked");
         assertEquals(1, errorModel.getCallCount(), "Failing agent should be invoked once");
     }
@@ -154,9 +161,11 @@ void shouldPropagateAllFailures() {
         // In concurrent execution, either error could be captured first
         String errorMsg = exception.getMessage();
         assertNotNull(errorMsg, "Should return an error");
-        assertEquals(
-                true,
-                Set.of("Error 1", "Error 2").contains(errorMsg),
+        assertTrue(
+                errorMsg.contains("Error 1"),
+                "Should return either Error 1 or Error 2, got: " + errorMsg);
+        assertTrue(
+                errorMsg.contains("Error 2"),
                 "Should return either Error 1 or Error 2, got: " + errorMsg);
         assertEquals(1, errorModel1.getCallCount(), "First failing agent should be invoked");
         assertEquals(1, errorModel2.getCallCount(), "Second failing agent should be invoked");