Commit

app.py

@@ -1,31 +1,76 @@
+import re, uuid
+import json, logging
 import gradio as gr
-from transformers import pipeline
-
-# Initialize the model pipeline only once
-text_generator = pipeline("text-generation", model="mistralai/Mixtral-8x7B-Instruct-v0.1", device_map="auto")
-
-class Inference:
-    def __init__(self, pipe):
-        self.pipe = pipe
-
-    def get_results(self, prompt, max_new_tokens=256, temperature=0.7):
-        output = self.pipe(prompt, max_new_tokens=max_new_tokens, temperature=temperature)
-        return output[0]["generated_text"]
-
-inference = Inference(text_generator)
-
-def generate_text(prompt, max_tokens, temperature):
-    return inference.get_results(prompt, max_tokens, temperature)
-
-# Create a simple Gradio interface
-gr.Interface(
-    fn=generate_text,
-    inputs=[
-        gr.Textbox(lines=4, label="Prompt"),
-        gr.Slider(minimum=10, maximum=1024, value=256, label="Max New Tokens"),
-        gr.Slider(minimum=0.1, maximum=1.5, value=0.7, step=0.1, label="Temperature"),
-    ],
-    outputs=gr.Textbox(label="Generated Text"),
-    title="Mixtral-8x7B Text Generator",
-    description="Enter a prompt to generate text using Mixtral-8x7B-Instruct."
-).launch()
+from typing import List, Dict
+
+from data.css import custom_css
+from src.controller.agent_cacher import AgentManager
+# Configure logging
+logging.basicConfig(level=logging.INFO, format="-->%(asctime)s [%(levelname)s] %(message)s")
+
+
+
+
+def create_gradio_interface():
+    gr.HTML('<link href="https://fonts.googleapis.com/css2?family=Fira+Code&family=Roboto&display=swap" rel="stylesheet">')
+    gr.HTML(custom_css)
+    manager = AgentManager()
+    
+    with gr.Blocks(title="AI Learning Assistant") as demo:
+        gr.Markdown("# 🧠 Learn with LlamaIndex Tools")
+        logging.info("Starting Learning Application")
+
+        with gr.Row():
+            session_id = gr.Textbox(label="Session ID", value=str(uuid.uuid4()), visible=True)
+            llm_selector = gr.Dropdown(
+                label="LLM Type",
+                choices=["Google", "OpenAI", "HuggingFace", "Mistral"],
+                value="Google",
+                interactive=True
+            )
+
+        with gr.Row():
+            with gr.Column(scale=3):
+                chatbot = gr.Chatbot(
+                    label="Learning Dialog",
+                    height=500,
+                    type="messages"
+                )
+                query_input = gr.Textbox(label="Your Learning Query", placeholder="Ask about ML/DL algorithms...")
+                submit_btn = gr.Button("Submit")
+
+            with gr.Column(scale=1):
+                gr.Markdown("### Tools Preview")
+                tool_output = gr.Textbox(label="Selected Tools", interactive=False)
+                response_output = gr.Textbox(label="Full Response", interactive=False, lines=10)
+
+        def process_query(session_id_val, query, chat_history, llm_val):
+            print("Session:", session_id)
+            print("Query:", query)
+            #print("Chat History:", chat_history)
+            print("Option selected:", llm_val)
+            agent = manager.get_agent(session_id_val, llm_type=llm_val)
+            chat_history, tools_used, response = agent.process_query(query)
+            manager.save_agent(session_id_val, agent)
+            return chat_history, tools_used, response, ""
+
+        submit_btn.click(
+            process_query,
+            inputs=[session_id, query_input, chatbot, llm_selector],  # ✅ 4 inputs
+            outputs=[chatbot, tool_output, response_output, query_input]
+        )
+
+        def load_history(session_id_val, llm_val):
+            agent = manager.get_agent(session_id_val, llm_val)
+            return agent.chat_history
+
+        session_id.change(
+            load_history,
+            inputs=[session_id, llm_selector],  # ✅ Pass component objects, not .value
+            outputs=chatbot
+        )
+
+    return demo
+if __name__ == "__main__":
+    interface = create_gradio_interface()
+    interface.launch()

