upload part of v1.2

.dockerignore

@@ -0,0 +1,11 @@
+keys.env
+.git
+.gitignore
+__pycache__/
+**/__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/

.gitattributes

@@ -1,35 +1,35 @@
-*.7z filter=lfs diff=lfs merge=lfs -text
-*.arrow filter=lfs diff=lfs merge=lfs -text
-*.bin filter=lfs diff=lfs merge=lfs -text
-*.bz2 filter=lfs diff=lfs merge=lfs -text
-*.ckpt filter=lfs diff=lfs merge=lfs -text
-*.ftz filter=lfs diff=lfs merge=lfs -text
-*.gz filter=lfs diff=lfs merge=lfs -text
-*.h5 filter=lfs diff=lfs merge=lfs -text
-*.joblib filter=lfs diff=lfs merge=lfs -text
-*.lfs.* filter=lfs diff=lfs merge=lfs -text
-*.mlmodel filter=lfs diff=lfs merge=lfs -text
-*.model filter=lfs diff=lfs merge=lfs -text
-*.msgpack filter=lfs diff=lfs merge=lfs -text
-*.npy filter=lfs diff=lfs merge=lfs -text
-*.npz filter=lfs diff=lfs merge=lfs -text
-*.onnx filter=lfs diff=lfs merge=lfs -text
-*.ot filter=lfs diff=lfs merge=lfs -text
-*.parquet filter=lfs diff=lfs merge=lfs -text
-*.pb filter=lfs diff=lfs merge=lfs -text
-*.pickle filter=lfs diff=lfs merge=lfs -text
-*.pkl filter=lfs diff=lfs merge=lfs -text
-*.pt filter=lfs diff=lfs merge=lfs -text
-*.pth filter=lfs diff=lfs merge=lfs -text
-*.rar filter=lfs diff=lfs merge=lfs -text
-*.safetensors filter=lfs diff=lfs merge=lfs -text
-saved_model/**/* filter=lfs diff=lfs merge=lfs -text
-*.tar.* filter=lfs diff=lfs merge=lfs -text
-*.tar filter=lfs diff=lfs merge=lfs -text
-*.tflite filter=lfs diff=lfs merge=lfs -text
-*.tgz filter=lfs diff=lfs merge=lfs -text
-*.wasm filter=lfs diff=lfs merge=lfs -text
-*.xz filter=lfs diff=lfs merge=lfs -text
-*.zip filter=lfs diff=lfs merge=lfs -text
-*.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1 @@
+keys.env

Consumer.py

