Added redis streams listener

app.py

@@ -32,6 +32,122 @@ index_name = "alert_index"
 stream_name = "alerts"
 redis_port = 16652
 
+# embedding model
+embedding_args = {
+    "model_name" : "BAAI/bge-large-en-v1.5",
+    "model_kwargs" : {"device": "cpu"},
+    "encode_kwargs" : {"normalize_embeddings": True}
+}
+embedding_model = HuggingFaceEmbeddings(**embedding_args)
+
+# Mongo Connection
+connection = pymongo.MongoClient(os.environ["MONGO_URI"])
+alert_collection = connection[database][collection]
+
+# redis connection
+r = redis.Redis(host=os.environ['REDIS_HOST'], password=os.environ['REDIS_PWD'], port=port)
+
+Sure, here's the entire code with the Streamlit app and the Redis stream listener combined:
+pythonCopy codeimport streamlit as st
+import os
+from collections.abc import Collection
+from langchain.memory import ChatMessageHistory
+from langchain_community.chat_message_histories import (
+    StreamlitChatMessageHistory,
+)
+from langchain_groq import ChatGroq
+from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
+from langchain_community.vectorstores import MongoDBAtlasVectorSearch
+from langchain_community.embeddings import HuggingFaceEmbeddings
+from langchain.chains import create_history_aware_retriever, create_retrieval_chain
+from langchain.chains.combine_documents import create_stuff_documents_chain
+from langchain.output_parsers import ResponseSchema, StructuredOutputParser
+from langchain_core.runnables.history import RunnableWithMessageHistory
+from langchain_core.chat_history import BaseChatMessageHistory
+from langchain.chains import RetrievalQA
+import nest_asyncio
+nest_asyncio.apply()
+
+import pymongo
+import logging
+import nest_asyncio
+from langchain.docstore.document import Document
+import redis
+import threading
+
+# Config
+nest_asyncio.apply()
+logging.basicConfig(level=logging.INFO)
+database = "AlertSimAndRemediation"
+collection = "alert_embed"
+stream_name = "alerts"
+
+# Embedding model
+embedding_args = {
+    "model_name": "BAAI/bge-large-en-v1.5",
+    "model_kwargs": {"device": "cpu"},
+    "encode_kwargs": {"normalize_embeddings": True}
+}
+embedding_model = HuggingFaceEmbeddings(**embedding_args)
+
+# Mongo Connection
+connection = pymongo.MongoClient(os.environ["MONGO_URI"])
+alert_collection = connection[database][collection]
+
+# Redis connection
+r = redis.Redis(host=os.environ['REDIS_HOST'], password=os.environ['REDIS_PWD'], port=16652)
+
+# Preprocessing
+async def create_textual_description(entry_data):
+    entry_dict = {k.decode(): v.decode() for k, v in entry_data.items()}
+    category = entry_dict["Category"]
+    created_at = entry_dict["CreatedAt"]
+    acknowledged = "Acknowledged" if entry_dict["Acknowledged"] == "1" else "Not Acknowledged"
+    remedy = entry_dict["Remedy"]
+    severity = entry_dict["Severity"]
+    source = entry_dict["Source"]
+    node = entry_dict["node"]
+    description = f"A {severity} alert of category {category} was raised from the {source} source for node {node} at {created_at}. The alert is {acknowledged}. The recommended remedy is: {remedy}."
+    return description, entry_dict
+
+# Saving alert doc
+async def save(entry):
+    vector_search = MongoDBAtlasVectorSearch.from_documents(
+        documents=[Document(
+            page_content=entry["content"],
+            metadata=entry["metadata"]
+        )],
+        embedding=embedding_model,
+        collection=alert_collection,
+        index_name="alert_index",
+    )
+    logging.info("Alerts stored successfully!")
+
+# Listening to alert stream
+async def listen_to_alerts(r):
+    logging.info("Listening to alerts...")
+    try:
+        last_id = '$'
+        while True:
+            entries = r.xread({stream_name: last_id}, block=0, count=None)
+            if entries:
+                stream, new_entries = entries[0]
+                for entry_id, entry_data in new_entries:
+                    description, entry_dict = await create_textual_description(entry_data)
+                    await save({"content": description, "metadata": entry_dict})
+                    st.toast(description, icon='🔔')
+                    # Update the last ID read
+                    last_id = entry_id
+    except KeyboardInterrupt:
+        print("Exiting...")
+
+# Start Redis listener in a separate thread
+def start_redis_listener():
+    try:
+        nest_asyncio.run(listen_to_alerts(r))
+    except Exception as e:
+        print(f"Error in Redis listener: {e}")
+
 # Streamlit Application
 st.set_page_config(
     page_title="ASMR Query Bot 🔔",
@@ -45,13 +161,9 @@ st.set_page_config(
 
 st.title('ASMR Query Bot 🔔')
 
-# embedding model
-embedding_args = {
-    "model_name" : "BAAI/bge-large-en-v1.5",
-    "model_kwargs" : {"device": "cpu"},
-    "encode_kwargs" : {"normalize_embeddings": True}
-}
-embedding_model = HuggingFaceEmbeddings(**embedding_args)
+# Start Redis listener in a separate thread
+redis_listener_thread = threading.Thread(target=start_redis_listener)
+redis_listener_thread.start()
 
 # vector search
 vector_search = MongoDBAtlasVectorSearch.from_connection_string(