diff --git a/doc/RAG_SERVICE.md b/doc/RAG_SERVICE.md
index fbd466e..66dde4c 100644
--- a/doc/RAG_SERVICE.md
+++ b/doc/RAG_SERVICE.md
@@ -1,4 +1,4 @@
-# The RAG ClusterService EJB
+# The RAG Cluster Service
With the EJB `ClusterService` clients can index and retrieve embeddings.
diff --git a/doc/RAG_WORKFLOW.md b/doc/RAG_WORKFLOW.md
new file mode 100644
index 0000000..0e91dc2
--- /dev/null
+++ b/doc/RAG_WORKFLOW.md
@@ -0,0 +1,63 @@
+# The RAG Workflow Integration
+
+The RAG system is integrated via the BPMN Model. There are several ways how the RAG information can be indexed, updated and deleted.
+
+## Indexing
+
+To index a process instance the Signal Adapter `ai.dynamixs.rag.RAGIndexAdapter` is used. The adapter expects a DataObject `PromptDefinition` associated with the corresponding event and a workflow configuration defining the LLM endpoint
+
+```xml
+
+ https://llama.cpp2.imixs.com/
+ true
+
+```
+
+The `debug` flag is optional and can be set to `true` to log index information.
+
+The PromptDefinition associated by a DataObject defines the text to generate the embeddings.
+
+```xml
+
+
+ {}
+ $workflowgroup: $workflowstatus
+$workflowsummary
+
+# Summary
+$workflowabstract
+
+]]>
+
+
+```
+
+The RAG System does not only store the embeddings but also the following Workflow Meta Information
+
+- ModelVersion - the `$modelversion` item assigned to the process instance at the time of the indexing
+- TaskId - the `$taskId` item assigned to the process instance at the time of the indexing
+
+**Note:** The workflow meta information is generated at the time of the indexing and does not necessary match the current workflow status!
+
+## Update
+
+To update the workflow metadata only, plugin class `ai.dynamixs.rag.RAGIndexPlugin` can be used.
+This plugin is used to update the workflow model version and the workflow status only but not the embedings. The update is only be performed if an index already exists.
+
+The update is performed automatically in each workflow processing cycle. To disabled a metadata update in single specific event you can define the following workflow definition:
+
+```xml
+
+```
+
+In this case the plugin will be disabled.
+
+## Delete
+
+To delete a RAG index the plugin class `ai.dynamixs.rag.RAGPlugin` can be used in the DELETE mode:
+
+```xml
+
+```
+
+This will remove the index for the current workitem.
diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGAdapter.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGAdapter.java
index 62bd288..8a60cc4 100644
--- a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGAdapter.java
+++ b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGAdapter.java
@@ -42,6 +42,9 @@ import org.imixs.workflow.engine.WorkflowService;
import org.imixs.workflow.exceptions.AdapterException;
import org.imixs.workflow.exceptions.PluginException;
+import ai.dynamixs.rag.cluster.ClusterException;
+import ai.dynamixs.rag.cluster.ClusterService;
+import ai.dynamixs.rag.cluster.RetrievalResult;
import jakarta.inject.Inject;
/**
diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGPlugin.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGPlugin.java
new file mode 100644
index 0000000..f75f1e2
--- /dev/null
+++ b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGPlugin.java
@@ -0,0 +1,107 @@
+/*
+ * Imixs-Workflow
+ *
+ * Copyright (C) 2001-2020 Imixs Software Solutions GmbH,
+ * http://www.imixs.com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You can receive a copy of the GNU General Public
+ * License at http://www.gnu.org/licenses/gpl.html
+ *
+ * Project:
+ * https://www.imixs.org
+ * https://github.com/imixs/imixs-workflow
+ *
+ * Contributors:
+ * Imixs Software Solutions GmbH - Project Management
+ * Ralph Soika - Software Developer
+ */
+
+package ai.dynamixs.rag;
+
+import java.util.List;
+import java.util.logging.Logger;
+
+import org.imixs.workflow.ItemCollection;
+import org.imixs.workflow.engine.EventLogService;
+import org.imixs.workflow.engine.WorkflowService;
+import org.imixs.workflow.engine.plugins.AbstractPlugin;
+import org.imixs.workflow.exceptions.PluginException;
+
+import ai.dynamixs.rag.events.RAGEventService;
+import jakarta.inject.Inject;
+
+/**
+ * The RAGIndexPlugin updates the workflow meta data for existing embeddings.
+ * The plugin can be deactivated with:
+ *
+ *
+ *
+ {@code
+
+ false
+
+ }
+ *
+ *
+ * The tag 'debug' is optional
+ *
+ *
+ * @author rsoika
+ *
+ */
+public class RAGPlugin extends AbstractPlugin {
+
+ private static final Logger logger = Logger.getLogger(RAGPlugin.class.getName());
+
+ @Inject
+ private WorkflowService workflowService;
+
+ @Inject
+ EventLogService eventLogService;
+
+ @Override
+ public ItemCollection run(ItemCollection workitem, ItemCollection event) throws PluginException {
+ List deleteDefinitions = null;
+ List disableDefinitions = null;
+
+ logger.finest("running TaxonomyPlugin");
+ String workflowResult = event.getItemValueString("workflow.result");
+ if (workflowResult.contains(" 0) {
+ // no op!
+ return workitem;
+ }
+
+ // delete mode?
+ if (deleteDefinitions != null && deleteDefinitions.size() > 0) {
+ eventLogService.createEvent(RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_DELETE, workitem.getUniqueID());
+ return workitem;
+ }
+
+ // default update meta information
+ eventLogService.createEvent(RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_UPDATE, workitem.getUniqueID());
+
+ return workitem;
+
+ }
+
+}
diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterException.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterException.java
similarity index 98%
rename from dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterException.java
rename to dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterException.java
index c00e525..0bda617 100644
--- a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterException.java
+++ b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterException.java
@@ -25,7 +25,7 @@
* Ralph Soika - Software Developer
*******************************************************************************/
-package ai.dynamixs.rag;
+package ai.dynamixs.rag.cluster;
/**
* The ClusterService provides methods to persist the content of a Imixs AI
diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterService.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterService.java
similarity index 88%
rename from dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterService.java
rename to dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterService.java
index 01e7c50..7e7784f 100644
--- a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterService.java
+++ b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterService.java
@@ -1,4 +1,4 @@
-package ai.dynamixs.rag;
+package ai.dynamixs.rag.cluster;
import java.net.InetSocketAddress;
import java.util.ArrayList;
@@ -137,6 +137,8 @@ public class ClusterService {
private final Map searchVectorStatements = new HashMap<>();
private PreparedStatement selectVectorStmt;
private PreparedStatement removeVectorStmt;
+ private PreparedStatement selectChunksStmt = null;
+ private PreparedStatement updateChunkStmt = null;
@Resource
private TimerService timerService;
@@ -414,29 +416,16 @@ public class ClusterService {
String deleteQuery = "DELETE FROM document_vectors WHERE id = ?";
removeVectorStmt = session.prepare(deleteQuery);
}
- if (selectVectorStmt == null) {
- String selectQuery = "SELECT COUNT(*) FROM document_vectors WHERE id = ?";
- selectVectorStmt = session.prepare(selectQuery);
- }
try {
// test if entries exist
- BoundStatement selectBoundStmt = selectVectorStmt.bind(uniqueID);
- ResultSet resultSet = session.execute(selectBoundStmt);
- Row row = resultSet.one();
-
- if (row != null) {
- long count = row.getLong(0);
-
- if (count > 0) {
- // remove all
- BoundStatement deleteBoundStmt = removeVectorStmt.bind(uniqueID);
- session.execute(deleteBoundStmt);
- logger.info("│ ├── ✅ Removed " + count + " entries for uniqueID: " + uniqueID);
- return true;
- } else {
- // logger.info("│ ├── ℹ️ No entries found for uniqueID: " + uniqueID);
- }
+ long count = countIndexEntries(uniqueID);
+ if (count > 0) {
+ // remove all
+ BoundStatement deleteBoundStmt = removeVectorStmt.bind(uniqueID);
+ session.execute(deleteBoundStmt);
+ logger.info("│ ├── ✅ Removed " + count + " entries for uniqueID: " + uniqueID);
+ return true;
} else {
// logger.info("│ ├── ℹ️ No entries found for uniqueID: " + uniqueID);
}
@@ -448,6 +437,36 @@ public class ClusterService {
return false;
}
+ /**
+ * This method returns the number of index entries with the given uniqueId. The
+ * method can be used to test if index data is available.
+ *
+ * @param uniqueId
+ * @return - count of entries, or 0 if no entries exist
+ * @throws ClusterException
+ */
+ public long countIndexEntries(String uniqueId) throws ClusterException {
+ if (selectVectorStmt == null) {
+ String selectQuery = "SELECT COUNT(*) FROM document_vectors WHERE id = ?";
+ selectVectorStmt = session.prepare(selectQuery);
+ }
+ try {
+ // test if entries exist
+ BoundStatement selectBoundStmt = selectVectorStmt.bind(uniqueId);
+ ResultSet resultSet = session.execute(selectBoundStmt);
+ Row row = resultSet.one();
+
+ if (row != null) {
+ long count = row.getLong(0);
+ return count;
+ }
+ } catch (Exception e) {
+ throw new ClusterException(ClusterException.CLUSTER_ERROR,
+ "Failed to lookup entries for uniqueID '" + uniqueId + "': " + e.getMessage(), e);
+ }
+ return 0;
+ }
+
/**
* Performs a semantic search on the document vector database using cosine
* similarity.
@@ -632,6 +651,63 @@ public class ClusterService {
return searchEmbeddings(embedding, maxResults, Arrays.asList(modelPattern), taskIds, null, null);
}
+ /**
+ * Update the workflow meta data. The method returns true if a index with the
+ * given uniqueId was found
+ *
+ * @param uniqueId
+ * @param modelVersion
+ * @param taskId
+ * @return true if an update was performed
+ * @throws ClusterException
+ */
+ public boolean updateMetaData(String uniqueId, String modelVersion, int taskId)
+ throws ClusterException {
+
+ // Prepare statements if not already done
+ if (selectChunksStmt == null) {
+ String selectQuery = "SELECT chunk_id FROM document_vectors WHERE id = ?";
+ selectChunksStmt = session.prepare(selectQuery);
+ }
+
+ if (updateChunkStmt == null) {
+ String updateQuery = "UPDATE document_vectors SET model_version = ?, task_id = ? WHERE id = ? AND chunk_id = ?";
+ updateChunkStmt = session.prepare(updateQuery);
+ }
+ try {
+ // 1. Alle chunk_ids für die uniqueId ermitteln
+ BoundStatement selectBoundStmt = selectChunksStmt.bind(uniqueId);
+ ResultSet resultSet = session.execute(selectBoundStmt);
+
+ List chunkIds = new ArrayList<>();
+ for (Row row : resultSet) {
+ chunkIds.add(row.getUuid("chunk_id"));
+ }
+
+ if (chunkIds.isEmpty()) {
+ logger.info("│ ├── ℹ️ No entries found for uniqueID: " + uniqueId);
+ return false;
+ }
+
+ // 2. Jede gefundene chunk_id einzeln updaten
+ int updateCount = 0;
+ for (UUID chunkId : chunkIds) {
+ BoundStatement updateBoundStmt = updateChunkStmt.bind(modelVersion, taskId, uniqueId, chunkId);
+ session.execute(updateBoundStmt);
+ updateCount++;
+ }
+
+ logger.info("│ ├── Updated metadata for " + updateCount + " entries with uniqueID: " +
+ uniqueId + " (model: " + modelVersion + ", task: " + taskId + ")");
+ return true;
+
+ } catch (Exception e) {
+ throw new ClusterException(ClusterException.CLUSTER_ERROR,
+ "Failed to update metadata for uniqueID '" + uniqueId + "': " + e.getMessage(), e);
+ }
+
+ }
+
/**
* Helper method to prepare a IN clause
*
diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RetrievalResult.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/RetrievalResult.java
similarity index 98%
rename from dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RetrievalResult.java
rename to dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/RetrievalResult.java
index ad2c689..4078c98 100644
--- a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RetrievalResult.java
+++ b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/RetrievalResult.java
@@ -1,4 +1,4 @@
-package ai.dynamixs.rag;
+package ai.dynamixs.rag.cluster;
/**
* Represents a single result from a semantic search operation in the RAG
diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/events/RAGEventScheduler.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/events/RAGEventScheduler.java
new file mode 100644
index 0000000..0e1b5e7
--- /dev/null
+++ b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/events/RAGEventScheduler.java
@@ -0,0 +1,145 @@
+/*
+ * Imixs-Workflow
+ *
+ * Copyright (C) 2001-2020 Imixs Software Solutions GmbH,
+ * http://www.imixs.com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You can receive a copy of the GNU General Public
+ * License at http://www.gnu.org/licenses/gpl.html
+ *
+ * Project:
+ * https://www.imixs.org
+ * https://github.com/imixs/imixs-workflow
+ *
+ * Contributors:
+ * Imixs Software Solutions GmbH - Project Management
+ * Ralph Soika - Software Developer
+ */
+
+package ai.dynamixs.rag.events;
+
+import static org.imixs.workflow.engine.AsyncEventSchedulerConfig.ASYNCEVENT_PROCESSOR_DEADLOCK;
+import static org.imixs.workflow.engine.AsyncEventSchedulerConfig.ASYNCEVENT_PROCESSOR_ENABLED;
+import static org.imixs.workflow.engine.AsyncEventSchedulerConfig.ASYNCEVENT_PROCESSOR_INITIALDELAY;
+import static org.imixs.workflow.engine.AsyncEventSchedulerConfig.ASYNCEVENT_PROCESSOR_INTERVAL;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.imixs.workflow.engine.AsyncEventService;
+import org.imixs.workflow.engine.EventLogService;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
+import jakarta.annotation.security.DeclareRoles;
+import jakarta.annotation.security.RunAs;
+import jakarta.ejb.Singleton;
+import jakarta.ejb.Startup;
+import jakarta.ejb.Timeout;
+import jakarta.ejb.Timer;
+import jakarta.ejb.TimerConfig;
+import jakarta.ejb.TimerService;
+import jakarta.inject.Inject;
+
+/**
+ * The RAGEventScheduler starts a scheduler service to process RAG events in an
+ * asynchronous way by calling the RAGEventService.
+ *
+ * The AsyncEventScheduler runs on a non-persistent ejb timer with the same
+ * interval settings as the AsyncEventProcessor
+ *
+ * RAGEventScheduler 'ASYNCEVENT_PROCESSOR_INTERVAL' and an optional delay
+ * defined by 'ASYNCEVENT_PROCESSOR_INITIALDELAY'. To enable the processor
+ * 'ASYNCEVENT_PROCESSOR_ENABLED' must be set to true (default=false).
+ * 'ASYNCEVENT_PROCESSOR_DEADLOCK' deadlock timeout
+ *
+ * In a clustered environment this timer runs in each cluster member that
+ * contains the EJB. So this means the non-persistent EJB Timer scales
+ * horizontal within a clustered environment – e.g. a Kubernetes cluster.
+ *
+ * @see AsyncEventService
+ * @version 1.1
+ * @author rsoika
+ *
+ */
+@DeclareRoles({ "org.imixs.ACCESSLEVEL.MANAGERACCESS" })
+@RunAs("org.imixs.ACCESSLEVEL.MANAGERACCESS")
+@Startup
+@Singleton
+public class RAGEventScheduler {
+
+ // enabled
+ @Inject
+ @ConfigProperty(name = ASYNCEVENT_PROCESSOR_ENABLED, defaultValue = "false")
+ boolean enabled;
+
+ // timeout interval in ms
+ @Inject
+ @ConfigProperty(name = ASYNCEVENT_PROCESSOR_INTERVAL, defaultValue = "1000")
+ long interval;
+
+ // initial delay in ms
+ @Inject
+ @ConfigProperty(name = ASYNCEVENT_PROCESSOR_INITIALDELAY, defaultValue = "0")
+ long initialDelay;
+
+ // deadlock timeout interval in ms
+ @Inject
+ @ConfigProperty(name = ASYNCEVENT_PROCESSOR_DEADLOCK, defaultValue = "60000")
+ long deadLockInterval;
+
+ private static final Logger logger = Logger.getLogger(RAGEventScheduler.class.getName());
+
+ @Resource
+ TimerService timerService;
+
+ @Inject
+ RAGEventService ragEventService;
+
+ @Inject
+ EventLogService eventLogService;
+
+ @PostConstruct
+ public void init() {
+ if (enabled) {
+
+ // Registering a non-persistent Timer Service.
+ final TimerConfig timerConfig = new TimerConfig();
+ timerConfig.setInfo("Imixs-Workflow AsyncEventScheduler");
+ timerConfig.setPersistent(false);
+ timerService.createIntervalTimer(initialDelay, interval, timerConfig);
+ logger.log(Level.INFO, "├── ✅ Started RAGEventScheduler - initalDelay={0} inverval={1}",
+ new Object[] { initialDelay, interval });
+
+ }
+ }
+
+ /**
+ * The method delegates the event processing to the stateless ejb
+ * AsyncEventProcessor.
+ *
+ * Before processing the eventLog the method releases possible dead locks first.
+ * Both methods are running in separate transactions
+ *
+ */
+ @Timeout
+ public void run(Timer timer) {
+ eventLogService.releaseDeadLocks(deadLockInterval,
+ RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_INDEX,
+ RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_UPDATE,
+ RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_DELETE);
+ ragEventService.processEventLog();
+ }
+
+}
diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/events/RAGEventService.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/events/RAGEventService.java
new file mode 100644
index 0000000..c1f0883
--- /dev/null
+++ b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/events/RAGEventService.java
@@ -0,0 +1,173 @@
+/*
+ * Imixs-Workflow
+ *
+ * Copyright (C) 2001-2020 Imixs Software Solutions GmbH,
+ * http://www.imixs.com
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public License
+ * as published by the Free Software Foundation; either version 2
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * General Public License for more details.
+ *
+ * You can receive a copy of the GNU General Public
+ * License at http://www.gnu.org/licenses/gpl.html
+ *
+ * Project:
+ * https://www.imixs.org
+ * https://github.com/imixs/imixs-workflow
+ *
+ * Contributors:
+ * Imixs Software Solutions GmbH - Project Management
+ * Ralph Soika - Software Developer
+ */
+
+package ai.dynamixs.rag.events;
+
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import org.apache.commons.compress.archivers.ArchiveException;
+import org.imixs.workflow.ItemCollection;
+import org.imixs.workflow.engine.AsyncEventScheduler;
+import org.imixs.workflow.engine.EventLogService;
+import org.imixs.workflow.engine.WorkflowService;
+import org.imixs.workflow.engine.jpa.EventLog;
+
+import ai.dynamixs.rag.cluster.ClusterException;
+import ai.dynamixs.rag.cluster.ClusterService;
+import jakarta.annotation.security.DeclareRoles;
+import jakarta.annotation.security.RolesAllowed;
+import jakarta.ejb.Stateless;
+import jakarta.ejb.TransactionAttribute;
+import jakarta.ejb.TransactionAttributeType;
+import jakarta.inject.Inject;
+import jakarta.persistence.OptimisticLockException;
+
+/**
+ * The AsyncEventService can be used to process workflow events in an
+ * asynchronous batch process. The AsyncEventService lookup eventLog entries of
+ * the topic "async.event".
+ *
+ * The processor look up the workItem and starts a processing life cycle.
+ *
+ * The AsyncEventService is called only by the AsyncEventScheduler which is
+ * implementing a ManagedScheduledExecutorService.
+ *
+ * To prevent concurrent processes to handle the same workitems the batch
+ * process uses a Optimistic lock strategy. After fetching new event log entries
+ * the processor updates the eventLog entry in a new transaction and set the
+ * topic to 'batch.process.lock'. After that update we can be sure that no other
+ * process is dealing with these entries. After completing the processing step
+ * the eventlog entry will be removed.
+ *
+ * To avoid ad deadlock the processor set an expiration time on the lock, so the
+ * lock will be auto-released after 1 minute (batch.processor.deadlock).
+ *
+ * @see AsyncEventScheduler
+ * @version 1.0
+ * @author rsoika
+ *
+ */
+@DeclareRoles({ "org.imixs.ACCESSLEVEL.NOACCESS", "org.imixs.ACCESSLEVEL.READERACCESS",
+ "org.imixs.ACCESSLEVEL.AUTHORACCESS", "org.imixs.ACCESSLEVEL.EDITORACCESS",
+ "org.imixs.ACCESSLEVEL.MANAGERACCESS" })
+@RolesAllowed({ "org.imixs.ACCESSLEVEL.NOACCESS", "org.imixs.ACCESSLEVEL.READERACCESS",
+ "org.imixs.ACCESSLEVEL.AUTHORACCESS", "org.imixs.ACCESSLEVEL.EDITORACCESS",
+ "org.imixs.ACCESSLEVEL.MANAGERACCESS" })
+@Stateless
+public class RAGEventService {
+
+ public static final String EVENTLOG_TOPIC_RAG_EVENT_INDEX = "rag.event.index";
+ public static final String EVENTLOG_TOPIC_RAG_EVENT_DELETE = "rag.event.delete";
+ public static final String EVENTLOG_TOPIC_RAG_EVENT_UPDATE = "rag.event.update";
+
+ private static final Logger logger = Logger.getLogger(RAGEventService.class.getName());
+
+ @Inject
+ EventLogService eventLogService;
+
+ @Inject
+ private WorkflowService workflowService;
+
+ @Inject
+ private ClusterService clusterService;
+
+ /**
+ * The method lookups for RAG event log entries and updates the RAG index
+ * information
+ *
+ * Each eventLogEntry provides optional a prompt template
+ *
+ * @throws ArchiveException
+ */
+ @TransactionAttribute(value = TransactionAttributeType.REQUIRES_NEW)
+ public void processEventLog() {
+ long l = System.currentTimeMillis();
+ boolean debug = logger.isLoggable(Level.FINE);
+
+ // test for new event log entries by timeout...
+ List events = eventLogService.findEventsByTimeout(100,
+ RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_INDEX,
+ RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_UPDATE,
+ RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_DELETE);
+
+ if (events.size() > 0) {
+ logger.log(Level.INFO, "├── 🔃 processing {0} RAG events....", events.size());
+
+ for (EventLog eventLogEntry : events) {
+ try {
+
+ // first try to lock the eventLog entry....
+ if (eventLogService.lock(eventLogEntry)) {
+ ItemCollection workitem = workflowService.getWorkItem(eventLogEntry.getRef());
+ if (workitem == null) {
+ logger.log(Level.WARNING, "│ ├── ⚠️ unable to read workitem: " + eventLogEntry.getRef());
+ eventLogService.removeEvent(eventLogEntry.getId());
+ continue;
+ }
+ switch (eventLogEntry.getTopic()) {
+ case RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_DELETE + ".lock":
+ // remove embeddings
+ clusterService.removeAllEmbeddings(eventLogEntry.getRef());
+ break;
+
+ case RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_UPDATE + ".lock":
+ // update workflow status
+ clusterService.updateMetaData(eventLogEntry.getRef(), workitem.getModelVersion(),
+ workitem.getTaskID());
+ break;
+ case RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_INDEX:
+ logger.log(Level.WARNING, "│ ├── ⚠️ RAG_INDEX Event is not yet supported!");
+ break;
+ default:
+ // no op
+ }
+
+ // finally remove the event log entry...
+ eventLogService.removeEvent(eventLogEntry.getId());
+ }
+
+ } catch (OptimisticLockException e) {
+ // lock was not possible - continue....
+ logger.log(Level.INFO, "...unable to lock AsyncEvent: {0}", e.getMessage());
+ } catch (ClusterException e) {
+ logger.log(Level.WARNING, e.getErrorCode() + " - " + e.getMessage());
+ // remove the event log entry...
+ eventLogService.removeEvent(eventLogEntry.getId());
+ }
+
+ }
+
+ logger.log(Level.INFO, "├── ✅ {0} RAGEvents processed in {1}ms",
+ new Object[] { events.size(), System.currentTimeMillis() - l });
+
+ }
+ }
+
+}
diff --git a/workflow/product-de-1.0.0.bpmn b/workflow/product-de-1.0.0.bpmn
index bd5ee03..2b3dd04 100644
--- a/workflow/product-de-1.0.0.bpmn
+++ b/workflow/product-de-1.0.0.bpmn
@@ -34,6 +34,7 @@
+
true