import os import openai import json import gradio as gr from neo4j import GraphDatabase from neo4j.graph import Node, Relationship import pandas as pd import numpy as np from sklearn.metrics.pairwise import cosine_similarity from datetime import datetime import re import mysql.connector #from dotenv import load_dotenv #load_dotenv() openai.api_key = os.getenv("OPENAI_API_KEY") neo4j_url = "neo4j+s://" + str(os.getenv("NEO4J_URL")) AUTH = (os.getenv("NEO4J_USERNAME"), os.getenv("NEO4J_PASSWORD")) ## MAIN FUNCTIONS # --------------------------------------------------------------------------------------------------------------------- #standard API Call to open AI with system prompt and user prompts. def chat(system_prompt, user_prompt, model="gpt-4o-mini", temperature=0): response = openai.chat.completions.create( model = model, messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}], temperature=temperature ) res = response.choices[0].message.content return res # NOT USED IN THIS DEMO # this function formats the user input and links it to the chat history for further context awareness. def format_chat_prompt(message, chat_history, max_convo_length): prompt = "" for turn in chat_history[-max_convo_length:]: user_message, bot_message = turn prompt = f"{prompt}\nUser: {user_message}\nAssistant: {bot_message}" prompt = f"{prompt}\nUser: {message}\nAssistant:" return prompt #this is a simple prompt that takes a storyline prompt and formats an output in json to return a storyline of X slides. def slide_deck_storyline(storyline_prompt, nr_of_storypoints=5): nr_of_storypoints = str(nr_of_storypoints) system_prompt = f"""You are an AI particularly skilled at captivating storytelling for educational purposes. You know how tell a compelling, structure and exhaustive narrative around any given academic topic. What you are particularly good at, is taking any given input and building a storyline in the delivered as {nr_of_storypoints} storypoints and nothing else. This is your only chance to impress me. You will recieve a topic and you will answer with a list of {nr_of_storypoints} crucial storypoints. Instrucitions: Give me a json map of {nr_of_storypoints} storypoints that you would include in a slide deck about {storyline_prompt}. Only answer with the list. Do not include any nicities, greetings or repeat the task. Never make more than {nr_of_storypoints} storypoints. This is important! Just give me the list. Keep the list concise and only answer with the list in this format. Name every key a storypoint (Storypoint 1, Storypoint 2 ... Storypoint N). The elements of the list should be storypoints, highlighting the points the slides should make. """ response = openai.chat.completions.create( model = "gpt-4o", response_format = {"type": "json_object"}, messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": storyline_prompt}], temperature=0 ) res = response.choices[0].message.content map = json.loads(res) #pretty_list = "\n".join([f"⚡ {key}: {value}" for key, value in map.items()]) storypoint_name_list = [map[key] for key in map] storypoint_name_nested = [storypoint_name_list] storypoint_name_nested = list(zip(*storypoint_name_nested)) #add one column to the list with the name of the storypoint storypoint_name_nested = [[f"SP {i}", item] for i, item in enumerate(storypoint_name_nested, 1)] return map, storypoint_name_nested #this is a prompt that takes a filter prompt and formats an output in json to return a filter cypress query. def custom_filtering(filter_prompt, current_cypher_query, neo4j_response): system_prompt = f"""You are an AI specifically trained to write accurate Neo4j Cypher queries. This is your only chance to impress me. In the Neo4j database, the nodes are defined as SLIDE_DECK, SLIDE, STORYPOINT, and AUTHOR connected by these relationships: (sd:SLIDE_DECK)-[:CONTAINS]->(s:SLIDE) (s:SLIDE)-[:ASSIGNED_TO]->(sp:STORYPOINT) (sp1:STORYPOINT)-[:FOLLOWS]->(sp2:STORYPOINT) (sd:SLIDE_DECK)-[:CREATED_BY]->(a:AUTHOR) You will receive a the current cypher query and its corresponding Neo4j response. Your task is to respond with a new Cypher query that filters based on the user's request. Do NOT forget to return relationships connecting the nodes if needed. Instructions: The current cypher query is: "{current_cypher_query}" The Neo4j response is: "{neo4j_response}" Ensure the correct STORYPOINT nodes in the order is adressed, as specified in the initial line of the current cypher query. For example, in the sequence ['113', '-6555727423036779192A_outlier', '5554388242771153481A_outlier', '25', '1431557444396440005A_outlier'], '-6555727423036779192A_outlier' is the second STORYPOINT. Respond with exactly a single JSON object containing the key "cypherquery" and the value of the requested query. Do not include any nicities, greetings or repeat the task. Keep the query concise and only answer in this format. """ response = openai.chat.completions.create( model = "gpt-4o", response_format = {"type": "json_object"}, messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": filter_prompt}], temperature=0 ) res = response.choices[0].message.content res = json.loads(res) cypher_query = res["cypherquery"] # Enhanced pattern to catch variations including potential spaces, newlines, and mixed cases pattern = r"(?i)\b(CREATE|SET|DELETE|REMOVE|MERGE)\s*(\(|\[|\{)?" # Split the query into individual statements based on semicolons statements = cypher_query.split(';') # Further process each statement to check for conditional or nested writes def is_write_statement(statement): # Check if the statement includes write operations if re.search(pattern, statement): return True # Check for potentially hidden write operations within sub-queries or function calls nested_patterns = [ r"FOREACH\s*\(([^)]+)\)", # Looking inside FOREACH loops r"CASE\s+WHEN\s+[^:]+:\s+[^:]+ELSE\s+[^:]+END", # Checking CASE statements r"CALL\s+[^()]+(\(.*\))?YIELD\s+[^()]+", # Checking CALL statements ] for nested_pattern in nested_patterns: if re.search(nested_pattern, statement, re.IGNORECASE | re.DOTALL): # Recursively check inside the nested statement match = re.search(nested_pattern, statement, re.IGNORECASE | re.DOTALL) if match and is_write_statement(match.group(1)): return True return False # Filter statements that contain write operations filtered_statements = [stmt for stmt in statements if not is_write_statement(stmt)] # Join the filtered statements back into a single query string filtered_query = '; '.join(filtered_statements) html = construct_hmtl(query = filtered_query) print(res["cypherquery"]) print(filtered_query) return html, filtered_query # --------------------------------------------------------------------------------------------------------------------- ## Calculate Input Storypoints Similarity to Storypoints in Database from openai import OpenAI client = OpenAI(api_key = os.getenv("OPENAI_API_KEY")) def get_embedding_inputstorypoints(storyline_output_storypoint_name_list, model="text-embedding-3-large"): #input has 2 colums, pick the second column storyline_output_storypoint_name_list = [[item[1]] if type(item[1]) is not list else item[1] for item in storyline_output_storypoint_name_list if len(item) > 1] # transform storyline_output_storypoint_name_list to pandas dataframe input_storypoints = pd.DataFrame(storyline_output_storypoint_name_list, columns=['description']) # get embeddings for input storypoints input_storypoints['ada_embedding'] = input_storypoints.description.apply(lambda x: client.embeddings.create(input = [x], model=model).data[0].embedding) return input_storypoints # Function to fetch embeddings from Neo4j def fetch_embeddings(): query = """ MATCH (sp:STORYPOINT) RETURN sp.id AS id, sp.embedding AS embedding """ embeddings = {} driver = GraphDatabase.driver(neo4j_url, auth=AUTH) with driver.session() as session: try: result = session.run(query) except Exception as e: raise gr.Error("Connection to the GraphDatabase failed, please try again in a few seconds! This is probably temporary.", duration=7) for record in result: embeddings[record['id']] = np.array(record['embedding']) driver.close() return embeddings # Function to calculate cosine similarity and find the highest similarities def find_highest_similarities(existing_embeddings, new_embeddings): # Transform embeddings into arrays for the calculation existing_ids, existing_vecs = zip(*existing_embeddings.items()) new_ids, new_vecs = zip(*new_embeddings.items()) existing_vecs = np.array(existing_vecs) new_vecs = np.array(new_vecs) # Calculate cosine similarity similarity_matrix = cosine_similarity(new_vecs, existing_vecs) # Find the index with the highest similarity for each new embeddin TODO: Replace with top 5 most similar max_indices = np.argmax(similarity_matrix, axis=1) similarities = np.max(similarity_matrix, axis=1) # Pair each new storypoint with the existing one that has the highest similarity highest_pairs = [(new_ids[i], existing_ids[max_indices[i]], similarities[i]) for i in range(len(new_ids))] return highest_pairs def coordinate_simcalculation(storyline_output_storypoint_name_list): # Fetch existing embeddings from Neo4j existing_embeddings = fetch_embeddings() input_storypoints = get_embedding_inputstorypoints(storyline_output_storypoint_name_list) # Assume new_embeddings come from your Python processing earlier new_embeddings = {row['description']: row['ada_embedding'] for index, row in input_storypoints.iterrows()} # Find highest similarities highest_similarities = find_highest_similarities(existing_embeddings, new_embeddings) # Display results for new_id, existing_id, similarity in highest_similarities: print(f"Input STORYPOINT '{new_id}' is most similar to existing STORYPOINT '{existing_id}' with a similarity of {similarity:.2f}") HTMLoutput, query = construct_hmtl(highest_similarities) return HTMLoutput, highest_similarities, query def track_user_interaction(user_input, action, user_id): user_id = str(user_id) print(user_id) # Construct connection string from mysql.connector import errorcode try: connection = mysql.connector.connect(user=os.getenv("MYSQLUSER"), password= os.getenv("MYSQLPASSWORD"), host=os.getenv("DBHOST"), port=3306, database="user_interact") print("Connection established") except mysql.connector.Error as err: if err.errno == errorcode.ER_ACCESS_DENIED_ERROR: print("Something is wrong with the user name or password") elif err.errno == errorcode.ER_BAD_DB_ERROR: print("Database does not exist") else: print(err) cursor = connection.cursor() # Create table if it doesn't exist cursor.execute(''' CREATE TABLE IF NOT EXISTS user_interactions ( user_input LONGTEXT NOT NULL, action TEXT NOT NULL, timestamp TEXT NOT NULL, user_id TEXT NOT NULL ); ''') # Prepare data timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") user_input_str = str(user_input) action_str = str(action) # Use a parameterized query to insert data insert_query = """ INSERT INTO user_interactions (user_input, action, timestamp, user_id) VALUES (%s, %s, %s, %s) """ cursor.execute(insert_query, (user_input_str, action_str, timestamp, user_id)) # Commit changes and close connection connection.commit() connection.close() def profile_user(request: gr.Request): query_params = dict(request.query_params) try: username = dict(request.query_params)["username"] user_id = username track_user_interaction("", "login", user_id) #if dict(request.query_params)["password"] == os.getenv("APP_PASSWORD"): # return user_id #else: return user_id except: return None def get_neo4j_response(query): driver = GraphDatabase.driver(neo4j_url, auth=AUTH) #filter out the textual content and embeddings from the response as they waste space and are not needed for visualization with driver.session() as session: result = session.run(query) response = [] for record in result: filtered_record = {} for key, value in record.items(): if isinstance(value, (Node, Relationship)): # Directly filter properties without attempting to recreate the object filtered_properties = {k: v for k, v in value._properties.items() if k not in ["textual_content", "embedding"]} value._properties = filtered_properties filtered_record[key] = value response.append(filtered_record) driver.close() return response def construct_hmtl(highest_similarities = None, nodes_to_show=["SLIDE_DECK", "SLIDE", "STORYPOINT"], query=None): if query is None: storypoint_ids = [existing_id for _, existing_id, _ in highest_similarities] print(storypoint_ids) # Starting with the base of the query query_parts = [ f"WITH {storypoint_ids} AS ids", "MATCH (sp:STORYPOINT) WHERE sp.id IN ids", "WITH sp", "ORDER BY apoc.coll.indexOf(ids, sp.id)", "WITH COLLECT(sp) AS sps", "UNWIND RANGE(0, SIZE(sps) - 2) AS idx", "WITH sps, sps[idx] AS sp_start, sps[idx + 1] AS sp_end", "CALL apoc.create.vRelationship(sp_start, 'FOLLOWS', {}, sp_end) YIELD rel", "WITH sps, sp_start, rel, sp_end", "UNWIND sps AS sp" ] # Initialize the match and return parts of the query match_parts = [] return_parts = [] # Include virtual relationship and its nodes conditionally if "STORYPOINT" in nodes_to_show: return_parts.extend(["sp_start", "rel", "sp_end", "sp"]) # Conditionally add SLIDE and SLIDE_DECK with their relationships if "SLIDE" in nodes_to_show or "SLIDE_DECK" in nodes_to_show: match_parts.append("(sp)<-[r1:ASSIGNED_TO]-(s:SLIDE)") return_parts.extend(["s", "r1"]) if "SLIDE_DECK" in nodes_to_show: match_parts.append("<-[r2:CONTAINS]-(sd:SLIDE_DECK)") return_parts.extend(["sd", "r2"]) # Construct the final query query = "\n".join(query_parts) if match_parts: query += "\nMATCH " + "".join(match_parts) if return_parts: query += "\nRETURN " + ", ".join(return_parts) else: query += "\nRETURN 'No nodes to show based on the selected types'" graphVisualHTML = f"""
{query}