Analytics pipeline functions (#2355)

maheshsattala · mahesh_sattala · web-flow · commit 6008726a1531 · 2026-03-17T11:45:22.000+05:30
* added key_vaults in analytics_pipeline

* added key_vaults in analytics_pipeline

* added analytic_pipeline functions.

* added analytic_pipeline functions.

* added analytic_pipeline functions.

* added analytic_pipeline functions.

* added tests for coverage.

* added tests for coverage.

* added tests for coverage.

* added tests for coverage.

* added tests for coverage.

* added tests for coverage.

---------

Co-authored-by: mahesh_sattala &lt;mahesh.sattala@digite.com&gt;
diff --git a/kairon/events/definitions/analytic_pipeline_handler.py b/kairon/events/definitions/analytic_pipeline_handler.py
@@ -61,7 +61,7 @@ def execute(self, event_id: str, **kwargs):
                 "pipeline_name": pipeline_name,
                 "callback_name": callback_name,
                 "event_id": event_id,
-                "slot": {"bot": self.bot},
+                "slot": {"bot": self.bot, "user": self.user},
                 "key_vault": key_vault
             }
 
diff --git a/kairon/shared/cognition/data_objects.py b/kairon/shared/cognition/data_objects.py
@@ -146,4 +146,21 @@ def validate(self, clean=True):
 
     def clean(self):
         if self.collection_name:
-            self.collection_name = self.collection_name.strip().lower()
+            self.collection_name = self.collection_name.strip().lower()
+
+
+@auditlogger.log
+@push_notification.apply
+class EmbeddingMetadata(Auditlog):
+    collection_name = StringField(required=True)
+    bot = StringField(required=True)
+    vector_config = DictField()
+    user = StringField(required=True)
+    timestamp = DateTimeField(default=datetime.utcnow)
+    knowledge_vault_name = StringField(required=True)
+    model_id = StringField(required=True)
+
+    meta = {"indexes": [{"fields": ["bot", "collection_name"]}]}
+
+    def clean(self):
+        self.collection_name = self.collection_name.strip().lower()
diff --git a/kairon/shared/concurrency/actors/analytics_runner.py b/kairon/shared/concurrency/actors/analytics_runner.py
@@ -46,6 +46,7 @@ def execute(self, source_code: Text, predefined_objects: Optional[Dict] = None,
             raise AppException(f"Validation failed: {e}")
 
         bot = predefined_objects.get("slot", {}).get("bot")
+        user = predefined_objects.get("slot", {}).get("user")
 
         safe_objects = {
             "add_data": partial(PyscriptSharedUtility.add_data, bot=bot),
@@ -57,6 +58,9 @@ def execute(self, source_code: Text, predefined_objects: Optional[Dict] = None,
             "mark_as_processed": partial(CallbackScriptUtility.mark_as_processed, bot=bot),
             "update_data_analytics": partial(CallbackScriptUtility.update_data_analytics, bot=bot),
             "delete_data_analytics": partial(CallbackScriptUtility.delete_data_analytics, bot=bot),
+            "extract_data": partial(CallbackScriptUtility.extract_data, bot=bot, user=user),
+            "process_instruction": partial(CallbackScriptUtility.process_instruction, bot=bot, user=user),
+            "create_vector_collection": partial(CallbackScriptUtility.create_vector_collection, bot=bot),
             "srtp_time": PyscriptUtility.srtptime,
             "srtf_time": PyscriptUtility.srtftime,
             "url_parse": PyscriptUtility.url_parse_quote_plus,
diff --git a/kairon/shared/pyscript/analytics_worker.py b/kairon/shared/pyscript/analytics_worker.py
@@ -50,6 +50,9 @@ def main():
                 "mark_as_processed": partial(CallbackScriptUtility.mark_as_processed, bot=bot),
                 "update_data_analytics": partial(CallbackScriptUtility.update_data_analytics, bot=bot),
                 "delete_data_analytics": partial(CallbackScriptUtility.delete_data_analytics, bot=bot),
+                "extract_data": CallbackScriptUtility.extract_data,
+                "process_instruction": CallbackScriptUtility.process_instruction,
+                "create_vector_collection": CallbackScriptUtility.create_vector_collection,
                 "srtp_time": PyscriptUtility.srtptime,
                 "srtf_time": PyscriptUtility.srtftime,
                 "url_parse": PyscriptUtility.url_parse_quote_plus,
diff --git a/kairon/shared/pyscript/callback_pyscript_utils.py b/kairon/shared/pyscript/callback_pyscript_utils.py
@@ -467,4 +467,158 @@ def update_data_analytics(collection_id: str, user: str, payload: dict, bot: str
         return {
             "message": "Record updated!",
             "data": {"_id": collection_id}
-        }
+        }
+
+    @staticmethod
+    def extract_data(input_source: str,
+                     prompt: str = None,
+                     result_type: str="markdown",
+                     llm_type: str = "openrouter",
+                     high_res_ocr: bool = False,
+                     language: str = "en",
+                     bot: str = None,
+                     user: str = None):
+
+        import requests
+
+        llm_server_url = Utility.environment['llm']['url']
+
+        payload = {
+            "input_source": input_source,
+            "llama_parser_api_key": Utility.environment['llama_parse']['key'],
+            "result_type": result_type,
+            "high_res_ocr": high_res_ocr,
+            "language": language,
+            "parsing_instruction": prompt,
+            "user": user,
+            "llm_type": llm_type
+        }
+
+        response = requests.post(
+            f"{llm_server_url}/{bot}/parse/{llm_type}",
+            json=payload
+        )
+
+        if response.status_code != 200:
+            raise Exception(response.text)
+
+        response = response.json()
+
+        if not response.get("success"):
+            raise Exception(response)
+
+        result = response.get("data")
+
+        return {
+            "full_text": result.get("full_text"),
+            "extracted_data": result.get("extracted_data")
+        }
+
+
+    @staticmethod
+    def process_instruction(data_list, prompt, operation_type, model_id, llm_type: str = "openrouter",
+                            bot: str = None, user: str = None):
+        import requests
+        from kairon.shared.admin.data_objects import LLMSecret
+
+        doc = LLMSecret.objects(llm_type="openrouter").first()
+        api_key = Utility.decrypt_message(doc.api_key)
+
+        if operation_type == "embedding":
+
+            llm_server_url = Utility.environment['llm']['url']
+            payload = {
+                "text": data_list,
+                "user": user,
+                "kwargs": {
+                    "model": model_id,
+                    "api_key": api_key
+                }
+            }
+
+            response = requests.request(method="POST",
+                                        url=f"{llm_server_url}/{bot}/aembedding/{llm_type}",
+                                        json=payload)
+            response.raise_for_status()
+            response = response.json()
+            logger.info(response)
+
+            return {
+                "embeddings": response
+            }
+
+        else:
+            text_input = data_list[0]
+            final_prompt = prompt.format(document=text_input)
+            payload = {
+                "user": user,
+                "hyperparameters": {"temperature": 0, "model": model_id},
+                "messages": [{"role": "user", "content": final_prompt}]
+            }
+            llm_server_url = Utility.environment['llm']['url']
+            response = requests.request(method="POST",
+                                        url=f"{llm_server_url}/{bot}/completion/{llm_type}",
+                                        json=payload)
+
+            response.raise_for_status()
+            response = response.json()
+            extracted_data = response['formatted_response']
+
+            logger.info(response)
+            logger.info(extracted_data)
+
+            return extracted_data
+
+
+    @staticmethod
+    def create_vector_collection(collection_name, model_id: str, user: str, emb_size: int = 3072,
+                                 overwrite: bool = False, metadata: list = None, bot: str = None):
+        from kairon.shared.cognition.data_objects import CognitionSchema, EmbeddingMetadata, ColumnMetadata
+        from qdrant_client.models import VectorParams, Distance
+        from qdrant_client import QdrantClient
+
+        db_url = Utility.environment['vector']['db']
+        knowledge_vault_name = collection_name
+        collection_name = f"{bot}_{collection_name}_faq_embd"
+        schema = {
+            "metadata": metadata,
+            "collection_name": knowledge_vault_name
+        }
+
+        client = QdrantClient(url=db_url)
+
+        collections = client.get_collections().collections
+        exists = any(c.name == collection_name for c in collections)
+        embed_config = {
+            "size": emb_size,
+            "distance": Distance.COSINE
+        }
+        vector_config = VectorParams(**embed_config)
+        if exists and overwrite:
+            client.delete_collection(collection_name=collection_name)
+            exist = CognitionSchema.objects(bot=bot, collection_name=knowledge_vault_name).first()
+            if exist:
+                exist.delete()
+
+        if not exists or overwrite:
+            client.create_collection(
+                collection_name=collection_name,
+                vectors_config=vector_config
+            )
+            metadata_obj = CognitionSchema(bot=bot, user=user)
+            metadata_obj.metadata = [ColumnMetadata(**meta) for meta in schema.get("metadata") or []]
+            metadata_obj.collection_name = schema.get("collection_name")
+            metadata_obj.save()
+        else:
+            return {
+                "message": "collection already exists"
+            }
+
+        exist = EmbeddingMetadata.objects(bot=bot, collection_name=collection_name, model_id=model_id,
+                                          knowledge_vault_name=knowledge_vault_name).first()
+        if not exist:
+            EmbeddingMetadata(bot=bot, collection_name=collection_name, model_id=model_id,
+                              knowledge_vault_name=knowledge_vault_name, user=user, vector_config=embed_config).save()
+        return {
+            "message": "collection created successfully"
+        }
diff --git a/requirements/dev.txt b/requirements/dev.txt
@@ -12,4 +12,4 @@ deepdiff==7.0.1
 pytest-cov==5.0.0
 pytest-html==4.1.1
 pytest-aioresponses==0.3.0
-aioresponses==0.7.6
+aioresponses==0.7.6
diff --git a/requirements/prod.txt b/requirements/prod.txt
@@ -74,4 +74,5 @@ nltk
 blacksheep==2.0.7
 fastembed==0.5.1
 markdown-pdf==1.7
-genson==1.3.0
+genson==1.3.0
+qdrant-client==1.13.3
diff --git a/system.yaml b/system.yaml
@@ -227,6 +227,9 @@ llm:
   url: ${LLM_SERVER_URL:http://localhost}
   request_timeout: ${LLM_REQUEST_TIMEOUT:30}
 
+llama_parse:
+  key: ${LLAMA_PARSER_API_KEY:"test-key"}
+
 vector:
   db: ${VECTOR_DB:http://localhost:6333}
   key: ${VECTOR_DB_KEY}
diff --git a/tests/unit_test/callback/pyscript_handler_test.py b/tests/unit_test/callback/pyscript_handler_test.py

Original file line number	Diff line number	Diff line change
`@@ -61,7 +61,7 @@ def execute(self, event_id: str, **kwargs):`
`61`	`61`	`"pipeline_name": pipeline_name,`
`62`	`62`	`"callback_name": callback_name,`
`63`	`63`	`"event_id": event_id,`
`64`		`- "slot": {"bot": self.bot},`
	`64`	`+ "slot": {"bot": self.bot, "user": self.user},`
`65`	`65`	`"key_vault": key_vault`
`66`	`66`	`}`
`67`	`67`