Merge pull request #27 from Prismadic/26-featperf-hash-entries

🧲 feat(perf): hashing headers for NATS

Author: mxchinegod

examples/distributed_charge/1_process.ipynb

@@ -2,26 +2,26 @@
  "cells": [
   {
    "cell_type": "code",
-   "execution_count": 1,
+   "execution_count": 4,
    "metadata": {},
    "outputs": [],
    "source": [
     "from magnet.ize.filings import Processor\n",
-    "source_data_file = \"./raw/kb_export_clean.csv\"\n",
-    "export_data_file = \"./data/filings_mistral_nlp.parquet\""
+    "source_data_file = \"../../raw/kb_export_clean.csv\"\n",
+    "export_data_file = \"../../data/filings_mistral_nlp.parquet\""
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 3,
+   "execution_count": 5,
    "metadata": {},
    "outputs": [
     {
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "\u001b[96m☕️ WAIT: loading - ./raw/kb_export_clean.csv\u001b[0m\n",
-      "\u001b[92m🌊 SUCCESS: loaded - ./raw/kb_export_clean.csv\u001b[0m\n"
+      "\u001b[96m☕️ WAIT: loading - ../raw/kb_export_clean.csv\u001b[0m\n",
+      "\u001b[91m☠️ FATAL: [Errno 2] No such file or directory: '../raw/kb_export_clean.csv'\u001b[0m\n"
      ]
     }
    ],
@@ -36,30 +36,19 @@
   },
   {
    "cell_type": "code",
-   "execution_count": 4,
+   "execution_count": 3,
    "metadata": {},
    "outputs": [
     {
-     "name": "stdout",
-     "output_type": "stream",
-     "text": [
-      "\u001b[96m☕️ WAIT: get coffee or tea - 65822 processing...\u001b[0m\n"
-     ]
-    },
-    {
-     "name": "stderr",
-     "output_type": "stream",
-     "text": [
-      "100%|██████████| 65822/65822 [20:08<00:00, 54.48it/s]   \n",
-      "100%|██████████| 65822/65822 [00:02<00:00, 29005.30it/s]\n"
-     ]
-    },
-    {
-     "name": "stdout",
-     "output_type": "stream",
-     "text": [
-      "\u001b[96m☕️ WAIT: saving to ./data/filings_mistral_nlp.parquet\u001b[0m\n",
-      "\u001b[92m🌊 SUCCESS: saved - ./data/filings_mistral_nlp.parquet\u001b[0m\n"
+     "ename": "AttributeError",
+     "evalue": "'NoneType' object has no attribute 'dropna'",
+     "output_type": "error",
+     "traceback": [
+      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
+      "\u001b[0;31mAttributeError\u001b[0m                            Traceback (most recent call last)",
+      "Cell \u001b[0;32mIn[3], line 1\u001b[0m\n\u001b[0;32m----> 1\u001b[0m \u001b[38;5;28;01mawait\u001b[39;00m clustered_filings\u001b[38;5;241m.\u001b[39mprocess(\n\u001b[1;32m      2\u001b[0m     export_data_file\n\u001b[1;32m      3\u001b[0m     , nlp\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mTrue\u001b[39;00m\n\u001b[1;32m      4\u001b[0m )\n",
+      "File \u001b[0;32m/opt/homebrew/lib/python3.11/site-packages/magnet/ize/filings.py:55\u001b[0m, in \u001b[0;36mProcessor.process\u001b[0;34m(self, path, splitter, nlp)\u001b[0m\n\u001b[1;32m     54\u001b[0m \u001b[38;5;28;01masync\u001b[39;00m \u001b[38;5;28;01mdef\u001b[39;00m \u001b[38;5;21mprocess\u001b[39m(\u001b[38;5;28mself\u001b[39m, path: \u001b[38;5;28mstr\u001b[39m \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m, splitter: \u001b[38;5;28many\u001b[39m \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;01mNone\u001b[39;00m, nlp\u001b[38;5;241m=\u001b[39m\u001b[38;5;28;01mTrue\u001b[39;00m):\n\u001b[0;32m---> 55\u001b[0m     \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mdf \u001b[38;5;241m=\u001b[39m \u001b[38;5;28;43mself\u001b[39;49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mdf\u001b[49m\u001b[38;5;241;43m.\u001b[39;49m\u001b[43mdropna\u001b[49m()\n\u001b[1;32m     56\u001b[0m     \u001b[38;5;28;01mif\u001b[39;00m \u001b[38;5;28mself\u001b[39m\u001b[38;5;241m.\u001b[39mdf \u001b[38;5;129;01mis\u001b[39;00m \u001b[38;5;129;01mnot\u001b[39;00m \u001b[38;5;28;01mNone\u001b[39;00m:\n\u001b[1;32m     57\u001b[0m         \u001b[38;5;28;01mtry\u001b[39;00m:\n",
+      "\u001b[0;31mAttributeError\u001b[0m: 'NoneType' object has no attribute 'dropna'"
      ]
     }
    ],

examples/distributed_charge/2_charge.ipynb

@@ -7,7 +7,7 @@
    "outputs": [],
    "source": [
     "from magnet.ize.filings import Processor\n",
-    "export_data_file = \"./data/filings_mistral_nlp.parquet\""
+    "export_data_file = \"../../data/filings_mistral_nlp.parquet\""
    ]
   },
   {

magnet/ic/field.py

@@ -4,6 +4,9 @@
 from nats.errors import TimeoutError
 from magnet.utils.data_classes import *
 from nats.js.api import StreamConfig
+import xxhash
+
+x = xxhash.xxh64()
 
 class Charge:
     """
@@ -80,6 +83,14 @@ async def pulse(self, payload):
         except Exception as e:
             _f('fatal', f'invalid JSON\n{e}')
         try:
+            _hash = x(bytes_).hexdigest()
+            await self.js.publish(
+                self.category
+                , bytes_
+                , headers={
+                    "Nats-Msg-Id": _hash
+                }
+            )
             await self.js.publish(self.category, bytes_, headers={"Nats-Msg-Id":})
         except Exception as e:
             _f('fatal', f'could not send data to {self.server}\n{e}')

magnet/utils/data_classes.py

@@ -24,7 +24,6 @@ class GeneratedPayload:
         result (str): The result generated by the system.
         model (str): The model used to generate the payload.
     """
-
     query: str
     prompt: str
     context: list
@@ -42,7 +41,6 @@ class EmbeddingPayload:
         text (list): The text of the data.
         model (str): The model used for embedding the text data.
     """
-
     document: str
     embedding: list
     text: list