@@ -1,138 +1,176 @@
-# Consumer
-import time
-import pika
-import os
-from Server import get_response
-import json
-from agent.agent_graph.StateTasks import ProblemState
-import argparse
-import redis
-from encryption_utils import decrypt_token_from_json 
-
-
-##################################################
-# VARIABLES 
-##################################################
-# args for this file
-argparse_model = argparse.ArgumentParser()
-argparse_model.add_argument("--id", type=int, default=0, help="Consumer ID")
-consumer_id = argparse_model.parse_args().id
-
-
-RABBITMQ_URL = os.environ["RABBITMQ_URL"]
-QUEUE_NAME = os.environ["QUEUE_NAME"] 
-redis_host = os.environ["REDIS_HOST"]
-redis_port = os.environ["REDIS_PORT"]   
-redis_password = os.environ["REDIS_PASSWORD"]
-
-
-##################################################
-# PROCESSING METHODS
-##################################################
-
-def redis_send(user_id,msg_id,answer):
-    r = redis.Redis(
-        host=redis_host,
-        port=redis_port,
-        decode_responses=True,
-        username="default",
-        password=redis_password,
-    )
- 
-    success = r.set(f'ANSWER_FOR_USER_ID{user_id}_OF_{msg_id}',json.dumps(answer))
-    return success
-
-
-def model_call(request, token): 
-    # تأكد إن request dict
-    if isinstance(request, str):
-        request = json.loads(request)
-
-    # fill with last state  
-    try: 
-        state = json.loads(request.get('last_state', "")) if request.get('last_state') else {}
-    except Exception: 
-        state = {}
-
-    # fallback لو مفيش state
-    if not state:
-        state = { 
-            "question": request.get('prompt', ""), 
-            "memory": request.get('memory', []) 
-        } 
-
-    answer = get_response(
-        request.get('prompt', ""), 
-        request.get('memory', []),
-        token,
-        state,
-        request.get('user_email', ""),
-        request.get('user_name', "")
-    ) 
-     
-    # drop unserializable keys 
-    for k in ["llm", "rag_model"]: 
-        if k in answer:
-            answer[k] = "" 
-
-    return answer
-
-def process_message(recieved_msg):
-    # decrypt token
-    token = decrypt_token_from_json(json.loads(recieved_msg['ht_token_encrypted_dumped']))
-    # call the model
-    model_answer = model_call(recieved_msg,token)
-    # send answer to redis
-    user_id = recieved_msg["user_id"]
-    msg_id = recieved_msg["msg_id"]
-    redis_send_res = redis_send(user_id,msg_id,model_answer)
-    print({"STATUS": redis_send_res , "CONSUMER": {consumer_id}}) # add monitoring but still hide user data
-
-
-##################################################
-# CONSUMER METHODS
-##################################################
-
-
-def get_connection():
-    params = pika.URLParameters(RABBITMQ_URL)
-    return pika.BlockingConnection(params)
-
-def callback(ch, method, properties, body):
-    ##### Recieve message and process it
-    recieved_msg = json.loads(body.decode())
-    print("-------------------------------------------------")
-    print(f"MSG AT CONSUMER {consumer_id}" )
-
-    ##### Process Message
-    process_message(recieved_msg)
-
-
-    ###### Finalize
-    ch.basic_ack(delivery_tag=method.delivery_tag)
-
-def start_consumer():
-    # when scalled each server has consumer
-    params = pika.URLParameters(RABBITMQ_URL)
-    connection = pika.BlockingConnection(params)
-    channel = connection.channel()
-
-    channel.queue_declare(queue=QUEUE_NAME, durable=True)
-
-    channel.basic_qos(prefetch_count=1)
-
-    channel.basic_consume(
-        queue=QUEUE_NAME,
-        on_message_callback=callback
-    )
-
-    print("Waiting for messages...")
-    channel.start_consuming()
-
-##################################################
-# MAIN
-##################################################
-
-if __name__ == "__main__":
-    print(f"Starting New Consumer {consumer_id}...")
-    start_consumer()
+# Consumer
+import time
+import pika
+import os
+from Server import get_response,initalizeGraph
+import json
+import argparse
+from encryption_utils import decrypt_token_from_json 
+from redis_utils import redis_send 
+import traceback
+import redis
+from langgraph.checkpoint.postgres import PostgresSaver
+from psycopg_pool import ConnectionPool
+from psycopg.rows import dict_row
+import yaml
+from utils import get_rabbit_connection
+
+
+##################################################
+# VARIABLES  and Initalization
+##################################################
+# args for this file
+argparse_model = argparse.ArgumentParser()
+argparse_model.add_argument("--id", type=int, default=0, help="Consumer ID")
+argparse_model.add_argument("--min_n_pool", type=int, default=2, help="Minimum number of connections in the pool")
+argparse_model.add_argument("--max_n_pool", type=int, default=10, help="Maximum number of connections in the pool")
+
+args = argparse_model.parse_args()
+consumer_id = args.id
+min_n_pool = args.min_n_pool
+max_n_pool = args.max_n_pool
+
+
+RABBITMQ_URL = os.environ["RABBITMQ_URL"]
+QUEUE_NAME = os.environ["QUEUE_NAME"] 
+redis_host = os.environ["REDIS_HOST"]
+redis_port = os.environ["REDIS_PORT"]   
+redis_password = os.environ["REDIS_PASSWORD"]
+
+
+
+##################################################
+# Connections Initalization
+##################################################
+REDIS_CONN = redis.Redis(
+            host=redis_host,
+            port=redis_port,
+            decode_responses=True,
+            username="default",
+            password=redis_password,
+        )
+
+pool = ConnectionPool(
+            conninfo=os.environ["NEON_POSTGRES_URL"],
+            min_size=min_n_pool,
+            max_size=max_n_pool,
+            kwargs={
+                "autocommit": True,
+                "row_factory": dict_row,
+            },
+        )
+
+CHECKPOINTER = PostgresSaver(pool)
+_, GRAPH = initalizeGraph(CHECKPOINTER)
+
+##################################################
+# PROCESSING METHODS
+##################################################
+
+
+def model_call(request, token): 
+    # تأكد إن request dict
+    if isinstance(request, str):
+        request = json.loads(request)
+
+    
+
+    answer = get_response(
+            GRAPH,
+            request.get('prompt', request.get('msg', "")), 
+            token,
+            request.get('user_email', ""),
+            request.get('user_id', ""),
+            request.get('chat_id', ""),
+
+        )
+
+    return answer
+
+def process_message(recieved_msg):
+    # decrypt token
+    token = decrypt_token_from_json(json.loads(recieved_msg['ht_token_encrypted_dumped']))
+    # call the model
+    model_answer = model_call(recieved_msg,token)
+    # send answer to redis
+    user_id = recieved_msg["user_id"]
+    chat_id = recieved_msg["chat_id"]
+    redis_send_res = redis_send(REDIS_CONN,user_id,chat_id,model_answer)
+    print({"STATUS": redis_send_res , "CONSUMER": {consumer_id}}) # add monitoring but still hide user data
+
+
+##################################################
+# CONSUMER METHODS
+##################################################
+
+
+
+def callback(ch, method, properties, body):
+    recieved_msg = None
+    try:
+        ##### Recieve message and process it
+        recieved_msg = json.loads(body.decode())
+        print("-------------------------------------------------")
+        print(f"MSG AT CONSUMER {consumer_id}" )
+
+        ##### Process Message
+        process_message(recieved_msg)
+
+    except Exception as e:
+        print({"STATUS": "error", "CONSUMER": consumer_id, "ERROR": type(e).__name__})
+        traceback.print_exc()
+        if isinstance(recieved_msg, dict) and "user_id" in recieved_msg and "chat_id" in recieved_msg:
+            try:
+                redis_send(REDIS_CONN,recieved_msg["user_id"], recieved_msg["chat_id"], {"status": "error", "message": "processing_failed"})
+            except Exception as redis_error:
+                print({"STATUS": "redis_error", "CONSUMER": consumer_id, "ERROR": type(redis_error).__name__})
+
+    finally:
+        ###### Finalize
+        ch.basic_ack(delivery_tag=method.delivery_tag)
+
+def start_consumer():
+    connection = get_rabbit_connection()
+    channel = connection.channel()
+
+    channel.queue_declare(
+        queue=QUEUE_NAME,
+        durable=True
+    )
+
+    channel.basic_qos(
+        prefetch_count=1
+    )
+
+    channel.basic_consume(
+        queue=QUEUE_NAME,
+        on_message_callback=callback
+    )
+
+    print("Waiting for RabbitMQ messages...")
+
+    channel.start_consuming()
+    
+
+##################################################
+# MAIN
+##################################################
+
+if __name__ == "__main__":
+    
+    print(f"Starting New Consumer {consumer_id}...")
+    try:
+        start_consumer()
+        print(f"Consumer {consumer_id} Done.")
+    except(Exception, KeyboardInterrupt) as e:
+        print(f"Consumer {consumer_id} stopped. Reason: {type(e).__name__}")
+    finally:
+        print(f"Consumer {consumer_id} is closing connections...")
+        # close all connections
+        pool.close()
+        REDIS_CONN.close()
+
+
+
+
+