app__.py

@@ -0,0 +1,59 @@
+# --- Gradio App ---
+import uuid
+import gradio as gr
+from src.controller.agent import ChatBot
+
+
+chatbot = ChatBot()
+session_store = {}
+session = ""
+
+def build_request(user_input):
+    return {
+        "query": user_input,
+        "metadata": {
+            "user_id": "123",
+            "timestamp": "2025-04-20T12:00:00"
+        }
+    }
+
+def chat(session_id, user_input, history):
+    if session_id not in session_store:
+        session_store[session_id] = []
+
+    if user_input.strip():
+        request = build_request(user_input)
+        agent = chatbot.agent_cacher.get_agent_for(session_id, request)
+        reply = agent.agentic_chat(user_input)
+
+        session_store[session_id].append(("You", user_input))
+        session_store[session_id].append(("Agent", reply))
+
+    updated_history = [(sender, message) for sender, message in session_store[session_id]]
+    return reply, ""
+
+with gr.Blocks() as demo:
+    gr.Markdown("# 🧠 Chat with LlamaIndex Agent")
+
+    with gr.Row():
+        if not session:
+            session = uuid.uuid4()
+        session_id = gr.Textbox(label="Session ID", value=session)
+
+    with gr.Row():
+        chatbot_ui = gr.Chatbot(label="Chat History", height=400)
+
+    with gr.Row():
+        user_input = gr.Textbox(label="Your Query", placeholder="Enter your query here...")
+
+    with gr.Row():
+        send_btn = gr.Button("Send")
+
+    send_btn.click(
+        chat,
+        inputs=[session_id, user_input, chatbot_ui],
+        outputs=[chatbot_ui, user_input],
+    )
+
+if __name__ == "__main__":
+    demo.launch()

data/coverage_doc.txt

@@ -0,0 +1,59 @@
+Plan Name: Elevate Health PPO 2025
+Type: Preferred Provider Organization (PPO)
+
+Summary:
+The Elevate Health PPO Plan offers flexibility in choosing healthcare providers. Members can visit any doctor or specialist without a referral, but in-network providers offer lower out-of-pocket costs.
+
+Key Benefits:
+- Annual deductible: $1,500 individual / $3,000 family
+- Out-of-pocket maximum: $6,000 individual / $12,000 family
+- Primary care visits: $25 copay in-network
+- Specialist visits: $50 copay in-network
+- Emergency room: $250 copay
+- Generic prescriptions: $10 copay
+- Brand name prescriptions: $40 copay
+
+Wellness Programs:
+- Free annual physical exams
+- Telehealth services with no copay
+- Mental health counseling (up to 10 visits/year covered)
+
+Covered Services:
+- Preventive care (100% in-network)
+- Maternity and newborn care
+- Surgery and hospitalization
+- Lab tests and X-rays
+
+Exclusions:
+- Cosmetic procedures
+- Fertility treatments
+- Long-term custodial care
+
+Plan Name: CoreCare HMO 2025
+Type: Health Maintenance Organization (HMO)
+
+Summary:
+CoreCare HMO Plan requires members to choose a Primary Care Physician (PCP) and get referrals for specialist care. This plan offers coordinated care at lower premiums.
+
+Key Benefits:
+- Annual deductible: $500 individual / $1,000 family
+- Out-of-pocket maximum: $4,000 individual / $8,000 family
+- Primary care visits: $15 copay
+- Specialist visits: $30 copay (with referral)
+- Emergency room: $200 copay
+- Generic prescriptions: $5 copay
+- Brand name prescriptions: $25 copay
+
+Covered Services:
+- Preventive and routine care
+- Prenatal and maternity services
+- Behavioral health support
+- Diagnostic imaging and lab services
+
+Exclusions:
+- Out-of-network care (except emergencies)
+- Cosmetic procedures
+- Alternative therapies (e.g., acupuncture)
+
+Contact Information:
+For questions about coverage, contact Member Services at 1-800-555-1234.

