Spaces:
Sleeping
Sleeping
File size: 2,767 Bytes
119a07a 6ad7021 119a07a 6ad7021 119a07a |
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 |
import faiss
import numpy as np
from langchain_community.embeddings import HuggingFaceEmbeddings
class QuestionRecommender:
def __init__(self, faiss_index_path, questions_path, embedding_model: HuggingFaceEmbeddings, top_k=5):
"""
MODIFIED: Now requires an embedding_model to handle new queries.
"""
print("Initializing Question Recommender...")
self.index = faiss.read_index(faiss_index_path)
self.questions = np.load(questions_path, allow_pickle=True)
self.embedding_model = embedding_model
self.top_k = top_k
self.start_questions = [
"What is Sat2Farm?",
"Can someone without farming background do farming using your advisories?",
"How to add my farm in the App?",
"Is the app available for Iphone?",
"Is the app free?",
]
self.history = []
self.current_recommendations = []
print("✅ Question Recommender initialized.")
def get_initial_questions(self):
"""Gets the initial set of questions and resets the state for a new session."""
self.history = []
self.current_recommendations = self.start_questions
return self.current_recommendations
def recommend(self, query: str):
"""
MODIFIED: Now handles any query string, not just a selected question.
"""
if self.current_recommendations:
self.history.append(self.current_recommendations)
embedding = None
try:
q_idx = np.where(self.questions == query)[0][0]
embedding = self.index.reconstruct(int(q_idx)).reshape(1, -1)
print(f"Recommending based on known question: '{query}'")
except IndexError:
print(f"Recommending based on new user query: '{query}'")
embedding = np.array(self.embedding_model.embed_query(query)).reshape(1, -1)
if embedding is not None:
distances, indices = self.index.search(embedding, self.top_k + 1)
recommended = [
self.questions[i] for i in indices[0]
if i < len(self.questions) and self.questions[i] != query
]
self.current_recommendations = recommended[:self.top_k]
return self.current_recommendations
return self.start_questions
def go_back(self):
"""Returns the previous set of recommended questions from history."""
if self.history:
self.current_recommendations = self.history.pop()
return self.current_recommendations
else:
print("💡 No more history. Returning to initial questions.")
return self.get_initial_questions() |