Dockerfile

@@ -1,18 +1,18 @@
-FROM python:3.11-slim
-
-ENV PYTHONDONTWRITEBYTECODE=1
-ENV PYTHONUNBUFFERED=1
-
-WORKDIR /app
-
-COPY requirements.txt ./
-
-RUN pip install --no-cache-dir --upgrade pip && \
-    pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu && \
-    pip install --no-cache-dir -r requirements.txt
-
-COPY . .
-
-
-
-CMD ["python", "-u", "app.py"]
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+
+WORKDIR /app
+
+COPY requirements.txt ./
+
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir torch --index-url https://download.pytorch.org/whl/cpu && \
+    pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+
+
+CMD ["python", "-u", "app.py"]

LICENSE.txt

@@ -1,21 +1,21 @@
-MIT License
-
-Copyright (c) 2025 TahaFawzyElshrif
-
-Permission is hereby granted, free of charge, to any person obtaining a copy
-of this software and associated documentation files (the "Software"), to deal
-in the Software without restriction, including without limitation the rights
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
-copies of the Software, and to permit persons to whom the Software is
-furnished to do so, subject to the following conditions:
-
-The above copyright notice and this permission notice shall be included in all
-copies or substantial portions of the Software.
-
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
-SOFTWARE.
+MIT License
+
+Copyright (c) 2025 TahaFawzyElshrif
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

Queue_Producer.py

