Spaces:
Sleeping
Sleeping
File size: 9,988 Bytes
e7a3876 | 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 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 | """
RAG Evaluation — lightweight implementation of RAGAS-style metrics.
Why not use the ragas library directly:
ragas has unstable langchain_community dependencies that break frequently.
This module implements the same four core metrics from first principles
using only sentence-transformers and the Groq LLM we already have.
Metrics implemented:
1. Faithfulness — are answer claims supported by context?
2. Answer Relevancy — does the answer address the question?
3. Context Precision — are retrieved chunks relevant to the question?
4. Context Recall — were all needed chunks retrieved?
"""
import numpy as np
from sentence_transformers import SentenceTransformer
from src.utils.logger import logger
# Use the same embedding model as the retrieval pipeline
_embedder = None
def _get_embedder():
global _embedder
if _embedder is None:
_embedder = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2")
return _embedder
def _cosine_similarity(a, b):
a, b = np.array(a), np.array(b)
return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
# ------------------------------------------------------------
# Metric 1: Faithfulness
# "Is every claim in the answer supported by the retrieved context?"
# Score 0-1. Low score = hallucination.
# ------------------------------------------------------------
def faithfulness(answer: str, context_chunks: list[str], threshold: float = 0.5) -> dict:
"""
Checks if each sentence in the answer has semantic support
in at least one retrieved chunk.
Args:
answer: The LLM-generated answer string.
context_chunks: List of retrieved chunk texts.
threshold: Minimum cosine similarity to consider a sentence supported.
Returns:
{
"score": float, # 0-1, fraction of sentences supported
"supported": int, # number of supported sentences
"total": int, # total sentences checked
"details": list[dict] # per-sentence breakdown
}
Note: Production RAGAS uses an LLM to decompose answers into atomic claims.
This implementation uses embedding similarity as a faster approximation.
"""
model = _get_embedder()
sentences = [s.strip() for s in answer.split(".") if s.strip()]
if not sentences:
return {"score": 0.0, "supported": 0, "total": 0, "details": []}
chunk_embeddings = model.encode(context_chunks)
details = []
supported = 0
for sentence in sentences:
sentence_embedding = model.encode(sentence)
max_similarity = max(
_cosine_similarity(sentence_embedding, chunk_emb)
for chunk_emb in chunk_embeddings
)
is_supported = max_similarity >= threshold
if is_supported:
supported += 1
details.append({
"sentence": sentence,
"max_similarity": round(max_similarity, 4),
"supported": is_supported,
})
score = supported / len(sentences)
logger.debug(f"Faithfulness: {score:.2f} ({supported}/{len(sentences)} sentences supported)")
return {
"score": round(score, 4),
"supported": supported,
"total": len(sentences),
"details": details,
}
# ------------------------------------------------------------
# Metric 2: Answer Relevancy
# "Does the answer actually address the question asked?"
# Score 0-1. Low score = answer is off-topic or evasive.
# ------------------------------------------------------------
def answer_relevancy(question: str, answer: str) -> dict:
"""
Measures semantic similarity between the question and answer.
High similarity = answer directly addresses the question.
Note: Production RAGAS generates multiple questions from the answer
and measures how well they reconstruct the original question.
This is a simpler direct similarity approximation.
"""
model = _get_embedder()
question_embedding = model.encode(question)
answer_embedding = model.encode(answer)
score = _cosine_similarity(question_embedding, answer_embedding)
# Clamp to [0, 1] — cosine can be slightly negative
score = max(0.0, min(1.0, score))
logger.debug(f"Answer relevancy: {score:.2f}")
return {
"score": round(score, 4),
"question": question,
"answer_preview": answer[:100] + "..." if len(answer) > 100 else answer,
}
# ------------------------------------------------------------
# Metric 3: Context Precision
# "Are the retrieved chunks actually relevant to the question?"
# Score 0-1. Low score = too many irrelevant chunks retrieved.
# ------------------------------------------------------------
def context_precision(question: str, context_chunks: list[str], threshold: float = 0.4) -> dict:
"""
Measures what fraction of retrieved chunks are relevant to the question.
Signal-to-noise ratio of retrieval.
Args:
question: The user's question.
context_chunks: Retrieved chunk texts.
threshold: Minimum similarity to consider a chunk relevant.
"""
if not context_chunks:
return {"score": 0.0, "relevant": 0, "total": 0, "details": []}
model = _get_embedder()
question_embedding = model.encode(question)
details = []
relevant = 0
for i, chunk in enumerate(context_chunks):
chunk_embedding = model.encode(chunk)
similarity = _cosine_similarity(question_embedding, chunk_embedding)
is_relevant = similarity >= threshold
if is_relevant:
relevant += 1
details.append({
"chunk_index": i,
"similarity": round(similarity, 4),
"relevant": is_relevant,
"preview": chunk[:80] + "..." if len(chunk) > 80 else chunk,
})
score = relevant / len(context_chunks)
logger.debug(f"Context precision: {score:.2f} ({relevant}/{len(context_chunks)} chunks relevant)")
return {
"score": round(score, 4),
"relevant": relevant,
"total": len(context_chunks),
"details": details,
}
# ------------------------------------------------------------
# Metric 4: Context Recall
# "Did we retrieve all the chunks needed to answer fully?"
# Score 0-1. Low score = answer is incomplete due to missing context.
# ------------------------------------------------------------
def context_recall(answer: str, context_chunks: list[str], threshold: float = 0.5) -> dict:
"""
Measures what fraction of the answer's content is attributable
to the retrieved context. Inverse of faithfulness framing —
here we measure coverage rather than support.
Note: Production RAGAS uses ground truth answers for recall.
Without ground truth, we approximate by checking how much of
the answer is semantically covered by the context.
"""
model = _get_embedder()
sentences = [s.strip() for s in answer.split(".") if s.strip()]
if not sentences or not context_chunks:
return {"score": 0.0, "covered": 0, "total": 0}
chunk_embeddings = model.encode(context_chunks)
covered = 0
for sentence in sentences:
sentence_embedding = model.encode(sentence)
max_similarity = max(
_cosine_similarity(sentence_embedding, chunk_emb)
for chunk_emb in chunk_embeddings
)
if max_similarity >= threshold:
covered += 1
score = covered / len(sentences)
logger.debug(f"Context recall: {score:.2f} ({covered}/{len(sentences)} sentences covered)")
return {
"score": round(score, 4),
"covered": covered,
"total": len(sentences),
}
# ------------------------------------------------------------
# Combined evaluation — run all four metrics at once
# ------------------------------------------------------------
def evaluate(
question: str,
answer: str,
context_chunks: list[str],
faithfulness_threshold: float = 0.5,
precision_threshold: float = 0.4,
) -> dict:
"""
Run all four metrics and return a combined evaluation report.
Args:
question: The user's question.
answer: The LLM-generated answer.
context_chunks: List of retrieved chunk texts.
Returns:
Dict with all four metric results and an overall summary.
"""
logger.info("Running RAG evaluation...")
faith = faithfulness(answer, context_chunks, faithfulness_threshold)
relevancy = answer_relevancy(question, answer)
precision = context_precision(question, context_chunks, precision_threshold)
recall = context_recall(answer, context_chunks, faithfulness_threshold)
# Overall score — average of four metrics
overall = np.mean([
faith["score"],
relevancy["score"],
precision["score"],
recall["score"],
])
report = {
"overall": round(float(overall), 4),
"faithfulness": faith["score"],
"answer_relevancy": relevancy["score"],
"context_precision": precision["score"],
"context_recall": recall["score"],
"details": {
"faithfulness": faith,
"answer_relevancy": relevancy,
"context_precision": precision,
"context_recall": recall,
}
}
logger.info(
f"Evaluation complete — "
f"overall={report['overall']:.2f} | "
f"faithfulness={faith['score']:.2f} | "
f"relevancy={relevancy['score']:.2f} | "
f"precision={precision['score']:.2f} | "
f"recall={recall['score']:.2f}"
)
return report |