setup cluster session

.vscode/launch.json

@@ -3,11 +3,11 @@
   "configurations": [
     {
       "type": "java",
-      "name": "Debug hai",
+      "name": "Debug Dynamixs-AI",
       "request": "attach",
       "hostName": "localhost",
       "port": "8787",
-      "projectName": "hai-office-demo"
+      "projectName": "dynamixs-ai"
     }
   ]
 }

docker/docker-compose-dev.yaml

@@ -26,7 +26,7 @@ services:
       LANG: "en_US.UTF-8"
       MAILGATEWAY: "localhost"
       ASYNCEVENT_PROCESSOR_ENABLED: "true"
-      ASYNCEVENT_PROCESSOR_INITIALDELAY: "10000"
+      ASYNCEVENT_PROCESSOR_INITIALDELAY: "30000"
       OCR_SERVICE_ENDPOINT: "http://tika:9998/tika"
       OCR_SERVICE_MODE: "MODEL"
       #OCR_STRATEGY: "NO_OCR, OCR_ONLY, OCR_AND_TEXT_EXTRACTION, AUTO (default)"

dynamixs-ai-app/src/main/resources/META-INF/beans.xml

@@ -1,6 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<beans xmlns="https://jakarta.ee/xml/ns/jakartaee"
-       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-       xsi:schemaLocation="https://jakarta.ee/xml/ns/jakartaee https://jakarta.ee/xml/ns/jakartaee/beans_3_0.xsd"
-       bean-discovery-mode="all" version="3.0">
-</beans>

dynamixs-ai-rag/src/main/java/ai/dynamixs/DynamixsSetupService.java

@@ -33,25 +33,25 @@ import java.util.logging.Logger;
 import org.imixs.workflow.engine.SetupEvent;
 import org.imixs.workflow.office.config.AccessRoleController;
 import org.imixs.workflow.office.config.PropertiesConfigSource;
-import org.imixs.workflow.services.rest.RestAPIException;
 
+import ai.dynamixs.rag.ClusterException;
 import ai.dynamixs.rag.ClusterService;
-import jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
 import jakarta.annotation.security.DeclareRoles;
 import jakarta.annotation.security.RolesAllowed;
 import jakarta.ejb.Stateless;
+import jakarta.ejb.Timeout;
+import jakarta.ejb.Timer;
+import jakarta.ejb.TimerConfig;
+import jakarta.ejb.TimerService;
 import jakarta.enterprise.event.Observes;
 import jakarta.inject.Inject;
 
 /**
- * This SetupController extends the ConfigController and holds the data from the
+ * This DynamixsSetupService reacts on the
- * configuration entity 'BASIC'. This is the general configuration entity used
+ * 'org.imixs.workflow.engine.SetupEvent' and initializes a Cassandra Session.
- * by Imixs-Office-Workflow.
+ * If the connection can not be established immediately the method retries a
- * <p>
+ * connection every 10 seconds.
- * The item 'properties' managed by the 'BASIC' configuration entity is also
- * used by the MartyConfigSource to provide optional config parameters.
- * <p>
- * Since version 4.5.1 the SetupController provides a list of fixed UserGroups.
  * 
  * @see AccessRoleController
  * 
@@ -69,32 +69,42 @@ import jakarta.inject.Inject;
 public class DynamixsSetupService {
     private static Logger logger = Logger.getLogger(DynamixsSetupService.class.getName());
 
-    private static final long serialVersionUID = 1L;
-
     @Inject
     ClusterService clusterService;
 
-    /**
+    @Resource
-     * Create a rest client instance
+    private TimerService timerService;
-     */
-    @PostConstruct
-    public void init() {
-
-    }
 
     /**
-     * This method verifies the schema of the cassandra database
+     * Try to connect to Cassandra
-     * <p>
-     * 
      * 
      * @param setupEvent
-     * @throws RestAPIException
      */