@@ -1,31 +1,26 @@
-# Producer
-import pika
-import os
-RABBITMQ_URL = os.environ["RABBITMQ_URL"]
-QUEUE_NAME = os.environ["QUEUE_NAME"] 
-
-def get_connection():
-    params = pika.URLParameters(RABBITMQ_URL)
-    return pika.BlockingConnection(params)
-
-connection = get_connection()
-def send_message(data: dict):
-    connection = get_connection()
-    channel = connection.channel()
-
-    channel.queue_declare(queue=QUEUE_NAME, durable=True)
-
-    message = str(data)
-
-    channel.basic_publish(
-        exchange='',
-        routing_key=QUEUE_NAME,
-        body=message,
-        properties=pika.BasicProperties(
-            delivery_mode=2,  # make message persistent
-        )
-    )
-
-    connection.close()
-
-    return {"status": "sent", "data": data}
+# Producer
+import pika
+import os
+QUEUE_NAME = os.environ["QUEUE_NAME"] 
+
+
+def send_message(RABBITMQ_CONNECTION,data: dict):
+    connection = RABBITMQ_CONNECTION
+    channel = connection.channel()
+
+    channel.queue_declare(queue=QUEUE_NAME, durable=True)
+
+    message = str(data)
+
+    channel.basic_publish(
+        exchange='',
+        routing_key=QUEUE_NAME,
+        body=message,
+        properties=pika.BasicProperties(
+            delivery_mode=2,  # make message persistent
+        )
+    )
+
+    
+
+    return {"status": "sent", "data": data}

README.md

@@ -1,12 +1,12 @@
----
-title: CodeBuddyAI
-emoji: 📚
-colorFrom: green
-colorTo: green
-sdk: docker
-pinned: false
-license: mit
-short_description: the agent model of CodeBuddy Project
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: CodeBuddyAI
+emoji: 📚
+colorFrom: green
+colorTo: green
+sdk: docker
+pinned: false
+license: mit
+short_description: the agent model of CodeBuddy Project
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

Server.py

