import json
import os
import shutil
import logging
import pandas as pd
from datetime import datetime as dt
from dotenv import load_dotenv
from score import *
from src.main import *
from src.QA_integration import QA_RAG
from langserve import add_routes
from graphdatascience import GraphDataScience
from src.entities.source_node import sourceNode
# Load environment variables if needed
load_dotenv()
# Constants
URI = ''
USERNAME = ''
PASSWORD = ''
DATABASE = 'persistent1'
CHUNK_DIR = os.path.join(os.path.dirname(__file__), "chunks")
MERGED_DIR = os.path.join(os.path.dirname(__file__), "merged_files")
# Initialize database connection
graph = create_graph_database_connection(URI, USERNAME, PASSWORD, DATABASE)
def create_source_node_local(graph, model, file_name):
"""Creates a source node for a local file."""
source_node = sourceNode()
source_node.file_name = file_name
source_node.file_type = 'pdf'
source_node.file_size = '1087'
source_node.file_source = 'local file'
source_node.model = model
source_node.created_at = dt.now()
graphDB_data_Access = graphDBdataAccess(graph)
graphDB_data_Access.create_source_node(source_node)
return source_node
def test_graph_from_file_local(model_name):
"""Test graph creation from a local file."""
file_name = 'About Amazon.pdf'
shutil.copyfile('/workspaces/llm-graph-builder/backend/files/About Amazon.pdf',
os.path.join(MERGED_DIR, file_name))
create_source_node_local(graph, model_name, file_name)
merged_file_path = os.path.join(MERGED_DIR, file_name)
local_file_result = extract_graph_from_file_local_file(
URI, USERNAME, PASSWORD, DATABASE, model_name, merged_file_path, file_name, '', '',None
)
logging.info("Local file processing complete")
print(local_file_result)
try:
assert local_file_result['status'] == 'Completed'
assert local_file_result['nodeCount'] > 0
assert local_file_result['relationshipCount'] > 0
print("Success")
except AssertionError as e:
print("Fail: ", e)
return local_file_result
def test_graph_from_wikipedia(model_name):
# try:
"""Test graph creation from a Wikipedia page."""
wiki_query = 'https://en.wikipedia.org/wiki/Ram_Mandir'
source_type = 'Wikipedia'
file_name = "Ram_Mandir"
create_source_node_graph_url_wikipedia(graph, model_name, wiki_query, source_type)
wiki_result = extract_graph_from_file_Wikipedia(URI, USERNAME, PASSWORD, DATABASE, model_name, file_name, 'en',file_name, '', '',None)
logging.info("Wikipedia test done")
print(wiki_result)
try:
assert wiki_result['status'] == 'Completed'
assert wiki_result['nodeCount'] > 0
assert wiki_result['relationshipCount'] > 0
print("Success")
except AssertionError as e:
print("Fail: ", e)
return wiki_result
# except Exception as ex:
# print(ex)
def test_graph_website(model_name):
"""Test graph creation from a Website page."""
#graph, model, source_url, source_type
source_url = 'https://www.amazon.com/'
source_type = 'web-url'
file_name = []
create_source_node_graph_web_url(graph, model_name, source_url, source_type)
weburl_result = extract_graph_from_web_page(URI, USERNAME, PASSWORD, DATABASE, model_name, source_url,file_name, '', '',None)
logging.info("WebUrl test done")
print(weburl_result)
try:
assert weburl_result['status'] == 'Completed'
assert weburl_result['nodeCount'] > 0
assert weburl_result['relationshipCount'] > 0
print("Success")
except AssertionError as e:
print("Fail: ", e)
return weburl_result
def test_graph_from_youtube_video(model_name):
"""Test graph creation from a YouTube video."""
source_url = 'https://www.youtube.com/watch?v=T-qy-zPWgqA'
source_type = 'youtube'
create_source_node_graph_url_youtube(graph, model_name, source_url, source_type)
youtube_result = extract_graph_from_file_youtube(
URI, USERNAME, PASSWORD, DATABASE, model_name, source_url, '', ''
)
logging.info("YouTube Video test done")
print(youtube_result)
try:
assert youtube_result['status'] == 'Completed'
assert youtube_result['nodeCount'] > 1
assert youtube_result['relationshipCount'] > 1
print("Success")
except AssertionError as e:
print("Failed: ", e)
return youtube_result
def test_chatbot_qna(model_name, mode='vector'):
"""Test chatbot QnA functionality for different modes."""
QA_n_RAG = QA_RAG(graph, model_name, 'Tell me about amazon', '[]', 1, mode)
print(QA_n_RAG)
print(len(QA_n_RAG['message']))
try:
assert len(QA_n_RAG['message']) > 20
return QA_n_RAG
print("Success")
except AssertionError as e:
print("Failed ", e)
return QA_n_RAG
#Get Test disconnected_nodes list
def disconected_nodes():
#graph = create_graph_database_connection(uri, userName, password, database)
graphDb_data_Access = graphDBdataAccess(graph)
nodes_list, total_nodes = graphDb_data_Access.list_unconnected_nodes()
print(nodes_list[0]["e"]["elementId"])
status = "False"
if total_nodes['total']>0:
status = "True"
else:
status = "False"
return nodes_list[0]["e"]["elementId"], status
#Test Delete delete_disconnected_nodes list
def delete_disconected_nodes(lst_element_id):
print(f'disconnect elementid list {lst_element_id}')
#graph = create_graph_database_connection(uri, userName, password, database)
graphDb_data_Access = graphDBdataAccess(graph)
result = graphDb_data_Access.delete_unconnected_nodes(json.dumps(lst_element_id))
print(f'delete disconnect api result {result}')
if not result:
return "True"
else:
return "False"
#Test Get Duplicate_nodes
def get_duplicate_nodes():
#graph = create_graph_database_connection(uri, userName, password, database)
graphDb_data_Access = graphDBdataAccess(graph)
nodes_list, total_nodes = graphDb_data_Access.get_duplicate_nodes_list()
if total_nodes['total']>0:
return "True"
else:
return "False"
#Test populate_graph_schema
def test_populate_graph_schema_from_text(model):
result_schema = populate_graph_schema_from_text('When Amazon was founded in 1993 by creator Jeff Benzos, it was mostly an online bookstore. Initially Amazon’s growth was very slow, not turning a profit until over 7 years after its founding. This was thanks to the great momentum provided by the dot-com bubble.', model, True)
print(result_schema)
return result_schema
# def compare_graph_results(results):
# """
# Compare graph results across different models.
# Add custom logic here to compare graph data, nodes, and relationships.
# """
# # Placeholder logic for comparison
# print("Comparing results...")
# for i in range(len(results) - 1):
# result_a = results[i]
# result_b = results[i + 1]
# if result_a == result_b:
# print(f"Result {i} is identical to result {i+1}")
# else:
# print(f"Result {i} differs from result {i+1}")
def run_tests():
final_list = []
error_list = []
models = ['openai-gpt-3.5','openai-gpt-4o','openai-gpt-4o-mini','gemini-1.5-pro','azure_ai_gpt_35','azure_ai_gpt_4o','ollama_llama3','groq_llama3_70b','anthropic_claude_3_5_sonnet','fireworks_v3p1_405b','bedrock_claude_3_5_sonnet']
for model_name in models:
try:
# final_list.append(test_graph_from_file_local(model_name))
# final_list.append(test_graph_from_wikipedia(model_name))
# final_list.append(test_populate_graph_schema_from_text(model_name))
# final_list.append(test_graph_website(model_name))
# final_list.append(test_graph_from_youtube_video(model_name))
final_list.append(test_chatbot_qna(model_name, mode='entity search+vector'))
except Exception as e:
error_list.append((model_name, str(e)))
# #Compare and log diffrences in graph results
# # compare_graph_results(final_list) # Pass the final_list to comapre_graph_results
# test_populate_graph_schema_from_text('openai-gpt-4o')
dis_elementid, dis_status = disconected_nodes()
lst_element_id = [dis_elementid]
delt = delete_disconected_nodes(lst_element_id)
# dup = get_duplicate_nodes()
print(final_list)
# schma = test_populate_graph_schema_from_text(model)
# Save final results to CSV
df = pd.DataFrame(final_list)
print(df)
df['execution_date'] = dt.today().strftime('%Y-%m-%d')
df['disconnected_nodes']=dis_status
# df['get_duplicate_nodes']=dup
df['delete_disconected_nodes']=delt
# df['test_populate_graph_schema_from_text'] = schma
df.to_csv(f"Integration_TestResult_{dt.now().strftime('%Y%m%d_%H%M%S')}.csv", index=False)
# Save error details to CSV
df_errors = pd.DataFrame(error_list, columns=['Model', 'Error'])
df_errors['execution_date'] = dt.today().strftime('%Y-%m-%d')
df_errors.to_csv(f"Error_details_{dt.now().strftime('%Y%m%d_%H%M%S')}.csv", index=False)
if __name__ == "__main__":
run_tests()
