from pathlib import Path import pandas as pd # import math, random from sentence_transformers import SentenceTransformer from langchain_huggingface import HuggingFaceEmbeddings from langchain_ollama import ChatOllama from langchain_core.messages import HumanMessage, SystemMessage import chromadb from langchain_chroma import Chroma from sklearn.metrics.pairwise import cosine_similarity from chroma.chatbot_retriever import ask from tqdm import tqdm import random import json import numpy as np import re model = HuggingFaceEmbeddings(model_name="multi-qa-mpnet-base-dot-v1") client = chromadb.PersistentClient(path="chroma/chroma_db") data_dir = Path("datasets") llm = ChatOllama(model="llama3.2", temperature=0) vectorstore = Chroma( client = client, collection_name = "bank_faq", embedding_function = model ) retriever = vectorstore.as_retriever( search_type = "similarity", search_kwargs = {"k": 3} ) # test_docs = vectorstore.similarity_search("what are the fees", k=3) # print("Search results:", test_docs) # test_docs_scores = vectorstore.similarity_search_with_score("what are the fees", k=3) # print("With scores:", test_docs_scores) def llm_judge(query, expected, predicted, llm): prompt = f""" You are evaluating a question answering system. Question: {query} Expected Answer: {expected} Predicted Answer: {predicted} IMPORTANT RULES: - Focus on whether the MEANING and KEY INFORMATION is the same - As long as most of the words matches with the expected answer, do not score as 0.0 - Ignore differences in formatting (bullet points vs numbered list vs plain text) - Ignore minor wording differences as long as the information is the same - Mark as correct if the predicted answer covers all the key points in the expected answer - Mark as incorrect ONLY if key information is missing or wrong Respond ONLY in this JSON format, no other text: - "correct": True if the meaning of the response matches with the expected answer, otherwise False - "score": 0.0 - 1.0 - "reason": one-sentence explanation """ response = llm.invoke(prompt) # If response has a 'content' attribute (e.g. an AIMessage object), use response.content. Otherwise, use response itself. # text = str(getattr(response, "content", response)) # print(text) # Remove Markdown code block markers and trim any leading/trailing whitespace. # text = re.sub(r"```json|```", "", text).strip() # Search for the first JSON object in the text. # \{.*\} matches everything between the first '{' and the last '}' raw = response.content.strip() raw = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw).strip() try: parsed = json.loads(raw) except json.JSONDecodeError as e: print(f"JSON parse failed: {e}\nRaw text: {raw!r}") parsed = {} return { "correct": False, "score": 0.0, "reason": "Predicted does not match with expected", **parsed } # from chroma.chatbot_retriever import ask train_dataset = pd.read_csv(data_dir/"processed"/"bank_faq" /"train_faq.csv") test_dataset = pd.read_csv(data_dir/"processed"/"bank_faq" /"test_faq.csv") train_dataset = train_dataset.rename(columns={ "Question": "query", "Answer": "expected_answer" }).to_dict(orient="records") test_dataset = test_dataset.rename(columns={ "Question": "query", "Answer": "expected_answer" }).to_dict(orient="records") def retrieval_success_check(expected, retrieved_text, threshold=0.6): # embed both the expected answer and the retrieved docs into vector space expected_vec = model.embed_query(expected) retrieved_vec = model.embed_query(retrieved_text) # measure how semantically similar they are (0 = no match, 1 = identical) score = cosine_similarity([expected_vec], [retrieved_vec])[0][0] # consider retrieval successful if similarity meets the threshold return score >= threshold def measure_hallucination(query: str, response: str, chunks: list[dict], llm) -> dict: """ Uses the LLM to check whether the response contains claims not supported by the retrieved chunks. Returns: - hallucinated: bool - confidence: "high" | "medium" | "low" - reason: short explanation """ context = "\n".join(c["text"] for c in chunks) if chunks else "No context retrieved." system = """ You are a hallucination detector. Given a context, a query, and a response, determine if the response contains information NOT supported by the context. Return ONLY valid JSON: - "hallucinated": true if the response contains unsupported claims, false otherwise - "confidence": "high", "medium", or "low" - "reason": one-sentence explanation Return ONLY the JSON object. No explanation. """ result = llm.invoke([ SystemMessage(content=system), HumanMessage(content=( f"Context:\n{context}\n\n" f"Query: {query}\n\n" f"Response: {response}" )) ]) raw = result.content.strip() raw = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw).strip() try: parsed = json.loads(raw) except json.JSONDecodeError: parsed = {} return { "hallucinated": True, "confidence": "low", "reason": "Hallucination check failed", **parsed } FALLBACK_PHRASES = [ "I don't have information on that.", ] def evaluate(dataset: list[dict], split: str = "", sample: int = None): data = random.sample(dataset, sample) if sample else dataset results = [] correct = 0 hallucinated = 0 retrieval_failures = 0 for item in tqdm(data, desc=f"Evaluating {split}", unit="q"): query = item["query"] expected = item["expected_answer"] # run chatbot predicted = ask(query=query, persist=False) # Score-aware retrieval docs_with_scores = vectorstore.similarity_search_with_score(query, k=3) # print("Docs: ", docs_with_scores) # --- retrieval check (simple proxy) --- # retrieved_docs = retriever.invoke(query) retrieved_docs = [doc for doc, _ in docs_with_scores] confidence_scores = [round(1 / (1 + score), 4) for _, score in docs_with_scores] avg_confidence = round(sum(confidence_scores) / len(confidence_scores), 4) if confidence_scores else 0.0 # retrieved_text = " ".join(retrieved_docs).lower() retrieved_text = " ".join([doc.page_content for doc in retrieved_docs]).lower() expected_in_context = expected.lower() in retrieved_text retrieval_success = retrieval_success_check(expected, retrieved_text) # --- correctness --- # correct_flag = is_correct(predicted, expected) is_fallback = any(phrase.lower() in predicted.lower() for phrase in FALLBACK_PHRASES) if is_fallback: judgement = { "correct": False, "score": 0.0, "reason": "Model returned a fallback response" } hallucinated_flag = False hallucination_detail = { "hallucinated": False, "confidence": "high", "reason": "Model does not know the answer" } else: judgement = llm_judge(query, expected, predicted, llm) if judgement["correct"]: hallucinated_flag = False hallucination_detail = { "hallucinated": False, "confidence": "high", "reason": "Response is correct" } correct += 1 else: chunks = [ { "text": doc.page_content, "similarity_score": score } for doc, score in docs_with_scores ] hallucination_detail = measure_hallucination(query, predicted, chunks, llm) hallucinated_flag = hallucination_detail["hallucinated"] # hallucinated_flag = not expected_in_context and not correct_flag and not is_fallback # if correct_flag: # correct += 1 if hallucinated_flag: hallucinated += 1 if not retrieval_success: retrieval_failures += 1 results.append({ "query": query, "expected": expected, "predicted": predicted, "retrieval_success": bool(retrieval_success), "correct": judgement["correct"], "score": judgement["score"], "reason": judgement["reason"], "is_fallback": is_fallback, "hallucination_detail": hallucination_detail, "confidence_scores": confidence_scores, "avg_confidence": avg_confidence }) metrics = { "split": split, "accuracy": correct / len(data), "hallucination_rate": hallucinated / len(data), "retrieval_failure_rate": retrieval_failures / len(data), "samples": len(data) } return metrics, results def save_results(metrics, results, filename="eval_results.json"): output = { "metrics": metrics, "detailed_results": results } with open(filename, "w") as f: json.dump(output, f, indent=2) train_metrics, train_results = evaluate(train_dataset, "train", sample=200) test_metrics, test_results = evaluate(test_dataset, "test", sample=20) # Bucket failures: # retrieval_failed = len([x for x in test_results if not x["retrieval_success"]]) # retrieval_ok_but_wrong = len([x for x in test_results if x["retrieval_success"] and not x["correct"]]) # fallbacks = len([x for x in test_results if x["is_fallback"]]) # print(retrieval_failed) # print(retrieval_ok_but_wrong) # print(fallbacks) # save_results(train_metrics, train_results, "train_eval.json") save_results(test_metrics, test_results, "test_eval.json")