@@ -1,75 +1,95 @@
-from langchain_huggingface import HuggingFaceEndpoint,ChatHuggingFace
-from langchain_core.messages import HumanMessage,SystemMessage
-import os
-import pandas as pd
-from agent.agent_graph.graph import compiled_graph
-from agent.rag.rag import rag_text_chooser
-import sys
-import os
-from agent.agent_graph.StateTasks import Available_Tasks
-from agent.tools.PDF import PDF_generator_Node
-from agent.tools.email import EMAIL_sender_Node
-from agent.agent_graph.Graph_Nodes import get_llm_answer
-from agent.llm.prompts import NODES_Prompts
-import dotenv
-
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')))
-dotenv.load_dotenv("/content/drive/MyDrive/study/Projects/keys.env")
-def get_response(prompt,memory,hf_key,state,user_email,user_name):
-        # Setting up models
-        os.environ["HF_TOKEN"]  = hf_key
-        llm_gpt = HuggingFaceEndpoint(
-            repo_id="openai/gpt-oss-20b",#"deepseek-ai/DeepSeek-V3.2-Exp",#"openai/gpt-oss-20b",
-            task='conversational',
-            provider="auto",
-                max_new_tokens=2048
-        )
-        llm_gpt = ChatHuggingFace(llm=llm_gpt)
-
-        print("RAG_PATH ",os.path.join(os.path.dirname(__file__), 'agent' ,'rag', 'rag.xlsx'), os.path.exists(os.path.join(os.path.dirname(__file__), 'agent' ,'rag', 'rag.xlsx')))
-        
-        rag_model = rag_text_chooser(os.path.join(os.path.dirname(__file__), 'agent' ,'rag', 'rag.xlsx'))
-
-        # update state
-        state["question"] = prompt
-        state["memory"] = memory
-        state["llm"] = llm_gpt
-        state["rag_model"] = rag_model
-
-        call = compiled_graph.invoke(state)
-
-        save_send_email(call,user_email,user_name)
-        os.environ["HF_TOKEN"]  = "" # to prevent keep it in env for other calls and for security
-
-        return call
-
-def save_send_email(call,user_email,user_name):
-    if ("all_ok" in call.keys()):
-        if (call['all_ok']== True):
-          if (call['question_type'] in [Available_Tasks.LAPTOP_CHOOSE.value ,  
-            Available_Tasks.QUESTION.value ,
-    Available_Tasks.ROADMAP.value]):
-              email_txt = get_llm_answer(model_llm=call['llm'],messages=[HumanMessage(content=("اسم الزميل لتستخدمه هو :  "+ user_name +"/n/n")+ NODES_Prompts.Email_text.value + call['question'] + str(call['memory']) +  call['question_type'] + call['answer'])])
-              title = get_llm_answer(model_llm=call['llm'],messages=[HumanMessage(content=NODES_Prompts.Email_title.value + call['question'] + str(call['memory']) +  call['question_type']+ call['answer'])])
-
-              import tempfile
-              path_pdf = ''
-              with tempfile.NamedTemporaryFile(suffix=".pdf", delete=False) as tmp_file:
-                  path_pdf = tmp_file.name
-                  # هنا تكتب الكود اللي بيولد الملف
-                  print("PDF path:", path_pdf)
-
-              # بعد ما تخلص من الملف ممكن تحذفه
-              # import os
-              # os.remove(path_pdf)
-
-
-              #path_pdf ="/content/drive/MyDrive/study/Projects/CodeBuddyAI/tmp.pdf"
-              PDF_generator_Node(call['answer'],title,path_pdf)
-              #EMAIL_sender_Node(user_email,email_txt,title,path_pdf)
-              import os
-              os.remove(path_pdf)
-              print("Done")
-
-
-
+from langchain_huggingface import HuggingFaceEndpoint,ChatHuggingFace
+from langchain_core.messages import HumanMessage,SystemMessage
+import os
+import pandas as pd
+import sys
+import os
+from agent.agent_graph.StateTasks import Available_Tasks
+import dotenv
+from agent.agent_graph.graph import  AgentGraph
+from langgraph.types import Command
+from langchain_core.messages import HumanMessage
+from utils import path_rag_file
+
+
+
+def initalizeGraph(checkpointer,hf_key=None):
+    if hf_key:
+        os.environ["HF_TOKEN"] = hf_key
+    graph_obj = AgentGraph(checkpointer,path_rag_file)
+    graph = graph_obj.get_graph()
+
+    return graph_obj,graph
+
+ 
+
+
+def get_response(graph, prompt,hf_key,user_email,user_id,chat_id):
+        # Setting up models
+        os.environ["HF_TOKEN"]  = hf_key
+
+        try:
+            config = {"configurable": {"thread_id": user_id+"_"+chat_id}}
+
+            state = graph.get_state(config)
+
+            waiting_resume  = len(state.tasks) > 0  # if there are old conversations, we need to resume the conversation instead of starting a new one , so meaning is do we need to resume the conversation or not
+
+            call = _chat_api(graph,prompt,config,waiting_resume )
+
+            #save_send_email(call,user_email,user_name)
+            return call
+
+        finally:
+            os.environ["HF_TOKEN"]  = "" # to prevent keep it in env for other calls and for security
+
+def _chat_api(
+    graph,
+    user_text: str,
+    config: dict,
+    waiting_resume: bool = False,
+) -> dict:
+    """
+    One API request -> One response.
+    """
+
+    if waiting_resume:
+        # resume the conversation
+        result = graph.invoke(
+            Command(resume=user_text),
+            config=config,
+        )
+    else:
+        # new conversation
+        result = graph.invoke(
+            {"messages": [HumanMessage(content=user_text)]},
+            config=config,
+        )
+
+    if "__interrupt__" in result:
+        interrupt = result["__interrupt__"][0]
+
+        return {
+            "status": "interrupt",
+            "question": interrupt.value["question"],
+        }
+
+    if "final_ans_report" in result and len(result["final_ans_report"]) > 400: #big enough to report not just the intermediate answer
+        return {
+            "status": "finished",
+            "answer": result["messages"][-1].content,
+            "report": result["final_ans_report"],
+        }
+
+    return {
+        "status": "finished",
+        "answer": result["messages"][-1].content,
+    }
+
+
+def save_send_email(call,user_email,user_name):
+    pass
+
+
+
+

app.py