data/css.py

@@ -0,0 +1,17 @@
+custom_css = """
+<style>
+body, .gradio-container {
+    font-family: 'Roboto', sans-serif;
+}
+
+h1, h2, h3, label {
+    font-family: 'Fira Code', monospace;
+    color: #2c3e50;
+}
+
+textarea, input, select, button {
+    font-family: 'Courier New', monospace;
+    font-size: 15px;
+}
+</style>
+"""

src/agent/agenttools.py

@@ -0,0 +1,211 @@
+from llama_index.core.tools import FunctionTool
+from src.setup.utils import retry
+from src.controller.customlogger import logging
+from src.llm.source_llm import LLMCall
+
+
+class ChatBotFunctionTools:
+    def __init__(self, llm_type="google"):
+        self.generator = LLMCall(llm_type).get_llm()
+
+    @retry(max_retries=5, delay=1)
+    def machine_learning_concept(self, query):
+        logging.info(f"Tool Call: machine_learning_concept('{query}')")
+        prompt = (
+            "You are a Machine Learning teacher.\n"
+            "Explain the following ML concept in:\n"
+            "The explaination should include geometrical or mathematical intuition"
+            "Try to keep answer crisp and compact"
+            f"User Query: {query}"
+        )
+        return self.generator.complete(prompt)
+
+    @retry(max_retries=5, delay=1)
+    def math_concept(self, query):
+        logging.info(f"Tool Call: math_concept('{query}')")
+        prompt = (
+            "You are a Math teacher.\n"
+            "Explain the following mathematics behind the Machine Learning algorithm in details with each step by step :\n"
+            "Try to keep answer crisp and compact"
+            f"User Query: {query}"
+        )
+        return self.generator.complete(prompt)
+
+    @retry(max_retries=5, delay=1)
+    def deep_learning_architecture(self, arch):
+        logging.info(f"Tool Call: deep_learning_architecture('{arch}')")
+        prompt = f"Explain the {arch} neural network architecture with diagram description"
+        return self.generator.complete(prompt)
+
+    @retry(max_retries=5, delay=1)
+    def visualize_algorithm(self, algo):
+        logging.info(f"Tool Call: visualize_algorithm('{algo}')")
+        prompt = f"Create visualization code that demonstrates how {algo} works"
+        return self.generator.complete(prompt)
+
+    @retry(max_retries=5, delay=1)
+    def concept_combiner(self, concepts):
+        logging.info(f"Tool Call: concept_combiner('{concepts}')")
+        prompt = f"Explain the relationship between these concepts: {', '.join(concepts)}"
+        return self.generator.complete(prompt)
+
+    @retry(max_retries=5, delay=1)
+    def llm_query(self, concepts):
+        logging.info(f"Tool Call: llm_query('{concepts}')")
+        prompt = f"You are an Expert to Answer the following question respond to the best of your knowledge: {', '.join(concepts)}"
+        return self.generator.complete(prompt)
+
+    @retry(max_retries=5, delay=1)
+    def get_tools(self):
+        return {
+            "ml_concept": FunctionTool.from_defaults(fn=self.machine_learning_concept),
+            "dl_architecture": FunctionTool.from_defaults(fn=self.deep_learning_architecture),
+            "algo_visualizer": FunctionTool.from_defaults(fn=self.visualize_algorithm),
+            "concept_combiner": FunctionTool.from_defaults(fn=self.concept_combiner),
+            "math_concept": FunctionTool.from_defaults(fn=self.math_concept),
+            "llm_query": FunctionTool.from_defaults(fn=self.llm_query)
+        }
+
+
+
+# class LearningAgent:
+#     """Core agent that orchestrates tool usage for learning system"""
+#
+#     def __init__(self, llm_value):
+#         self.llm = LLMCall(llm_type=llm_value).get_llm()
+#         self.tools = self._setup_tools()
+#         self.chat_history: List[Dict[str, str]] = []  # Stores properly formatted messages
+#         self.max_history = 20
+#
+#     def _setup_tools(self) -> Dict[str, FunctionTool]:
+#         """Initialize all learning tools"""
+#         return {
+#             **self._setup_ml_tools(),
+#             **self._setup_dl_tools(),
+#             **self._setup_graph_tools(),
+#             **self._setup_utility_tools()
+#         }
+#
+#     def _setup_ml_tools(self) -> Dict[str, FunctionTool]:
+#         """Machine Learning tools"""
+#
+#         def ml_concept_explainer(query: str) -> str:
+#             prompt = f"Explain this ML concept in simple terms with examples: {query}"
+#             return self.llm.complete(prompt).text
+#
+#         return {
+#             "ml_concept": FunctionTool.from_defaults(fn=ml_concept_explainer)
+#         }
+#
+#     def _setup_dl_tools(self) -> Dict[str, FunctionTool]:
+#         """Deep Learning tools"""
+#
+#         def dl_architecture(arch: str) -> str:
+#             prompt = f"Explain the {arch} neural network architecture with diagram description"
+#             return self.llm.complete(prompt).text
+#
+#         return {
+#             "dl_architecture": FunctionTool.from_defaults(fn=dl_architecture)
+#         }
+#
+#     def _setup_graph_tools(self) -> Dict[str, FunctionTool]:
+#         """Graph/Visualization tools"""
+#
+#         def visualize_algorithm(algo: str) -> str:
+#             prompt = f"Create visualization code that demonstrates how {algo} works"
+#             return self.llm.complete(prompt).text
+#
+#         return {
+#             "algo_visualizer": FunctionTool.from_defaults(fn=visualize_algorithm)
+#         }
+#
+#     def _setup_utility_tools(self) -> Dict[str, FunctionTool]:
+#         """Utility tools"""
+#
+#         def concept_combiner(concepts: List[str]) -> str:
+#             prompt = f"Explain the relationship between these concepts: {', '.join(concepts)}"
+#             return self.llm.complete(prompt).text
+#
+#         return {
+#             "concept_combiner": FunctionTool.from_defaults(fn=concept_combiner)
+#         }
+#
+#     def extract_json_from_markdown(self, markdown_text):
+#         """
+#         Extract and parse JSON content from a markdown-style code block.
+#         Handles formats like ```json ... ```
+#         """
+#         try:
+#             # Extract JSON block using regex
+#             match = re.search(r"```json\s*(\{.*?\})\s*```", markdown_text, re.DOTALL)
+#             if not match:
+#                 raise ValueError("No JSON block found in markdown")
+#
+#             json_str = match.group(1)
+#             return json.loads(json_str)
+#         except Exception as e:
+#             print(f"[ERROR] Could not parse JSON: {e}")
+#             return None
+#
+#     def determine_tools(self, query: str):
+#         """Decide which tools to use based on query"""
+#         previous_questions = ""
+#         if self.chat_history:
+#             previous_questions = "\n".join([msg["content"] for msg in self.chat_history if msg["role"] == "user"][:-1]) \
+#                 if self.chat_history else "No previous questions"
+#         prompt = f"""Analyze this learning query and select appropriate tools also form the condensed query based on
+#                     previous_questions and Query:
+#                     Query: {query}
+#                     Previous Query: {previous_questions}
+#                     Available Tools: {list(self.tools.keys())}
+#                     Return dictionary of tool names and condensed query as dictionary in the below format:
+#                             ```json
+#                             {{
+#                               "condensed_query": condensed query considering chat history and user query as string,
+#                               "tool_names": tool names as comma-separated list
+#                             }}
+#                             ```
+#                     Do Not add additional text"""
+#
+#         response = self.llm.complete(prompt).text
+#         response = self.extract_json_from_markdown(response)
+#         return [t.strip() for t in response['tool_names'].split(",") if t.strip() in self.tools], response["condensed_query"]
+#
+#     def execute_tools(self, tools: List[str], query: str) -> tuple[str, str]:
+#         """Execute multiple tools and combine results"""
+#         tool_results = []
+#         content_results = []
+#
+#         for tool in tools:
+#             try:
+#                 tool_output = self.tools[tool](query)
+#                 content = tool_output.content if isinstance(tool_output, ToolOutput) else str(tool_output)
+#                 tool_results.append(tool)
+#                 content_results.append(content)
+#             except Exception as e:
+#                 logging.error(f"Tool {tool} failed: {str(e)}")
+#                 tool_results.append(tool)
+#                 content_results.append(f"Error: {str(e)}")
+#
+#         if len(tool_results) > 1:
+#             combined = "\n\n".join(f"**{t}**:\n{c}" for t, c in zip(tool_results, content_results))
+#             explanation = self.tools["concept_combiner"](content_results)
+#             return "multiple", f"{combined}\n\n**Combined Analysis**:\n{explanation}"
+#         return tool_results[0], content_results[0]
+#
+#     def process_query(self, query: str) -> tuple[List[Dict[str, str]], str, str]:
+#         """Process query and return properly formatted messages"""
+#         tools, condensed_query = self.determine_tools(query)
+#         logging.info(f"Selected tools: {tools}")
+#
+#         tool_used, response = self.execute_tools(tools, condensed_query)
+#
+#         # Format messages for Gradio Chatbot
+#         user_msg = {"role": "user", "content": query.title()}
+#         assistant_msg = {"role": "assistant", "content": response}
+#
+#         self.chat_history.extend([user_msg, assistant_msg])
+#         if len(self.chat_history) > self.max_history * 2:  # *2 for user+assistant pairs
+#             self.chat_history = self.chat_history[-(self.max_history * 2):]
+#
+#         return self.chat_history, tool_used, response

