This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch CAMEL-26543-huggingface-chat-lock in repository https://gitbox.apache.org/repos/asf/camel.git
commit aa056cb7fe6957e33e34b27ccbe0587a7577cd96 Author: Claus Ibsen <[email protected]> AuthorDate: Thu Nov 5 21:03:42 2026 +0100 CAMEL-26543: camel-huggingface - chat turns of one memory ID run one at a time The chat task read the history before the model call and wrote the whole returned history back after it, outside any lock per memory ID. Two overlapping turns of one conversation each sent the model a history without the other turn, and the last write dropped the other turn. A clear sent while a turn ran was undone when that turn wrote its history back. ChatPredictor now holds a lock per memory ID for the whole turn. The locks are reference counted and removed when no turn of the memory ID is in flight. AbstractTaskPredictor gets an infer hook for the model call, so the tests can play the model. Co-Authored-By: Claude Opus 5.5 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/catalog/docs/huggingface-component.adoc | 4 +- components/camel-ai/camel-huggingface/pom.xml | 6 + .../src/main/docs/huggingface-component.adoc | 4 +- .../huggingface/tasks/AbstractTaskPredictor.java | 14 ++- .../component/huggingface/tasks/ChatPredictor.java | 50 +++++++- .../huggingface/tasks/ChatHistoryTest.java | 129 ++++++++++++++++++++- 6 files changed, 195 insertions(+), 12 deletions(-) diff --git a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/huggingface-component.adoc b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/huggingface-component.adoc index 355ecfe3f1c2..8a7c223a4f35 100644 --- a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/huggingface-component.adoc +++ b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/huggingface-component.adoc @@ -279,7 +279,9 @@ otherwise): the oldest messages are dropped first, the system prompt is kept, an message, so an odd value keeps one message less. For long-running conversations, keep `maxHistoryMessages` within what the context of the model can take: a history that outgrows it makes every message of that conversation fail until the history is cleared. Set the `CamelChatClearHistory` header to `true` to clear the history of the conversation before -the message is sent. +the message is sent. The messages of one conversation are processed one at a time, so each message sees the history +of the message before it, also when a concurrent consumer receives them; messages of different conversations are not +held up by each other. .Custom Task For a custom task (e.g., _translation_): diff --git a/components/camel-ai/camel-huggingface/pom.xml b/components/camel-ai/camel-huggingface/pom.xml index bb698d9348e9..26caca739175 100644 --- a/components/camel-ai/camel-huggingface/pom.xml +++ b/components/camel-ai/camel-huggingface/pom.xml @@ -63,6 +63,12 @@ <artifactId>camel-test-junit6</artifactId> <scope>test</scope> </dependency> + <dependency> + <groupId>org.awaitility</groupId> + <artifactId>awaitility</artifactId> + <version>${awaitility-version}</version> + <scope>test</scope> + </dependency> </dependencies> <profiles> <profile> diff --git a/components/camel-ai/camel-huggingface/src/main/docs/huggingface-component.adoc b/components/camel-ai/camel-huggingface/src/main/docs/huggingface-component.adoc index 355ecfe3f1c2..8a7c223a4f35 100644 --- a/components/camel-ai/camel-huggingface/src/main/docs/huggingface-component.adoc +++ b/components/camel-ai/camel-huggingface/src/main/docs/huggingface-component.adoc @@ -279,7 +279,9 @@ otherwise): the oldest messages are dropped first, the system prompt is kept, an message, so an odd value keeps one message less. For long-running conversations, keep `maxHistoryMessages` within what the context of the model can take: a history that outgrows it makes every message of that conversation fail until the history is cleared. Set the `CamelChatClearHistory` header to `true` to clear the history of the conversation before -the message is sent. +the message is sent. The messages of one conversation are processed one at a time, so each message sees the history +of the message before it, also when a concurrent consumer receives them; messages of different conversations are not +held up by each other. .Custom Task For a custom task (e.g., _translation_): diff --git a/components/camel-ai/camel-huggingface/src/main/java/org/apache/camel/component/huggingface/tasks/AbstractTaskPredictor.java b/components/camel-ai/camel-huggingface/src/main/java/org/apache/camel/component/huggingface/tasks/AbstractTaskPredictor.java index bf3ccb0d0375..1624bad6fbd3 100644 --- a/components/camel-ai/camel-huggingface/src/main/java/org/apache/camel/component/huggingface/tasks/AbstractTaskPredictor.java +++ b/components/camel-ai/camel-huggingface/src/main/java/org/apache/camel/component/huggingface/tasks/AbstractTaskPredictor.java @@ -158,21 +158,27 @@ public abstract class AbstractTaskPredictor implements TaskPredictor { @Override public void predict(Exchange exchange) throws Exception { Input input = prepareInput(exchange); + Output output = infer(input); + processOutput(exchange, output); + } + + /** + * Runs the model on the input. With pooling, the calls share one predictor and run one at a time. + */ + protected Output infer(Input input) throws Exception { if (config.isPooling()) { lock.lock(); try { if (predictor == null) { predictor = model.newPredictor(); } - Output output = predictor.predict(input); - processOutput(exchange, output); + return predictor.predict(input); } finally { lock.unlock(); } } else { try (Predictor<Input, Output> djlPredictor = model.newPredictor()) { - Output output = djlPredictor.predict(input); - processOutput(exchange, output); + return djlPredictor.predict(input); } } } diff --git a/components/camel-ai/camel-huggingface/src/main/java/org/apache/camel/component/huggingface/tasks/ChatPredictor.java b/components/camel-ai/camel-huggingface/src/main/java/org/apache/camel/component/huggingface/tasks/ChatPredictor.java index 5c410a3a0887..9fc1d48a146d 100644 --- a/components/camel-ai/camel-huggingface/src/main/java/org/apache/camel/component/huggingface/tasks/ChatPredictor.java +++ b/components/camel-ai/camel-huggingface/src/main/java/org/apache/camel/component/huggingface/tasks/ChatPredictor.java @@ -21,6 +21,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReentrantLock; import ai.djl.modality.Input; import ai.djl.modality.Output; @@ -44,6 +45,11 @@ import org.apache.camel.support.LRUCacheFactory; * </p> * * <p> + * The turns of one memory ID run one at a time, so each turn sees the history of the turn before it. Turns of different + * memory IDs are not held up by each other. + * </p> + * + * <p> * <b>Input Contract (Camel Message Body):</b> * </p> * <ul> @@ -105,6 +111,10 @@ public class ChatPredictor extends AbstractTaskPredictor { // In-memory chat history (key = memoryId, value = JSON string of list of dicts) private final Map<String, String> chatHistories; + // A turn reads the history, runs the model and writes the history back, so the turns of one memory ID hold its + // lock. A lock is kept only while turns of its memory ID are in flight, so the map needs no bound. + private final Map<String, TurnLock> turnLocks = new ConcurrentHashMap<>(); + public ChatPredictor(HuggingFaceEndpoint endpoint) { super(endpoint); int maxHistoryMessages = config.getMaxHistoryMessages(); @@ -130,10 +140,37 @@ public class ChatPredictor extends AbstractTaskPredictor { config.getMaxTokens(), doSample, temperature); } + @Override + public void predict(Exchange exchange) throws Exception { + String memoryId = memoryId(exchange); + TurnLock turnLock = turnLocks.compute(memoryId, (id, existing) -> { + TurnLock answer = existing != null ? existing : new TurnLock(); + answer.turns++; + return answer; + }); + turnLock.lock.lock(); + try { + super.predict(exchange); + } finally { + turnLock.lock.unlock(); + turnLocks.computeIfPresent(memoryId, (id, existing) -> --existing.turns == 0 ? null : existing); + } + } + + /** + * The number of memory IDs with turns in flight. + */ + int conversationsInFlight() { + return turnLocks.size(); + } + + private String memoryId(Exchange exchange) { + return exchange.getIn().getHeader(config.getMemoryIdHeader(), "default", String.class); + } + @Override protected Input prepareInput(Exchange exchange) throws Exception { - String memoryIdHeader = config.getMemoryIdHeader(); - String memoryId = exchange.getIn().getHeader(memoryIdHeader, "default", String.class); + String memoryId = memoryId(exchange); // Clear if requested Boolean clear = exchange.getIn().getHeader("CamelChatClearHistory", Boolean.class); @@ -209,8 +246,7 @@ public class ChatPredictor extends AbstractTaskPredictor { exchange.getMessage().setHeader(HuggingFaceConstants.OUTPUT, generatedText); // Update history with full returned history - String memoryIdHeader = config.getMemoryIdHeader(); - String memoryId = exchange.getIn().getHeader(memoryIdHeader, "default", String.class); + String memoryId = memoryId(exchange); if (node.isArray() && !node.isEmpty()) { JsonNode first = node.get(0); if (first.has("generated_text") && first.get("generated_text").isArray()) { @@ -245,4 +281,10 @@ public class ChatPredictor extends AbstractTaskPredictor { LOG.debug("Dropped {} old messages from the chat history", history.size() - trimmed.size()); return trimmed; } + + private static final class TurnLock { + private final ReentrantLock lock = new ReentrantLock(); + // the turns that hold or wait for the lock, only changed inside a compute of turnLocks + private int turns; + } } diff --git a/components/camel-ai/camel-huggingface/src/test/java/org/apache/camel/component/huggingface/tasks/ChatHistoryTest.java b/components/camel-ai/camel-huggingface/src/test/java/org/apache/camel/component/huggingface/tasks/ChatHistoryTest.java index b7a4d82169b8..13c78746bdb3 100644 --- a/components/camel-ai/camel-huggingface/src/test/java/org/apache/camel/component/huggingface/tasks/ChatHistoryTest.java +++ b/components/camel-ai/camel-huggingface/src/test/java/org/apache/camel/component/huggingface/tasks/ChatHistoryTest.java @@ -18,6 +18,11 @@ package org.apache.camel.component.huggingface.tasks; import java.util.ArrayList; import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.FutureTask; +import java.util.concurrent.TimeUnit; import ai.djl.modality.Input; import ai.djl.modality.Output; @@ -33,6 +38,7 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import static org.awaitility.Awaitility.await; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; @@ -45,6 +51,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; class ChatHistoryTest { private final ObjectMapper mapper = new ObjectMapper(); + private final Map<FutureTask<Void>, Thread> threads = new ConcurrentHashMap<>(); private DefaultCamelContext context; @BeforeEach @@ -163,6 +170,60 @@ class ChatHistoryTest { assertTrue(negative.getMessage().contains("maxHistoryMessages"), negative.getMessage()); } + @Test + void turnsOfOneConversationRunOneAtATime() throws Exception { + CountDownLatch release = new CountDownLatch(1); + HeldModelChatPredictor predictor = new HeldModelChatPredictor(release); + FutureTask<Void> one = start(predictor, "alice", "One", false); + await().atMost(10, TimeUnit.SECONDS).until(() -> predictor.sent.containsKey("One")); + FutureTask<Void> two = start(predictor, "alice", "Two", false); + // the second turn must wait for the first one, and not be sent to the model without the first turn + await().atMost(10, TimeUnit.SECONDS).until(() -> isWaiting(two)); + release.countDown(); + one.get(10, TimeUnit.SECONDS); + two.get(10, TimeUnit.SECONDS); + + assertEquals(List.of("user:One", "assistant:Reply to One", "user:Two"), predictor.sent.get("Two")); + assertEquals(List.of("user:One", "assistant:Reply to One", "user:Two", "assistant:Reply to Two", "user:Three"), + turn(predictor, "alice", "Three", "Reply three")); + assertEquals(0, predictor.conversationsInFlight()); + } + + @Test + void clearWaitsForTheRunningTurnOfTheConversation() throws Exception { + CountDownLatch release = new CountDownLatch(1); + HeldModelChatPredictor predictor = new HeldModelChatPredictor(release); + turn(predictor, "alice", "Zero", "Reply zero"); + FutureTask<Void> one = start(predictor, "alice", "One", false); + await().atMost(10, TimeUnit.SECONDS).until(() -> predictor.sent.containsKey("One")); + FutureTask<Void> two = start(predictor, "alice", "Two", true); + await().atMost(10, TimeUnit.SECONDS).until(() -> isWaiting(two)); + release.countDown(); + one.get(10, TimeUnit.SECONDS); + two.get(10, TimeUnit.SECONDS); + + // the history written by the running turn is cleared too + assertEquals(List.of("user:Two"), predictor.sent.get("Two")); + assertEquals(List.of("user:Two", "assistant:Reply to Two", "user:Three"), + turn(predictor, "alice", "Three", "Reply three")); + } + + @Test + void turnsOfDifferentConversationsRunAtTheSameTime() throws Exception { + CountDownLatch release = new CountDownLatch(1); + HeldModelChatPredictor predictor = new HeldModelChatPredictor(release); + FutureTask<Void> alice = start(predictor, "alice", "Hello from alice", false); + FutureTask<Void> bob = start(predictor, "bob", "Hello from bob", false); + // both turns are with the model at the same time + await().atMost(10, TimeUnit.SECONDS) + .until(() -> predictor.sent.containsKey("Hello from alice") && predictor.sent.containsKey("Hello from bob")); + release.countDown(); + alice.get(10, TimeUnit.SECONDS); + bob.get(10, TimeUnit.SECONDS); + + assertEquals(0, predictor.conversationsInFlight()); + } + private ChatPredictor predictor(String options) { return new ChatPredictor( context.getEndpoint("huggingface:chat?modelId=test&" + options, HuggingFaceEndpoint.class)); @@ -179,19 +240,83 @@ class ChatHistoryTest { Input input = predictor.prepareInput(exchange); JsonNode sent = mapper.readTree(input.getAsString("data")); + predictor.processOutput(exchange, answer(sent, reply)); + assertEquals(reply, exchange.getMessage().getBody(String.class)); + return messages(sent); + } + + /** + * Answers the conversation with the whole conversation and the reply, as the text-generation pipeline does. + */ + private Output answer(JsonNode sent, String reply) { ArrayNode conversation = sent.deepCopy(); conversation.addObject().put("role", "assistant").put("content", reply); ArrayNode result = mapper.createArrayNode(); result.addObject().set("generated_text", conversation); Output output = new Output(); output.add("data", result.toString()); - predictor.processOutput(exchange, output); - assertEquals(reply, exchange.getMessage().getBody(String.class)); + return output; + } + private static List<String> messages(JsonNode sent) { List<String> messages = new ArrayList<>(); for (JsonNode node : sent) { messages.add(node.get("role").asText() + ":" + node.get("content").asText()); } return messages; } + + /** + * Starts a turn of the conversation of the memory ID in its own thread, through the predict method that the + * producer calls. + */ + private FutureTask<Void> start(ChatPredictor predictor, String memoryId, String message, boolean clear) { + FutureTask<Void> task = new FutureTask<>(() -> { + Exchange exchange = new DefaultExchange(context); + exchange.getIn().setHeader("CamelChatMemoryId", memoryId); + if (clear) { + exchange.getIn().setHeader("CamelChatClearHistory", true); + } + exchange.getIn().setBody(message); + predictor.predict(exchange); + return null; + }); + Thread thread = new Thread(task, "turn-" + message); + thread.setDaemon(true); + threads.put(task, thread); + thread.start(); + return task; + } + + /** + * Whether the thread of the turn waits, for the turn before it or for the model to be released. + */ + private boolean isWaiting(FutureTask<Void> task) { + Thread.State state = threads.get(task).getState(); + return state == Thread.State.WAITING || state == Thread.State.TIMED_WAITING; + } + + /** + * Plays the model: keeps the conversation sent for each user message, holds every turn until the test releases the + * model, and then answers it. + */ + private final class HeldModelChatPredictor extends ChatPredictor { + + private final Map<String, List<String>> sent = new ConcurrentHashMap<>(); + private final CountDownLatch release; + + HeldModelChatPredictor(CountDownLatch release) { + super(new HuggingFaceEndpoint(null, null, new HuggingFaceConfiguration())); + this.release = release; + } + + @Override + protected Output infer(Input input) throws Exception { + JsonNode conversation = mapper.readTree(input.getAsString("data")); + String message = conversation.get(conversation.size() - 1).get("content").asText(); + sent.put(message, messages(conversation)); + assertTrue(release.await(10, TimeUnit.SECONDS), "the model was not released"); + return answer(conversation, "Reply to " + message); + } + } }
