import ast import re import pandas as pd import tempfile import os from intent_classification.fd_classification import find_intent from intent_classification.retrieval_classification import handle_query_classification,find_matching_card,generate_card_response_with_context from recommender.retrieval_ranking import generate_multi_queries,convert_to_direct_query_gradio,retrieve_and_rank_cards,generate_credit_card_recommendation_gemini,cross_encoder from data import eligibility_lookup,card_features_lookup,all_card_names,features from recommender.graph_retrieval_vectordb import generate_cypher,run_cypher_query,Neo4jConnectionError #function to pass the retrieved cards and generated response to the UI def recommend_cards_gradio(user_query, preferences, income, cibil, age, min_joining_fee, max_joining_fee, min_annual_fee, max_annual_fee, use_eligibility=True,): try: # print(user_query) if(user_query): result = handle_query_classification(user_query) if result["intent"] == "no_retrieval": return ( f"
" f"
{result['response']}
", [["No retrieval required", "Answered using LLM"]], None, [], {}, "Answered without retrieval" ) elif result["intent"] == "specific": matched_card = find_matching_card(user_query) if matched_card: gemini_answer = generate_card_response_with_context(user_query, matched_card) card_name = matched_card["name"] card_desc = matched_card["description"] card_lookup = {card_name: card_desc} # Constructing eligibility info if available eligibility_info = eligibility_lookup.get(card_name, "No eligibility or fee information available.") chat_history_entry = f"{card_name}:\n{card_desc}\n\nEligibility & Fees:\n{eligibility_info}" return ( f"
" f"
{gemini_answer}
", [["Specific card detected", card_name]], None, [], card_lookup, user_query ) else: return ( "Card mentioned not found in database.", [["Card not found", "Try another card name."]], None, [], {}, "Card not found" ) direct_query,excluded_cards = convert_to_direct_query_gradio(user_query, preferences, all_card_names=all_card_names,feature_list=features) queries = generate_multi_queries(direct_query) if cibil < 700 and use_eligibility: query_intent = True else: query_intent = find_intent(user_query) print(query_intent) cypher_query = generate_cypher(direct_query, query_intent) print("Generated Cypher:\n", cypher_query) try: faiss_index, filtered_mapping = run_cypher_query( user_query, cypher_query, use_eligibility, income, cibil, age, min_joining_fee, max_joining_fee, min_annual_fee, max_annual_fee,excluded_cards ) except Neo4jConnectionError as graph_err: return ( "Graph database connection failed. Please try again later.", [["Graph database error", str(graph_err)]], None, [], {}, "Graph DB connection error" ) cards = retrieve_and_rank_cards(faiss_index, filtered_mapping, direct_query, queries, top_k=10) gemini_summary = generate_credit_card_recommendation_gemini(user_query, direct_query, cards) if not cards: return ( "No eligible cards found.", [["No eligible cards found", "Please try a different query or check your input values."]], None, [], {}, "No eligible card found" ) match = re.search(r"f\.name IN (\[.*?\])", cypher_query) query_features = set(ast.literal_eval(match.group(1))) if match else set() card_rows = [] for score, card in sorted( zip(cross_encoder.predict([[direct_query, card["description"]] for card in cards]), cards), reverse=True, key=lambda x: x[0] ): card_name = card["name"] card_desc = card["description"] matched_features = query_features.intersection(card_features_lookup.get(card_name, set())) feature_str = ", ".join(matched_features) if matched_features else "None" card_rows.append([card_name, feature_str, card_desc]) card_names = [row[0] for row in card_rows] card_lookup = {row[0]: row[2] for row in card_rows} top_card_html = f"""
{gemini_summary}
""" df_cards = pd.DataFrame(card_rows, columns=["Card Name", "Matched Features", "Description"]) filename = "recommended_cards.csv" temp_dir = tempfile.gettempdir() file_path = os.path.join(temp_dir, filename) df_cards.to_csv(file_path, index=False) return top_card_html, card_rows, file_path, card_names, card_lookup, direct_query except Exception as e: print("Error:", e) return ( "An unexpected error occurred. Please try again in a few minutes.", [["Something went wrong", "Please try again."]], None, [], {}, "Unexpected error occurred, please try again in a while" )