@@ -1,116 +1,170 @@
-from urllib import request
-
-from fastapi import FastAPI
-from fastapi.responses import JSONResponse
-import sentry_sdk
-import uvicorn
-import sys
-import os
-import json
-sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')))
-from agent.agent_graph.StateTasks import ProblemState
-import subprocess
-from Queue_Producer import send_message
-import redis
-from utils import RequestModel, RequestAnswer
-from Consumer import redis_send
-import psutil
-
-
-##################################################
-# VARIABLES
-##################################################
-redis_host = os.environ["REDIS_HOST"]
-redis_port = os.environ["REDIS_PORT"]   
-redis_password = os.environ["REDIS_PASSWORD"]
-
-
-##################################################
-# START CONSUMERS in a separate process
-##################################################
-for i in range(1,4):  # Start 3 consumers
-    subprocess.Popen(['python','-u','Consumer.py', '--id', str(i)])
-
-
-##################################################
-# START API and METHODS
-##################################################
-
-# Create Redis connection (global to make the get very light)
-redis_conn = redis.Redis(
-        host=redis_host,
-        port=redis_port,
-        decode_responses=True,
-        username="default",
-        password=redis_password,
-    )
-
-# model and rag are not global for better security ,at least for this version
-
-# Create app instance
-app = FastAPI()
-
-# Create Sentry Monitoring for better error tracking and performance monitoring
-sentry_sdk.init(
-    dsn=os.environ["SENTRY_DSN"],
-    # see https://docs.sentry.io/platforms/python/data-management/data-collected/ for more info
-    send_default_pii=True,
-)
-
-app = FastAPI()
-#redis_send(request.user_id,request.msg_id,{"status": "pending"})  # for performance test
-
-print("Starting API Server...")
-
-
-##################################################
-# ROUTES
-##################################################
-
-@app.get("/")
-def read_root():
-    return {"message": "Hello From CodeBuddyAI!"}
-
-
-@app.get("/metrics")
-def metrics():
-    return {
-        "cpu": psutil.cpu_percent(),
-        "ram": psutil.virtual_memory().percent
-    }
-
-
-@app.api_route("/health", methods=["GET", "HEAD", "POST", "OPTIONS"])
-def get_health():
-    return JSONResponse({"status": "ok"})
-
-  
-@app.post("/Message/Send/")
-def call(request: RequestModel): 
-    redis_send(request.user_id,request.msg_id,{"status": "pending"}) 
-    return send_message(json.dumps(request.model_dump()))
-
-
-@app.post("/Message/Answer/")
-def call(request: RequestAnswer): 
-    ## MUST BE LIGHTWEIGHT, JUST CHECK IF ANSWER IS READY IN REDIS, IF YES RETURN IT, ELSE RETURN PENDING
-    try:
-        answer = redis_conn.get(f'ANSWER_FOR_USER_ID{request.user_id}_OF_{request.msg_id}')
-        if answer is None:
-            return {"status": "error"}
-        elif  "status" in answer and json.loads(answer)["status"] == "pending":
-            return {"status": "pending"}
-        else:
-
-            redis_conn.delete(f'ANSWER_FOR_USER_ID{request.user_id}_OF_{request.msg_id}')  # Clean up after fetching  for memory and better secure as double call is wrong   
-            return {"status": "ready", "data": json.loads(answer)}
-        
-    except Exception as e:
-        print(f"Error fetching answer from Redis: {e}")
-        return {"status": "error", "message": str(e)}
-    
-    
-
-
-if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=7860)
+from contextlib import asynccontextmanager
+from urllib import request
+
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse
+import sentry_sdk
+import uvicorn
+import sys
+import os
+import json
+import dotenv
+from langgraph.checkpoint.postgres import PostgresSaver
+from psycopg_pool import ConnectionPool
+from psycopg.rows import dict_row
+import pika 
+import yaml
+from utils import get_rabbit_connection
+
+parameters_obj = None
+
+##################################################
+# VARIABLES
+##################################################
+from utils import * # by default utils in same directory as app.py, so no need to add it to sys.path
+
+sys.path.append(os.path.abspath(current_dir))
+#sys.path.append(os.path.abspath(os.path.join(current_dir, '..', '..'))) # directory of the app.py with 2 upper folders
+dotenv.load_dotenv(path_env) # won't affet when the env variables are already set in the system, but will set them if they are not set yet
+
+
+redis_host = os.environ["REDIS_HOST"]
+redis_port = os.environ["REDIS_PORT"]   
+redis_password = os.environ["REDIS_PASSWORD"]
+
+
+
+
+import subprocess
+from Queue_Producer import send_message
+import redis
+from redis_utils import redis_send
+import psutil
+ 
+
+
+
+##################################################
+# START API and METHODS
+##################################################
+
+# Create Redis connection (global to make the get very light)
+REDIS_CONN = None
+# Create app instance
+CONSUMER_PROCESSES = []
+RABBITMQ_CONNECTION = None
+
+
+RABBITMQ_URL = os.environ["RABBITMQ_URL"]
+
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    global REDIS_CONN, CHECKPOINTER, RABBITMQ_CONNECTION,parameters_obj
+    
+    with open(yaml_path, 'r') as file:
+        parameters_obj = yaml.safe_load(file)
+
+    min_n_pool = parameters_obj.get("min_n_pool", 2)
+    max_n_pool = parameters_obj.get("max_n_pool", 10)
+    n_consumers = parameters_obj.get("n_consumers", 3)
+
+    for i in range(1,1+n_consumers):  # Start 3 consumers
+        print(f"Starting consumer {i}")
+        CONSUMER_PROCESSES.append(subprocess.Popen(['python','-u',consumer_path, '--id', str(i), '--min_n_pool', str(min_n_pool), '--max_n_pool', str(max_n_pool)]))  # Start consumer.py with id and pool size as arguments
+    
+
+
+    RABBITMQ_CONNECTION = get_rabbit_connection()
+
+    
+    # Init Redis
+    REDIS_CONN = redis.Redis(
+            host=redis_host,
+            port=redis_port,
+            decode_responses=True,
+            username="default",
+            password=redis_password,
+        )
+
+    
+    # Create Sentry Monitoring for better error tracking and performance monitoring
+    sentry_sdk.init(
+        dsn=os.environ["SENTRY_DSN"],
+        # see https://docs.sentry.io/platforms/python/data-management/data-collected/ for more info
+        send_default_pii=False, # more secure, don't send user info, but can be set to True if needed
+    )
+    
+
+    yield
+
+    # Terminate processes and connections on shutdown
+    for process in CONSUMER_PROCESSES:
+        process.terminate()
+
+    RABBITMQ_CONNECTION.close()
+
+app = FastAPI(lifespan=lifespan)
+
+
+
+#redis_send(REDIS_CONN,request.user_id,request.chat_id,{"status": "pending"})  # for performance test
+
+print("Starting API Server...")
+
+
+##################################################
+# ROUTES
+##################################################
+
+@app.get("/")
+def read_root():
+    return {"message": "Hello From CodeBuddyAI!"}
+
+
+@app.get("/metrics")
+def metrics():
+    return {
+        "cpu": psutil.cpu_percent(),
+        "ram": psutil.virtual_memory().percent,
+        "consumers": [{"pid": p.pid, "status": p.poll()} for p in CONSUMER_PROCESSES],
+
+    }
+
+
+@app.api_route("/health", methods=["GET", "HEAD", "POST", "OPTIONS"])
+def get_health():
+    return JSONResponse({"status": "ok"})
+
+  
+@app.post("/Message/Send/")
+def call(request: RequestModel): 
+    redis_send(REDIS_CONN,request.user_id,request.chat_id,{"status": "pending"}) 
+    return send_message(RABBITMQ_CONNECTION, json.dumps(request.model_dump()))
+
+
+@app.post("/Message/Answer/")
+def call(request: RequestAnswer): 
+    ## MUST BE LIGHTWEIGHT, JUST CHECK IF ANSWER IS READY IN REDIS, IF YES RETURN IT, ELSE RETURN PENDING
+    try:
+        answer = REDIS_CONN.get(f'ANSWER_FOR_USER_ID{request.user_id}_OF_{request.chat_id}')
+        if answer is None:
+            return {"status": "error"}
+        elif  "status" in answer and json.loads(answer)["status"] == "pending":
+            return {"status": "pending"}
+        else:
+
+            REDIS_CONN.delete(f'ANSWER_FOR_USER_ID{request.user_id}_OF_{request.chat_id}')  # Clean up after fetching  for memory and better secure as double call is wrong   
+            return {"status": "ready", "data": json.loads(answer)}
+        
+    except Exception as e:
+        print(f"Error fetching answer from Redis: {e}")
+        return {"status": "error", "message": str(e)}
+    
+    
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=7860)
+