-    public void setup(@Observes SetupEvent setupEvent) throws RestAPIException {
+    public void setup(@Observes SetupEvent setupEvent) {
-
+        scheduleClusterCheck();
-        logger.log(Level.INFO, "├── verify cassandra database...");
+    }
 
+    /**
+     * Versucht, die Session herzustellen. Bei Fehler wird ein neuer Timer geplant.
+     */
+    private void scheduleClusterCheck() {
+        try {
+            logger.log(Level.INFO, "├── Connecting Cassandra cluster...");
             clusterService.getSession();
+            logger.log(Level.INFO, "├── ✅ Successfully connected to Cassandra!");
+        } catch (ClusterException e) {
+            logger.log(Level.WARNING, "├── " + e.getMessage());
+            logger.log(Level.WARNING, "├── ⚠️ Cluster not ready yet. Retrying in 10 seconds...");
+            // retry in 10 sec
+            timerService.createSingleActionTimer(10000, new TimerConfig());
+        }
     }
 
+    /**
+     * called by timer. Tries to connect cassandra clauster
+     */
+    @Timeout
+    public void retryClusterCheck(Timer timer) {
+        scheduleClusterCheck(); // retry Check
+    }
 }

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterException.java

@@ -46,6 +46,7 @@ public class ClusterException extends Exception {
 
     public static final String INVALID_DOCUMENT_OBJECT = "INVALID_DOCUMENT_OBJECT";
     public static final String INVALID_KEYSPACE = "INVALID_KEYSPACE";
+    public static final String CLUSTER_ERROR = "CLUSTER_ERROR";
     public static final String INVALID_WORKITEM = "INVALID_WORKITEM";
     public static final String MD5_ERROR = "MD5_ERROR";
 

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterService.java

@@ -15,7 +15,6 @@ import com.datastax.oss.driver.api.core.cql.BoundStatement;
 import com.datastax.oss.driver.api.core.cql.PreparedStatement;
 import com.datastax.oss.driver.api.core.servererrors.InvalidQueryException;
 
-import jakarta.annotation.PostConstruct;
 import jakarta.annotation.PreDestroy;
 import jakarta.ejb.Singleton;
 import jakarta.inject.Inject;
@@ -108,19 +107,25 @@ public class ClusterService {
     @ConfigProperty(name = ENV_EMBEDDINGS_CLUSTER_SSL_KEYSTOREPASSWORD)
     Optional<String> keystorePwd;
 
-    // private Cluster cluster;
     private CqlSession session;
 
-    @PostConstruct
+    /**
-    private void init() {
+     * This method tries to initalize the cassandara session object.
-        try {
+     * 
-            // cluster = initCluster();
+     * Note: the initialization is only successful if the cassandra cluster is fully
+     * started.
+     * 
+     * @throws ClusterException
+     */
+    private void init() throws ClusterException {
+        logger.info("├── initializing cluster...");
         session = initSession();
+        if (session != null) {
+            logger.info("│   ├── keySpace connection status = OK");
             initKeySpace();
-        } catch (ClusterException e) {
+            logger.info("├── ✅ initializing cluster successful.");
-            logger.severe("Failed to init cassandra session!");
-            e.printStackTrace();
         }
+
     }
 
     @PreDestroy
@@ -129,12 +134,23 @@ public class ClusterService {
         if (session != null) {
             session.close();
         }
-
     }
 
-    public CqlSession getSession() {
+    /**
+     * Returns the Cassandra Session object. If the session is not yet created, the
+     * method tries to init the session the first time. If initialization failed the
+     * method throws a ClusterException
+     * 
+     * @return
+     */
+    public CqlSession getSession() throws ClusterException {
         if (session == null) {
+            try {
                 init();
+            } catch (Exception e) {
+                throw new ClusterException(ClusterException.CLUSTER_ERROR,
+                        "Failed to init cassandra session: " + e.getMessage(), e);
+            }
         }
         return session;
     }
@@ -156,20 +172,16 @@ public class ClusterService {
             throw new ClusterException(ClusterException.INVALID_KEYSPACE,
                     "keyspace '" + keySpacename + "' name invalid.");
         }
-        logger.info("├── initializing cluster...");
+
         logger.info("│   ├── connecting keyspace='" + keySpacename + "'...");
         // try to open keySpace
         try {
             // session = cluster.connect(keySpace.get());
         } catch (InvalidQueryException e) {
             logger.warning("......connecting keySpace '" + keySpacename + "' failed: " + e.getMessage());
-
             createKeySpace(keySpacename);
         }
-        if (session != null) {
+
-            logger.info("│   ├── keySpace connection status = OK");
-        }
-        // return session;
     }
 
     /**
@@ -190,7 +202,7 @@ public class ClusterService {
                     "missing cluster contact points - verify configuration!");
         }
 
-        logger.info("│   ├── cluster conecting: " + contactPoint.get());
+        logger.info("│   ├── cluster connecting: " + contactPoint.get());
 
         CqlSessionBuilder builder = CqlSession.builder();
         String[] hosts = contactPoint.get().split(",");
@@ -333,7 +345,7 @@ public class ClusterService {
         session.execute(statement);
         logger.info("│   ├── keyspace created...");
         if (session != null) {
-            logger.info("│   ├──  create scheam...");
+            logger.info("│   ├──  create schema...");
             // now create table schemas
             createArchiveTableSchema();
         }

dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/RAGAdapter.java

@@ -34,6 +34,8 @@ import java.util.logging.Level;
 import java.util.logging.Logger;
 
 import org.eclipse.microprofile.config.inject.ConfigProperty;
+import org.imixs.ai.workflow.ImixsAIPromptEvent;
+import org.imixs.ai.workflow.OpenAIAPIAdapter;
 import org.imixs.ai.workflow.OpenAIAPIService;
 import org.imixs.workflow.ItemCollection;
 import org.imixs.workflow.SignalAdapter;
@@ -41,6 +43,7 @@ import org.imixs.workflow.engine.WorkflowService;
 import org.imixs.workflow.exceptions.AdapterException;
 import org.imixs.workflow.exceptions.PluginException;
 
+import jakarta.enterprise.event.Event;
 import jakarta.inject.Inject;
 
 /**
@@ -117,6 +120,9 @@ public class RAGAdapter implements SignalAdapter {
     @Inject
     private ClusterService clusterService;
 
+    @Inject
+    private Event<ImixsAIPromptEvent> llmPromptEventObservers = null;
+
     /**
      * Default Constructor
      */
@@ -167,9 +173,37 @@ public class RAGAdapter implements SignalAdapter {
             if (llmIndexDefinitions != null && llmIndexDefinitions.size() > 0) {
 
                 for (ItemCollection indexDefinition : llmIndexDefinitions) {
-                    String llmText = indexDefinition.getItemValueString("text");
+                    llmAPIEndpoint = parseLLMEndpointByBPMN(indexDefinition);
-                    logger.info("POST index text: " + llmText);
+                    // do we have a valid endpoint?
-                    float[] indexResult = llmService.processRAGIndex(llmText, llmAPIEndpoint);
+                    if (llmAPIEndpoint == null || llmAPIEndpoint.isEmpty()) {
+                        throw new PluginException(OpenAIAPIAdapter.class.getSimpleName(), API_ERROR,
+                                "imixs-ai configuration error: no llm service endpoint defined!");
+                    }
+                    if ("true".equalsIgnoreCase(indexDefinition.getItemValueString("debug"))) {
+                        llmAPIDebug = true;
+                    }
+
+                    logger.info("post Llama-cpp request: " + llmAPIEndpoint);
+
+                    String promptTemplate = llmService.loadPromptTemplate(event);
+                    logger.info("===> PromptTemplate: ");
+                    logger.info(promptTemplate);
+
+                    String llmPrompt = llmService.buildPrompt(promptTemplate, workitem);
+
+                    if (llmPrompt.isBlank()) {
+                        throw new PluginException(
+                                RAGAdapter.class.getSimpleName(), API_ERROR,
+                                "Unable to parse prompt for RAG indexing");
+                    }
+
+                    if (llmAPIDebug) {
+                        logger.info("===> Total Prompt Length = " + llmPrompt.length());
+                        logger.info("===> Prompt: ");
+                        logger.info(llmPrompt);
+                    }
+
+                    float[] indexResult = llmService.postEmbedding(llmPrompt, llmAPIEndpoint);
 
                     // clusterService.insertVector
                     logger.info("POST result: " + indexResult);

workflow/bedarfsmeldung-de-1.0.0.bpmn

@@ -2062,7 +2062,7 @@ result.isValid=true;
     <bpmn2:sequenceFlow id="sequenceFlow_ZsCUdg" sourceRef="IntermediateCatchEvent_3" targetRef="Task_2">
       <bpmn2:documentation id="documentation_hziJNw"/>
     </bpmn2:sequenceFlow>
-    <bpmn2:intermediateCatchEvent id="event_qTIbqA" imixs:activityid="50" name="RAG Test">
+    <bpmn2:intermediateCatchEvent id="event_qTIbqA" imixs:activityid="50" name="RAG Test2">
       <bpmn2:extensionElements>
         <imixs:item name="keylogtimeformat" type="xs:string">
           <imixs:value><![CDATA[2]]></imixs:value>
@@ -2133,26 +2133,26 @@ result.isValid=true;
           <imixs:value>false</imixs:value>
         </imixs:item>
       </bpmn2:extensionElements>
-      <bpmn2:documentation id="documentation_405Wvg"><![CDATA[Zwischenspeichern]]></bpmn2:documentation>
+      <bpmn2:documentation id="documentation_405Wvg"><![CDATA[RAG Test]]></bpmn2:documentation>
       <bpmn2:outgoing>sequenceFlow_08JtcA</bpmn2:outgoing>
       <bpmn2:signalEventDefinition id="signalEventDefinition_OmSXlg" signalRef="signal_4"/>
     </bpmn2:intermediateCatchEvent>
     <bpmn2:sequenceFlow id="sequenceFlow_08JtcA" sourceRef="event_qTIbqA" targetRef="Task_2">
       <bpmn2:documentation id="documentation_mnWPUQ"/>
     </bpmn2:sequenceFlow>
-    <bpmn2:dataObject id="dataObject_axJA7Q" name="Data Object-4">
+    <bpmn2:dataObject id="dataObject_axJA7Q" imixs:datatype="prompt-data" name="Data Object-4">
       <bpmn2:documentation id="documentation_3VwL8w"><![CDATA[<?xml version="1.0" encoding="UTF-8"?>
-<IndexDefinition>
+<PromptDefinition>
-  <index_options>{"n_predict": 4096, "temperature": 0}</index_options>
+  <prompt_options>{}</prompt_options>
-<![CDATA[<text><itemvalue>$workflowgroup</itemvalue>: <itemvalue>$workflowstatus</itemvalue>
+  <prompt><![CDATA[<itemvalue>$workflowgroup</itemvalue>: <itemvalue>$workflowstatus</itemvalue>
 
 # Summary
 <itemvalue>$workflowsummary</itemvalue>
 
 <itemvalue>offer.summary</itemvalue>
-
+]]]]><![CDATA[>
-]]]]><![CDATA[></text>
+</prompt>
-</IndexDefinition>    ]]></bpmn2:documentation>
+</PromptDefinition>  ]]></bpmn2:documentation>
     </bpmn2:dataObject>
     <bpmn2:association id="association_JGXYyg" sourceRef="event_qTIbqA" targetRef="dataObject_axJA7Q">
       <bpmn2:documentation id="documentation_mvyBDQ"/>
@@ -2710,8 +2710,6 @@ result.isValid=true;
       <bpmndi:BPMNEdge bpmnElement="Association_4" id="BPMNEdge_Association_4" sourceElement="BPMNShape_DataObject_2" targetElement="BPMNShape_Task_2">
         <bpmndi:BPMNLabel id="BPMNLabel_95"/>
         <di:waypoint x="248.0" y="855.0"/>
-        <di:waypoint x="248.0" y="855.0"/>
-        <di:waypoint x="248.0" y="855.0"/>
         <di:waypoint x="248.0" y="636.0"/>
         <di:waypoint x="310.0" y="636.0"/>
       </bpmndi:BPMNEdge>
@@ -2808,7 +2806,6 @@ result.isValid=true;
       </bpmndi:BPMNShape>
       <bpmndi:BPMNEdge bpmnElement="association_JGXYyg" id="BPMNEdge_bHPaEg" sourceElement="BPMNShape_0OS0TQ" targetElement="BPMNShape_3PbNFQ">
         <di:waypoint x="513.0" y="175.0"/>
-        <di:waypoint x="559.0" y="175.0"/>
         <di:waypoint x="570.0" y="175.0"/>
       </bpmndi:BPMNEdge>
     </bpmndi:BPMNPlane>