diff --git a/doc/CASSANDRA.md b/doc/CASSANDRA.md new file mode 100644 index 0000000..abe966d --- /dev/null +++ b/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 ; + +### Create a dev keyspace with cqlsh + + cqlsh> CREATE KEYSPACE WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1}; diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterService.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterService.java index 655d12b..0c259b1 100644 --- a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterService.java +++ b/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..."); - session = initSession(); - if (session != null) { - logger.info("│ ├── keySpace connection status = OK"); - initKeySpace(); - } + logger.info("├── initializing cluster and keyspace..."); + session = initSessionWithKeyspace(); + logger.info("│ ├── ✅ schema status = OK"); } 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 - * by the environmetn variable EMBEDDINGS__CLUSTER_KEYSPACE. If no keyspace with - * this given keyspace name exists, the method creates the keyspace and table - * schemas. + * This method opens the configured keyspace. If the keyspace does no yet exist + * it creates a new keyspace with the table schema * - * @throws ArchiveException + * @return + * @throws ClusterException */ - private void initKeySpace() throws ClusterException { - String keySpacename = keySpace.orElse(null); - if (keySpacename == null) { - throw new ClusterException(ClusterException.INVALID_KEYSPACE, "EMBEDDINGS_CLUSTER_KEYSPACE not set!"); - } + private CqlSession initSessionWithKeyspace() throws ClusterException { + // validate keyspace name + String keySpacename = keySpace.orElseThrow( + () -> 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()); - } catch (InvalidQueryException e) { - logger.warning("......connecting keySpace '" + keySpacename + "' failed: " + e.getMessage()); + // try to connect keyspace ... + session = createSession(keySpacename); + 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 (bUseSSL) { - // 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(); - // } - // } - + if (keyspace != null) { + builder.withKeyspace(keyspace); + } 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. - *

- * At least the truststorePath is needed to establish a one way SSL connection. - *

- * 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. - *

- * 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. + * This method creates a cassandra keySpace * * @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..."); - if (session != null) { - logger.info("│ ├── create schema..."); - // now create table schemas - createArchiveTableSchema(); - } + logger.info("│ ├── ✅ keyspace '" + keySpace + "' created."); } /** - * This helper method creates the ImixsArchive document table schema if not yet - * exists + * This method creates the keySpace schema. + * + * @param cluster + * @throws ArchiveException */ - protected void createArchiveTableSchema() { + private void createSchema() throws ClusterException { - logger.info("│ ├── Create Data Table..."); - - String query = "CREATE TABLE IF NOT EXISTS embeddings.document_vectors (\n" + // + logger.info("│ ├── create schema..."); + // now create table schemas + 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 - UUID id = UUID.randomUUID(); - Instant createdAt = Instant.now(); + // CQL-Query mit Platzhaltern + String insertQuery = "INSERT INTO " + keySpacename + ".document_vectors " + + "(record_id, id, content_chunk, content_vector, created_at) VALUES (now(), ?, ?, ?, ?)"; - // CQL-Query mit Platzhaltern - String insertQuery = """ - INSERT INTO embeddings.document_vectors - (record_id, id, content_chunk, content_vector, created_at) - VALUES (now(), ?, ?, ?, ?) - """; + // Vorbereiten und Ausführen der Query + PreparedStatement preparedStmt = session.prepare(insertQuery); + BoundStatement boundStmt = preparedStmt.bind(id, content, vector, createdAt); + session.execute(boundStmt); - // Vorbereiten und Ausführen der Query - PreparedStatement preparedStmt = session.prepare(insertQuery); - BoundStatement boundStmt = preparedStmt.bind(id, content, vector, createdAt); - session.execute(boundStmt); + // id = uuid.uuid4() + // content_chunk = content + // content_vector = vector + // 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() - // content_chunk = content - // content_vector = vector - // 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 - // ]) + } catch (Exception e) { + throw new ClusterException(ClusterException.CLUSTER_ERROR, + "Failed to insert embeddings into keyspace: " + e.getMessage(), e); + } } } 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 6682a4d..17b7e89 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 @@ -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("===> Prompt: "); + logger.info("│ ├── Total Prompt Length = " + llmPrompt.length()); + logger.info("│ ├── Prompt: "); logger.info(llmPrompt); } - float[] indexResult = llmService.postEmbedding(llmPrompt, llmAPIEndpoint); + float[] indexResult = llmService.postEmbedding(llmPrompt, llmAPIEndpoint, llmAPIDebug); - // clusterService.insertVector - logger.info("POST result: " + indexResult); - logger.info("result size= " + indexResult.length); + // write to cassandra + clusterService.insertVector(promptTemplate, indexResult); + + 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()); - throw new PluginException( - RAGAdapter.class.getSimpleName(), e.getErrorCode(), - "Unable to parse item definitions for 'imixs-ai', verify model - " + e.getMessage(), e); - } - - if (llmAPIDebug) { - logger.info("===> Total Processing Time=" + (System.currentTimeMillis() - processingTime) + "ms"); + logger.severe("├── ⚠️ Failed to post embeddings: " + e.getMessage()); + throw e; + } catch (ClusterException e) { + throw new PluginException(RAGAdapter.class.getSimpleName(), API_ERROR, e.getMessage(), e); } return workitem; }