encryption_utils.py

@@ -1,44 +1,44 @@
-import os
-import hashlib
-from cryptography.hazmat.primitives.ciphers.aead import AESGCM
-
-
-# ===== Key (same idea as Node) =====
-def get_key():
-    raw_key = os.environ["TOKEN_KEY_ENCRYPTION"].encode()
-    return hashlib.sha256(raw_key).digest()  # 32 bytes
-
-
-# ===== Encrypt =====
-def encrypt_token_to_json(token: str) -> dict:
-    key = get_key()
-    aesgcm = AESGCM(key)
-
-    iv = os.urandom(12)  # same as crypto.randomBytes(12)
-
-    encrypted = aesgcm.encrypt(iv, token.encode(), None)
-
-    ciphertext = encrypted[:-16]
-    tag = encrypted[-16:]
-
-    return {
-        "iv": iv.hex(),
-        "data": ciphertext.hex(),
-        "tag": tag.hex(),
-    }
-
-
-# ===== Decrypt =====
-def decrypt_token_from_json(enc: dict) -> str:
-    key = get_key()
-    aesgcm = AESGCM(key)
-
-    iv = bytes.fromhex(enc["iv"])
-    ciphertext = bytes.fromhex(enc["data"])
-    tag = bytes.fromhex(enc["tag"])
-
-    encrypted = ciphertext + tag
-
-    decrypted = aesgcm.decrypt(iv, encrypted, None)
-
-    return decrypted.decode()
+import os
+import hashlib
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+
+
+# ===== Key (same idea as Node) =====
+def get_key():
+    raw_key = os.environ["TOKEN_KEY_ENCRYPTION"].encode()
+    return hashlib.sha256(raw_key).digest()  # 32 bytes
+
+
+# ===== Encrypt =====
+def encrypt_token_to_json(token: str) -> dict:
+    key = get_key()
+    aesgcm = AESGCM(key)
+
+    iv = os.urandom(12)  # same as crypto.randomBytes(12)
+
+    encrypted = aesgcm.encrypt(iv, token.encode(), None)
+
+    ciphertext = encrypted[:-16]
+    tag = encrypted[-16:]
+
+    return {
+        "iv": iv.hex(),
+        "data": ciphertext.hex(),
+        "tag": tag.hex(),
+    }
+
+
+# ===== Decrypt =====
+def decrypt_token_from_json(enc: dict) -> str:
+    key = get_key()
+    aesgcm = AESGCM(key)
+
+    iv = bytes.fromhex(enc["iv"])
+    ciphertext = bytes.fromhex(enc["data"])
+    tag = bytes.fromhex(enc["tag"])
+
+    encrypted = ciphertext + tag
+
+    decrypted = aesgcm.decrypt(iv, encrypted, None)
+
+    return decrypted.decode()