src/agent/custom_agent.py

@@ -0,0 +1,86 @@
+import re, ast
+import logging
+
+import re
+import json
+import logging
+from typing import List, Dict
+
+from llama_index.core.tools.types import ToolOutput
+from src.agent.agenttools import ChatBotFunctionTools
+
+
+class LearningAgent:
+    def __init__(self, llm_value: str):
+        tool_builder = ChatBotFunctionTools(llm_type=llm_value)
+        self.llm = tool_builder.generator
+        self.tools = tool_builder.get_tools()
+        self.chat_history: List[Dict[str, str]] = []
+        self.max_history = 20
+
+    def extract_json_from_markdown(self, markdown_text: str):
+        try:
+            match = re.search(r"```json\s*(\{.*?\})\s*```", markdown_text, re.DOTALL)
+            if not match:
+                raise ValueError("No JSON block found in markdown")
+            return json.loads(match.group(1))
+        except Exception as e:
+            logging.error(f"Could not parse JSON: {e}")
+            return None
+
+    def determine_tools(self, query: str):
+        previous_questions = "\n".join(
+            [msg["content"] for msg in self.chat_history if msg["role"] == "user"]
+        ) if self.chat_history else "No previous questions"
+
+        prompt = f"""Analyze this learning query and select appropriate tools also form the condensed query based on
+                    previous_questions and Query:
+                    For Choosing tool properly analyze the condensed query and then decide. Choose multiple if its necessary based on condensed query
+                    Its mandatory to select to atleast 1 tool
+                    Query: {query}
+                    Previous Query: {previous_questions}
+                    Available Tools: {list(self.tools.keys())}
+                    Return dictionary of tool names and condensed query as dictionary in the below format:
+                            ```json
+                            {{
+                              "condensed_query": condensed query considering chat history and user query as string,
+                              "tool_names": tool names as comma-separated list
+                            }}
+                            ```
+                 Do Not add additional text"""
+        response = self.llm.complete(prompt)
+        try:
+            response = response.text
+        except:
+            response = response
+        parsed = self.extract_json_from_markdown(response)
+        condensed_query = parsed["condensed_query"]
+        tools = [t.strip() for t in parsed['tool_names'].split(",") if t.strip() in self.tools]
+        return tools if tools else ['llm_query'], condensed_query
+
+    def execute_tools(self, tools: List[str], query: str) -> tuple[str, str]:
+        tool_results, content_results = [], []
+        for tool in tools:
+            try:
+                tool_output = self.tools[tool](query)
+                content = tool_output.content if isinstance(tool_output, ToolOutput) else str(tool_output)
+                tool_results.append(tool)
+                content_results.append(content)
+            except Exception as e:
+                logging.error(f"Tool {tool} failed: {str(e)}")
+                tool_results.append(tool)
+                content_results.append(f"Error: {str(e)}")
+
+        if len(tool_results) > 1:
+            combined = "\n\n".join(f"**{t}**:\n{c}" for t, c in zip(tool_results, content_results))
+            explanation = self.tools["concept_combiner"](content_results)
+            return "multiple", f"{combined}\n\n**Combined Analysis**:\n{explanation}"
+        return tool_results[0], content_results[0]
+
+    def process_query(self, query: str) -> tuple[List[Dict[str, str]], str, str]:
+        tools, condensed_query = self.determine_tools(query)
+        tool_used, response = self.execute_tools(tools, condensed_query)
+        self.chat_history += [{"role": "user", "content": query}, {"role": "assistant", "content": response}]
+        self.chat_history = self.chat_history[-(self.max_history * 2):]
+        return self.chat_history, tool_used, response
+

