cassandra setup and docu

doc/CASSANDRA.md

@@ -0,0 +1,49 @@
+# Cassandra RAG Database
+
+The Cassandra database used by dynamixs.ai is created automatically during startup. This includes the Cassandra keyspace as also the data base schema.
+Aside from that, the Cassandra cluster can also be managed through the Command Line Console (CLI).
+
+## Using the Cassandra CLI
+
+To connect to your cluster you can use the cqlsh tool.
+To run cqlsh from your started docker environment run:
+
+```bash
+$ docker exec -it cassandra cqlsh
+Connected to dynamixsAiCluster at 127.0.0.1:9042
+[cqlsh 6.2.0 | Cassandra 5.0.4 | CQL spec 3.4.7 | Native protocol v5]
+Use HELP for help.
+cqlsh>
+```
+
+## CQL Examples
+
+The following section contains some basic cqlsh commands. For full description see the [Cassandra CQL refernce](https://docs.datastax.com/en/dse/6.0/cql/).
+
+**show key spaces:**
+
+Show all available keyspaces:
+
+    cqlsh> DESC KEYSPACES;
+
+**Switch to Keysapce:**
+
+Select a keyspace be name to interact with this keyspace:
+
+    cqlsh> use embeddings;
+
+**Show tables in a keyspace:**
+
+Show tables schemas in current keyspace:
+
+    cqlsh:embeddings> DESC TABLES;
+
+**Drop Keyspace:**
+
+Drop the keyspace:
+
+    DROP KEYSPACE IF EXISTS <keyspace_name>;
+
+### Create a dev keyspace with cqlsh
+
+    cqlsh> CREATE KEYSPACE <keyspace_name> WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};

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

@@ -12,6 +12,7 @@ import org.eclipse.microprofile.config.inject.ConfigProperty;
 
 import com.datastax.oss.driver.api.core.CqlSession;
 import com.datastax.oss.driver.api.core.CqlSessionBuilder;
+import com.datastax.oss.driver.api.core.InvalidKeyspaceException;
 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;