parameters.yaml

@@ -0,0 +1,5 @@
+# actually n_pool = sum of n_pool for  n_consumer , each consumer has pool oof min: max number of connections
+
+min_n_pool : 2  
+max_n_pool : 10 
+n_consumer : 1

redis_utils.py

@@ -0,0 +1,8 @@
+import json
+import os
+
+import redis
+
+
+def redis_send(redis_conn,user_id, chat_id, answer):
+    return redis_conn.set(f"ANSWER_FOR_USER_ID{user_id}_OF_{chat_id}", json.dumps(answer))

requirements.txt

@@ -1,18 +1,27 @@
-langchain-huggingface
-langchain-core
-langchain-text-splitters
-langgraph
-python-dotenv
-pandas
-openpyxl
-sentence-transformers
-requests
-ipython
-fastapi 
-uvicorn
-markdown_pdf
-pika
-redis
-cryptography
-sentry-sdk
-psutil
+fastapi[standard]
+arabic-reshaper==3.0.1
+langchain==1.3.14
+langchain-google-genai==4.3.2
+langchain-huggingface==1.2.2
+langchain-openai==1.4.1
+langchain-text-splitters==1.1.2
+langgraph==1.2.10
+langgraph-checkpoint-postgres==3.1.1
+python-bidi==0.6.11
+reportlab==5.0.0
+psycopg[binary]
+psycopg[pool]
+python-dotenv
+pandas
+openpyxl
+sentence-transformers
+requests
+ipython
+fastapi 
+uvicorn
+markdown_pdf
+pika
+redis
+cryptography
+sentry-sdk
+psutil

utils.py

@@ -1,17 +1,33 @@
-from pydantic import BaseModel
-
-
-class RequestModel(BaseModel):
-    prompt: str
-    ht_token_encrypted_dumped : str
-    user_email : str
-    user_name : str
-    user_id : str
-    msg_id : str
-    memory: list[str]
-    last_state : str
-
-class RequestAnswer(BaseModel):
-    user_id: str
-    msg_id:str
- 
+from pydantic import BaseModel
+import os 
+import dotenv
+import  pika
+current_dir = os.path.dirname(os.path.abspath(__file__)) #  directory of main file utils.py ######## So This must be in app.py only
+
+
+path_env = os.path.join(current_dir, "keys.env") 
+path_rag_file =  os.path.join(current_dir,"agent","rag","rag.xlsx")
+consumer_path = os.path.join(current_dir, "Consumer.py") 
+yaml_path = os.path.join(current_dir, "parameters.yaml")
+
+dotenv.load_dotenv(path_env) #
+RABBITMQ_URL = os.environ["RABBITMQ_URL"]
+
+
+class RequestModel(BaseModel):
+    prompt: str
+    ht_token_encrypted_dumped : str
+    user_email : str
+    user_name : str
+    user_id : str
+    chat_id : str
+
+
+class RequestAnswer(BaseModel):
+    user_id: str
+    chat_id:str
+
+
+def get_rabbit_connection():
+    params = pika.URLParameters(RABBITMQ_URL)
+    return pika.BlockingConnection(params)