src/controller/agent_cacher.py

@@ -0,0 +1,26 @@
+
+from src.setup.cache import RedisDataSource
+from src.agent.custom_agent import LearningAgent
+
+
+class AgentManager:
+    def __init__(self):
+        self.cache = RedisDataSource()
+
+    def get_agent(self, session_id, llm_type):
+        cached = self.cache.read(f"agent:{session_id}")
+        agent = LearningAgent(llm_type)
+        if cached and "chat_history" in cached:
+            agent.chat_history = [
+                msg for msg in cached["chat_history"]
+                if isinstance(msg, dict) and "role" in msg and "content" in msg
+            ]
+        return agent
+
+    def save_agent(self, session_id, agent):
+        self.cache.write(f"agent:{session_id}", {
+            "chat_history": [
+                {"role": msg["role"], "content": msg["content"]}
+                for msg in agent.chat_history
+            ]
+        })

src/controller/customlogger.py

@@ -0,0 +1,2 @@
+import logging
+logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

src/controller/utils.py

@@ -0,0 +1 @@
+SYSTEM_PROMPT = "You are helpful, intelligent, and reroute queries efficiently."

src/llm/source_llm.py

@@ -0,0 +1,74 @@
+import os
+from src.controller.customlogger import logging
+from llama_index.llms.google_genai import GoogleGenAI
+from llama_index.llms.openai import OpenAI
+from llama_index.llms.huggingface import HuggingFaceLLM
+from mistralai import Mistral
+from llama_index.core.llms import ChatMessage
+from dotenv import load_dotenv
+load_dotenv()
+
+
+class MistralWrapper:
+    def __init__(self):
+        self.client = Mistral(api_key=os.environ["MISTRAL_API_KEY"])
+        self.model = "mistral-large-latest"
+
+    def complete(self, messages, **kwargs):
+        chat_response = self.client.chat.complete(
+                        model= self.model,
+                        messages = [{"role":"user", "content":messages}])
+        return chat_response.choices[0].message.content
+
+## --- LLMCall ---
+class LLMCall:
+    def __init__(self, llm_type="OpenAi", config=None):
+        self.llm_type = llm_type
+        self.config = config or {}
+        self.api_key_google = os.getenv("GOOGLE_API_KEY")
+        self.client = self.get_llm()
+
+    def get_llm(self):
+        if self.llm_type == "OpenAi":
+            logging.info("Initializing OpenAI LLM")
+            return self._get_openai_llm()
+        elif self.llm_type == "Google":
+            logging.info("Initializing Google Gemini LLM")
+            return self._get_google_llm()
+        elif self.llm_type == "HuggingFace":
+            logging.info("Initializing HuggingFace LLM")
+            return self._get_huggingface_llm()
+        elif self.llm_type == "Mistral":
+            logging.info("Initializing HuggingFace LLM")
+            return self._get_mistral()
+        else:
+            raise ValueError(f"Unsupported LLM type: {self.llm_type}")
+
+    def _get_openai_llm(self):
+        return OpenAI(
+            model=self.config.get("model", "gpt-3.5-turbo-0613"),
+            api_key=self.config.get("api_key"),
+            temperature=self.config.get("temperature", 0.7),
+            max_tokens=self.config.get("max_tokens", 1024)
+        )
+
+    def _get_google_llm(self):
+        return GoogleGenAI(
+            model="models/gemini-1.5-flash",
+        )
+    
+    def _get_mistral(self):
+        return MistralWrapper()
+
+
+    def _get_huggingface_llm(self):
+        return HuggingFaceLLM(
+            model_name="HuggingFaceH4/zephyr-7b-beta",
+            tokenizer_name="HuggingFaceH4/zephyr-7b-beta",
+            context_window=self.config.get("context_window", 2048),
+            max_new_tokens=self.config.get("max_new_tokens", 256)
+        )
+    
+    
+    def get_client(self):
+        return self.client

