File size: 6,965 Bytes
15be85a 6ec6e89 15be85a 8ccdfc2 15be85a 8ccdfc2 8badc83 15be85a 8ccdfc2 15be85a 8ccdfc2 15be85a 6390a72 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | 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"<div style='background-color:#e3f2fd;padding:20px;border-radius:10px;'>"
f"<pre style='white-space:pre-wrap;font-size:13px;color:#212121;'>{result['response']}</pre></div>",
[["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"<div style='background-color:#fffde7;padding:20px;border-radius:10px;'>"
f"<pre style='white-space:pre-wrap;font-size:13px;color:#212121;'>{gemini_answer}</pre></div>",
[["Specific card detected", card_name]],
None,
[],
card_lookup,
user_query
)
else:
return (
"<b style='color:red;'>Card mentioned not found in database.</b>",
[["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 (
"<b style='color:red;'>Graph database connection failed. Please try again later.</b>",
[["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 (
"<b style='color:red;'>No eligible cards found.</b>",
[["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"""
<div style="
background-color: #fff3e0;
color: #212121;
border-radius: 16px;
padding: 20px;
border: 2px solid #ffa726;
box-shadow: 2px 2px 8px rgba(0,0,0,0.1);
margin-bottom: 16px;
font-family: sans-serif;
font-size: 8px;
">
<pre style="white-space: pre-wrap; font-size: 13px; color: #212121;">{gemini_summary}</pre>
</div>
"""
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"
) |