This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 8ea839f45875 CAMEL-23952: Support executeToolsConcurrently for
langchain4j-agent
8ea839f45875 is described below
commit 8ea839f4587588da6c51c06fa0b4b022aa78745a
Author: Omar Atie <[email protected]>
AuthorDate: Tue Aug 4 01:56:42 2026 -0700
CAMEL-23952: Support executeToolsConcurrently for langchain4j-agent
Expose executeToolsConcurrently on AgentConfiguration with optional
user-supplied or Camel-managed executor. Uses duplicate() to protect
shared registry beans from mutation. Includes concurrency tests and
component documentation.
Co-authored-by: Cursor <[email protected]>
---
.../catalog/docs/langchain4j-agent-component.adoc | 19 +++
.../camel-ai/camel-langchain4j-agent-api/pom.xml | 5 +
.../langchain4j/agent/api/AbstractAgent.java | 9 ++
.../langchain4j/agent/api/AgentConfiguration.java | 79 ++++++++++
.../agent/api/AgentConfigurationTest.java | 106 ++++++++++++-
.../src/main/docs/langchain4j-agent-component.adoc | 19 +++
.../agent/LangChain4jAgentProducer.java | 28 +++-
...ngChain4jAgentExecuteToolsConcurrentlyTest.java | 168 +++++++++++++++++++++
8 files changed, 431 insertions(+), 2 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-agent-component.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-agent-component.adoc
index 143a168fd3d4..4afe1aa5b07b 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-agent-component.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/langchain4j-agent-component.adoc
@@ -158,6 +158,25 @@ Agents are configured using the `AgentConfiguration` class
which provides a flue
* Chat Memory Provider (for memory-enabled agents)
* Retrieval Augmentor (for RAG functionality)
* Input and Output Guardrails
+* Concurrent tool execution (`withExecuteToolsConcurrently`) for parallel
Camel route tools and MCP tools within one LLM round trip
+
+==== Concurrent tool execution
+
+When the LLM requests multiple tools in a single turn, LangChain4j can invoke
them in parallel via `AgentConfiguration.withExecuteToolsConcurrently()`. Camel
route tools (`ai-tool:` consumers discovered through `tags`) run each
invocation on an isolated exchange copy, so concurrent execution does not
corrupt the producer exchange.
+
+._Java-only: enable parallel tool calls with a managed Camel executor_
+[source,java]
+----
+AgentConfiguration configuration = new AgentConfiguration()
+ .withChatModel(chatModel)
+ .withExecuteToolsConcurrently();
+
+// When created through langchain4j-agent (agentConfiguration on the endpoint),
+// Camel registers a managed thread pool from ExecutorServiceManager
automatically.
+// Pass your own Executor with withExecuteToolsConcurrently(executor) to
override.
+----
+
+Tool-route header side-effects are not merged back onto the main exchange when
tools run concurrently; only the tool result text is returned to the LLM.
==== Creating an Agent without Memory
diff --git a/components/camel-ai/camel-langchain4j-agent-api/pom.xml
b/components/camel-ai/camel-langchain4j-agent-api/pom.xml
index 3bc1fb80dc59..0a1f9f9bb6b7 100644
--- a/components/camel-ai/camel-langchain4j-agent-api/pom.xml
+++ b/components/camel-ai/camel-langchain4j-agent-api/pom.xml
@@ -64,6 +64,11 @@
<artifactId>junit-jupiter</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
\ No newline at end of file
diff --git
a/components/camel-ai/camel-langchain4j-agent-api/src/main/java/org/apache/camel/component/langchain4j/agent/api/AbstractAgent.java
b/components/camel-ai/camel-langchain4j-agent-api/src/main/java/org/apache/camel/component/langchain4j/agent/api/AbstractAgent.java
index 9c0689f8f51a..34fd61f2a118 100644
---
a/components/camel-ai/camel-langchain4j-agent-api/src/main/java/org/apache/camel/component/langchain4j/agent/api/AbstractAgent.java
+++
b/components/camel-ai/camel-langchain4j-agent-api/src/main/java/org/apache/camel/component/langchain4j/agent/api/AbstractAgent.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.langchain4j.agent.api;
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.Executor;
import java.util.function.BiPredicate;
import dev.langchain4j.agent.tool.ToolSpecification;
@@ -184,6 +185,14 @@ public abstract class AbstractAgent<S> implements Agent {
if (configuration.getCompensateOnToolErrors() != null) {
builder.compensateOnToolErrors(configuration.getCompensateOnToolErrors());
}
+ if (Boolean.TRUE.equals(configuration.getExecuteToolsConcurrently())) {
+ Executor executor = configuration.getExecuteToolsExecutor();
+ if (executor != null) {
+ builder.executeToolsConcurrently(executor);
+ } else {
+ builder.executeToolsConcurrently();
+ }
+ }
// Custom AiServices builder customizer (escape hatch for any builder
option not directly exposed)
if (configuration.getAiServicesCustomizer() != null) {
diff --git
a/components/camel-ai/camel-langchain4j-agent-api/src/main/java/org/apache/camel/component/langchain4j/agent/api/AgentConfiguration.java
b/components/camel-ai/camel-langchain4j-agent-api/src/main/java/org/apache/camel/component/langchain4j/agent/api/AgentConfiguration.java
index c813d7454ae5..c311894921c1 100644
---
a/components/camel-ai/camel-langchain4j-agent-api/src/main/java/org/apache/camel/component/langchain4j/agent/api/AgentConfiguration.java
+++
b/components/camel-ai/camel-langchain4j-agent-api/src/main/java/org/apache/camel/component/langchain4j/agent/api/AgentConfiguration.java
@@ -20,6 +20,8 @@ package org.apache.camel.component.langchain4j.agent.api;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.Executor;
import java.util.function.BiPredicate;
import java.util.function.Consumer;
import java.util.function.Function;
@@ -79,6 +81,8 @@ public class AgentConfiguration {
private ToolExecutionErrorHandler toolExecutionErrorHandler;
private ToolArgumentsErrorHandler toolArgumentsErrorHandler;
private Boolean compensateOnToolErrors;
+ private Boolean executeToolsConcurrently;
+ private Executor executeToolsExecutor;
private Consumer<AiServices<?>> aiServicesCustomizer;
/**
@@ -463,6 +467,81 @@ public class AgentConfiguration {
return this;
}
+ /**
+ * Gets whether concurrent tool execution is enabled for this agent.
+ *
+ * @return {@code true} if enabled, {@code false} if explicitly disabled,
or {@code null} if not configured
+ */
+ public Boolean getExecuteToolsConcurrently() {
+ return executeToolsConcurrently;
+ }
+
+ /**
+ * Gets the executor used for concurrent tool execution, if configured.
+ *
+ * @return the executor, or {@code null} if not configured
+ */
+ public Executor getExecuteToolsExecutor() {
+ return executeToolsExecutor;
+ }
+
+ /**
+ * Enables parallel execution of all tool calls within a single LLM round
trip via LangChain4j's
+ * {@code executeToolsConcurrently()} mode. Requires Camel route tools to
run on isolated exchange copies (see
+ * {@code LangChain4jAgentProducer}).
+ * <p>
+ * When no explicit executor is supplied, the langchain4j-agent component
resolves a managed thread pool from the
+ * Camel {@code ExecutorServiceManager} at startup.
+ *
+ * @return this configuration instance for method chaining
+ */
+ public AgentConfiguration withExecuteToolsConcurrently() {
+ this.executeToolsConcurrently = true;
+ return this;
+ }
+
+ /**
+ * Enables parallel tool execution using the given executor.
+ *
+ * @param executeToolsExecutor the executor for concurrent tool
invocations (must not be {@code null}; use
+ * {@link #withExecuteToolsConcurrently()}
for the managed executor)
+ * @return this configuration instance for method
chaining
+ * @throws NullPointerException if {@code executeToolsExecutor} is {@code
null}
+ */
+ public AgentConfiguration withExecuteToolsConcurrently(Executor
executeToolsExecutor) {
+ this.executeToolsConcurrently = true;
+ this.executeToolsExecutor =
Objects.requireNonNull(executeToolsExecutor, "executeToolsExecutor");
+ return this;
+ }
+
+ /**
+ * Creates a shallow copy of this configuration. Used by the
langchain4j-agent producer to attach a managed tool
+ * executor without mutating registry-held configuration beans.
+ *
+ * @return a new configuration instance with the same settings
+ * @since 4.22
+ */
+ public AgentConfiguration duplicate() {
+ AgentConfiguration copy = new AgentConfiguration();
+ copy.chatModel = chatModel;
+ copy.chatMemoryProvider = chatMemoryProvider;
+ copy.retrievalAugmentor = retrievalAugmentor;
+ copy.inputGuardrailClasses = inputGuardrailClasses;
+ copy.outputGuardrailClasses = outputGuardrailClasses;
+ copy.customTools = customTools;
+ copy.mcpClients = mcpClients;
+ copy.mcpToolProviderFilter = mcpToolProviderFilter;
+ copy.maxToolCallingRoundTrips = maxToolCallingRoundTrips;
+ copy.hallucinatedToolNameStrategy = hallucinatedToolNameStrategy;
+ copy.toolExecutionErrorHandler = toolExecutionErrorHandler;
+ copy.toolArgumentsErrorHandler = toolArgumentsErrorHandler;
+ copy.compensateOnToolErrors = compensateOnToolErrors;
+ copy.executeToolsConcurrently = executeToolsConcurrently;
+ copy.executeToolsExecutor = executeToolsExecutor;
+ copy.aiServicesCustomizer = aiServicesCustomizer;
+ return copy;
+ }
+
/**
* Gets the custom AiServices builder customizer.
*
diff --git
a/components/camel-ai/camel-langchain4j-agent-api/src/test/java/org/apache/camel/component/langchain4j/agent/api/AgentConfigurationTest.java
b/components/camel-ai/camel-langchain4j-agent-api/src/test/java/org/apache/camel/component/langchain4j/agent/api/AgentConfigurationTest.java
index 297ac1ccc71e..5f9c45e10e69 100644
---
a/components/camel-ai/camel-langchain4j-agent-api/src/test/java/org/apache/camel/component/langchain4j/agent/api/AgentConfigurationTest.java
+++
b/components/camel-ai/camel-langchain4j-agent-api/src/test/java/org/apache/camel/component/langchain4j/agent/api/AgentConfigurationTest.java
@@ -17,23 +17,34 @@
package org.apache.camel.component.langchain4j.agent.api;
import java.io.Serializable;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import dev.langchain4j.agent.tool.ToolExecutionRequest;
+import dev.langchain4j.data.message.AiMessage;
import dev.langchain4j.data.message.ToolExecutionResultMessage;
+import dev.langchain4j.model.chat.ChatModel;
+import dev.langchain4j.model.chat.request.ChatRequest;
+import dev.langchain4j.model.chat.response.ChatResponse;
import dev.langchain4j.service.tool.ToolArgumentsErrorHandler;
import dev.langchain4j.service.tool.ToolExecutionErrorHandler;
import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;
-public class AgentConfigurationTest {
+class AgentConfigurationTest {
@Test
public void testParseGuardrailClasses_WithValidClasses() {
@@ -331,6 +342,99 @@ public class AgentConfigurationTest {
assertNotNull(config.getAiServicesCustomizer());
}
+ @Test
+ void duplicateCopiesAllDeclaredInstanceFields() throws Exception {
+ Executor executor = Executors.newSingleThreadExecutor();
+ try {
+ ChatModel chatModel = new ChatModel() {
+ @Override
+ public ChatResponse doChat(ChatRequest request) {
+ return
ChatResponse.builder().aiMessage(AiMessage.from("ok")).build();
+ }
+ };
+ Function<ToolExecutionRequest, ToolExecutionResultMessage> strategy
+ = request -> ToolExecutionResultMessage.from(request,
"unknown");
+ ToolExecutionErrorHandler execHandler = (error, context) -> null;
+ ToolArgumentsErrorHandler argsHandler = (error, context) -> null;
+
+ AgentConfiguration original = new AgentConfiguration()
+ .withChatModel(chatModel)
+ .withMaxToolCallingRoundTrips(11)
+ .withHallucinatedToolNameStrategy(strategy)
+ .withToolExecutionErrorHandler(execHandler)
+ .withToolArgumentsErrorHandler(argsHandler)
+ .withCompensateOnToolErrors(true)
+ .withExecuteToolsConcurrently(executor)
+ .withInputGuardrailClassesArray(new String[] {
"java.lang.String" })
+ .withOutputGuardrailClassesArray(new String[] {
"java.io.Serializable" })
+ .withCustomTools(List.of("custom-tool"))
+ .withAiServicesCustomizer(builder -> {
+ });
+
+ AgentConfiguration copy = original.duplicate();
+
+ for (Field field : AgentConfiguration.class.getDeclaredFields()) {
+ int modifiers = field.getModifiers();
+ if (Modifier.isStatic(modifiers)) {
+ continue;
+ }
+ field.setAccessible(true);
+ assertThat(field.get(copy))
+ .as("duplicate() should copy field %s",
field.getName())
+ .isEqualTo(field.get(original));
+ }
+ } finally {
+ ((ExecutorService) executor).shutdownNow();
+ }
+ }
+
+ @Test
+ void withExecuteToolsConcurrentlyRejectsNullExecutor() {
+ assertThatThrownBy(() -> new
AgentConfiguration().withExecuteToolsConcurrently((Executor) null))
+ .isInstanceOf(NullPointerException.class)
+ .hasMessageContaining("executeToolsExecutor");
+ }
+
+ @Test
+ void testDuplicateCopiesExecuteToolsConcurrentlySettings() {
+ Executor executor = Executors.newSingleThreadExecutor();
+ try {
+ AgentConfiguration original = new AgentConfiguration()
+ .withExecuteToolsConcurrently(executor)
+ .withMaxToolCallingRoundTrips(3);
+
+ AgentConfiguration copy = original.duplicate();
+
+ assertThat(copy.getExecuteToolsConcurrently()).isTrue();
+ assertThat(copy.getExecuteToolsExecutor()).isSameAs(executor);
+ assertThat(copy.getMaxToolCallingRoundTrips()).isEqualTo(3);
+ } finally {
+ ((ExecutorService) executor).shutdownNow();
+ }
+ }
+
+ @Test
+ void testExecuteToolsConcurrently() {
+ AgentConfiguration config = new AgentConfiguration();
+ assertThat(config.getExecuteToolsConcurrently()).isNull();
+ assertThat(config.getExecuteToolsExecutor()).isNull();
+
+ AgentConfiguration enabled = config.withExecuteToolsConcurrently();
+ assertThat(enabled).isSameAs(config);
+ assertThat(config.getExecuteToolsConcurrently()).isTrue();
+ assertThat(config.getExecuteToolsExecutor()).isNull();
+
+ Executor executor = Executors.newSingleThreadExecutor();
+ try {
+ AgentConfiguration withExecutor =
config.withExecuteToolsConcurrently(executor);
+ assertThat(withExecutor).isSameAs(config);
+ assertThat(config.getExecuteToolsConcurrently()).isTrue();
+ assertThat(config.getExecuteToolsExecutor()).isSameAs(executor);
+ } finally {
+ ((ExecutorService) executor).shutdownNow();
+ }
+ }
+
@Test
public void testFluentChaining() {
ToolExecutionErrorHandler execHandler = (error, context) -> null;
diff --git
a/components/camel-ai/camel-langchain4j-agent/src/main/docs/langchain4j-agent-component.adoc
b/components/camel-ai/camel-langchain4j-agent/src/main/docs/langchain4j-agent-component.adoc
index 143a168fd3d4..4afe1aa5b07b 100644
---
a/components/camel-ai/camel-langchain4j-agent/src/main/docs/langchain4j-agent-component.adoc
+++
b/components/camel-ai/camel-langchain4j-agent/src/main/docs/langchain4j-agent-component.adoc
@@ -158,6 +158,25 @@ Agents are configured using the `AgentConfiguration` class
which provides a flue
* Chat Memory Provider (for memory-enabled agents)
* Retrieval Augmentor (for RAG functionality)
* Input and Output Guardrails
+* Concurrent tool execution (`withExecuteToolsConcurrently`) for parallel
Camel route tools and MCP tools within one LLM round trip
+
+==== Concurrent tool execution
+
+When the LLM requests multiple tools in a single turn, LangChain4j can invoke
them in parallel via `AgentConfiguration.withExecuteToolsConcurrently()`. Camel
route tools (`ai-tool:` consumers discovered through `tags`) run each
invocation on an isolated exchange copy, so concurrent execution does not
corrupt the producer exchange.
+
+._Java-only: enable parallel tool calls with a managed Camel executor_
+[source,java]
+----
+AgentConfiguration configuration = new AgentConfiguration()
+ .withChatModel(chatModel)
+ .withExecuteToolsConcurrently();
+
+// When created through langchain4j-agent (agentConfiguration on the endpoint),
+// Camel registers a managed thread pool from ExecutorServiceManager
automatically.
+// Pass your own Executor with withExecuteToolsConcurrently(executor) to
override.
+----
+
+Tool-route header side-effects are not merged back onto the main exchange when
tools run concurrently; only the tool result text is returned to the LLM.
==== Creating an Agent without Memory
diff --git
a/components/camel-ai/camel-langchain4j-agent/src/main/java/org/apache/camel/component/langchain4j/agent/LangChain4jAgentProducer.java
b/components/camel-ai/camel-langchain4j-agent/src/main/java/org/apache/camel/component/langchain4j/agent/LangChain4jAgentProducer.java
index 63d98402041c..197481eb9054 100644
---
a/components/camel-ai/camel-langchain4j-agent/src/main/java/org/apache/camel/component/langchain4j/agent/LangChain4jAgentProducer.java
+++
b/components/camel-ai/camel-langchain4j-agent/src/main/java/org/apache/camel/component/langchain4j/agent/LangChain4jAgentProducer.java
@@ -26,6 +26,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.ExecutorService;
import java.util.function.BiPredicate;
import java.util.stream.Collectors;
@@ -61,6 +62,7 @@ import
org.apache.camel.component.langchain4j.agent.api.AgentWithoutMemory;
import org.apache.camel.component.langchain4j.agent.api.AiAgentBody;
import org.apache.camel.component.langchain4j.agent.api.CompositeToolProvider;
import org.apache.camel.component.langchain4j.agent.api.Headers;
+import org.apache.camel.spi.ThreadPoolProfile;
import org.apache.camel.support.DefaultProducer;
import org.apache.camel.support.ExchangeHelper;
import org.apache.camel.support.ResourceHelper;
@@ -76,6 +78,7 @@ public class LangChain4jAgentProducer extends DefaultProducer
{
private AgentFactory agentFactory;
private Agent agent;
private List<McpClient> materializedMcpClients;
+ private ExecutorService managedToolExecutionExecutor;
public LangChain4jAgentProducer(LangChain4jAgentEndpoint endpoint) {
super(endpoint);
@@ -114,7 +117,8 @@ public class LangChain4jAgentProducer extends
DefaultProducer {
if (endpoint.getConfiguration().getAgent() != null) {
agent = endpoint.getConfiguration().getAgent();
} else if (endpoint.getConfiguration().getAgentConfiguration() !=
null) {
- AgentConfiguration agentConfiguration =
endpoint.getConfiguration().getAgentConfiguration();
+ AgentConfiguration agentConfiguration =
endpoint.getConfiguration().getAgentConfiguration().duplicate();
+ resolveExecuteToolsConcurrentlyExecutor(agentConfiguration);
agent = agentConfiguration.getChatMemoryProvider() != null
? new AgentWithMemory(agentConfiguration)
: new AgentWithoutMemory(agentConfiguration);
@@ -164,6 +168,24 @@ public class LangChain4jAgentProducer extends
DefaultProducer {
}
}
+ /**
+ * When concurrent tool execution is enabled without an explicit executor,
register a Camel-managed thread pool so
+ * tool parallelism uses the framework lifecycle and naming conventions.
+ */
+ private void resolveExecuteToolsConcurrentlyExecutor(AgentConfiguration
agentConfiguration) {
+ if
(!Boolean.TRUE.equals(agentConfiguration.getExecuteToolsConcurrently())) {
+ return;
+ }
+ if (agentConfiguration.getExecuteToolsExecutor() != null) {
+ return;
+ }
+ ThreadPoolProfile profile =
endpoint.getCamelContext().getExecutorServiceManager().getDefaultThreadPoolProfile();
+ managedToolExecutionExecutor =
endpoint.getCamelContext().getExecutorServiceManager()
+ .newThreadPool(this, "LangChain4jAgentToolExecution", profile);
+
agentConfiguration.withExecuteToolsConcurrently(managedToolExecutionExecutor);
+ LOG.debug("Registered Camel-managed executor for concurrent
LangChain4j tool execution");
+ }
+
/**
* Creates a composed tool provider that aggregates tools from all
configured sources: Camel route tools (via tags)
* and MCP tools (via endpoint-level mcpClients and mcpServers
configuration).
@@ -580,6 +602,10 @@ public class LangChain4jAgentProducer extends
DefaultProducer {
}
materializedMcpClients = null;
}
+ if (managedToolExecutionExecutor != null) {
+
endpoint.getCamelContext().getExecutorServiceManager().shutdownGraceful(managedToolExecutionExecutor);
+ managedToolExecutionExecutor = null;
+ }
super.doStop();
}
}
diff --git
a/components/camel-ai/camel-langchain4j-agent/src/test/java/org/apache/camel/component/langchain4j/agent/LangChain4jAgentExecuteToolsConcurrentlyTest.java
b/components/camel-ai/camel-langchain4j-agent/src/test/java/org/apache/camel/component/langchain4j/agent/LangChain4jAgentExecuteToolsConcurrentlyTest.java
new file mode 100644
index 000000000000..90fe96527fab
--- /dev/null
+++
b/components/camel-ai/camel-langchain4j-agent/src/test/java/org/apache/camel/component/langchain4j/agent/LangChain4jAgentExecuteToolsConcurrentlyTest.java
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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 org.apache.camel.component.langchain4j.agent;
+
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import dev.langchain4j.agent.tool.ToolExecutionRequest;
+import dev.langchain4j.data.message.AiMessage;
+import dev.langchain4j.model.chat.ChatModel;
+import dev.langchain4j.model.chat.request.ChatRequest;
+import dev.langchain4j.model.chat.response.ChatResponse;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.langchain4j.agent.api.AgentConfiguration;
+import org.apache.camel.component.langchain4j.agent.api.AiAgentBody;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.spi.Registry;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for CAMEL-23952: {@link
AgentConfiguration#withExecuteToolsConcurrently()} and exchange-safe Camel
route tools.
+ */
+class LangChain4jAgentExecuteToolsConcurrentlyTest extends CamelTestSupport {
+
+ private static final String TAG = "concurrent-tools";
+
+ private final TwoToolRoundTripChatModel chatModel = new
TwoToolRoundTripChatModel();
+ private final AtomicInteger chatRound = new AtomicInteger();
+ private CountDownLatch bothToolsEntered = new CountDownLatch(2);
+ private final AtomicInteger inFlight = new AtomicInteger();
+ private final AtomicInteger maxConcurrent = new AtomicInteger();
+
+ @BeforeEach
+ void resetConcurrencyState() {
+ chatRound.set(0);
+ bothToolsEntered = new CountDownLatch(2);
+ maxConcurrent.set(0);
+ inFlight.set(0);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:agent")
+
.to("langchain4j-agent:test?agentConfiguration=#agentConfig&tags=" + TAG);
+
+ from("ai-tool:slowToolA?tags=" + TAG + "&description=Slow tool
A")
+ .process(exchange -> recordConcurrentEntry("A",
exchange));
+
+ from("ai-tool:slowToolB?tags=" + TAG + "&description=Slow tool
B")
+ .process(exchange -> recordConcurrentEntry("B",
exchange));
+ }
+ };
+ }
+
+ @Override
+ protected void bindToRegistry(Registry registry) {
+ registry.bind("agentConfig", new AgentConfiguration()
+ .withChatModel(chatModel)
+ .withExecuteToolsConcurrently()
+ .withMaxToolCallingRoundTrips(5));
+ }
+
+ @Test
+ void concurrentToolExecutionRunsBothToolsInParallel() throws Exception {
+ String response = template.requestBody("direct:agent", new
AiAgentBody<>("run tools"), String.class);
+
+ assertThat(response).isEqualTo("done");
+ assertThat(bothToolsEntered.await(10, TimeUnit.SECONDS))
+ .as("both tools should be invoked")
+ .isTrue();
+ assertThat(maxConcurrent.get())
+ .as("tools should overlap when executeToolsConcurrently is
enabled")
+ .isGreaterThanOrEqualTo(2);
+ }
+
+ @Test
+ void producerResolvesCamelManagedExecutorWhenNoneConfigured() throws
Exception {
+ try (DefaultCamelContext ctx = new DefaultCamelContext()) {
+ ChatModel noopModel = new ChatModel() {
+ @Override
+ public ChatResponse doChat(ChatRequest request) {
+ return
ChatResponse.builder().aiMessage(AiMessage.from("ok")).build();
+ }
+ };
+ AgentConfiguration config = new AgentConfiguration()
+ .withChatModel(noopModel)
+ .withExecuteToolsConcurrently();
+ ctx.getRegistry().bind("cfg", config);
+
+ ctx.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:x").to("langchain4j-agent:test?agentConfiguration=#cfg");
+ }
+ });
+
+ ctx.start();
+
+ assertThat(config.getExecuteToolsExecutor())
+ .as("registry AgentConfiguration must not be mutated with
a managed executor")
+ .isNull();
+ }
+ }
+
+ @Test
+ void managedExecutorWorksAfterContextStopAndRestart() throws Exception {
+ context.stop();
+ context.getRegistry().bind("agentConfig", new AgentConfiguration()
+ .withChatModel(chatModel)
+ .withExecuteToolsConcurrently()
+ .withMaxToolCallingRoundTrips(5));
+ context.start();
+
+ String response = template.requestBody("direct:agent", new
AiAgentBody<>("run tools"), String.class);
+
+ assertThat(response).isEqualTo("done");
+ }
+
+ private void recordConcurrentEntry(String label, Exchange exchange) throws
InterruptedException {
+ int concurrent = inFlight.incrementAndGet();
+ maxConcurrent.updateAndGet(current -> Math.max(current, concurrent));
+ bothToolsEntered.countDown();
+ assertThat(bothToolsEntered.await(10, TimeUnit.SECONDS))
+ .as("both tools should enter before either completes")
+ .isTrue();
+ inFlight.decrementAndGet();
+ exchange.getMessage().setBody(label);
+ }
+
+ private final class TwoToolRoundTripChatModel implements ChatModel {
+ @Override
+ public ChatResponse doChat(ChatRequest request) {
+ if (chatRound.getAndIncrement() == 0) {
+ List<ToolExecutionRequest> requests = List.of(
+
ToolExecutionRequest.builder().id("a").name("slowToolA").arguments("{}").build(),
+
ToolExecutionRequest.builder().id("b").name("slowToolB").arguments("{}").build());
+ return ChatResponse.builder()
+
.aiMessage(AiMessage.builder().toolExecutionRequests(requests).build())
+ .build();
+ }
+ return
ChatResponse.builder().aiMessage(AiMessage.from("done")).build();
+ }
+ }
+}
