Update src/RAGSample.py

src/RAGSample.py

@@ -117,6 +117,38 @@ class SmartFAQRetriever(BaseRetriever):
     @property
     def k(self):
         return self._k
+
+    def get_documents_with_confidence(self, query: str) -> List[dict]:
+    """Return top documents and their confidence (similarity) scores."""
+    results = self._get_relevant_documents_with_scores(query)
+    return [{"document": doc.page_content, "confidence": round(score, 3)} for doc, score in results]
+
+    
+    def _get_relevant_documents_with_scores(self, query: str) -> List[tuple[Document, float]]:
+    """Retrieve documents along with similarity scores."""
+    if not hasattr(self, '_vectorizer') or self._vectorizer is None or not hasattr(self._vectorizer, 'vocabulary_') or not self._vectorizer.vocabulary_:
+        self._vectorizer = TfidfVectorizer(
+            max_features=3000,
+            stop_words='english',
+            ngram_range=(1, 2),
+            min_df=1,
+            max_df=0.9
+        )
+        questions = [doc.page_content.split("ANSWER:")[0].replace("QUESTION:", "").strip()
+                     if "QUESTION:" in doc.page_content else doc.page_content
+                     for doc in self._documents]
+        self._vectorizer.fit(questions)
+
+    query_vector = self._vectorizer.transform([query.lower().strip()])
+    question_texts = [doc.page_content.split("ANSWER:")[0].replace("QUESTION:", "").strip()
+                      if "QUESTION:" in doc.page_content else doc.page_content
+                      for doc in self._documents]
+    question_vectors = self._vectorizer.transform(question_texts)
+    similarities = cosine_similarity(query_vector, question_vectors).flatten()
+
+    top_indices = similarities.argsort()[-self._k:][::-1]
+    return [(self._documents[i], float(similarities[i])) for i in top_indices if similarities[i] > 0.1]
+
     
     def _get_relevant_documents(self, query: str) -> List[Document]:
         """Retrieve documents based on semantic similarity."""
@@ -447,48 +479,82 @@ class RAGApplication:
         self.retriever = retriever
         self.rag_chain = rag_chain
 
-    def run(self, question: str) -> str:
-        """Runs the RAG pipeline for a given question."""
-        try:
-            # Input validation
-            if not question or not question.strip():
-                return "Please provide a valid question."
+    # def run(self, question: str) -> str:
+    #     """Runs the RAG pipeline for a given question."""
+    #     try:
+    #         # Input validation
+    #         if not question or not question.strip():
+    #             return "Please provide a valid question."
             
-            question = question.strip()
-            print(f"\nProcessing question: '{question}'")
+    #         question = question.strip()
+    #         print(f"\nProcessing question: '{question}'")
             
-            # Retrieve relevant documents
-            documents = self.retriever.invoke(question)
+    #         # Retrieve relevant documents
+    #         documents = self.retriever.invoke(question)
             
-            # Debug: Print retrieved documents
-            print(f"DEBUG: Retrieved {len(documents)} documents")
-            for i, doc in enumerate(documents):
-                print(f"DEBUG: Document {i+1}: {doc.page_content[:200]}...")
+    #         # Debug: Print retrieved documents
+    #         print(f"DEBUG: Retrieved {len(documents)} documents")
+    #         for i, doc in enumerate(documents):
+    #             print(f"DEBUG: Document {i+1}: {doc.page_content[:200]}...")
             
-            # Extract content from retrieved documents
-            doc_texts = "\n\n".join([doc.page_content for doc in documents])
+    #         # Extract content from retrieved documents
+    #         doc_texts = "\n\n".join([doc.page_content for doc in documents])
             
-            # Limit the total input length to prevent token overflow
-            max_input_length = 500  # Conservative limit
-            if len(doc_texts) > max_input_length:
-                doc_texts = doc_texts[:max_input_length] + "..."
-                print(f"DEBUG: Truncated document text to {max_input_length} characters")
+    #         # Limit the total input length to prevent token overflow
+    #         max_input_length = 500  # Conservative limit
+    #         if len(doc_texts) > max_input_length:
+    #             doc_texts = doc_texts[:max_input_length] + "..."
+    #             print(f"DEBUG: Truncated document text to {max_input_length} characters")
             
-            print(f"DEBUG: Combined document text length: {len(doc_texts)}")
+    #         print(f"DEBUG: Combined document text length: {len(doc_texts)}")
             
-            # Get the answer from the language model
-            print("DEBUG: Calling language model...")
-            answer = self.rag_chain.invoke({"question": question, "documents": doc_texts})
-            print(f"DEBUG: Language model response: {answer}")
+    #         # Get the answer from the language model
+    #         print("DEBUG: Calling language model...")
+    #         answer = self.rag_chain.invoke({"question": question, "documents": doc_texts})
+    #         print(f"DEBUG: Language model response: {answer}")
             
-            return answer
+    #         return answer
             
+    #     except Exception as e:
+    #         print(f"Error in RAG application: {str(e)}")
+    #         import traceback
+    #         traceback.print_exc()
+    #         return f"I apologize, but I encountered an error processing your question: {str(e)}. Please try rephrasing it or ask a different question."
+    def run(self, question: str) -> str:
+    try:
+        if not question.strip():
+            return "Please provide a valid question."
+
+        print(f"\nProcessing question: '{question}'")
+
+        if hasattr(self.retriever, "get_documents_with_confidence"):
+            docs_with_scores = self.retriever.get_documents_with_confidence(question)
+            documents = [Document(page_content=d["document"]) for d in docs_with_scores]
+            confidence_info = "\n".join([f"- Score: {d['confidence']}, Snippet: {d['document'][:100]}..." for d in docs_with_scores])
+        else:
+            documents = self.retriever.invoke(question)
+            confidence_info = "Confidence scoring not available."
+
+        print(f"Retrieved {len(documents)} documents")
+        print(confidence_info)
+
+        doc_texts = "\n\n".join([doc.page_content for doc in documents])
+        if len(doc_texts) > 500:
+            doc_texts = doc_texts[:500] + "..."
+
+        answer = self.rag_chain.invoke({"question": question, "documents": doc_texts})
+
+        # Append confidence footer
+        footer = "\n\n(Note: This answer is based on documents with confidence scores. Review full context if critical.)"
+        return answer.strip() + footer
+
         except Exception as e:
             print(f"Error in RAG application: {str(e)}")
             import traceback
             traceback.print_exc()
             return f"I apologize, but I encountered an error processing your question: {str(e)}. Please try rephrasing it or ask a different question."
 
+
 # Main execution block
 if __name__ == "__main__":
     load_dotenv()