from src.use_llm import main_generate, get_embeddings from src.questions_queries import * import time import uuid import chromadb import spacy import numpy as np import os #os.environ["TOKENIZERS_PARALLELISM"] = "false" import spacy def get_nlp(): try: return spacy.load("en_core_web_sm") except OSError: from spacy.cli import download download("en_core_web_sm") return spacy.load("en_core_web_sm") nlp = get_nlp() wikibase_properties_id = {'instance of': 'P2', 'reference URL': 'P24', 'start time': 'P15', 'end time': 'P16', 'occupation title': 'P25', 'educated at': 'P9', 'employer': 'P10', 'work location': 'P7', 'award received': 'P18', 'point in time': 'P28', 'exact match': 'P23', 'date of birth': 'P3', 'place of birth': 'P4', 'date of death': 'P5', 'country of citizenship': 'P6', 'occupation': 'P19', 'sex or gender': 'P8', 'official website': 'P17', 'perfumes': 'P27', 'who wears it': 'P26', 'inception': 'P11', 'headquarters location': 'P12', 'parent organization': 'P13', 'founded by': 'P14', 'owned by': 'P22', 'industry': 'P20', 'country': 'P30', 'total revenue': 'P21', 'designer employed': 'P29', 'country of origin': 'P30', 'fashion collection': 'P31', 'fashion season': 'P32', 'fashion show location': 'P33', 'description of fashion collection': 'P34', 'image of fashion collection': 'P35', 'editor of fashion collection description': 'P36', 'date of fashion collection': 'P37', 'fashion show category': 'P38', 'fashion house X fashion collection': 'P39'} classes_wikibase = {'fashion designer': 'Q5', 'fashion house': 'Q1', 'business': 'Q9', 'academic institution': 'Q2', 'geographic location': 'Q4', 'fashion award': 'Q8', 'gender': 'Q6', 'occupation': 'Q7', 'human': 'Q36', 'organization': 'Q3', 'brand': 'Q38', 'lifestyle brand': 'Q3417', 'privately held company': 'Q1729', 'fashion season': 'Q8199', 'fashion show category': 'Q8200', 'fashion season collection': 'Q8201', 'fashion journalist': 'Q8207'} questions_queries_all = [{ "question": education_most_popular_question, "query": education_most_popular_query}, { "question": how_many_designers_per_fashion_house_question, "query": how_many_designers_per_fashion_house_query}, {"question": how_many_directors_per_fashion_house_question, "query": how_many_directors_per_fashion_house_query}, {"question": designers_multiple_houses_question, "query":designers_multiple_houses_query }, {"question": award_question, "query": award_question}, {"question": fashion_houses_with_collections_question, "query": fashion_houses_with_collections_query}, {"question": popular_year_inception_question, "query": popular_year_inception_query}, {"question": longest_serving_director_question, "query": longest_serving_director_query}, {"question": houses_most_collections_question, "query": houses_most_collections_query}, {"question": collections_sustainability_theme_question, "query": collections_sustainability_theme_query}, {"question": collections_jeans_question, "query": collections_jeans_query}, {"question": creative_directors_school_question, "query": creative_directors_school_query}, {"question": fashion_houses_thematic_collection_question, "query": fashion_houses_thematic_collection_query}, # {"question": fashion_house_directors_question.substitute({ "x": f"{"Chanel"}"}), "query": fashion_house_directors_query.substitute({ "x": f"'{"Chanel"}'"})}, # { "question": designer_fashion_house_directors_question.substitute({ "x": f"{"Alexander McQueen"}"}), "query": designer_fashion_house_directors_query.substitute({ "x": f"'{"Alexander McQueen"}'"})}, # {"question": country_designer_question.substitute({ "x": f"{"Italy"}"}), "query": country_designer_query.substitute({ "x": f"'{"Italy"}'"})}, # { "question": designer_order_fashion_collection_question.substitute({ "x": f"{"Alexander McQueen"}"}), "query": designer_order_fashion_collection_query.substitute({ "x": f"'{"Alexander McQueen"}'"})}, # {"question": designer_fashion_director_question2.substitute({ "x": f"{"Alexander McQueen"}"}), "query": designer_fashion_director_query2.substitute({ "x": f"'{"Alexander McQueen"}'"})}, # { "question": year_designers_birth_question.substitute({ "x": 1970}), "query": year_designers_birth_query.substitute({ "x": 1970})}, # { "question": fashion_collection_images_question.substitute({ "x": f"{"Chanel"}"}), "query": fashion_collection_images_query.substitute({ "x": f"'{"Chanel"}'"})} ] if os.path.exists("web_app/query_log.json"): other_pairs = pd.read_json("web_app/query_log.json") other_pairs_success = other_pairs[(other_pairs["status"] == "Success") & (other_pairs["feedback"] == "good")] other_pairs_success = other_pairs_success[["question", "query"]] questions_queries_all = questions_queries_all + other_pairs_success.to_dict(orient='records') #print only the questions #print([q["question"] for q in questions_queries_all]) def mask_entities(text, nlp): doc = nlp(text) masked_text = text for ent in doc.ents: masked_text = masked_text.replace(ent.text, "[ENTITY]") return masked_text import re import spacy def replace_entity(original_question, to_do_question, query): """ Replaces entities in the query using entities from to_do_question while preserving quotation marks. Handles multiple entity replacements and numerical entity replacements. """ nlp = spacy.load("en_core_web_sm") original_doc = nlp(original_question) to_do_doc = nlp(to_do_question) # Extract entities from both questions original_entities = [ent.text for ent in original_doc.ents] to_do_entities = [ent.text for ent in to_do_doc.ents] # print("Original Entities:", original_entities) # print("To-Do Entities:", to_do_entities) # Create entity mapping entity_mapping = {} for orig_ent, new_ent in zip(original_entities, to_do_entities): # Find numbers in each entity orig_numbers = re.findall(r"\d+", orig_ent) new_numbers = re.findall(r"\d+", new_ent) if orig_numbers and new_numbers and len(orig_numbers) == len(new_numbers): # If multiple numbers, replace each one for orig_num, new_num in zip(orig_numbers, new_numbers): entity_mapping[orig_num] = new_num else: # Otherwise, replace entire entity entity_mapping[orig_ent] = new_ent #print("Entity Mapping:", entity_mapping) # Replace entities in the query for orig, new in entity_mapping.items(): query = re.sub(rf'("{orig}"|\b{re.escape(orig)}\b)', lambda match: f'"{new}"' if match.group(0).startswith('"') else new, query) return query def capitalize_sentences(sentences): """ Ensures that each sentence in a list starts with an uppercase letter. """ capitalized_sentences = [] for sentence in sentences: sentence = sentence.strip() # Remove leading/trailing spaces if sentence: # Check if the sentence is not empty sentence = sentence[0].upper() + sentence[1:] # Capitalize first letter capitalized_sentences.append(sentence) return capitalized_sentences def similarity_question(question, questions_queries_dictionary, collection, n_results=5, threshold=0.15): """ Removes duplicate embeddings and retrieves similar questions. """ nlp = spacy.load("en_core_web_sm") # Load spaCy model for entity recognition original_documents = [questions_queries_dictionary[i]["question"] for i in range(len(questions_queries_dictionary))] masked_documents = [mask_entities(q, nlp) for q in original_documents] # Dictionary to store unique embeddings unique_embeddings = {} # Store each unique document in the vector embedding database for i, d in enumerate(masked_documents): response = get_embeddings(d) embedding = response["embeddings"][0] # Extract the first (and only) embedding from the nested list # Check if embedding is unique is_duplicate = any(np.allclose(embedding, np.array(e), atol=1e-6) for e in unique_embeddings.values()) if not is_duplicate: unique_embeddings[str(i)] = embedding # Store unique embedding as a list collection.add( ids=[str(i)], embeddings=[embedding], # Ensure this is a list of lists documents=[d] ) # Compute the embedding for the input question masked_question = mask_entities(question, nlp) response = get_embeddings(d) query_embedding = response["embeddings"][0] # Extract embedding results = collection.query( query_embeddings=[query_embedding], # Ensure correct format n_results=n_results ) triples = [] for i in range(len(results['documents'][0])): masked_similar_question = results['documents'][0][i] distance = results['distances'][0][i] print(distance) paraphrase = distance < threshold # Find the corresponding original question index_similar_query = masked_documents.index(masked_similar_question) original_similar_question = original_documents[index_similar_query] similar_query = questions_queries_dictionary[index_similar_query]["query"] if paraphrase and "[ENTITY]" in masked_similar_question and "[ENTITY]" in masked_question: to_do_query = replace_entity(original_similar_question, question, similar_query) else: to_do_query = None triples.append((original_similar_question, similar_query, to_do_query)) return triples def similarity_question_no_masking(question, questions_queries_dictionary, collection, n_results=5, threshold=0.15): """ Removes duplicate embeddings and retrieves similar questions. """ original_documents = [questions_queries_dictionary[i]["question"] for i in range(len(questions_queries_dictionary))] # Dictionary to store unique embeddings unique_embeddings = {} # Store each unique document in the vector embedding database for i, d in enumerate(original_documents): response = get_embeddings(d) embedding = response["embeddings"][0] # Extract the first (and only) embedding from the nested list # Check if embedding is unique is_duplicate = any(np.allclose(embedding, np.array(e), atol=1e-6) for e in unique_embeddings.values()) if not is_duplicate: unique_embeddings[str(i)] = embedding # Store unique embedding as a list collection.add( ids=[str(i)], embeddings=[embedding], # Ensure this is a list of lists documents=[d] ) # Compute the embedding for the input question response = get_embeddings(question) query_embedding = response["embeddings"][0] # Extract embedding results = collection.query( query_embeddings=[query_embedding], # Ensure correct format n_results=n_results ) triples = [] for i in range(len(results['documents'][0])): similar_question = results['documents'][0][i] distance = results['distances'][0][i] print(distance) paraphrase = distance < threshold # Find the corresponding original question index_similar_query = original_documents.index(similar_question) original_similar_question = original_documents[index_similar_query] similar_query = questions_queries_dictionary[index_similar_query]["query"] to_do_query = similar_query if paraphrase else None triples.append((original_similar_question, similar_query, to_do_query)) return triples def select_dict(dict, keys): return {k: dict[k] for k in keys if k in dict} def prompt_template(to_do_question,triples_examples,wikibase_properties_id,how_many_examples = 1, ): questions = [triples_examples[i][0] for i in range(len(triples_examples))][:how_many_examples] print("EXAMPLE QUESTION(s): ",questions) classes_wikibase_selection = select_dict(classes_wikibase, ["fashion house", "fashion designer"]) general_properties = select_dict(wikibase_properties_id, ["instance of", "reference URL", "start time", "end time", "occupation title", "point in time", "official website"]) general_properties["rdfs:label"] = "rdfs:label" designer_properties = select_dict(wikibase_properties_id, ["employer", "educated at", "work location", "award received", "date of birth", "date of death", "place of birth", "country of citizenship", "occupation", "sex or gender"]) fashion_house_properties = select_dict(wikibase_properties_id, ["inception","headquarters location", "parent organization", "founded by","owned by", "industry", "country", "total revenue", "designer employed", "fashion collection", "description of fashion collection","image of fashion collection"]) fashion_collection_properties = select_dict(wikibase_properties_id, ["fashion show category", "fashion show location", "fashion season"]) qualifier_properties = select_dict(wikibase_properties_id, ["start time", "end time", "occupation title", "point in time","description of fashion collection","image of fashion collection"]) prompt = f"""You are an expert in translating natural language questions into SPARQL queries for FashionDB - a knwoledge graph about Fashion. I provide you with the ontology of FashionDB. The properties are stored in a dictionary as property_label: property_id. The classes are stored in a dictionary as class_label: class_id. General Properties: {general_properties}, Fashion Designer Properties: {designer_properties}, Fashion House Properties: {fashion_house_properties}, Fashion Collection Properties: {fashion_collection_properties}. In particular the following properties are always qualifiers thus their prefix is always pq: {qualifier_properties}. Classes: {classes_wikibase_selection}. Remember to use the entities presented in Natural language question to translate , when generating the corresponding SPARQL query. I provide you with example.""" for i in range(len(questions)): prompt += f""" Example question: {triples_examples[i][0]} Corresponding SPARQL query:{triples_examples[i][1]} """ prompt += f""" Question to translate to SPARQL: {to_do_question} Remember that the use case is FASHION: if there is a mispelling of a fashion designer or house, you can adjust it according to your knowledge of fashion. Example: "balenciaho" should be "Balenciaga". Your generated corresponding SPARQL query: """ return prompt def prompt_template_gemma2(to_do_question, triples_examples, wikibase_properties_id, how_many_examples=1): questions = [triples_examples[i][0] for i in range(len(triples_examples))][:how_many_examples] print("EXAMPLE QUESTION(s): ",questions) classes_wikibase_selection = select_dict(classes_wikibase, ["fashion house", "fashion designer"]) general_properties = select_dict(wikibase_properties_id, ["instance of", "reference URL", "start time", "end time", "occupation title", "point in time", "official website"]) general_properties["rdfs:label"] = "rdfs:label" designer_properties = select_dict(wikibase_properties_id, ["employer", "educated at", "work location", "award received", "date of birth", "date of death", "place of birth", "country of citizenship", "occupation", "sex or gender"]) fashion_house_properties = select_dict(wikibase_properties_id, ["inception", "headquarters location", "parent organization", "founded by", "owned by", "industry", "country", "total revenue", "designer employed", "fashion collection", "description of fashion collection", "image of fashion collection"]) fashion_collection_properties = select_dict(wikibase_properties_id, ["fashion show category", "fashion show location", "fashion season"]) qualifier_properties = select_dict(wikibase_properties_id, ["start time", "end time", "occupation title", "point in time", "description of fashion collection", "image of fashion collection"]) prompt = f""" You are an expert in translating natural language fashion-related questions into **SPARQL queries** for **FashionDB**, a knowledge graph about fashion. --- ## **FashionDB Ontology** - **Classes**: {classes_wikibase_selection} - **General Properties**: {general_properties} - **Fashion Designer Properties**: {designer_properties} - **Fashion House Properties**: {fashion_house_properties} - **Fashion Collection Properties**: {fashion_collection_properties} - **Qualifier Properties** (always prefixed with `pq:`): {qualifier_properties} --- ## **Instructions** - **Fix misspellings** of fashion brands and designers before generating the query. - Example: "Guxci" → **"Gucci"**, "Balenciaho" → **"Balenciaga"**. - If a brand or designer **isn't recognized**, **make a reasonable correction** based on common fashion knowledge. - Handle **abstract or conceptual fashion questions**, such as: - "Which fashion houses have had the most influence in the 20th century?" - "What are the key design trends in haute couture from the 1990s?" - **Always return a valid SPARQL query** using the provided ontology. --- ## **Example(s)** """ for i in range(len(questions)): prompt += f""" **Example {i+1}** - **Question**: {triples_examples[i][0]} - **SPARQL Query**: ```sparql {triples_examples[i][1]} ``` """ prompt += f""" --- ## **Your Task** **Question**: {to_do_question} **SPARQL Query:** ```sparql """ return prompt def prompt_template_gpt4o_mini(to_do_question, triples_examples, wikibase_properties_id, how_many_examples=1): questions = [triples_examples[i][0] for i in range(len(triples_examples))][:how_many_examples] classes_wikibase_selection = select_dict(classes_wikibase, ["fashion house", "fashion designer"]) general_properties = select_dict(wikibase_properties_id, ["instance of", "reference URL", "start time", "end time", "occupation title", "point in time", "official website"]) general_properties["rdfs:label"] = "rdfs:label" designer_properties = select_dict(wikibase_properties_id, ["employer", "educated at", "work location", "award received", "date of birth", "date of death", "place of birth", "country of citizenship", "occupation", "sex or gender"]) fashion_house_properties = select_dict(wikibase_properties_id, ["inception", "headquarters location", "parent organization", "founded by", "owned by", "industry", "country", "total revenue", "designer employed", "fashion collection", "description of fashion collection", "image of fashion collection"]) fashion_collection_properties = select_dict(wikibase_properties_id, ["fashion show category", "fashion show location", "fashion season"]) qualifier_properties = select_dict(wikibase_properties_id, ["start time", "end time", "occupation title", "point in time", "description of fashion collection", "image of fashion collection"]) prompt = f""" You are a **SPARQL expert** specializing in **FashionDB**, a knowledge graph about fashion. ### **Your Task** - Translate the given **natural language question** into a **valid SPARQL query**. - **Fix spelling mistakes** of fashion brands and designers. - Example: "Guxci" → "Gucci", "Balenciaho" → "Balenciaga". - If a brand or designer isn't recognized, **guess the correct name** based on fashion industry knowledge. - Support **abstract fashion questions**, such as: - "How did Dior's designs evolve over the decades?" - "Which fashion houses had the biggest impact on 21st-century streetwear?" - Your **SPARQL query must use the correct ontology**. --- ### **FashionDB Ontology** - **Classes**: {classes_wikibase_selection} - **General Properties**: {general_properties} - **Fashion Designer Properties**: {designer_properties} - **Fashion House Properties**: {fashion_house_properties} - **Fashion Collection Properties**: {fashion_collection_properties} - **Qualifier Properties (always prefixed with `pq:`)**: {qualifier_properties} --- ### **Example(s)** """ for i in range(len(questions)): prompt += f""" **Example {i+1}** - **Question**: {triples_examples[i][0]} - **SPARQL Query**: ```sparql {triples_examples[i][1]} ``` """ prompt += f""" --- ### **Now Translate This Question** **Question**: {to_do_question} **SPARQL Query:** ```sparql """ return prompt #validate def replace_last_occurrence(s, pattern, replacement): pos = s.rfind(pattern) # Find the last occurrence of the pattern if pos != -1: return s[:pos] + s[pos:].replace(pattern, replacement, 1) def validation_query(sparql_query): if sparql_query.startswith("sparql"): sparql_query = sparql_query[6:] #if last character is \n remove it while sparql_query[-1] == "\n" or sparql_query[-1] == " ": sparql_query = sparql_query[:-1] if sparql_query[-1] == ".": sparql_query = sparql_query[:-1] sparql_query = sparql_query.encode().decode('unicode_escape') sparql_query = sparql_query.replace("wdt", "wbt") if "SERVICE" not in sparql_query: sparql_query = replace_last_occurrence(sparql_query, "}", "SERVICE wikibase:label { bd:serviceParam wikibase:language 'en'. } \n }") return sparql_query def safe_get_results(query, max_retries=3): """ Safely executes a SPARQL query, handling HTTP errors gracefully. Parameters: - query (str): The SPARQL query to execute. - max_retries (int): Number of retries before failing. Returns: - DataFrame: Query results, or an empty DataFrame if the query fails. """ for attempt in range(max_retries): try: return get_results_to_df(query) # Attempt to execute the query except requests.exceptions.HTTPError as e: print(f"Attempt {attempt + 1}: Query failed with HTTPError {e}") time.sleep(2) # Wait before retrying except Exception as e: print(f"Attempt {attempt + 1}: Unexpected error {e}") time.sleep(2) print("All attempts failed. Returning empty DataFrame.") return pd.DataFrame() # Return empty DataFrame if all retries fail def correction_question_prompt(to_do_question): correction_prompt = f""" You are an expert in **fashion brand and designer names**. Your task is to **correct misspellings** in the given question while keeping its original meaning. If you recognize a fashion-related name that is misspelled, **fix it**. If nothing is wrong, generate the Question to Correct. Don't generate **. ### **Examples** - "Who founded Guxci?" → "Who founded Gucci?" - "What is balenciaho famous for?" → "What is Balenciaga famous for?" - "Who is the head designer of gucxi?" → "Who is the head designer of Gucci?" ### **Question to Correct** {to_do_question} ### **Corrected Version** """ return correction_prompt def initialize_collection(): # Initialize ChromaDB client client = chromadb.Client() # If the collection already exists, delete it to start fresh. try: client.delete_collection(name="docs") # Delete the existing collection except: pass # Re-create the collection for each query collection = client.create_collection(name="docs") return collection def main_generate_queries(to_do_question): # # Initialize ChromaDB client # client = chromadb.Client() # # If the collection already exists, delete it to start fresh. # try: # client.delete_collection(name="docs") # Delete the existing collection # except: # pass # # Re-create the collection for each query # collection = client.create_collection(name="docs") collection = initialize_collection() triples_examples = similarity_question(to_do_question, questions_queries_all, collection) if triples_examples[0][2] is not None: print("it's a paraphrase :)") sparql_query = triples_examples[0][2] print(triples_examples[0][0]) result_query = safe_get_results(sparql_query) if result_query.empty: to_do_question = main_generate(correction_question_prompt(to_do_question), "gemma2", "You have to fix the mispellings of the Question to Correct") print(to_do_question) sparql_query = replace_entity(triples_examples[0][0], to_do_question, triples_examples[0][1]) result_query = safe_get_results(sparql_query) print(sparql_query) if not result_query.empty: return result_query.to_dict(orient='records'), sparql_query prompt = prompt_template_gemma2(to_do_question, triples_examples, wikibase_properties_id, how_many_examples=1) sparql_query = main_generate(prompt, "gemma2", "You are a natural language to SPARQL language translator. Do only generate the SPARQL query, nothing else.") sparql_query = validation_query(sparql_query) result_query = safe_get_results(sparql_query) print(sparql_query) if result_query.empty: to_do_question = main_generate(correction_question_prompt(to_do_question), "gemma2", "You have to fix the mispellings of the Question to Correct") print(to_do_question) prompt = prompt_template_gemma2(to_do_question, triples_examples, wikibase_properties_id, how_many_examples=2) sparql_query = main_generate(prompt, "gemma2", "You are a natural language to SPARQL language translator. Do only generate the SPARQL query, nothing else.") sparql_query = validation_query(sparql_query) result_query = safe_get_results(sparql_query) if result_query.empty: new_collection = initialize_collection() triples_examples_no_masked = similarity_question_no_masking(to_do_question, questions_queries_all, new_collection) prompt = prompt_template_gemma2(to_do_question, triples_examples_no_masked, wikibase_properties_id, how_many_examples=2) sparql_query = main_generate(prompt, "gemma2", "You are a natural language to SPARQL language translator. Do only generate the SPARQL query, nothing else.") sparql_query = validation_query(sparql_query) result_query = safe_get_results(sparql_query) print(sparql_query) if result_query.empty: text_generated = main_generate(to_do_question, "gemma2", "You are an expert in fashion. Just provide the answer to the question.") return text_generated, sparql_query print(sparql_query) print(result_query) return result_query.to_dict(orient='records'), sparql_query # #main("What is the inception of Chanel?") # if __name__ == "__main__": # #main("Which fashion designers being creative directors were born in Italy?") # #main_generate_queries("Which fashion houses had collections with jeans in their descriptions and how many of the collections have jeans?") # main_generate_queries("Which designers were born in 1970?")