src/setup/cache.py

@@ -0,0 +1,32 @@
+import os, redis
+from dotenv import load_dotenv
+from src.controller.customlogger import logging
+
+load_dotenv()
+
+
+class RedisDataSource:
+    def __init__(self, host='localhost', port=6379, db=0):
+        self.host = os.environ.get("REDIS_HOST")
+        self.username = os.environ.get("REDIS_USERNAME")
+        self.password = os.environ.get("REDIS_PASSWORD")
+        self.port = os.environ.get("REDIS_PORT")
+        self.client = redis.Redis(host=self.host,
+                                  username=self.username,
+                                  password=self.password,
+                                  port=self.port,
+                                  decode_responses=True)
+
+    def read(self, session_id):
+        cache_data = self.client.get(session_id)
+        logging.info(f"Redis Read: session_id={session_id}, found={bool(cache_data)}")
+        return eval(cache_data) if cache_data else {}
+
+    def write(self, session_id, data):
+        try:
+            self.client.set(session_id, str(data))
+            logging.info(f"Redis Write: session_id={session_id}, data_keys={list(data.keys())}")
+        except Exception as e:
+            logging.error(f"Redis Write Error: {e}")
+            print(data)
+

src/setup/utils.py

@@ -0,0 +1,22 @@
+
+
+
+import time
+import functools
+
+def retry(max_retries=3, delay=1, exceptions=(Exception,)):
+    def decorator(func):
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            retries = 0
+            while True:
+                try:
+                    return func(*args, **kwargs)
+                except exceptions as e:
+                    retries += 1
+                    if retries > max_retries:
+                        raise
+                    print(f"Retrying {func.__name__} due to {e} (attempt {retries}/{max_retries})...")
+                    time.sleep(delay)
+        return wrapper
+    return decorator