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 74380be01ab1 CAMEL-26543: camel-huggingface - chat turns of one memory
ID run one at a time (#29279)
74380be01ab1 is described below
commit 74380be01ab17bac8f360702a0e38ae0ea916e81
Author: Claus Ibsen <[email protected]>
AuthorDate: Fri Nov 6 08:28:12 2026 +0100
CAMEL-26543: camel-huggingface - chat turns of one memory ID run one at a
time (#29279)
The chat turns of one memory ID now run one at a time, so each turn sees
the history of the one before it, while different memory IDs stay concurrent.
Co-Authored-By: Claude Opus 5.5 <[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);
+ }
+ }
}
