diff --git a/.vscode/launch.json b/.vscode/launch.json index a861aeb..515c614 100644 --- a/.vscode/launch.json +++ b/.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" } ] } \ No newline at end of file diff --git a/docker/docker-compose-dev.yaml b/docker/docker-compose-dev.yaml index e279b43..1ad7701 100644 --- a/docker/docker-compose-dev.yaml +++ b/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)" diff --git a/dynamixs-ai-app/src/main/resources/META-INF/beans.xml b/dynamixs-ai-app/src/main/resources/META-INF/beans.xml deleted file mode 100644 index debd97c..0000000 --- a/dynamixs-ai-app/src/main/resources/META-INF/beans.xml +++ /dev/null @@ -1,6 +0,0 @@ - - - \ No newline at end of file diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/DynamixsSetupService.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/DynamixsSetupService.java index 97c4021..36bbea2 100644 --- a/dynamixs-ai-rag/src/main/java/ai/dynamixs/DynamixsSetupService.java +++ b/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 - * configuration entity 'BASIC'. This is the general configuration entity used - * by Imixs-Office-Workflow. - *

- * The item 'properties' managed by the 'BASIC' configuration entity is also - * used by the MartyConfigSource to provide optional config parameters. - *

- * Since version 4.5.1 the SetupController provides a list of fixed UserGroups. + * This DynamixsSetupService reacts on the + * 'org.imixs.workflow.engine.SetupEvent' and initializes a Cassandra Session. + * If the connection can not be established immediately the method retries a + * connection every 10 seconds. * * @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; - /** - * Create a rest client instance - */ - @PostConstruct - public void init() { - - } + @Resource + private TimerService timerService; /** - * This method verifies the schema of the cassandra database - *

- * + * Try to connect to Cassandra * * @param setupEvent - * @throws RestAPIException */ - public void setup(@Observes SetupEvent setupEvent) throws RestAPIException { - - logger.log(Level.INFO, "├── verify cassandra database..."); - - clusterService.getSession(); + public void setup(@Observes SetupEvent setupEvent) { + scheduleClusterCheck(); } + /** + * 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 + } } diff --git a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterException.java b/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterException.java index 472b13f..c00e525 100644 --- a/dynamixs-ai-rag/src/main/java/ai/dynamixs/rag/ClusterException.java +++ b/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"; 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 4397659..f099115 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 @@ -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 keystorePwd; - // private Cluster cluster; private CqlSession session; - @PostConstruct - private void init() { - try { - // cluster = initCluster(); - session = initSession(); + /** + * This method tries to initalize the cassandara session object. + * + * 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.severe("Failed to init cassandra session!"); - e.printStackTrace(); + logger.info("├── ✅ initializing cluster successful."); } + } @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) { - init(); + 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(); } 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 87c6392..6682a4d 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 @@ -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 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"); - logger.info("POST index text: " + llmText); - float[] indexResult = llmService.processRAGIndex(llmText, llmAPIEndpoint); + llmAPIEndpoint = parseLLMEndpointByBPMN(indexDefinition); + // do we have a valid endpoint? + 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); diff --git a/workflow/bedarfsmeldung-de-1.0.0.bpmn b/workflow/bedarfsmeldung-de-1.0.0.bpmn index f8870ee..42936fe 100644 --- a/workflow/bedarfsmeldung-de-1.0.0.bpmn +++ b/workflow/bedarfsmeldung-de-1.0.0.bpmn @@ -2062,7 +2062,7 @@ result.isValid=true; - + @@ -2133,26 +2133,26 @@ result.isValid=true; false - + sequenceFlow_08JtcA - + - - {"n_predict": 4096, "temperature": 0} -$workflowgroup: $workflowstatus + + {} + $workflowgroup: $workflowstatus # Summary $workflowsummary offer.summary - -]]]]> - ]]> +]]]]> + + ]]> @@ -2710,8 +2710,6 @@ result.isValid=true; - - @@ -2808,7 +2806,6 @@ result.isValid=true; -