Merge pull request #221 from neo4j-labs/binary-removal-changes

Binary removal changes to DEV

Author: karanchellani

.gitignore

@@ -164,4 +164,4 @@ google-cloud-sdk
 google-cloud-cli-469.0.0-linux-x86_64.tar.gz
 /data/llm-experiments-387609-c73d512ca3b1.json
 /backend/src/merged_files
-/backend/chunks
+/backend/src/chunks

backend/score.py

@@ -5,12 +5,14 @@
 from src.main import *
 from src.QA_integration import *
 from src.entities.user_credential import user_credential
+from src.shared.common_fn import *
 import uvicorn
 import asyncio
 import base64
 from langserve import add_routes
 from langchain_google_vertexai import ChatVertexAI
 from src.api_response import create_api_response
+from src.graphDB_dataAccess import graphDBdataAccess
 
 def healthy_condition():
     output = {"healthy": True}
@@ -36,37 +38,37 @@ def sick():
 
 app.add_api_route("/health", health([healthy_condition, healthy]))
 
-@app.post("/sources")
-async def create_source_knowledge_graph(
-    uri=Form(None), userName=Form(None), password=Form(None), file: UploadFile = File(...), model=Form(),database=Form(None), 
-):
-    """
-    Calls 'create_source_node_graph' function in a new thread to create
-    source node in Neo4jGraph when a new file is uploaded.
+# @app.post("/sources")
+# async def create_source_knowledge_graph(
+#     uri=Form(None), userName=Form(None), password=Form(None), file: UploadFile = File(...), model=Form(),database=Form(None), 
+# ):
+#     """
+#     Calls 'create_source_node_graph' function in a new thread to create
+#     source node in Neo4jGraph when a new file is uploaded.
 
-    Args:
-         uri: URI of Graph Service to connect to
-         userName: Username to connect to Graph Service with ( default : None )
-         password: Password to connect to Graph Service with ( default : None )
-         file: File object containing the PDF file
+#     Args:
+#          uri: URI of Graph Service to connect to
+#          userName: Username to connect to Graph Service with ( default : None )
+#          password: Password to connect to Graph Service with ( default : None )
+#          file: File object containing the PDF file
 
