Add a new events app (#199)

Author: vegito22

llmstack/events/__init__.py

@@ -0,0 +1,28 @@
+import importlib
+import logging
+import uuid
+
+from django.conf import settings
+from rest_framework import viewsets
+
+from llmstack.jobs.adhoc import ProcessingJob
+
+logger = logging.getLogger(__name__)
+
+
+class EventProcessingJob(ProcessingJob):
+    @classmethod
+    def generate_job_id(cls):
+        return "{}".format(str(uuid.uuid4()))
+
+
+class EventsViewSet(viewsets.ViewSet):
+    def create(self, topic, event_data):
+        if topic in settings.EVENT_TOPIC_MAPPING:
+            for processor_fn_name in settings.EVENT_TOPIC_MAPPING[topic]:
+                module_name = ".".join(processor_fn_name.split(".")[:-1])
+                fn_name = processor_fn_name.split(".")[-1]
+                module = importlib.import_module(module_name)
+                fn = getattr(module, fn_name)
+
+                EventProcessingJob.create(func=fn, args=[event_data]).add_to_queue()

llmstack/events/apis.py

@@ -0,0 +1,34 @@
+import importlib
+
+from django.apps import AppConfig
+
+
+class EventsConfig(AppConfig):
+    name = "llmstack.events"
+    label = "events"
+
+    def ready(self) -> None:
+        from django.conf import settings
+
+        # Ensure Event topic mapping is correctly configured in settings, if not make it empty
+        if not hasattr(settings, "EVENT_TOPIC_MAPPING"):
+            setattr(settings, "EVENT_TOPIC_MAPPING", {})
+
+        for topic in settings.EVENT_TOPIC_MAPPING:
+            if not isinstance(settings.EVENT_TOPIC_MAPPING[topic], list):
+                raise ValueError(f"EVENT_TOPIC_MAPPING[{topic}] must be a list of processor functions")
+            for processor_fn_name in settings.EVENT_TOPIC_MAPPING[topic]:
+                try:
+                    module_name = ".".join(processor_fn_name.split(".")[:-1])
+                    fn_name = processor_fn_name.split(".")[-1]
+                    module = importlib.import_module(module_name)
+                    if not callable(getattr(module, fn_name)):
+                        raise ValueError(
+                            f"EVENT_TOPIC_MAPPING[{topic}] contains an invalid processor function: {processor_fn_name}"
+                        )
+                except ImportError:
+                    raise ValueError(
+                        f"EVENT_TOPIC_MAPPING[{topic}] contains an invalid processor function: {processor_fn_name}"
+                    )
+
+        return super().ready()

llmstack/events/apps.py

@@ -2,20 +2,10 @@
 
 from llmstack.processors.models import RunEntry
 
-from .history import HistoryStore
-
 logger = logging.getLogger(__name__)
 
 
-def persist_history_task_internal(processors, bookkeeping_data_map):
-    if "input" not in bookkeeping_data_map or (
-        "output" not in bookkeeping_data_map and "agent" not in bookkeeping_data_map
-    ):
-        logger.error(
-            f"Could not persist history {bookkeeping_data_map} for {processors} because input or output is missing",
-        )
-        return
-
+def _process_app_run_data(processors, bookkeeping_data_map):
     input = bookkeeping_data_map["input"]["run_data"]
     output = (
         bookkeeping_data_map["agent"]["run_data"]
@@ -118,17 +108,19 @@ def persist_history_task_internal(processors, bookkeeping_data_map):
         processor_runs=processor_runs,
         platform_data=platform_data,
     )
+    return input, processor_runs, run_entry
 
-    # Persist history
-    HistoryStore.persist(
-        run_entry,
-        processor_runs=processor_runs,
-        input=input,
-    )
 
+def persist_app_run_history(event_data):
+    processors = event_data.get("processors", [])
+    bookkeeping_data_map = event_data.get("bookkeeping_data_map", {})
+    if "input" not in bookkeeping_data_map or (
+        "output" not in bookkeeping_data_map and "agent" not in bookkeeping_data_map
+    ):
+        logger.error(
+            f"Could not persist history {bookkeeping_data_map} for {processors} because input or output is missing",
+        )
+        return
 
-def persist_history_task(processors, bookkeeping_data_map):
-    try:
-        persist_history_task_internal(processors, bookkeeping_data_map)
-    except Exception as e:
-        logger.error(f"Error persisting history: {e}")
+    input, processor_runs, run_entry = _process_app_run_data(processors, bookkeeping_data_map)
+    run_entry.save()

llmstack/events/consumers/__init__.py

@@ -2,10 +2,9 @@
 from typing import Any
 
 from llmstack.apps.app_session_utils import save_app_session_data
-from llmstack.jobs.adhoc import HistoryPersistenceJob
+from llmstack.events.apis import EventsViewSet
 from llmstack.play.actor import Actor
 from llmstack.play.output_stream import Message, MessageType
-from llmstack.processors.tasks import persist_history_task
 
 logger = logging.getLogger(__name__)
 
@@ -81,16 +80,13 @@ def on_stop(self) -> None:
             ):
                 logger.info("Not persisting history since disable_history is set to True")
                 return super().on_stop()
-
-            HistoryPersistenceJob.create(
-                func=persist_history_task,
-                args=[
-                    list(
-                        self._processor_configs.keys(),
-                    ),
-                    self._bookkeeping_data_map,
-                ],
-            ).add_to_queue()
+            EventsViewSet().create(
+                "app.run.finished",
+                {
+                    "processors": list(self._processor_configs.keys()),
+                    "bookkeeping_data_map": self._bookkeeping_data_map,
+                },
+            )
         except Exception as e:
             logger.error(f"Error adding history persistence job: {e}")
         return super().on_stop()

llmstack/processors/tasks.py renamed to llmstack/events/consumers/app_run_finished.py

@@ -378,7 +378,6 @@
 INDEX_VIEW_MODULE = "llmstack.base.views"
 EMAIL_SENDER_CLASS = "llmstack.emails.sender.DefaultEmailSender"
 EMAIL_TEMPLATE_FACTORY_CLASS = "llmstack.emails.templates.factory.DefaultEmailTemplateFactory"
-HISTORY_STORE_CLASS = "llmstack.processors.history.DefaultHistoryStore"
 FLAG_SOURCES = ["llmstack.base.flags.FlagSource"]
 
 # Make sure name and slug are unique
@@ -573,6 +572,10 @@
     },
 }
 
+EVENT_TOPIC_MAPPING = {
+    "app.run.finished": ["llmstack.events.consumers.app_run_finished.persist_app_run_history"],
+}
+
 ANONYMOUS_USER_RATELIMIT = os.getenv("ANONYMOUS_USER_RATELIMIT", None)
 
 LIMITER_MODULE = "llmstack.server.limiter"