gnodet-bot commented on code in PR #28354: URL: https://github.com/apache/camel/pull/28354#discussion_r4366566430
########## components/camel-ai/camel-openai/src/main/java/org/apache/camel/component/openai/OpenAIToolSearch.java: ########## @@ -0,0 +1,329 @@ +/* + * 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.openai; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.openai.models.chat.completions.ChatCompletionFunctionTool; +import com.openai.models.chat.completions.ChatCompletionMessageToolCall; +import com.openai.models.chat.completions.ChatCompletionTool; +import com.openai.models.embeddings.CreateEmbeddingResponse; +import com.openai.models.embeddings.EmbeddingCreateParams; +import org.apache.camel.Exchange; +import org.apache.camel.component.ai.tool.AiToolSearch; +import org.apache.camel.component.ai.tool.AiToolSearchCandidate; +import org.apache.camel.component.ai.tool.AiToolSearchEmbedder; +import org.apache.camel.component.ai.tool.AiToolSearchIndex; +import org.apache.camel.component.ai.tool.AiToolSearchSession; +import org.apache.camel.component.ai.tool.AiToolSearchStrategy; +import org.apache.camel.component.ai.tool.AiToolSpec; +import org.apache.camel.util.ObjectHelper; + +/** + * Selects the OpenAI tool definitions for one model call when tool search is enabled. + * <p> + * The shared {@link McpToolState} snapshot is not modified. Discovered names live on the exchange, and only when + * conversation memory is enabled, so concurrent requests do not share them. Full definitions are read again from the + * latest snapshot on every call. MCP {@code tools/list} is unchanged: search only changes which definitions are sent to + * the model. + */ +final class OpenAIToolSearch { + + static final String SESSION_PROPERTY = "CamelOpenAIToolSearchSession"; + + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() { + }; + + private static final String KEYWORD_SCHEMA = "{\"type\":\"object\",\"properties\":{\"terms\":{\"type\":\"array\"," + + "\"description\":\"A list of individual search terms used to find relevant tools\"," + + "\"items\":{\"type\":\"string\"}}},\"required\":[\"terms\"]}"; + private static final String VECTOR_SCHEMA = "{\"type\":\"object\",\"properties\":{\"query\":{\"type\":\"string\"," + + "\"description\":\"Natural language query describing the desired tool\"}}," + + "\"required\":[\"query\"]}"; + + private final Object indexLock = new Object(); + private String cachedFingerprint; + private AiToolSearchIndex cachedIndex; + + static void validateConfiguration(OpenAIConfiguration configuration) { + if (!configuration.isToolSearch()) { + return; + } + if (configuration.getToolSearchMaxResults() < 1) { + throw new IllegalArgumentException("toolSearchMaxResults must be at least 1"); + } + if (configuration.getToolSearchStrategy() == null) { + throw new IllegalArgumentException("toolSearchStrategy is required when toolSearch=true"); + } + if (configuration.getToolSearchStrategy() == AiToolSearchStrategy.VECTOR + && ObjectHelper.isEmpty(configuration.getToolSearchEmbeddingModel())) { + throw new IllegalArgumentException( + "toolSearchStrategy=vector requires toolSearchEmbeddingModel and does not fall back to keyword search"); + } + } + + static void rejectReservedName(McpToolState state) { + if (state.knownToolNames().contains(AiToolSearch.TOOL_NAME)) { + throw new IllegalArgumentException( + "Tool name '" + AiToolSearch.TOOL_NAME + "' is reserved while toolSearch=true"); + } + } + + AiToolSearchSession openSession(Exchange exchange, OpenAIConfiguration configuration) { + if (configuration.isConversationMemory()) { + AiToolSearchSession existing = exchange.getProperty(SESSION_PROPERTY, AiToolSearchSession.class); + if (existing != null) { + return existing; + } + AiToolSearchSession created = new AiToolSearchSession(); + exchange.setProperty(SESSION_PROPERTY, created); + return created; + } + AiToolSearchSession created = new AiToolSearchSession(); + exchange.setProperty(SESSION_PROPERTY, created); + return created; + } + + void closeSession(Exchange exchange, OpenAIConfiguration configuration) { + if (!configuration.isConversationMemory()) { + exchange.removeProperty(SESSION_PROPERTY); + } + } + + List<ChatCompletionTool> toolsFor(OpenAIEndpoint endpoint, AiToolSearchSession session) { + McpToolState state = endpoint.getMcpToolState(); + rejectReservedName(state); + return select(endpoint.getConfiguration(), state, session).stream().map(ChatCompletionTool::ofFunction).toList(); + } + + McpToolCallExecutor.ToolResult execute(OpenAIEndpoint endpoint, ChatCompletionMessageToolCall toolCall, Exchange exchange) { + OpenAIConfiguration configuration = endpoint.getConfiguration(); + AiToolSearchSession session = exchange.getProperty(SESSION_PROPERTY, AiToolSearchSession.class); + if (session == null) { + session = openSession(exchange, configuration); + } + McpToolState state = endpoint.getMcpToolState(); + String arguments = toolCall.asFunction().function().arguments(); + String content; + try { + content = search(configuration, state, session, arguments, index(endpoint, state)); + } catch (IllegalArgumentException e) { + content = "Error: " + e.getMessage(); + } + return new McpToolCallExecutor.ToolResult( + toolCall.asFunction().id(), AiToolSearch.TOOL_NAME, content, false, 0, !content.startsWith("Error:")); + } + + static String search( + OpenAIConfiguration configuration, McpToolState state, AiToolSearchSession session, String argumentsJson, + AiToolSearchIndex index) { + List<AiToolSearchCandidate> hits; + if (configuration.getToolSearchStrategy() == AiToolSearchStrategy.VECTOR) { + String query = readString(argumentsJson, AiToolSearch.QUERY_ARGUMENT); + if (query == null) { + return "Error: missing required tool argument '" + AiToolSearch.QUERY_ARGUMENT + "'"; + } + hits = index.searchVector(query); + } else { + List<String> terms = readTerms(argumentsJson); + if (terms == null) { + return "Error: missing required tool argument '" + AiToolSearch.TERMS_ARGUMENT + "'"; + } + hits = index.searchKeyword(terms); + } + List<String> names = hits.stream().map(AiToolSearchCandidate::name).toList(); + session.remember(names); + return AiToolSearch.resultMessage(names); + } + + static List<AiToolSearchCandidate> candidates(McpToolState state) { + List<AiToolSearchCandidate> candidates = new ArrayList<>(); + for (ChatCompletionFunctionTool tool : toolsByName(state).values()) { + String name = tool.function().name(); + if (!isSearchable(state, name)) { + continue; + } + AiToolSpec route = state.routeTools().get(name); + String description = route != null ? route.getDescription() : tool.function().description().orElse(null); + String schema = route != null + ? route.getParametersJsonSchema() + : tool.function().parameters().map(Object::toString).orElse(null); + Object source = route != null ? route : name; + candidates.add(new AiToolSearchCandidate(name, description, schema, true, source)); + } + return List.copyOf(candidates); + } + + static boolean isSearchable(McpToolState state, String name) { + AiToolSpec route = state.routeTools().get(name); + if (route != null) { + return route.isSearchable(); + } + return state.toolClientMap().containsKey(name); + } + + static ChatCompletionFunctionTool synthetic(AiToolSearchStrategy strategy) { + boolean vector = strategy == AiToolSearchStrategy.VECTOR; + return AiToolSpecToOpenAI.function( + AiToolSearch.TOOL_NAME, + vector + ? "Finds available tools using semantic vector search. Matching tools become available to call on the next turn." + : "Finds available tools whose name or description contains the given search terms. Matching tools become available to call on the next turn.", + vector ? VECTOR_SCHEMA : KEYWORD_SCHEMA); + } + + private AiToolSearchIndex index(OpenAIEndpoint endpoint, McpToolState state) { + OpenAIConfiguration configuration = endpoint.getConfiguration(); + String fingerprint = fingerprint(configuration, state); + synchronized (indexLock) { + if (fingerprint.equals(cachedFingerprint) && cachedIndex != null) { + return cachedIndex; + } + } + AiToolSearchEmbedder embedder = configuration.getToolSearchStrategy() == AiToolSearchStrategy.VECTOR + ? embedder(endpoint) + : null; + AiToolSearchIndex built = AiToolSearchIndex.create( + configuration.getToolSearchStrategy(), candidates(state), configuration.getToolSearchMaxResults(), + embedder); + synchronized (indexLock) { Review Comment: 🔴 **TOCTOU race — still open from prior review.** `index()` uses two disjoint `synchronized(indexLock)` blocks (lines 198–202 and 209–213) with unguarded `AiToolSearchIndex.create()` between them. Thread A checks the fingerprint, releases the lock, starts building. Thread B observes the same state, also starts building. Thread A writes its result; Thread B overwrites it. The final `cachedIndex` is non-deterministic and redundant indexes are built under burst load. Fix: consolidate the cache check, build, and update into a single lock acquisition: ```suggestion private AiToolSearchIndex index(OpenAIEndpoint endpoint, McpToolState state) { OpenAIConfiguration configuration = endpoint.getConfiguration(); String fingerprint = fingerprint(configuration, state); synchronized (indexLock) { if (fingerprint.equals(cachedFingerprint) && cachedIndex != null) { return cachedIndex; } AiToolSearchEmbedder embedder = configuration.getToolSearchStrategy() == AiToolSearchStrategy.VECTOR ? embedder(endpoint) : null; AiToolSearchIndex built = AiToolSearchIndex.create( configuration.getToolSearchStrategy(), candidates(state), configuration.getToolSearchMaxResults(), embedder); cachedFingerprint = fingerprint; cachedIndex = built; return built; } } ``` ########## components/camel-ai/camel-langchain4j-agent/src/main/java/org/apache/camel/component/langchain4j/agent/CamelToolSearchStrategy.java: ########## @@ -0,0 +1,318 @@ +/* + * 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.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.langchain4j.agent.tool.ToolSpecification; +import dev.langchain4j.data.segment.TextSegment; +import dev.langchain4j.invocation.InvocationContext; +import dev.langchain4j.model.chat.request.json.JsonArraySchema; +import dev.langchain4j.model.chat.request.json.JsonObjectSchema; +import dev.langchain4j.model.chat.request.json.JsonStringSchema; +import dev.langchain4j.model.embedding.EmbeddingModel; +import dev.langchain4j.service.tool.search.ToolSearchRequest; +import dev.langchain4j.service.tool.search.ToolSearchResult; +import dev.langchain4j.service.tool.search.ToolSearchStrategy; +import org.apache.camel.component.ai.tool.AiToolSearch; +import org.apache.camel.component.ai.tool.AiToolSearchCandidate; +import org.apache.camel.component.ai.tool.AiToolSearchEmbedder; +import org.apache.camel.component.ai.tool.AiToolSearchIndex; +import org.apache.camel.component.ai.tool.AiToolSearchStrategy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Ranks the tools LangChain4j has already combined from route and MCP providers. + * <p> + * LangChain4j still executes a found tool through its original executor and restores names found earlier from chat + * memory. This strategy only replaces the ranking engine. Dynamic tool providers registered outside Camel's composite + * are refreshed after search and are not part of this catalogue. + */ +final class CamelToolSearchStrategy implements ToolSearchStrategy { + + private static final Logger LOG = LoggerFactory.getLogger(CamelToolSearchStrategy.class); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() { + }; + + private static final int MIN_CACHED_EMBEDDINGS = 4096; + private static final int CACHE_HEADROOM = 256; + + private final AiToolSearchStrategy strategy; + private final int maxResults; + + /* + * The cache belongs to this strategy instance and therefore to its final embeddingModel. + * Access is serialized by search(), including batch population and query embedding. + */ + private final EmbeddingModel embeddingModel; + private int maxCachedEmbeddings = MIN_CACHED_EMBEDDINGS; + + private final Map<String, float[]> embeddingCache = new LinkedHashMap<>(256, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<String, float[]> eldest) { + return size() > maxCachedEmbeddings; + } + }; + + /* + * Embedding texts belonging to the previously searched tool catalogue. + * This lets each search discard entries for tool content that has disappeared or changed. + */ + private Set<String> cachedToolEmbeddingKeys = Set.of(); + + CamelToolSearchStrategy(AiToolSearchStrategy strategy, int maxResults, EmbeddingModel embeddingModel) { + this.strategy = strategy; + this.maxResults = maxResults; + this.embeddingModel = embeddingModel; + } + + @Override + public List<ToolSpecification> getToolSearchTools(InvocationContext invocationContext) { + return List.of(synthetic()); + } + + @Override + public synchronized ToolSearchResult search(ToolSearchRequest request) { Review Comment: 🔴 **New: `search()` holds the instance monitor for the entire embedding lifecycle — throughput bottleneck.** `search()` is `synchronized`, meaning one thread holds the instance lock from entry through `AiToolSearchIndex.create()` and the full `index.searchVector(query)` call — which calls `embeddingModel.embed(text)` for each cache miss. For remote embedding models (OpenAI, Cohere, etc.), each `embed()` call is an outbound HTTP request taking 100–500ms. Under concurrent load, all tool-search requests queue behind a single monitor. Throughput degrades to 1 / (embedding latency) regardless of thread count — far worse than the non-synchronized path. The cache (`embeddingCache`) is the shared mutable state worth protecting, not the embedding calls themselves. Solution: hold the lock only for cache reads/writes: ```suggestion public ToolSearchResult search(ToolSearchRequest request) { ``` Then replace `embeddingCache.get(text)` / `embeddingCache.put(text, ...)` with `synchronized(this)` blocks scoped around only the map access, and let `embeddingModel.embed()` / `embeddingModel.embedAll()` run outside the lock. ########## components/camel-spring-parent/camel-spring-ai/camel-spring-ai-chat/src/main/java/org/apache/camel/component/springai/chat/SpringAiToolSearch.java: ########## @@ -0,0 +1,382 @@ +/* + * 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.springai.chat; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.camel.component.ai.tool.AiToolSearch; +import org.apache.camel.component.ai.tool.AiToolSearchCandidate; +import org.apache.camel.component.ai.tool.AiToolSearchEmbedder; +import org.apache.camel.component.ai.tool.AiToolSearchIndex; +import org.apache.camel.component.ai.tool.AiToolSearchSession; +import org.apache.camel.component.ai.tool.AiToolSearchSessions; +import org.apache.camel.component.ai.tool.AiToolSearchStrategy; +import org.apache.camel.util.json.JsonObject; +import org.apache.camel.util.json.Jsoner; +import org.springframework.ai.chat.client.ChatClientRequest; +import org.springframework.ai.chat.client.advisor.ToolCallingAdvisor; +import org.springframework.ai.chat.client.advisor.api.CallAdvisorChain; +import org.springframework.ai.chat.client.advisor.api.StreamAdvisorChain; +import org.springframework.ai.chat.prompt.Prompt; +import org.springframework.ai.embedding.EmbeddingModel; +import org.springframework.ai.model.tool.ToolCallingChatOptions; +import org.springframework.ai.model.tool.ToolCallingManager; +import org.springframework.ai.model.tool.ToolExecutionEligibilityChecker; +import org.springframework.ai.tool.ToolCallback; +import org.springframework.ai.tool.definition.ToolDefinition; + +/** + * Rebuilds the tool list inside Spring AI's existing tool-calling loop. + * <p> + * The advisor does not run its own loop. {@code doBeforeCall} and {@code doBeforeStream} replace the immutable callback + * snapshot with the directly visible tools, the synthetic search tool, and tools already discovered for this request or + * conversation. Execution stays on the original callback. + */ +final class SpringAiToolSearch { + + static final String CATALOGUE = "CamelSpringAiToolSearchCatalogue"; + static final String SEARCHABLE = "CamelSpringAiToolSearchSearchableNames"; + static final String CONVERSATION = "CamelSpringAiToolSearchConversationId"; + static final String SESSION = "CamelSpringAiToolSearchSession"; + + private static final String KEYWORD_SCHEMA = "{\"type\":\"object\",\"properties\":{\"terms\":{\"type\":\"array\"," + + "\"description\":\"A list of individual search terms used to find relevant tools\"," + + "\"items\":{\"type\":\"string\"}}},\"required\":[\"terms\"]}"; + private static final String VECTOR_SCHEMA = "{\"type\":\"object\",\"properties\":{\"query\":{\"type\":\"string\"," + + "\"description\":\"Natural language query describing the desired tool\"}}," + + "\"required\":[\"query\"]}"; + + private final AiToolSearchStrategy strategy; + private final int maxResults; + private final EmbeddingModel embeddingModel; + private final AiToolSearchSessions sessions = new AiToolSearchSessions(); + private final Object indexLock = new Object(); + private String cachedFingerprint; + private AiToolSearchIndex cachedIndex; + + SpringAiToolSearch(SpringAiChatConfiguration configuration) { + this.strategy = configuration.getToolSearchStrategy(); + this.maxResults = configuration.getToolSearchMaxResults(); + this.embeddingModel = configuration.getToolSearchEmbeddingModel(); + } + + static void validate(SpringAiChatConfiguration configuration) { + if (!configuration.isToolSearch()) { + return; + } + if (configuration.getChatClient() != null) { + throw new IllegalArgumentException( + "toolSearch=true requires Camel to build the ChatClient from chatModel. " + + "A prebuilt chatClient cannot install the tool search advisor."); + } + if (configuration.getToolSearchMaxResults() < 1) { + throw new IllegalArgumentException("toolSearchMaxResults must be at least 1"); + } + if (configuration.getToolSearchStrategy() == null) { + throw new IllegalArgumentException("toolSearchStrategy is required when toolSearch=true"); + } + if (configuration.getToolSearchStrategy() == AiToolSearchStrategy.VECTOR + && configuration.getToolSearchEmbeddingModel() == null) { + throw new IllegalArgumentException( + "toolSearchStrategy=vector requires toolSearchEmbeddingModel and does not fall back to keyword search"); + } + } + + ChatClientRequest initialize(ChatClientRequest request) { + if (!request.context().containsKey(CATALOGUE)) { + return request; + } + if (conversationId(request) != null || request.context().get(SESSION) instanceof AiToolSearchSession) { + return request; + } + Map<String, Object> context = new HashMap<>(request.context()); + context.put(SESSION, new AiToolSearchSession()); + return request.mutate().context(context).build(); + } + + ChatClientRequest prepare(ChatClientRequest request) { + if (!request.context().containsKey(CATALOGUE)) { + return request; + } + List<ToolCallback> catalogue = catalogue(request); + Set<String> searchable = searchable(request); + Map<String, ToolCallback> byName = new LinkedHashMap<>(); + for (ToolCallback callback : catalogue) { + byName.putIfAbsent(callback.getToolDefinition().name(), callback); + } + AiToolSearchSession session = session(request); + session.retainAvailable(byName.keySet()); + + List<ToolCallback> selected = new ArrayList<>(); + for (ToolCallback callback : byName.values()) { + if (!searchable.contains(callback.getToolDefinition().name())) { + selected.add(callback); + } + } + selected.add(searchCallback(catalogue, searchable, session)); + for (String name : session.discoveredNames()) { + ToolCallback discovered = byName.get(name); + if (discovered != null && searchable.contains(name)) { + selected.add(discovered); + } + } + ToolCallingChatOptions options = ((ToolCallingChatOptions) request.prompt().getOptions()).mutate() + .toolCallbacks(List.copyOf(selected)) + .build(); + Prompt prompt = request.prompt().mutate().chatOptions(options).build(); + return request.mutate().prompt(prompt).build(); + } + + private ToolCallback searchCallback( + List<ToolCallback> catalogue, Set<String> searchable, AiToolSearchSession session) { + boolean vector = strategy == AiToolSearchStrategy.VECTOR; + ToolDefinition definition = ToolDefinition.builder() + .name(AiToolSearch.TOOL_NAME) + .description(vector + ? "Finds available tools using semantic vector search. Matching tools become available to call on the next turn." + : "Finds available tools whose name or description contains the given search terms. Matching tools become available to call on the next turn.") + .inputSchema(vector ? VECTOR_SCHEMA : KEYWORD_SCHEMA) + .build(); + return new ToolCallback() { + @Override + public ToolDefinition getToolDefinition() { + return definition; + } + + @Override + public String call(String toolInput) { + try { + List<AiToolSearchCandidate> hits = vector + ? index(catalogue, searchable).searchVector(requiredQuery(toolInput)) + : index(catalogue, searchable).searchKeyword(requiredTerms(toolInput)); + List<String> names = hits.stream().map(AiToolSearchCandidate::name).toList(); + session.remember(names); + return AiToolSearch.resultMessage(names); + } catch (IllegalArgumentException e) { + return "Error: " + e.getMessage(); + } + } + }; + } + + private AiToolSearchIndex index(List<ToolCallback> catalogue, Set<String> searchable) { + List<AiToolSearchCandidate> candidates = new ArrayList<>(); + for (ToolCallback callback : catalogue) { + String name = callback.getToolDefinition().name(); + if (!searchable.contains(name) || AiToolSearch.TOOL_NAME.equals(name)) { + continue; + } + candidates.add(new AiToolSearchCandidate( + name, callback.getToolDefinition().description(), callback.getToolDefinition().inputSchema(), true, + callback)); + } + String fingerprint = fingerprint(candidates); + synchronized (indexLock) { + if (fingerprint.equals(cachedFingerprint) && cachedIndex != null) { + return cachedIndex; + } + } + AiToolSearchEmbedder embedder = strategy == AiToolSearchStrategy.VECTOR ? this::embed : null; + AiToolSearchIndex built = AiToolSearchIndex.create(strategy, candidates, maxResults, embedder); + synchronized (indexLock) { Review Comment: 🔴 **Same TOCTOU race as `OpenAIToolSearch` — still open from prior review.** Identical two-lock pattern. Fix: same consolidation — hold the lock through the build: ```suggestion private AiToolSearchIndex index(List<ToolCallback> catalogue, Set<String> searchable) { List<AiToolSearchCandidate> candidates = new ArrayList<>(); for (ToolCallback callback : catalogue) { String name = callback.getToolDefinition().name(); if (!searchable.contains(name) || AiToolSearch.TOOL_NAME.equals(name)) { continue; } candidates.add(new AiToolSearchCandidate( name, callback.getToolDefinition().description(), callback.getToolDefinition().inputSchema(), true, callback)); } String fingerprint = fingerprint(candidates); synchronized (indexLock) { if (fingerprint.equals(cachedFingerprint) && cachedIndex != null) { return cachedIndex; } AiToolSearchEmbedder embedder = strategy == AiToolSearchStrategy.VECTOR ? this::embed : null; AiToolSearchIndex built = AiToolSearchIndex.create(strategy, candidates, maxResults, embedder); cachedFingerprint = fingerprint; cachedIndex = built; return built; } } ``` ########## components/camel-ai/camel-langchain4j-agent/src/main/java/org/apache/camel/component/langchain4j/agent/CamelToolSearchStrategy.java: ########## @@ -0,0 +1,318 @@ +/* + * 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.ArrayList; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import dev.langchain4j.agent.tool.ToolSpecification; +import dev.langchain4j.data.segment.TextSegment; +import dev.langchain4j.invocation.InvocationContext; +import dev.langchain4j.model.chat.request.json.JsonArraySchema; +import dev.langchain4j.model.chat.request.json.JsonObjectSchema; +import dev.langchain4j.model.chat.request.json.JsonStringSchema; +import dev.langchain4j.model.embedding.EmbeddingModel; +import dev.langchain4j.service.tool.search.ToolSearchRequest; +import dev.langchain4j.service.tool.search.ToolSearchResult; +import dev.langchain4j.service.tool.search.ToolSearchStrategy; +import org.apache.camel.component.ai.tool.AiToolSearch; +import org.apache.camel.component.ai.tool.AiToolSearchCandidate; +import org.apache.camel.component.ai.tool.AiToolSearchEmbedder; +import org.apache.camel.component.ai.tool.AiToolSearchIndex; +import org.apache.camel.component.ai.tool.AiToolSearchStrategy; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Ranks the tools LangChain4j has already combined from route and MCP providers. + * <p> + * LangChain4j still executes a found tool through its original executor and restores names found earlier from chat + * memory. This strategy only replaces the ranking engine. Dynamic tool providers registered outside Camel's composite + * are refreshed after search and are not part of this catalogue. + */ +final class CamelToolSearchStrategy implements ToolSearchStrategy { + + private static final Logger LOG = LoggerFactory.getLogger(CamelToolSearchStrategy.class); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final TypeReference<Map<String, Object>> MAP_TYPE = new TypeReference<>() { + }; + + private static final int MIN_CACHED_EMBEDDINGS = 4096; + private static final int CACHE_HEADROOM = 256; + + private final AiToolSearchStrategy strategy; + private final int maxResults; + + /* + * The cache belongs to this strategy instance and therefore to its final embeddingModel. + * Access is serialized by search(), including batch population and query embedding. + */ + private final EmbeddingModel embeddingModel; + private int maxCachedEmbeddings = MIN_CACHED_EMBEDDINGS; + + private final Map<String, float[]> embeddingCache = new LinkedHashMap<>(256, 0.75f, true) { + @Override + protected boolean removeEldestEntry(Map.Entry<String, float[]> eldest) { + return size() > maxCachedEmbeddings; + } + }; + + /* + * Embedding texts belonging to the previously searched tool catalogue. + * This lets each search discard entries for tool content that has disappeared or changed. + */ + private Set<String> cachedToolEmbeddingKeys = Set.of(); + + CamelToolSearchStrategy(AiToolSearchStrategy strategy, int maxResults, EmbeddingModel embeddingModel) { + this.strategy = strategy; + this.maxResults = maxResults; + this.embeddingModel = embeddingModel; + } + + @Override + public List<ToolSpecification> getToolSearchTools(InvocationContext invocationContext) { + return List.of(synthetic()); + } + + @Override + public synchronized ToolSearchResult search(ToolSearchRequest request) { + List<AiToolSearchCandidate> candidates = new ArrayList<>(); + for (ToolSpecification specification : request.searchableTools()) { + if (AiToolSearch.TOOL_NAME.equals(specification.name())) { + throw new IllegalArgumentException( + "Tool name '" + AiToolSearch.TOOL_NAME + "' is reserved while toolSearch=true"); + } + String schema = specification.parameters() == null ? null : specification.parameters().toString(); + candidates.add(new AiToolSearchCandidate( + specification.name(), specification.description(), schema, true, specification)); + } + + // Validate search arguments early before performing any embedding work. + String arguments = request.toolExecutionRequest() == null ? null : request.toolExecutionRequest().arguments(); + String query = null; + List<String> terms = null; + + if (strategy == AiToolSearchStrategy.VECTOR) { + query = readString(arguments, AiToolSearch.QUERY_ARGUMENT); + if (query == null) { + return new ToolSearchResult( + List.of(), + "Error: missing required tool argument '" + AiToolSearch.QUERY_ARGUMENT + "'"); + } + } else { + terms = readTerms(arguments); + if (terms == null) { + return new ToolSearchResult( + List.of(), + "Error: missing required tool argument '" + AiToolSearch.TERMS_ARGUMENT + "'"); + } + } + + if (strategy == AiToolSearchStrategy.VECTOR && embeddingModel != null) { + List<AiToolSearchCandidate> searchable = searchableCandidates(candidates); + Set<String> currentToolKeys = new LinkedHashSet<>(); + for (AiToolSearchCandidate candidate : searchable) { + currentToolKeys.add(embeddingText(candidate)); + } + + /* + * Keep enough room for every distinct current tool embedding plus a bounded + * amount of query-cache headroom. This avoids evicting current tool vectors + * while the index is being constructed, even for catalogues larger than the + * minimum cache capacity. + */ + maxCachedEmbeddings = Math.max( + MIN_CACHED_EMBEDDINGS, + currentToolKeys.size() + CACHE_HEADROOM); + + // Remove embeddings for tools no longer present or whose content has changed. + for (String previousKey : cachedToolEmbeddingKeys) { + if (!currentToolKeys.contains(previousKey)) { + embeddingCache.remove(previousKey); + } + } + cachedToolEmbeddingKeys = Set.copyOf(currentToolKeys); + + List<String> missingTexts = new ArrayList<>(); + for (String text : currentToolKeys) { + if (!embeddingCache.containsKey(text)) { + missingTexts.add(text); + } + } + + if (!missingTexts.isEmpty()) { + try { + List<TextSegment> segments = missingTexts.stream() + .map(TextSegment::from) + .toList(); + var response = embeddingModel.embedAll(segments); + if (response != null && response.content() != null + && response.content().size() == missingTexts.size()) { + boolean valid = true; + List<float[]> vectors = new ArrayList<>(missingTexts.size()); + for (int i = 0; i < missingTexts.size(); i++) { + var embedding = response.content().get(i); + if (embedding == null || embedding.vector() == null || embedding.vector().length == 0) { + valid = false; + break; + } + vectors.add(embedding.vector()); + } + if (valid) { + for (int i = 0; i < missingTexts.size(); i++) { + embeddingCache.putIfAbsent(missingTexts.get(i), vectors.get(i)); + } + } else { + LOG.debug( + "Batch embedding returned invalid or empty vectors, falling back to individual embedding"); + } + } else { + LOG.debug("Batch embedding returned null or mismatched size, falling back to individual embedding"); + } + } catch (UnsupportedOperationException e) { + LOG.debug( + "Batch embedding (embedAll) not supported by embedding model, falling back to individual embedding"); + } catch (Exception e) { + LOG.warn("Batch embedding failed unexpectedly, falling back to individual embedding: {}", e.getMessage()); + } + } + } + + AiToolSearchEmbedder embedder = embeddingModel == null ? null : text -> { + float[] cached = embeddingCache.get(text); + if (cached != null) { + return cached; + } + + var response = embeddingModel.embed(text); + if (response == null || response.content() == null || response.content().vector() == null) { + throw new IllegalStateException("Embedding model returned no vector"); + } + float[] vector = response.content().vector(); + if (vector.length == 0) { + throw new IllegalStateException("Embedding model returned an empty vector"); + } + embeddingCache.put(text, vector); + return vector; + }; + + AiToolSearchIndex index = AiToolSearchIndex.create(strategy, candidates, maxResults, embedder); + if (strategy == AiToolSearchStrategy.VECTOR) { + return result(index.searchVector(query)); + } + return result(index.searchKeyword(terms)); + } + + private ToolSpecification synthetic() { + boolean vector = strategy == AiToolSearchStrategy.VECTOR; + JsonObjectSchema parameters = vector + ? JsonObjectSchema.builder() + .addProperty(AiToolSearch.QUERY_ARGUMENT, JsonStringSchema.builder() + .description("Natural language query describing the desired tool") + .build()) + .required(AiToolSearch.QUERY_ARGUMENT) + .build() + : JsonObjectSchema.builder() + .addProperty(AiToolSearch.TERMS_ARGUMENT, JsonArraySchema.builder() + .description("A list of individual search terms used to find relevant tools") + .items(JsonStringSchema.builder().build()) + .build()) + .required(AiToolSearch.TERMS_ARGUMENT) + .build(); + return ToolSpecification.builder() + .name(AiToolSearch.TOOL_NAME) + .description(vector + ? "Finds available tools using semantic vector search. Matching tools become available to call on the next turn." + : "Finds available tools whose name or description contains the given search terms. Matching tools become available to call on the next turn.") + .parameters(parameters) + .build(); + } + + private static ToolSearchResult result(List<AiToolSearchCandidate> hits) { + List<String> names = hits.stream().map(AiToolSearchCandidate::name).toList(); + return new ToolSearchResult(names, AiToolSearch.resultMessage(names)); + } + + private static List<AiToolSearchCandidate> searchableCandidates(List<AiToolSearchCandidate> candidates) { + if (candidates == null || candidates.isEmpty()) { + return List.of(); + } + Map<String, AiToolSearchCandidate> byName = new LinkedHashMap<>(); + for (AiToolSearchCandidate candidate : candidates) { + if (candidate == null || !candidate.searchable() || candidate.name().isBlank()) { + continue; + } + byName.putIfAbsent(candidate.name(), candidate); + } + return List.copyOf(byName.values()); + } + + private static String embeddingText(AiToolSearchCandidate candidate) { + String description = candidate.description(); + if (description == null || description.isBlank()) { + return candidate.name(); + } + return candidate.name() + "\n" + description; + } + + private static List<String> readTerms(String argumentsJson) { + Map<String, Object> arguments = readArguments(argumentsJson); + if (arguments == null || !arguments.containsKey(AiToolSearch.TERMS_ARGUMENT)) { + return null; + } + Object value = arguments.get(AiToolSearch.TERMS_ARGUMENT); + if (value instanceof List<?> list) { + List<String> terms = new ArrayList<>(); + for (Object item : list) { + if (item != null) { + terms.add(item.toString()); + } + } + return terms; + } + if (value instanceof String text) { + return List.of(text); + } + return null; + } + + private static String readString(String argumentsJson, String name) { + Map<String, Object> arguments = readArguments(argumentsJson); + if (arguments == null || arguments.get(name) == null) { + return null; + } + String text = arguments.get(name).toString().trim(); + return text.isEmpty() ? null : text; + } + + private static Map<String, Object> readArguments(String argumentsJson) { + if (argumentsJson == null || argumentsJson.isBlank()) { + return null; + } + try { + return OBJECT_MAPPER.readValue(argumentsJson, MAP_TYPE); + } catch (Exception e) { + return null; + } + } Review Comment: 🟡 **DRY violation — still open from prior review.** `readTerms()`, `readString()`, and `readArguments()` are copy-pasted verbatim across: - `CamelToolSearchStrategy` (lines 278–317) - `OpenAIToolSearch` (lines 285–328) - `SpringAiToolSearch` (lines 269–308) Extract to a shared utility in `camel-ai-tool` (e.g. `AiToolSearchArguments`). Three copies of the same JSON parser means any fix must be applied in 3 places. ########## components/camel-ai/camel-langchain4j-agent/src/main/java/org/apache/camel/component/langchain4j/agent/LangChain4jAgentProducer.java: ########## @@ -291,6 +295,35 @@ private static void applyEndpointToolCallingOptions( if (endpointConfiguration.getExecuteToolsConcurrently() != null) { agentConfiguration.withExecuteToolsConcurrentlyEnabled(endpointConfiguration.getExecuteToolsConcurrently()); } + if (endpointConfiguration.isToolSearch()) { + if (endpointConfiguration.getToolSearchMaxResults() < 1) { + throw new IllegalArgumentException("toolSearchMaxResults must be at least 1"); + } + if (endpointConfiguration.getToolSearchStrategy() == null) { + throw new IllegalArgumentException("toolSearchStrategy is required when toolSearch=true"); + } + if (endpointConfiguration.getToolSearchStrategy() == AiToolSearchStrategy.VECTOR + && endpointConfiguration.getToolSearchEmbeddingModel() == null) { + throw new IllegalArgumentException( + "toolSearchStrategy=vector requires toolSearchEmbeddingModel and does not fall back to keyword search"); + } + agentConfiguration.withToolSearchStrategy(new CamelToolSearchStrategy( + endpointConfiguration.getToolSearchStrategy(), + endpointConfiguration.getToolSearchMaxResults(), + endpointConfiguration.getToolSearchEmbeddingModel())); + } + } + + private void rejectReservedToolName(LangChain4jAgentConfiguration configuration) { + if (!configuration.isToolSearch() || configuration.getTags() == null || configuration.getTags().isBlank()) { + return; + } + for (AiToolSpec spec : discoverAiToolsByName(configuration.getTags()).values()) { + if (AiToolSearch.TOOL_NAME.equals(spec.getName())) { + throw new IllegalArgumentException( + "Tool name '" + AiToolSearch.TOOL_NAME + "' is reserved while toolSearch=true"); + } + } Review Comment: 🟡 **MCP tools bypass the reserved-name guard — still open from prior review.** `rejectReservedToolName` iterates only `discoverAiToolsByName(configuration.getTags())` (route tools). MCP tools from `mcpClients` / `mcpServers` are never validated. If an MCP server exposes a tool named `ai-tool-search` (the `AiToolSearch.TOOL_NAME` value), the reserved name check silently passes and the conflict corrupts the tool-search mechanism at runtime. Either validate MCP tool names in this guard, or document the limitation explicitly so operators know MCP servers must not expose reserved names. ########## components/camel-ai/camel-ai-tool/src/main/java/org/apache/camel/component/ai/tool/AiToolSearchSessions.java: ########## @@ -0,0 +1,201 @@ + +/* + * 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.ai.tool; + +import java.util.Iterator; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.concurrent.TimeUnit; + +/** + * Conversation-scoped {@link AiToolSearchSession}s. + * <p> + * Sessions are retained only after a tool has been discovered. Retained sessions expire after a period of inactivity + * and are subject to a maximum size. Access and mutation are synchronized so expiration and capacity eviction cannot + * corrupt the session catalogue. + * <p> + * Expired entries are removed during subsequent operations; no background thread is required. Callers can explicitly + * remove a conversation or clear all retained sessions when their conversation or chat-memory lifecycle ends. + * + * @since 4.24 + */ +public final class AiToolSearchSessions { + + static final int DEFAULT_MAX_SESSIONS = 10_000; + static final long DEFAULT_EXPIRATION_MILLIS = TimeUnit.MINUTES.toMillis(30); + + private final int maxSessions; + private final long expirationNanos; + + /* + * Access-order LinkedHashMap provides least-recently-used eviction. + * All access is guarded by this instance's monitor. + */ + private final LinkedHashMap<String, Entry> sessions = new LinkedHashMap<>(16, 0.75f, true); + + public AiToolSearchSessions() { + this(DEFAULT_MAX_SESSIONS, DEFAULT_EXPIRATION_MILLIS); + } + + AiToolSearchSessions(int maxSessions, long expirationMillis) { + if (maxSessions < 1) { + throw new IllegalArgumentException("maxSessions must be greater than zero"); + } + if (expirationMillis < 1) { + throw new IllegalArgumentException("expirationMillis must be greater than zero"); + } + this.maxSessions = maxSessions; + this.expirationNanos = TimeUnit.MILLISECONDS.toNanos(expirationMillis); + } + + /** + * Returns the session for a conversation id. + * <p> + * An unregistered session is returned without retaining it. It is registered only when a tool is remembered. + * + * @param conversationId stable id for one conversation; must not be blank + * @return the session for that id + */ + public synchronized AiToolSearchSession session(String conversationId) { + validateConversationId(conversationId); + + long now = System.nanoTime(); + purgeExpired(now); + + Entry entry = sessions.get(conversationId); + if (entry != null) { + entry.lastAccessNanos = now; + return entry.session; + } + + return new AiToolSearchSession(this, conversationId); + } + + /** + * Removes the retained session for a conversation. + * + * @param conversationId conversation id + * @return true if a retained session was removed + */ + public synchronized boolean remove(String conversationId) { + if (conversationId == null || conversationId.isBlank()) { + return false; + } + purgeExpired(System.nanoTime()); + return sessions.remove(conversationId) != null; + } + + /** + * Removes all retained sessions. + */ + public synchronized void clear() { + sessions.clear(); + } + + /** + * Records discovered tool names, registering the session only when names are present. + */ + synchronized void remember(String conversationId, java.util.Collection<String> names) { + if (names == null || names.isEmpty()) { + return; + } + + long now = System.nanoTime(); + purgeExpired(now); + + Entry entry = sessions.get(conversationId); + if (entry == null) { + AiToolSearchSession stored = new AiToolSearchSession(); + stored.rememberLocal(names); + sessions.put(conversationId, new Entry(stored, now)); + evictExcessSessions(); + } else { + entry.session.rememberLocal(names); + entry.lastAccessNanos = now; + } + } + + synchronized java.util.List<String> discoveredNames(String conversationId) { + long now = System.nanoTime(); + purgeExpired(now); + + Entry entry = sessions.get(conversationId); + if (entry == null) { + return java.util.List.of(); + } + + entry.lastAccessNanos = now; + return entry.session.discoveredNamesLocal(); + } + + synchronized void retainAvailable(String conversationId, java.util.Set<String> availableNames) { + long now = System.nanoTime(); + purgeExpired(now); + + Entry entry = sessions.get(conversationId); + if (entry == null) { + return; + } + + entry.session.retainAvailableLocal(availableNames); + if (entry.session.discoveredNamesLocal().isEmpty()) { + sessions.remove(conversationId); + } else { + entry.lastAccessNanos = now; + } + } + + synchronized int size() { + purgeExpired(System.nanoTime()); + return sessions.size(); + } + + private void purgeExpired(long now) { + Iterator<Map.Entry<String, Entry>> iterator = sessions.entrySet().iterator(); + while (iterator.hasNext()) { + Entry entry = iterator.next().getValue(); + if (now - entry.lastAccessNanos >= expirationNanos) { + iterator.remove(); + } + } Review Comment: 🟡 **`purgeExpired()` doesn't exploit access-order to short-circuit — minor efficiency issue.** `sessions` is an access-order `LinkedHashMap`, so the front of the iteration order is the least-recently-used entry. If the front entry is not expired, no subsequent entry can be expired either. The current scan continues checking all entries regardless. ```suggestion while (iterator.hasNext()) { Map.Entry<String, Entry> mapEntry = iterator.next(); if (now - mapEntry.getValue().lastAccessNanos < expirationNanos) { break; // access-order: remaining entries are newer } iterator.remove(); } ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