-    Returns:
-         'Source' Node creation in Neo4j database
-    """
-    try:
-        result = await asyncio.to_thread(
-            create_source_node_graph_local_file, uri, userName, password, file, model, database
-        )
-        return create_api_response("Success",message="Source Node created successfully",file_source=result.file_source, file_name=result.file_name)
-    except Exception as e:
-        # obj_source_node = sourceNode()
-        job_status = "Failed"
-        message = "Unable to create source node"
-        error_message = str(e)
-        logging.error(f"Error in creating document node: {error_message}")
-        #update exception in source node
-        # obj_source_node.update_exception_db(file.filename, error_message)
-        return create_api_response(job_status, message=message,error=error_message,file_source='local file',file_name=file.filename)
+#     Returns:
+#          'Source' Node creation in Neo4j database
+#     """
+#     try:
+#         result = await asyncio.to_thread(
+#             create_source_node_graph_local_file, uri, userName, password, file, model, database
+#         )
+#         return create_api_response("Success",message="Source Node created successfully",file_source=result.file_source, file_name=result.file_name)
+#     except Exception as e:
+#         # obj_source_node = sourceNode()
+#         job_status = "Failed"
+#         message = "Unable to create source node"
+#         error_message = str(e)
+#         logging.error(f"Error in creating document node: {error_message}")
+#         #update exception in source node
+#         # obj_source_node.update_exception_db(file.filename, error_message)
+#         return create_api_response(job_status, message=message,error=error_message,file_source='local file',file_name=file.filename)
 
 @app.post("/url/scan")
 async def create_source_knowledge_graph_url(
@@ -84,22 +86,23 @@ async def create_source_knowledge_graph_url(
     source_type=Form(None)
     ):
     try:
+        graph = create_graph_database_connection(uri, userName, password, database)
+        graphDb_data_Access = graphDBdataAccess(graph)
         if source_type == 's3 bucket' and aws_access_key_id and aws_secret_access_key:
-            lst_file_name,success_count,failed_count = create_source_node_graph_url_s3(
-                uri, userName, password, database, model, source_url, aws_access_key_id, aws_secret_access_key, source_type
+            lst_file_name,success_count,failed_count = create_source_node_graph_url_s3(graph, model, source_url, aws_access_key_id, aws_secret_access_key, source_type
             )
         elif source_type == 'gcs bucket':
-            lst_file_name,success_count,failed_count = create_source_node_graph_url_gcs(
-                uri, userName, password, database, model, source_url, gcs_bucket_name, gcs_bucket_folder, source_type
+            lst_file_name,success_count,failed_count = create_source_node_graph_url_gcs(graph, model, source_url, gcs_bucket_name, gcs_bucket_folder, source_type
             )
         elif source_type == 'youtube':
-            lst_file_name,success_count,failed_count = create_source_node_graph_url_youtube(
-                uri, userName, password, database, model, source_url, source_type
+            lst_file_name,success_count,failed_count = create_source_node_graph_url_youtube(graph, model, source_url, source_type
             )
         elif source_type == 'Wikipedia':
-            lst_file_name,success_count,failed_count = create_source_node_graph_url_wikipedia(
-                uri, userName, password, database, model, wiki_query, source_type
+            lst_file_name,success_count,failed_count = create_source_node_graph_url_wikipedia(graph, model, wiki_query, source_type
             )
+        else:
+            return create_api_response('Failed',message='source_type is other than accepted source')
+
         if source_url is not None:
             source = source_url
         else:
@@ -108,11 +111,10 @@ async def create_source_knowledge_graph_url(
         message = f"Source Node created successfully for source type: {source_type} and source: {source}"
         return create_api_response("Success",message=message,success_count=success_count,failed_count=failed_count,file_name=lst_file_name)    
     except Exception as e:
-        job_status = "Failed"
-        message = f"Unable to create source node for source type: {source_type} and source: {source_url}{wiki_query}"
+        message = f"Unable to create source node for source type: {source_type} and source: {source}"
         error_message = str(e)
         logging.exception(f'Exception Stack trace:')
-        return create_api_response(job_status,message=message,error=error_message,file_source=source_type)
+        return create_api_response('Failed',message=message,error=error_message,file_source=source_type)
 
 
 @app.post("/extract")
@@ -122,7 +124,6 @@ async def extract_knowledge_graph_from_file(
     password=Form(None),
     model=Form(None),
     database=Form(None),
-    file: UploadFile = File(None),
     source_url=Form(None),
     aws_access_key_id=Form(None),
     aws_secret_access_key=Form(None),
@@ -131,7 +132,8 @@ async def extract_knowledge_graph_from_file(
     gcs_bucket_name=Form(None),
     gcs_bucket_folder=Form(None),
     gcs_blob_filename=Form(None),
-    source_type=Form(None)
+    source_type=Form(None),
+    file_name=Form(None)
 ):
     """
     Calls 'extract_graph_from_file' in a new thread to create Neo4jGraph from a
@@ -148,35 +150,38 @@ async def extract_knowledge_graph_from_file(
           Nodes and Relations created in Neo4j databse for the pdf file
     """
     try:
+        graph = create_graph_database_connection(uri, userName, password, database)   
+        graphDb_data_Access = graphDBdataAccess(graph)
         if source_type == 'local file':
-            return await asyncio.to_thread(
-                extract_graph_from_file_local_file, uri, userName, password, model, database, file=file)
+            result = await asyncio.to_thread(
+                extract_graph_from_file_local_file, graph, model, file_name)
 
         elif source_type == 's3 bucket' and source_url:
             result = await asyncio.to_thread(
-                extract_graph_from_file_s3, uri, userName, password, model, database,
-                source_url, aws_access_key_id, aws_secret_access_key)
+                extract_graph_from_file_s3, graph, model, source_url, aws_access_key_id, aws_secret_access_key)
 
         elif source_type == 'youtube' and source_url:
             result = await asyncio.to_thread(
-                extract_graph_from_file_youtube, uri, userName, password, model, database, source_url)
+                extract_graph_from_file_youtube, graph, model, source_url)
 
         elif source_type == 'Wikipedia' and wiki_query:
             result = await asyncio.to_thread(
-                extract_graph_from_file_Wikipedia, uri, userName, password, model, database, wiki_query, max_sources)
+                extract_graph_from_file_Wikipedia, graph, model, wiki_query, max_sources)
 
         elif source_type == 'gcs bucket' and gcs_bucket_name:
             result = await asyncio.to_thread(
-                extract_graph_from_file_gcs, uri, userName, password, model, database, 
-                gcs_bucket_name, gcs_bucket_folder, gcs_blob_filename)
+                extract_graph_from_file_gcs, graph, model, gcs_bucket_name, gcs_bucket_folder, gcs_blob_filename)
+        else:
+            return create_api_response('Failed',message='source_type is other than accepted source')
 
-        return create_api_response('Success',data=result)
+        return create_api_response('Success', data=result)
     except Exception as e:
-      message=f"Failed To Process File or LLM Unable To Parse Content"
-      job_status = "Failed"
-      error_message = str(e)
-      return create_api_response(job_status,message=message,error=error_message)
-    
+        message=f"Failed To Process File:{file_name} or LLM Unable To Parse Content"
+        logging.info(message)
+        error_message = str(e)
+        graphDb_data_Access.update_exception_db(file_name,error_message)
+        logging.exception(f'Exception Stack trace: {error_message}')
+        return create_api_response('Failed', message=message, error=error_message, file_name = file_name)
 
 @app.get("/sources_list")
 async def get_source_list(uri:str, userName:str, password:str, database:str=None):
@@ -194,7 +199,7 @@ async def get_source_list(uri:str, userName:str, password:str, database:str=None
         message="Unable to fetch source list"
         error_message = str(e)
         logging.exception(f'Exception:{error_message}')
-        return create_api_response(job_status,message=message,error=error_message)
+        return create_api_response(job_status, message=message, error=error_message)
 
 @app.post("/update_similarity_graph")
 async def update_similarity_graph(uri=Form(None), userName=Form(None), password=Form(None), database=Form(None)):
@@ -210,7 +215,7 @@ async def update_similarity_graph(uri=Form(None), userName=Form(None), password=
         message="Unable to update KNN Graph"
         error_message = str(e)
         logging.exception(f'Exception in update KNN graph:{error_message}')
-        return create_api_response(job_status,message=message,error=error_message)
+        return create_api_response(job_status, message=message, error=error_message)
 
 @app.post("/chat_bot")
 async def chat_bot(uri=Form(None),model=Form(None),userName=Form(None), password=Form(None), question=Form(None), session_id=Form(None)):
@@ -224,7 +229,7 @@ async def chat_bot(uri=Form(None),model=Form(None),userName=Form(None), password
         message="Unable to get chat response"
         error_message = str(e)
         logging.exception(f'Exception in chat bot:{error_message}')
-        return create_api_response(job_status,message=message,error=error_message)
+        return create_api_response(job_status, message=message, error=error_message)
 
 @app.post("/connect")
 async def connect(uri=Form(None), userName=Form(None), password=Form(None), database=Form(None)):
@@ -236,7 +241,21 @@ async def connect(uri=Form(None), userName=Form(None), password=Form(None), data
         message="Connection failed to connect Neo4j database"
         error_message = str(e)
         logging.exception(f'Connection failed to connect Neo4j database:{error_message}')
-        return create_api_response(job_status,message=message,error=error_message)
+        return create_api_response(job_status, message=message, error=error_message)
+
+@app.post("/upload")
+async def upload_large_file_into_chunks(file:UploadFile = File(...), chunkNumber=Form(None), totalChunks=Form(None), 
+                                        originalname=Form(None), model=Form(None), uri=Form(None), userName=Form(None), 
+                                        password=Form(None), database=Form(None)):
+    try:
+        result = await asyncio.to_thread(upload_file,uri,userName,password,database,model,file,chunkNumber,totalChunks,originalname)
+        return create_api_response('Success', message=result)
+    except Exception as e:
+        job_status = "Failed"
+        message="Unable to upload large file into chunks or saving the chunks"
+        error_message = str(e)
+        logging.info(message)
+        logging.exception(f'Exception:{error_message}')
 
 def decode_password(pwd):
     sample_string_bytes = base64.b64decode(pwd)

backend/src/document_sources/local_file.py

@@ -4,13 +4,25 @@
 from tempfile import NamedTemporaryFile
 from langchain_community.document_loaders import PyPDFLoader
 
-def get_documents_from_file(file):
-    file_name = file.filename
-    logging.info(f"get_documents_from_file called for filename = {file_name}")
-    suffix = Path(file.filename).suffix
-    with NamedTemporaryFile(delete=True, suffix=suffix) as tmp:
-        shutil.copyfileobj(file.file, tmp)
-        tmp_path = Path(tmp.name)
-        loader = PyPDFLoader(str(tmp_path))
+# def get_documents_from_file_by_bytes(file):
+#     file_name = file.filename
+#     logging.info(f"get_documents_from_file called for filename = {file_name}")
+#     suffix = Path(file.filename).suffix
+#     with NamedTemporaryFile(delete=True, suffix=suffix) as tmp:
+#         shutil.copyfileobj(file.file, tmp)
+#         tmp_path = Path(tmp.name)
+#         loader = PyPDFLoader(str(tmp_path))
+#         pages = loader.load_and_split()
+#     return file_name, pages
+
+def get_documents_from_file_by_path(file_path,file_name):
+    file_path = Path(file_path)
+    if file_path.exists():
+        logging.info(f'file {file_name} processing')
+        loader = PyPDFLoader(str(file_path))
         pages = loader.load_and_split()
-    return file_name, pages
+    else:
+        logging.info(f'File {file_name} does not exist')
+        raise Exception(f'File {file_name} does not exist')
+    
+    return file_name, pages 

backend/src/graphDB_dataAccess.py

@@ -81,16 +81,20 @@ def update_KNN_graph(self):
         """
         Update the graph node with SIMILAR relationship where embedding scrore match
         """
+        index = self.graph.query("""show indexes yield * where type = 'VECTOR' and name = 'vector'""")
+        # logging.info(f'show index vector: {index}')
         knn_min_score = os.environ.get('KNN_MIN_SCORE')
-        result = self.graph.query("""MATCH (c:Chunk)
-                                WHERE c.embedding IS NOT NULL AND count { (c)-[:SIMILAR]-() } < 5
-                                CALL db.index.vector.queryNodes('vector', 6, c.embedding) yield node, score
-                                WHERE node <> c and score >= $score MERGE (c)-[rel:SIMILAR]-(node) SET rel.score = score
-                            """,
-                            {"score":knn_min_score}
-                            )
+        if index[0]['name'] == 'vector':
+            logging.info('update KNN graph')
+            result = self.graph.query("""MATCH (c:Chunk)
+                                    WHERE c.embedding IS NOT NULL AND count { (c)-[:SIMILAR]-() } < 5
+                                    CALL db.index.vector.queryNodes('vector', 6, c.embedding) yield node, score
+                                    WHERE node <> c and score >= $score MERGE (c)-[rel:SIMILAR]-(node) SET rel.score = score
+                                """,
+                                {"score":knn_min_score}
+                                )
+            logging.info(f"result : {result}")
 
-        
     def connection_check(self):
         """
         Args: