A Coding Implementation to Automating LLM High quality Assurance with DeepEval, Customized Retrievers, and LLM-as-a-Choose Metrics

import sys, os, textwrap, json, math, re
from getpass import getpass


print("🔧 Hardening surroundings (prevents frequent Colab/py3.12 numpy corruption)...")


!pip -q uninstall -y numpy || true
!pip -q set up --no-cache-dir --force-reinstall "numpy==1.26.4"


!pip -q set up -U deepeval openai scikit-learn pandas tqdm


print("✅ Packages put in.")




import numpy as np
import pandas as pd
from tqdm.auto import tqdm


from sklearn.feature_extraction.textual content import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity


from deepeval import consider
from deepeval.test_case import LLMTestCase, LLMTestCaseParams
from deepeval.metrics import (
   AnswerRelevancyMetric,
   FaithfulnessMetric,
   ContextualRelevancyMetric,
   ContextualPrecisionMetric,
   ContextualRecallMetric,
   GEval,
)


print("✅ Imports loaded efficiently.")




OPENAI_API_KEY = getpass("🔑 Enter OPENAI_API_KEY (depart empty to run with out OpenAI): ").strip()
openai_enabled = bool(OPENAI_API_KEY)


if openai_enabled:
   os.environ["OPENAI_API_KEY"] = OPENAI_API_KEY
print(f"🔌 OpenAI enabled: {openai_enabled}")

DOCS = [
   {
       "id": "doc_01",
       "title": "DeepEval Overview",
       "text": (
           "DeepEval is an open-source LLM evaluation framework for unit testing LLM apps. "
           "It supports LLM-as-a-judge metrics, custom metrics like G-Eval, and RAG metrics "
           "such as contextual precision and faithfulness."
       ),
   },
   {
       "id": "doc_02",
       "title": "RAG Evaluation: Why Faithfulness Matters",
       "text": (
           "Faithfulness checks whether the answer is supported by retrieved context. "
           "In RAG, hallucinations occur when the model states claims not grounded in context."
       ),
   },
   {
       "id": "doc_03",
       "title": "Contextual Precision",
       "text": (
           "Contextual precision evaluates how well retrieved chunks are ranked by relevance "
           "to a query. High precision means relevant chunks appear earlier in the ranked list."
       ),
   },
   {
       "id": "doc_04",
       "title": "Contextual Recall",
       "text": (
           "Contextual recall measures whether the retriever returns enough relevant context "
           "to answer the query. Low recall means key information was missed in retrieval."
       ),
   },
   {
       "id": "doc_05",
       "title": "Answer Relevancy",
       "text": (
           "Answer relevancy measures whether the generated answer addresses the user's query. "
           "Even grounded answers can be irrelevant if they don't respond to the question."
       ),
   },
   {
       "id": "doc_06",
       "title": "G-Eval (GEval) Custom Rubrics",
       "text": (
           "G-Eval lets you define evaluation criteria in natural language. "
           "It uses an LLM judge to score outputs against your rubric (e.g., correctness, tone, policy)."
       ),
   },
   {
       "id": "doc_07",
       "title": "What a DeepEval Test Case Contains",
       "text": (
           "A test case typically includes input (query), actual_output (model answer), "
           "expected_output (gold answer), and retrieval_context (ranked retrieved passages) for RAG."
       ),
   },
   {
       "id": "doc_08",
       "title": "Common Pitfall: Missing expected_output",
       "text": (
           "Some RAG metrics require expected_output in addition to input and retrieval_context. "
           "If expected_output is None, evaluation fails for metrics like contextual precision/recall."
       ),
   },
]




EVAL_QUERIES = [
   {
       "query": "What is DeepEval used for?",
       "expected": "DeepEval is used to evaluate and unit test LLM applications using metrics like LLM-as-a-judge, G-Eval, and RAG metrics.",
   },
   {
       "query": "What does faithfulness measure in a RAG system?",
       "expected": "Faithfulness measures whether the generated answer is supported by the retrieved context and avoids hallucinations not grounded in that context.",
   },
   {
       "query": "What does contextual precision mean?",
       "expected": "Contextual precision evaluates whether relevant retrieved chunks are ranked higher than irrelevant ones for a given query.",
   },
   {
       "query": "What does contextual recall mean in retrieval?",
       "expected": "Contextual recall measures whether the retriever returns enough relevant context to answer the query, capturing key missing information issues.",
   },
   {
       "query": "Why might an answer be relevant but still low quality in RAG?",
       "expected": "An answer can address the question (relevant) but still be low quality if it is not grounded in retrieved context or misses important details.",
   },
]

class TfidfRetriever:
   def __init__(self, docs):
       self.docs = docs
       self.texts = [f"{d['title']}n{d['text']}" for d in docs]
       self.vectorizer = TfidfVectorizer(stop_words="english", ngram_range=(1, 2))
       self.matrix = self.vectorizer.fit_transform(self.texts)


   def retrieve(self, question, okay=4):
       qv = self.vectorizer.remodel([query])
       sims = cosine_similarity(qv, self.matrix).flatten()
       top_idx = np.argsort(-sims)[:k]
       outcomes = []
       for i in top_idx:
           outcomes.append(
               {
                   "id": self.docs[i]["id"],
                   "rating": float(sims[i]),
                   "textual content": self.texts[i],
               }
           )
       return outcomes


retriever = TfidfRetriever(DOCS)

def extractive_baseline_answer(question, retrieved_contexts):
   """
   Offline fallback: we create a brief reply by extracting probably the most related sentences.
   This retains the pocket book runnable even with out OpenAI.
   """
   joined = "n".be a part of(retrieved_contexts)
   sents = re.break up(r"(?<=[.!?])s+", joined)
   key phrases = [w.lower() for w in re.findall(r"[a-zA-Z]{4,}", question)]
   scored = []
   for s in sents:
       s_l = s.decrease()
       rating = sum(1 for okay in key phrases if okay in s_l)
       if len(s.strip()) > 20:
           scored.append((rating, s.strip()))
   scored.type(key=lambda x: (-x[0], -len(x[1])))
   greatest = [s for sc, s in scored[:3] if sc > 0]
   if not greatest:
       greatest = [s.strip() for s in sents[:2] if len(s.strip()) > 20]
   ans = " ".be a part of(greatest).strip()
   if not ans:
       ans = "I couldn't discover sufficient context to reply confidently."
   return ans


def openai_answer(question, retrieved_contexts, mannequin="gpt-4.1-mini"):
   """
   Easy RAG immediate for demonstration. DeepEval metrics can nonetheless consider even when
   your era immediate differs; the bottom line is we retailer retrieval_context individually.
   """
   from openai import OpenAI
   consumer = OpenAI()


   context_block = "nn".be a part of([f"[CTX {i+1}]n{c}" for i, c in enumerate(retrieved_contexts)])
   immediate = f"""You're a concise technical assistant.
Use ONLY the supplied context to reply the question. If the reply will not be in context, say you do not know.


Question:
{question}


Context:
{context_block}


Reply:"""
   resp = consumer.chat.completions.create(
       mannequin=mannequin,
       messages=[{"role": "user", "content": prompt}],
       temperature=0.2,
   )
   return resp.decisions[0].message.content material.strip()


def rag_answer(question, retrieved_contexts):
   if openai_enabled:
       attempt:
           return openai_answer(question, retrieved_contexts)
       besides Exception as e:
           print(f"⚠️ OpenAI era failed, falling again to extractive baseline. Error: {e}")
           return extractive_baseline_answer(question, retrieved_contexts)
   else:
       return extractive_baseline_answer(question, retrieved_contexts)

print("n🚀 Working RAG to create check circumstances...")


test_cases = []
Okay = 4


for merchandise in tqdm(EVAL_QUERIES):
   q = merchandise["query"]
   anticipated = merchandise["expected"]


   retrieved = retriever.retrieve(q, okay=Okay)
   retrieval_context = [r["text"] for r in retrieved] 


   precise = rag_answer(q, retrieval_context)


   tc = LLMTestCase(
       enter=q,
       actual_output=precise,
       expected_output=anticipated,
       retrieval_context=retrieval_context,
   )
   test_cases.append(tc)


print(f"✅ Constructed {len(test_cases)} LLMTestCase objects.")


print("n✅ Metrics configured.")


metrics = [
   AnswerRelevancyMetric(threshold=0.5, model="gpt-4.1", include_reason=True, async_mode=True),
   FaithfulnessMetric(threshold=0.5, model="gpt-4.1", include_reason=True, async_mode=True),
   ContextualRelevancyMetric(threshold=0.5, model="gpt-4.1", include_reason=True, async_mode=True),
   ContextualPrecisionMetric(threshold=0.5, model="gpt-4.1", include_reason=True, async_mode=True),
   ContextualRecallMetric(threshold=0.5, model="gpt-4.1", include_reason=True, async_mode=True),


   GEval(
       name="RAG Correctness Rubric (GEval)",
       criteria=(
           "Score the answer for correctness and usefulness. "
           "The answer must directly address the query, must not invent facts not supported by context, "
           "and should be concise but complete."
       ),
       evaluation_params=[
           LLMTestCaseParams.INPUT,
           LLMTestCaseParams.ACTUAL_OUTPUT,
           LLMTestCaseParams.EXPECTED_OUTPUT,
           LLMTestCaseParams.RETRIEVAL_CONTEXT,
       ],
       mannequin="gpt-4.1",
       threshold=0.5,
       async_mode=True,
   ),
]


if not openai_enabled:
   print("n⚠️ You probably did NOT present an OpenAI API key.")
   print("DeepEval's LLM-as-a-judge metrics (AnswerRelevancy/Faithfulness/Contextual* and GEval) require an LLM decide.")
   print("Re-run this cell and supply OPENAI_API_KEY to run DeepEval metrics.")
   print("n✅ Nevertheless, your RAG pipeline + check case development succeeded end-to-end.")
   rows = []
   for i, tc in enumerate(test_cases):
       rows.append({
           "id": i,
           "question": tc.enter,
           "actual_output": tc.actual_output[:220] + ("..." if len(tc.actual_output) > 220 else ""),
           "expected_output": tc.expected_output[:220] + ("..." if len(tc.expected_output) > 220 else ""),
           "contexts": len(tc.retrieval_context or []),
       })
   show(pd.DataFrame(rows))
   elevate SystemExit("Stopped earlier than analysis (no OpenAI key).")

print("n🧪 Working DeepEval consider(...) ...")


outcomes = consider(test_cases=test_cases, metrics=metrics)


summary_rows = []
for idx, tc in enumerate(test_cases):
   row = {
       "case_id": idx,
       "question": tc.enter,
       "actual_output": tc.actual_output[:200] + ("..." if len(tc.actual_output) > 200 else ""),
   }
   for m in metrics:
       row[m.__class__.__name__ if hasattr(m, "__class__") else str(m)] = None


   summary_rows.append(row)


def try_extract_case_metrics(results_obj):
   extracted = []
   candidates = []
   for attr in ["test_results", "results", "evaluations"]:
       if hasattr(results_obj, attr):
           candidates = getattr(results_obj, attr)
           break
   if not candidates and isinstance(results_obj, record):
       candidates = results_obj


   for case_i, case_result in enumerate(candidates or []):
       merchandise = {"case_id": case_i}
       metrics_list = None
       for attr in ["metrics_data", "metrics", "metric_results"]:
           if hasattr(case_result, attr):
               metrics_list = getattr(case_result, attr)
               break
       if isinstance(metrics_list, dict):
           for okay, v in metrics_list.objects():
               merchandise[f"{k}_score"] = getattr(v, "rating", None) if v will not be None else None
               merchandise[f"{k}_reason"] = getattr(v, "cause", None) if v will not be None else None
       else:
           for mr in metrics_list or []:
               identify = getattr(mr, "identify", None) or getattr(getattr(mr, "metric", None), "identify", None)
               if not identify:
                   identify = mr.__class__.__name__
               merchandise[f"{name}_score"] = getattr(mr, "rating", None)
               merchandise[f"{name}_reason"] = getattr(mr, "cause", None)
       extracted.append(merchandise)
   return extracted


case_metrics = try_extract_case_metrics(outcomes)


df_base = pd.DataFrame([{
   "case_id": i,
   "query": tc.input,
   "actual_output": tc.actual_output,
   "expected_output": tc.expected_output,
} for i, tc in enumerate(test_cases)])


df_metrics = pd.DataFrame(case_metrics) if case_metrics else pd.DataFrame([])
df = df_base.merge(df_metrics, on="case_id", how="left")


score_cols = [c for c in df.columns if c.endswith("_score")]
compact = df[["case_id", "query"] + score_cols].copy()


print("n📊 Compact rating desk:")
show(compact)


print("n🧾 Full particulars (contains causes):")
show(df)


print("n✅ Achieved. Tip: if contextual precision/recall are low, enhance retriever rating/protection; if faithfulness is low, tighten era to solely use context.")