From a942f9f15f328ef2e14318a650d6c9d3095a7642 Mon Sep 17 00:00:00 2001 From: Ralph Soika Date: Sun, 15 Jun 2025 16:23:00 +0200 Subject: [PATCH] async rag index --- .../main/java/ai/dynamixs/rag/RAGAdapter.java | 149 ++-- .../dynamixs/rag/cluster/ClusterService.java | 10 +- .../dynamixs/rag/events/RAGEventService.java | 49 +- pom.xml | 2 +- workflow/offer-de-1.0.0.bpmn | 760 +++++++++++++++--- workflow/offer-proposal-de.xml | 32 + workflow/product-de-1.0.0.bpmn | 6 +- 7 files changed, 827 insertions(+), 181 deletions(-) create mode 100644 workflow/offer-proposal-de.xml 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 8a60cc4..bad8f47 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 @@ -38,6 +38,7 @@ import org.eclipse.microprofile.config.inject.ConfigProperty; import org.imixs.ai.workflow.OpenAIAPIService; import org.imixs.workflow.ItemCollection; import org.imixs.workflow.SignalAdapter; +import org.imixs.workflow.engine.EventLogService; import org.imixs.workflow.engine.WorkflowService; import org.imixs.workflow.exceptions.AdapterException; import org.imixs.workflow.exceptions.PluginException; @@ -45,52 +46,60 @@ 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 ai.dynamixs.rag.events.RAGEventService; import jakarta.inject.Inject; /** * The RAGAdapter is used for Retrieval-Augmented Generation workflows. *

- * The adapter automatically index and embed text data. + * The adapter operates in two modes: *

- * The imixs-ai prompt definition defines the API endpoint and processing events - * to adapt the indexing and the prompt template. - *

- * The Adapter can be configured by a event entity. - *

- * INDEX configuration - to add embeddings: + * In the 'INDEX' mode the adapter evaluates an associated imixs-ai prompt + * definition and generates - based on the given prompt template - a text to be + * indexed in the RAG database. * * 

  * {@code
-        
-            llm.service.endpoint
-            true
-        
+
+    llm.service.endpoint
+    true
+
+ * }
+ *
+ *

+ * In the 'RETRIEVAL' mode, the adapter evaluates an associated imixs-ai prompt + * definition to get retrieval embeddings from the RAG database. + * + * 

+ * {@code
+
+    llm.service.endpoint
+    true
+
  * }
  *
* - * The Endpoint defines the Rest API endpoint of the llama-cpp http server or - * any compatible OpenAI / Open API rest service endpoint. + * In both modes the Endpoint defines the Rest API endpoint of the llama-cpp + * http server or any compatible OpenAI / Open API rest service endpoint. *

- * Example Prompt Template: - * + * The Prompt Template is based on the concepts of Imixs-AI: * * 

  * {@code
-    
-    
-    {"n_predict": 4096, "temperature": 0}
+
+
+    < index_options>{"n_predict": 4096, "temperature": 0}< /index_options>
     $workflowgroup: $workflowstatus
-
-    # Summary
-    $workflowsummary
-
-    offer.summary
-
+        # Summary
+        $workflowsummary
+        offer.summary
     ]]>
-     
+ 
  * }
  *
* + * The index Process is + * * @author Ralph Soika * @version 1.0 * @@ -101,7 +110,6 @@ public class RAGAdapter implements SignalAdapter { public static final String API_ERROR = "API_ERROR"; public static final String RAG_INDEX = "INDEX"; - public static final String RAG_DELETE = "DELETE"; public static final String RAG_RETRIEVAL = "RETRIEVAL"; private static Logger logger = Logger.getLogger(RAGAdapter.class.getName()); @@ -113,6 +121,9 @@ public class RAGAdapter implements SignalAdapter { @Inject private WorkflowService workflowService; + @Inject + EventLogService eventLogService; + @Inject private OpenAIAPIService llmService; @@ -160,18 +171,6 @@ public class RAGAdapter implements SignalAdapter { List llmRetrievalDefinitions = workflowService.evalWorkflowResultXML(event, "imixs-ai", RAG_RETRIEVAL, workitem, false); - List llmDeleteDefinitions = workflowService.evalWorkflowResultXML(event, "imixs-ai", - RAG_DELETE, workitem, false); - - // verify if we have an DELETE configuration - if (llmDeleteDefinitions != null && llmDeleteDefinitions.size() > 0) { - try { - clusterService.removeAllEmbeddings(workitem.getUniqueID()); - } catch (ClusterException e) { - throw new PluginException(RAGAdapter.class.getSimpleName(), API_ERROR, e.getMessage(), e); - } - } - // verify if we have an INDEX configuration if (llmIndexDefinitions != null && llmIndexDefinitions.size() > 0) { createIndex(llmIndexDefinitions, workitem, event); @@ -186,7 +185,7 @@ public class RAGAdapter implements SignalAdapter { } /** - * Creates a RAG Index for a Worktiem based on IndexDefinitions + * Creates a RAG Index for a Workitem based on IndexDefinitions * * @param llmIndexDefinitions * @param workitem @@ -213,41 +212,57 @@ public class RAGAdapter implements SignalAdapter { logger.info("├── post RAG index request: " + llmAPIEndpoint); String promptTemplate = llmService.loadPromptTemplate(event); - logger.info("│ ├── PromptTemplate: "); - logger.info(promptTemplate); + indexDefinition.setItemValue("promptTemplate", promptTemplate); + indexDefinition.setItemValue("debug", llmAPIDebug); + indexDefinition.setItemValue("endpoint", llmAPIEndpoint); + + if (llmAPIDebug) { + logger.info("│ ├── PromptTemplate: "); + logger.info(promptTemplate); + } + + // Build the prompt template... String llmPrompt = llmService.buildPrompt(promptTemplate, workitem); if (llmPrompt.isBlank()) { throw new PluginException( RAGAdapter.class.getSimpleName(), API_ERROR, "Unable to parse prompt for RAG indexing"); } + indexDefinition.setItemValue("prompt", llmPrompt); - if (llmAPIDebug) { - logger.info("│ ├── Total Prompt Length = " + llmPrompt.length()); - logger.info("│ ├── Prompt: "); - logger.info(llmPrompt); - } + // send eventLog .... + eventLogService.createEvent( + RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_INDEX, + workitem.getUniqueID(), + indexDefinition); - // remove old embeddings - clusterService.removeAllEmbeddings(workitem.getUniqueID()); - // chunk text.... - List chunk_list = RAGUtil.chunkMarkupDocument(llmPrompt); - for (String chunk : chunk_list) { - if (llmAPIDebug) { - logger.info("│ ├── 🔸 chunk: "); - logger.info(chunk); - } - List indexResult = llmService.postEmbedding(chunk, llmAPIEndpoint, llmAPIDebug); - // write to cassandra - clusterService.insertVector(workitem.getUniqueID(), - workitem.getModelVersion(), - workitem.getTaskID(), - chunk, - indexResult); - if (llmAPIDebug) { - logger.info("│ ├── ⇨ " + indexResult.size() + " floats stored in RAG db."); - } - } + // if (llmAPIDebug) { + // logger.info("│ ├── Total Prompt Length = " + llmPrompt.length()); + // logger.info("│ ├── Prompt: "); + // logger.info(llmPrompt); + // } + + // // remove old embeddings + // clusterService.removeAllEmbeddings(workitem.getUniqueID()); + // // chunk text.... + // List chunk_list = RAGUtil.chunkMarkupDocument(llmPrompt); + // for (String chunk : chunk_list) { + // if (llmAPIDebug) { + // logger.info("│ ├── 🔸 chunk: "); + // logger.info(chunk); + // } + // List indexResult = llmService.postEmbedding(chunk, llmAPIEndpoint, + // llmAPIDebug); + // // write to cassandra + // clusterService.insertVector(workitem.getUniqueID(), + // workitem.getModelVersion(), + // workitem.getTaskID(), + // chunk, + // indexResult); + // if (llmAPIDebug) { + // logger.info("│ ├── ⇨ " + indexResult.size() + " floats stored in RAG db."); + // } + // } if (llmAPIDebug) { logger.info( @@ -257,8 +272,6 @@ public class RAGAdapter implements SignalAdapter { } catch (PluginException e) { logger.severe("├── ⚠️ Failed to post embeddings: " + e.getMessage()); throw e; - } catch (ClusterException e) { - throw new PluginException(RAGAdapter.class.getSimpleName(), API_ERROR, e.getMessage(), e); } } diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterService.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterService.java index 7e7784f..3bb2671 100644 --- a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterService.java +++ b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/cluster/ClusterService.java @@ -424,7 +424,7 @@ public class ClusterService { // remove all BoundStatement deleteBoundStmt = removeVectorStmt.bind(uniqueID); session.execute(deleteBoundStmt); - logger.info("│ ├── ✅ Removed " + count + " entries for uniqueID: " + uniqueID); + logger.info("│ ├── ✅ Removed " + count + " entries for uniqueID: " + uniqueID); return true; } else { // logger.info("│ ├── ℹ️ No entries found for uniqueID: " + uniqueID); @@ -602,7 +602,7 @@ public class ClusterService { if (existingMatch == null || similarity > existingMatch.getScore()) { // better match found! bestMatches.put(id, new RetrievalResult(id, contentChunk, similarity)); - logger.info("│ ├── Found/Updated document: " + id + " with score: " + similarity); + logger.info("│ ├── Found/Updated document: " + id + " with score: " + similarity); } } // Sortiere nach Score (absteigend) und gebe Top 5 zurück @@ -620,7 +620,7 @@ public class ClusterService { }) .collect(Collectors.toList()); - logger.info("│ ├── ✅ Retrieved " + result.size() + " relevant documents"); + logger.info("│ ├── ✅ Retrieved " + result.size() + " relevant documents"); return result; } catch (Exception e) { @@ -685,7 +685,7 @@ public class ClusterService { } if (chunkIds.isEmpty()) { - logger.info("│ ├── ℹ️ No entries found for uniqueID: " + uniqueId); + logger.info("│ ├── ℹ️ No entries found for uniqueID: " + uniqueId); return false; } @@ -697,7 +697,7 @@ public class ClusterService { updateCount++; } - logger.info("│ ├── Updated metadata for " + updateCount + " entries with uniqueID: " + + logger.info("│ ├── Updated metadata for " + updateCount + " entries with uniqueID: " + uniqueId + " (model: " + modelVersion + ", task: " + taskId + ")"); return true; 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 index c1f0883..d185fb7 100644 --- 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 @@ -33,12 +33,15 @@ import java.util.logging.Level; import java.util.logging.Logger; import org.apache.commons.compress.archivers.ArchiveException; +import org.imixs.ai.workflow.OpenAIAPIService; 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 org.imixs.workflow.exceptions.PluginException; +import ai.dynamixs.rag.RAGUtil; import ai.dynamixs.rag.cluster.ClusterException; import ai.dynamixs.rag.cluster.ClusterService; import jakarta.annotation.security.DeclareRoles; @@ -98,6 +101,9 @@ public class RAGEventService { @Inject private ClusterService clusterService; + @Inject + private OpenAIAPIService llmService; + /** * The method lookups for RAG event log entries and updates the RAG index * information @@ -109,7 +115,6 @@ public class RAGEventService { @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, @@ -142,8 +147,42 @@ public class RAGEventService { 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!"); + case RAGEventService.EVENTLOG_TOPIC_RAG_EVENT_INDEX + ".lock": + + // get indexDefinition + ItemCollection indexDefinition = new ItemCollection(eventLogEntry.getData()); + String llmAPIEndpoint = indexDefinition.getItemValueString("endpoint"); + boolean debug = indexDefinition.getItemValueBoolean("debug"); + // String llmPromptTemplate = + // indexDefinition.getItemValueString("promptTemplate"); + String llmPrompt = indexDefinition.getItemValueString("prompt"); + + if (debug) { + logger.info("│ ├── Total Prompt Length = " + llmPrompt.length()); + logger.info("│ ├── Prompt: "); + logger.info(llmPrompt); + } + + // remove old embeddings + clusterService.removeAllEmbeddings(workitem.getUniqueID()); + // chunk text.... + List chunk_list = RAGUtil.chunkMarkupDocument(llmPrompt); + for (String chunk : chunk_list) { + if (debug) { + logger.info("│ ├── 🔸 chunk: "); + logger.info(chunk); + } + List indexResult = llmService.postEmbedding(chunk, llmAPIEndpoint, debug); + // write to cassandra + clusterService.insertVector(workitem.getUniqueID(), + workitem.getModelVersion(), + workitem.getTaskID(), + chunk, + indexResult); + if (debug) { + logger.info("│ ├── ⇨ " + indexResult.size() + " floats stored in RAG db."); + } + } break; default: // no op @@ -160,6 +199,10 @@ public class RAGEventService { logger.log(Level.WARNING, e.getErrorCode() + " - " + e.getMessage()); // remove the event log entry... eventLogService.removeEvent(eventLogEntry.getId()); + } catch (PluginException e) { + logger.log(Level.WARNING, e.getErrorCode() + " - " + e.getMessage()); + // remove the event log entry... + eventLogService.removeEvent(eventLogEntry.getId()); } } diff --git a/pom.xml b/pom.xml index 50e3015..ba60719 100644 --- a/pom.xml +++ b/pom.xml @@ -61,7 +61,7 @@ 10.0.0 6.0 - 6.2.2-SNAPSHOT + 6.2.3-SNAPSHOT 3.1.2-SNAPSHOT 5.1.3-SNAPSHOT 5.0.3 diff --git a/workflow/offer-de-1.0.0.bpmn b/workflow/offer-de-1.0.0.bpmn index 93a3e59..73a18b5 100644 --- a/workflow/offer-de-1.0.0.bpmn +++ b/workflow/offer-de-1.0.0.bpmn @@ -69,6 +69,15 @@ event_qTIbqA dataObject_axJA7Q event_y0smQw + task_ol2R9A + event_P1MFZQ + event_NLZbsw + task_0THxmg + event_KwpMKw + dataObject_iq4agg + dataObject_u4ZytA + event_zsQpoQ + event_EyDvQQ @@ -127,6 +136,7 @@ Color: _color
sequenceFlow_8NDJxA sequenceFlow_UIpo9Q sequenceFlow_WRPdIA + sequenceFlow_ytbNEQ @@ -193,7 +203,9 @@ Color: _color
$workflowgroup - $workflowstatus]]> SequenceFlow_42 sequenceFlow_HCxYvw - sequenceFlow_4DUFiA + sequenceFlow_RppP5Q + sequenceFlow_VlP5cw + sequenceFlow_ZtZaBQ @@ -295,7 +307,7 @@ Color: _color
- + @@ -365,6 +377,7 @@ Betrag: _amount (Brutto € _amount_brutto SequenceFlow_56 sequenceFlow_HCxYvw + @@ -1061,7 +1074,7 @@ result.isValid=true; - + @@ -1113,7 +1126,7 @@ result.isValid=true; - + home Vertrieb]]> @@ -1132,8 +1145,9 @@ result.isValid=true; sequenceFlow_4DUFiA sequenceFlow_UIpo9Q + sequenceFlow_VlP5cw - + @@ -1148,11 +1162,447 @@ result.isValid=true; + + + + + + + Product: _name
+EAN: _ean
+Category: _category
+Brand: _brand
+
+Description:
+_description
+
+Price: _price _currency
+Stock: _stock
+Size: _size
+Color: _color
+]]> + + + + + + + + + request.subject]]> + + + true + + + + + + + + + + + + + + + + + + $workflowgroup - $workflowstatus]]> + sequenceFlow_WRPdIA + sequenceFlow_4DUFiA + sequenceFlow_0Q20mA + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + https://llama.cpp2.imixs.com/ + true + +]]> + + + + + + + + + + + + false + + + + + sequenceFlow_0Q20mA + sequenceFlow_InfPdQ + sequenceFlow_hZg0iw + + + + + sequenceFlow_InfPdQ + + + + + + + + + + + + + + Product: _name
+EAN: _ean
+Category: _category
+Brand: _brand
+
+Description:
+_description
+
+Price: _price _currency
+Stock: _stock
+Size: _size
+Color: _color
+]]> + + + + + + + + + request.subject]]> + + + true + + + + + + + + + + + + + + + + + + $workflowgroup - $workflowstatus]]> + sequenceFlow_4DUFiA + sequenceFlow_hZg0iw + sequenceFlow_HfDRCQ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + llm.service.endpoint + offer.proposal + true + +false +]]> + + + + + + + + + + + + false + + + + + sequenceFlow_InfPdQ + sequenceFlow_HfDRCQ + sequenceFlow_RppP5Q + sequenceFlow_JZ6sYQ + + + + + + + + + + + + + + {"n_predict": 4096, "temperature": 0} + request.description + +[INST] + +Zu dem hier vorliegenden Angebot kommen folgende Produkte in Frage: + + + + $workflowabstract + + + +Prüfe welches dieser Produkte am besten für die Anfrage geeignet ist und erstelle ein Angebot sowie eine kurze Begründung für deine Auswahl die für den Kunden verständlich ist. + +Füge eine Tabelle mit den folgenden Spalten für die Angebotserstellung mit ein: + +- Produkt +- Menge +- Preis pro Stück +- Mehrwertsteuer +- Gesamtpreis + + +[/INST] +]]]]> + + ]]> + + + + + + + + + + + + + + + + + + + + + + +]]> + + + + + + + + + + + + sequenceFlow_JZ6sYQ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + ]]> + + + + + + + + + false + + + + sequenceFlow_ZtZaBQ + sequenceFlow_ytbNEQ + + + + + + + - + @@ -1160,15 +1610,15 @@ result.isValid=true; - + - + - + - + @@ -1178,188 +1628,187 @@ result.isValid=true; - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - + - - - + + + - - - + + + - - + + - - + + - - + + - - + + - - + + - - - + + + - - - + + + - - + + - - - + + + - - + + - - + + - - + + - - - + + + - - + + - - + + - - - + + @@ -1368,46 +1817,155 @@ result.isValid=true; - + - + - + - + - - + + + - + - + - - - + + + + - - + + - - + + - - + + - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/workflow/offer-proposal-de.xml b/workflow/offer-proposal-de.xml new file mode 100644 index 0000000..8c7bec0 --- /dev/null +++ b/workflow/offer-proposal-de.xml @@ -0,0 +1,32 @@ + + + {"n_predict": 4096, "temperature": 0} + request.description + +[INST] + +Zu dem hier vorliegenden Angebot kommen folgende Produkte in Frage: + + + + $workflowabstract + + + +Prüfe welches dieser Produkte am besten für die Anfrage geeignet ist und erstelle ein Angebot sowie eine kurze Begründung für deine Auswahl die für den Kunden verständlich ist. + +Füge eine Tabelle mit den folgenden Spalten für die Angebotserstellung mit ein: + +- Produkt +- Menge +- Preis pro Stück +- Mehrwertsteuer +- Gesamtpreis + + +[/INST] +]]> + + \ No newline at end of file diff --git a/workflow/product-de-1.0.0.bpmn b/workflow/product-de-1.0.0.bpmn index 2b3dd04..5528a27 100644 --- a/workflow/product-de-1.0.0.bpmn +++ b/workflow/product-de-1.0.0.bpmn @@ -1333,10 +1333,10 @@ Betrag: _amount (Brutto € _amount_brutto home - + https://llama.cpp2.imixs.com/ true - + ]]> @@ -1355,7 +1355,7 @@ Betrag: _amount (Brutto € _amount_brutto sequenceFlow_Jkgs0A sequenceFlow_RgOnDQ - +