243 lines
11 KiB
Java
243 lines
11 KiB
Java
/****************************************************************************
|
|
* Copyright (c) 2022-2025 Imixs Software Solutions GmbH and others.
|
|
*
|
|
* This program and the accompanying materials are made available under the
|
|
* terms of the Eclipse Public License v. 2.0 which is available at
|
|
* https://www.eclipse.org/legal/epl-2.0
|
|
*
|
|
* This Source Code may also be made available under the terms of the
|
|
* GNU General Public License, version 2 or later (GPL-2.0-or-later),
|
|
* which is available at https://www.gnu.org/licenses/old-licenses/gpl-2.0.html
|
|
*
|
|
* SPDX-License-Identifier: EPL-2.0 OR GPL-2.0-or-later
|
|
****************************************************************************/
|
|
|
|
package ai.dynamixs.cognition;
|
|
|
|
import java.util.List;
|
|
import java.util.Optional;
|
|
import java.util.logging.Level;
|
|
import java.util.logging.Logger;
|
|
|
|
import org.eclipse.microprofile.config.inject.ConfigProperty;
|
|
import org.imixs.ai.ImixsAIContextHandler;
|
|
import org.imixs.ai.workflow.OpenAIAPIConnector;
|
|
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.AdapterException;
|
|
import org.imixs.workflow.exceptions.PluginException;
|
|
|
|
import jakarta.annotation.security.DeclareRoles;
|
|
import jakarta.annotation.security.RunAs;
|
|
import jakarta.ejb.LocalBean;
|
|
import jakarta.ejb.Stateless;
|
|
import jakarta.ejb.TransactionAttribute;
|
|
import jakarta.ejb.TransactionAttributeType;
|
|
import jakarta.inject.Inject;
|
|
import jakarta.persistence.OptimisticLockException;
|
|
|
|
/**
|
|
* The CognitoEventService is responsible to process Dynamixs-AI Cognito events
|
|
* in an asynchronous batch process. The CognitoEventService automatically
|
|
* lookup eventLog entries of the topic "rag.event.*". The RAGEventService is
|
|
* called only by the RAGEventScheduler which is implementing a
|
|
* ManagedScheduledExecutorService.
|
|
* <p>
|
|
* The CognitoEventService reacts on events from the following types:
|
|
* <ul>
|
|
* <li>cognito.event.prompt - resoles the llm prompt based on the given workitem
|
|
* and creates a cache event</li>
|
|
* <li>cognito.event.cache - indicates that this entry is still cached in the
|
|
* event log database</li>
|
|
* </ul>
|
|
* The processor look up the workItem by the $uniqueId.
|
|
* <p>
|
|
* To prevent concurrent processes to handle the same workitems the batch
|
|
* process uses a Optimistic lock strategy. After fetching new event log entries
|
|
* the processor updates the eventLog entry in a new transaction and set the
|
|
* topic to 'batch.process.lock'. After that update we can be sure that no other
|
|
* process is dealing with these entries. After completing the processing step
|
|
* the eventlog entry will be removed.
|
|
* <p>
|
|
* To avoid ad deadlock the processor set an expiration time on the lock, so the
|
|
* lock will be auto-released after 1 minute (batch.processor.deadlock).
|
|
*
|
|
* @see AsyncEventScheduler
|
|
* @version 1.0
|
|
* @author rsoika
|
|
*
|
|
*/
|
|
@DeclareRoles({ "org.imixs.ACCESSLEVEL.MANAGERACCESS" })
|
|
@RunAs("org.imixs.ACCESSLEVEL.MANAGERACCESS")
|
|
@Stateless
|
|
@LocalBean
|
|
public class CognitoEventService {
|
|
|
|
public static final String EVENTLOG_TOPIC_COGNITO_PROMPT = "cognito.event.prompt";
|
|
public static final String EVENTLOG_TOPIC_COGNITO_CACHE = "cognito.event.cache";
|
|
|
|
public static final String ITEM_PROMPT_DEFINITION = "prompt.definition";
|
|
|
|
private static final Logger logger = Logger.getLogger(CognitoEventService.class.getName());
|
|
|
|
@Inject
|
|
EventLogService eventLogService;
|
|
|
|
@Inject
|
|
private WorkflowService workflowService;
|
|
|
|
@Inject
|
|
private OpenAIAPIService llmService;
|
|
|
|
@Inject
|
|
ImixsAIContextHandler imixsAIContextHandler;
|
|
|
|
@Inject
|
|
CognitoLockHandler cognitoLockHandler;
|
|
|
|
@Inject
|
|
@ConfigProperty(name = OpenAIAPIConnector.ENV_LLM_SERVICE_ENDPOINT)
|
|
Optional<String> serviceEndpoint;
|
|
|
|
/**
|
|
* The method lookups for cognito prompt event log entries and updates the ....
|
|
* information
|
|
* <p>
|
|
* Each eventLogEntry provides optional a prompt template
|
|
*
|
|
*/
|
|
@TransactionAttribute(value = TransactionAttributeType.REQUIRES_NEW)
|
|
public void processEventLog() {
|
|
long l = System.currentTimeMillis();
|
|
|
|
// test for new event log entries by timeout...
|
|
List<EventLog> events = eventLogService.findEventsByTimeout(100,
|
|
CognitoEventService.EVENTLOG_TOPIC_COGNITO_PROMPT, CognitoEventService.EVENTLOG_TOPIC_COGNITO_CACHE);
|
|
|
|
if (events.size() > 0) {
|
|
logger.log(Level.INFO, "├── 🔃 processing {0} Cognito events....", events.size());
|
|
|
|
// do we know the endpoint?
|
|
if (!serviceEndpoint.isPresent()) {
|
|
logger.log(Level.WARNING,
|
|
"├── ⚠️ Missing environment var 'llm.service.endpoint' - can not process cognito events!");
|
|
return;
|
|
}
|
|
|
|
for (EventLog eventLogEntry : events) {
|
|
try {
|
|
boolean llmAPIDebug = true;
|
|
logger.log(Level.INFO,
|
|
"│ ├── Event: " + eventLogEntry.getTopic() + " - " + eventLogEntry.getRef());
|
|
// first try to lock the eventLog entry....
|
|
if (eventLogService.lock(eventLogEntry)) {
|
|
|
|
// Index / Update event?
|
|
ItemCollection workitem = workflowService.getWorkItem(eventLogEntry.getRef());
|
|
if (workitem == null) {
|
|
logger.log(Level.WARNING, "│ ├── ⚠️ unable to read workitem: " + eventLogEntry.getRef());
|
|
eventLogService.removeEvent(eventLogEntry.getId());
|
|
continue;
|
|
}
|
|
|
|
ItemCollection promptItemCol = new ItemCollection(eventLogEntry.getData());
|
|
|
|
switch (eventLogEntry.getTopic()) {
|
|
|
|
case CognitoEventService.EVENTLOG_TOPIC_COGNITO_PROMPT + ".lock":
|
|
|
|
// load the promptDefinition from the event data object
|
|
|
|
String promptTemplate = promptItemCol.getItemValueString(ITEM_PROMPT_DEFINITION);
|
|
|
|
imixsAIContextHandler.setWorkItem(workitem);
|
|
imixsAIContextHandler.addPromptDefinition(promptTemplate);
|
|
|
|
String llmPrompt = imixsAIContextHandler.toString();
|
|
// if we have a prompt we call the llm api endpoint
|
|
if (llmPrompt != null && !llmPrompt.isBlank()) {
|
|
String llmAPIEndpoint = serviceEndpoint.get();
|
|
// postPromptCompletion
|
|
// JsonObject jsonPrompt =
|
|
// llmService.buildJsonPromptObjectV1(imixsAIContextHandler);
|
|
String completionResult = llmService.postPromptCompletion(imixsAIContextHandler,
|
|
llmAPIEndpoint);
|
|
// process the ai.result....
|
|
if (llmAPIDebug) {
|
|
logger.info("===> Completion Result: ");
|
|
logger.info(completionResult);
|
|
}
|
|
|
|
String resultMessage = llmService.processPromptResult(completionResult, "", workitem);
|
|
|
|
// if the workitem is not locked we can safety update the $workflowabstarct
|
|
if (!cognitoLockHandler.isLocked(workitem.getUniqueID())) {
|
|
logger.log(Level.INFO,
|
|
"│ ├── ✓ workitem not locked - updating $workflowabstract...");
|
|
workitem.setItemValue("$workflowabstract", resultMessage);
|
|
workflowService.getDocumentService().save(workitem);
|
|
} else {
|
|
// workitem is still in edit mode - so we create a cache event that just stores
|
|
// the result. This is an async caching mechanism for the front end.
|
|
logger.log(Level.INFO,
|
|
"│ ├── ⚐ workitem locked - create COGNITO_CACHE event...");
|
|
promptItemCol.setItemValue("$workflowabstract", resultMessage);
|
|
// create a new event
|
|
eventLogService.createEvent(CognitoEventService.EVENTLOG_TOPIC_COGNITO_CACHE,
|
|
workitem.getUniqueID(),
|
|
promptItemCol);
|
|
}
|
|
} else {
|
|
logger.finest(
|
|
"......no prompt definition found for " + workitem.getUniqueID());
|
|
}
|
|
// finally remove the event log entry...
|
|
eventLogService.removeEvent(eventLogEntry.getId());
|
|
break;
|
|
|
|
case CognitoEventService.EVENTLOG_TOPIC_COGNITO_CACHE + ".lock":
|
|
logger.log(Level.INFO,
|
|
"│ ├── process COGNITO_CACHE event...");
|
|
if (!cognitoLockHandler.isLocked(workitem.getUniqueID())) {
|
|
logger.log(Level.INFO,
|
|
"│ ├── ✓ workitem not locked - updating $workflowabstract...");
|
|
// ItemCollection promptItemCol = new ItemCollection(eventLogEntry.getData());
|
|
String resultMessage = promptItemCol.getItemValueString("$workflowabstract");
|
|
workitem.setItemValue("$workflowabstract", resultMessage);
|
|
workflowService.getDocumentService().save(workitem);
|
|
// finally remove the event log entry...
|
|
eventLogService.removeEvent(eventLogEntry.getId());
|
|
} else {
|
|
// repeat event until workitem is unlocked
|
|
eventLogService.unlock((eventLogEntry));
|
|
}
|
|
|
|
break;
|
|
default:
|
|
// no op
|
|
}
|
|
|
|
}
|
|
|
|
} catch (OptimisticLockException e) {
|
|
// lock was not possible - continue....
|
|
logger.log(Level.INFO, "│ ├── ⚠️ unable to lock AsyncEvent: {0}", e.getMessage());
|
|
} catch (PluginException | AdapterException e) {
|
|
logger.log(Level.WARNING, e.getErrorCode() + " - " + e.getMessage());
|
|
// remove the event log entry...
|
|
eventLogService.removeEvent(eventLogEntry.getId());
|
|
}
|
|
|
|
}
|
|
|
|
logger.log(Level.INFO, "├── ✅ {0} CognitoEvents processed in {1}ms",
|
|
new Object[] { events.size(), System.currentTimeMillis() - l });
|
|
|
|
}
|
|
}
|
|
|
|
}
|