@@ -131,12 +132,12 @@ public class ClusterService {
      */
     private void scheduleClusterCheck() {
         try {
-            logger.log(Level.INFO, "├── Connecting Cassandra cluster...");
+            logger.log(Level.INFO, "├── connecting Cassandra cluster...");
             getSession();
-            logger.log(Level.INFO, "├── ✅ Successfully connected to Cassandra!");
+            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...");
+            logger.log(Level.WARNING, "├── ⚠️ cluster not ready yet. retrying in 10 seconds...");
             // retry in 10 sec
             timerService.createSingleActionTimer(10000, new TimerConfig());
         }
@@ -174,12 +175,9 @@ public class ClusterService {
     public CqlSession getSession() throws ClusterException {
         if (session == null) {
             try {
-                logger.info("├── initializing cluster...");
+                logger.info("├── initializing cluster and keyspace...");
-                session = initSession();
+                session = initSessionWithKeyspace();
-                if (session != null) {
+                logger.info("│   ├── ✅ schema status = OK");
-                    logger.info("│   ├── keySpace connection status = OK");
-                    initKeySpace();
-                }
             } catch (Exception e) {
                 throw new ClusterException(ClusterException.CLUSTER_ERROR,
                         "Failed to init cassandra session: " + e.getMessage(), e);
@@ -189,32 +187,38 @@ public class ClusterService {
     }
 
     /**
-     * Returns a cassandra session for the archive KeySpace. The keyspace is defined
+     * This method opens the configured keyspace. If the keyspace does no yet exist
-     * by the environmetn variable EMBEDDINGS__CLUSTER_KEYSPACE. If no keyspace with
+     * it creates a new keyspace with the table schema
-     * this given keyspace name exists, the method creates the keyspace and table
-     * schemas.
      * 
-     * @throws ArchiveException
+     * @return
+     * @throws ClusterException
      */
-    private void initKeySpace() throws ClusterException {
+    private CqlSession initSessionWithKeyspace() throws ClusterException {
-        String keySpacename = keySpace.orElse(null);
+        // validate keyspace name
-        if (keySpacename == null) {
+        String keySpacename = keySpace.orElseThrow(
-            throw new ClusterException(ClusterException.INVALID_KEYSPACE, "EMBEDDINGS_CLUSTER_KEYSPACE not set!");
+                () -> new ClusterException(ClusterException.INVALID_KEYSPACE, "EMBEDDINGS_CLUSTER_KEYSPACE not set!"));
-        }
+
         if (!isValidKeyspaceName(keySpacename)) {
             throw new ClusterException(ClusterException.INVALID_KEYSPACE,
                     "keyspace '" + keySpacename + "' name invalid.");
         }
 
-        logger.info("│   ├── connecting keyspace='" + keySpacename + "'...");
-        // try to open keySpace
         try {
-            // session = cluster.connect(keySpace.get());
+            // try to connect keyspace ...
-        } catch (InvalidQueryException e) {
+            session = createSession(keySpacename);
-            logger.warning("......connecting keySpace '" + keySpacename + "' failed: " + e.getMessage());
+            logger.info("│   ├── ✅ keyspace status = OK");
+            return session;
+        } catch (InvalidQueryException | InvalidKeyspaceException e) {
+            logger.warning("│   ├── ⚠️ Keyspace does not yet exist, creating new keyspace...");
+            // Create a basic session object to create the keyspace....
+            session = createSession(null);
             createKeySpace(keySpacename);
+            // close and reconnect session...
+            session.close();
+            session = createSession(keySpacename);
+            createSchema();
+            return session;
         }
-
     }
 
     /**
@@ -226,7 +230,9 @@ public class ClusterService {
      * @return Cassandra Cluster instacne
      * @throws ArchiveException
      */
-    protected CqlSession initSession() throws ClusterException {
+
+    private CqlSession createSession(String keyspace) throws ClusterException {
+        CqlSessionBuilder builder = CqlSession.builder();
         // Boolean used to check if at least one host could be resolved
         boolean found = false;
 
@@ -236,8 +242,6 @@ public class ClusterService {
         }
 
         logger.info("│   ├── cluster connecting: " + contactPoint.get());
-
-        CqlSessionBuilder builder = CqlSession.builder();
         String[] hosts = contactPoint.get().split(",");
         for (String host : hosts) {
             try {
@@ -254,96 +258,18 @@ public class ClusterService {
             // No host could be resolved so we throw an exception
             throw new IllegalStateException("All provided hosts are unknown - check cluster status and configuration!");
         }
-
-        // builder.withLoadBalancingPolicy(new RoundRobinPolicy());
-        // builder.withRetryPolicy(DefaultRetryPolicy.INSTANCE);
-
         // set optional credentials...
         if (userid.isPresent() && !userid.get().isEmpty()) {
             builder = builder.withAuthCredentials(userid.get(), password.get());
         }
 
-        // use SSL ?
+        if (keyspace != null) {
-        // if (bUseSSL) {
+            builder.withKeyspace(keyspace);
-        // try {
+        }
-        // // create ssl options based on environment settings...
-        // SSLOptions options = createSSLOptions();
-        // logger.info("...creating SSL cluster session...");
-        // builder.withSSL(options);
-        // } catch (KeyManagementException | UnrecoverableKeyException |
-        // KeyStoreException | NoSuchAlgorithmException
-        // | CertificateException | IOException e) {
-        // logger.severe("Failed to connect withSSL: " + e.getMessage());
-        // e.printStackTrace();
-        // }
-        // }
-
         builder.withLocalDatacenter("datacenter1");
         return builder.build();
-        // cluster = builder.build();
-        // cluster.init();
-        // logger.info("...cluster conection status = OK");
-
-        // return cluster;
-
     }
 
-    /**
-     * Helper Method to generate sslOpens object for cassandra dataStax driver
-     * withSSL.
-     * <p>
-     * At least the truststorePath is needed to establish a one way SSL connection.
-     * <p>
-     * The keystorePath is only needed in case 'require_client_auth = true' is set
-     * in the cassandra.yaml file
-     * 
-     * @return sslOptons
-     * @throws KeyStoreException
-     * @throws FileNotFoundException
-     * @throws IOException
-     * @throws NoSuchAlgorithmException
-     * @throws KeyManagementException
-     * @throws CertificateException
-     * @throws UnrecoverableKeyException
-     */
-    // private SSLOptions createSSLOptions() throws KeyStoreException,
-    // FileNotFoundException, IOException,
-    // NoSuchAlgorithmException, KeyManagementException, CertificateException,
-    // UnrecoverableKeyException {
-
-    // TrustManagerFactory tmf = null;
-    // if (truststorePath.isPresent() && !truststorePath.get().isEmpty()) {
-    // KeyStore tks = KeyStore.getInstance("JKS");
-    // tks.load((InputStream) new FileInputStream(new File(truststorePath.get())),
-    // truststorePwd.get().toCharArray());
-    // tmf =
-    // TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
-    // tmf.init(tks);
-    // } else {
-    // logger.info("SSLOptions without truststore...");
-    // }
-
-    // KeyManagerFactory kmf = null;
-    // if (keystorePath.isPresent() && !keystorePath.get().isEmpty()) {
-    // KeyStore kks = KeyStore.getInstance("JKS");
-    // kks.load((InputStream) new FileInputStream(new File(keystorePath.get())),
-    // keystorePwd.get().toCharArray());
-    // kmf = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
-    // kmf.init(kks, keystorePwd.get().toCharArray());
-    // } else {
-    // logger.info("SSLOptions without keystore...");
-    // }
-
-    // SSLContext sslContext = SSLContext.getInstance("TLS");
-    // sslContext.init(kmf != null ? kmf.getKeyManagers() : null, tmf != null ?
-    // tmf.getTrustManagers() : null,
-    // new SecureRandom());
-    // RemoteEndpointAwareJdkSSLOptions sslOptions =
-    // RemoteEndpointAwareJdkSSLOptions.builder()
-    // .withSSLContext(sslContext).build();
-    // return sslOptions;
-    // }
-
     /**
      * Test if the keyspace name is valid.
      * 
@@ -354,46 +280,37 @@ public class ClusterService {
         if (keySpace == null || keySpace.isEmpty()) {
             return false;
         }
-
         return keySpace.matches(KEYSPACE_REGEX);
-
     }
 
     /**
-     * This method creates a cassandra keySpace.
+     * This method creates a cassandra keySpace
-     * <p>
-     * Depending on the KeyspaceType, the method creates the core table schema to
-     * store configurations, or the extended Archive table schcema which is used to
-     * store imixs documents.
      * 
      * @param cluster
      * @throws ArchiveException
      */
-    protected void createKeySpace(String keySpace) throws ClusterException {
+    private void createKeySpace(String keySpace) throws ClusterException {
         logger.info("│   ├── creating new keyspace '" + keySpace + "'...");
 
         String statement = "CREATE KEYSPACE IF NOT EXISTS " + keySpace + " WITH replication = {'class': '" + repClass
                 + "', 'replication_factor': " + repFactor + "};";
 
         session.execute(statement);
-        logger.info("│   ├── keyspace created...");
+        logger.info("│   ├── ✅ keyspace '" + keySpace + "' created.");
-        if (session != null) {
-            logger.info("│   ├──  create schema...");
-            // now create table schemas
-            createArchiveTableSchema();
-        }
 
     }
 
     /**
-     * This helper method creates the ImixsArchive document table schema if not yet
+     * This method creates the keySpace schema.
-     * exists
+     * 
+     * @param cluster
+     * @throws ArchiveException
      */
-    protected void createArchiveTableSchema() {
+    private void createSchema() throws ClusterException {
 
-        logger.info("│   ├── Create Data Table...");
+        logger.info("│   ├── create schema...");
-
+        // now create table schemas
-        String query = "CREATE TABLE IF NOT EXISTS embeddings.document_vectors (\n" + //
+        String query = "CREATE TABLE IF NOT EXISTS document_vectors (\n" + //
                 "  record_id timeuuid,\n" + //
                 "  id uuid,\n" + //
                 "  content_chunk text,\n" + //
@@ -404,48 +321,51 @@ public class ClusterService {
                 "WITH CLUSTERING ORDER BY (created_at DESC);";
         session.execute(query);
 
-        logger.info("│   ├── Create Index...");
+        logger.info("│   ├── create index...");
         query = "CREATE INDEX IF NOT EXISTS edv_ann_index\n" + //
-                "  ON embeddings.document_vectors(content_vector) USING 'sai';";
+                "  ON document_vectors(content_vector) USING 'sai';";
 
         session.execute(query);
-
+        logger.info("│   ├── ✅ database schema created.");
     }
 
-    public void insertVector(String content, float[] vector) {
+    public void insertVector(String content, float[] vector) throws ClusterException {
+        String keySpacename = keySpace.orElse(null);
+        try {
+            // Generiere UUID und aktuelle Zeit
+            UUID id = UUID.randomUUID();
+            Instant createdAt = Instant.now();
 
-        // Generiere UUID und aktuelle Zeit
+            // CQL-Query mit Platzhaltern
-        UUID id = UUID.randomUUID();
+            String insertQuery = "INSERT INTO " + keySpacename + ".document_vectors "
-        Instant createdAt = Instant.now();
+                    + "(record_id, id, content_chunk, content_vector, created_at)  VALUES (now(), ?, ?, ?, ?)";
 
-        // CQL-Query mit Platzhaltern
+            // Vorbereiten und Ausführen der Query
-        String insertQuery = """
+            PreparedStatement preparedStmt = session.prepare(insertQuery);
-                INSERT INTO embeddings.document_vectors
+            BoundStatement boundStmt = preparedStmt.bind(id, content, vector, createdAt);
-                    (record_id, id, content_chunk, content_vector, created_at)
+            session.execute(boundStmt);
-                VALUES (now(), ?, ?, ?, ?)
-                """;
 
-        // Vorbereiten und Ausführen der Query
+            // id = uuid.uuid4()
-        PreparedStatement preparedStmt = session.prepare(insertQuery);
+            // content_chunk = content
-        BoundStatement boundStmt = preparedStmt.bind(id, content, vector, createdAt);
+            // content_vector = vector
-        session.execute(boundStmt);
+            // created_at = datetime.datetime.now()
+            // insert_query = """
+            // INSERT INTO embeddings.document_vectors (record_id, id, content_chunk,
+            // content_vector, created_at)
+            // VALUES (now(), ?, ?, ?, ?)
+            // """
+            // prepared_stmt = session.prepare(insert_query)
+            // session.execute(prepared_stmt, [
+            // id,
+            // content_chunk,
+            // content_vector,
+            // created_at
+            // ])
 
-        // id = uuid.uuid4()
+        } catch (Exception e) {
-        // content_chunk = content
+            throw new ClusterException(ClusterException.CLUSTER_ERROR,
-        // content_vector = vector
+                    "Failed to insert embeddings into keyspace: " + e.getMessage(), e);
-        // created_at = datetime.datetime.now()
+        }
-        // insert_query = """
-        // INSERT INTO embeddings.document_vectors (record_id, id, content_chunk,
-        // content_vector, created_at)
-        // VALUES (now(), ?, ?, ?, ?)
-        // """
-        // prepared_stmt = session.prepare(insert_query)
-        // session.execute(prepared_stmt, [
-        // id,
-        // content_chunk,
-        // content_vector,
-        // created_at
-        // ])
     }
 
 }

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

@@ -35,7 +35,6 @@ 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;
@@ -156,8 +155,6 @@ public class RAGAdapter implements SignalAdapter {
     public ItemCollection execute(ItemCollection workitem, ItemCollection event)
             throws AdapterException, PluginException {
         String llmAPIEndpoint = null;
-        String llmAPIResultEvent = null;
-        String llmAPIResultItem = null;
         boolean llmAPIDebug = false;
         long processingTime = System.currentTimeMillis();
 
@@ -176,17 +173,17 @@ public class RAGAdapter implements SignalAdapter {
                     llmAPIEndpoint = parseLLMEndpointByBPMN(indexDefinition);
                     // do we have a valid endpoint?
                     if (llmAPIEndpoint == null || llmAPIEndpoint.isEmpty()) {
-                        throw new PluginException(OpenAIAPIAdapter.class.getSimpleName(), API_ERROR,
+                        throw new PluginException(RAGAdapter.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);
+                    logger.info("├── post Llama-cpp request: " + llmAPIEndpoint);
 
                     String promptTemplate = llmService.loadPromptTemplate(event);
-                    logger.info("===> PromptTemplate: ");
+                    logger.info("│   ├── PromptTemplate: ");
                     logger.info(promptTemplate);
 
                     String llmPrompt = llmService.buildPrompt(promptTemplate, workitem);
@@ -198,28 +195,29 @@ public class RAGAdapter implements SignalAdapter {
                     }
 
                     if (llmAPIDebug) {
-                        logger.info("===> Total Prompt Length = " + llmPrompt.length());
+                        logger.info("│   ├── Total Prompt Length = " + llmPrompt.length());
-                        logger.info("===> Prompt: ");
+                        logger.info("│   ├── Prompt: ");
                         logger.info(llmPrompt);
                     }
 
-                    float[] indexResult = llmService.postEmbedding(llmPrompt, llmAPIEndpoint);
+                    float[] indexResult = llmService.postEmbedding(llmPrompt, llmAPIEndpoint, llmAPIDebug);
 
-                    // clusterService.insertVector
+                    // write to cassandra
-                    logger.info("POST result: " + indexResult);
+                    clusterService.insertVector(promptTemplate, indexResult);
-                    logger.info("result size= " + indexResult.length);
+
+                    if (llmAPIDebug) {
+                        logger.info("├── ✅ " + indexResult.length + " floats stored in RAG db.");
+                        logger.info(
+                                "├── ✅ Total processing time: " + (System.currentTimeMillis() - processingTime) + "ms");
+                    }
                 }
             }
 
         } catch (PluginException e) {
-            logger.severe("Unable to parse item definitions for 'imixs-ai', verify model - " + e.getMessage());
+            logger.severe("├── ⚠️ Failed to post embeddings: " + e.getMessage());
-            throw new PluginException(
+            throw e;
-                    RAGAdapter.class.getSimpleName(), e.getErrorCode(),
+        } catch (ClusterException e) {
-                    "Unable to parse item definitions for 'imixs-ai', verify model - " + e.getMessage(), e);
+            throw new PluginException(RAGAdapter.class.getSimpleName(), API_ERROR, e.getMessage(), e);
-        }
-
-        if (llmAPIDebug) {
-            logger.info("===> Total Processing Time=" + (System.currentTimeMillis() - processingTime) + "ms");
         }
         return workitem;
     }