Spaces:
Sleeping
Sleeping
File size: 7,638 Bytes
6af9658 f3e893e 6af9658 f3e893e 6af9658 f3e893e 6af9658 f3e893e 6af9658 a422c4a 6af9658 f3e893e 6af9658 | 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 | """Validate generated answers for quality, specificity, and citation accuracy.
Provides:
- AnswerValidator: checks generated answers against quality thresholds
- estimate_confidence: estimates response confidence from chunks + classification
- get_response_prefix: provides appropriate hedging based on confidence
"""
import re
class AnswerValidator:
"""Validates generated answers against retrieved chunks and query context."""
MIN_ANSWER_LENGTH = 100
MIN_CITATIONS = 2
MIN_RELEVANT_PHRASES = {
"obligation": ["shall", "must", "required", "obligation", "duty", "obliged", "mandatory"],
"definition": ["means", "refers to", "is defined as", "shall mean", "constitutes"],
"procedural": ["step", "process", "procedure", "shall", "must", "deadline", "period"],
}
CELEX_PATTERN = r"3\d{4}[A-Z]\d{4}"
STOP_WORDS = {
"what", "is", "are", "the", "a", "an", "of", "in", "to", "for",
"under", "by", "and", "or", "does", "do", "did", "was", "were",
"it", "its", "this", "that", "with", "on", "at", "from", "as",
"be", "been", "being", "have", "has", "had", "not", "no", "but",
}
def validate(self, query: str, answer: str, chunks: list[dict],
classification: dict = None) -> tuple[bool, str]:
"""Validate answer quality. Returns (passes, reason)."""
checks = []
# Check 1: Answer exists and is substantive
if not answer or len(answer.strip()) < self.MIN_ANSWER_LENGTH:
return False, "answer_too_short_or_empty"
checks.append("has_min_length")
# Check 2: Answer cites CELEX numbers from chunks
chunk_celexes = {c.get("celex") for c in chunks if c.get("celex")}
answer_celexes = set(re.findall(self.CELEX_PATTERN, answer))
mentioned = chunk_celexes & answer_celexes
if len(mentioned) < self.MIN_CITATIONS and len(chunk_celexes) >= self.MIN_CITATIONS:
return False, "insufficient_citation_of_retrieved_sources"
checks.append("has_citations")
# Check 3: For obligation queries, verify answer contains deontic language
if classification and classification.get("obligation_seeking"):
has_obligation_lang = any(
word in answer.lower()
for word in self.MIN_RELEVANT_PHRASES["obligation"]
)
if not has_obligation_lang:
return False, "obligation_query_without_obligation_language"
checks.append("has_obligation_language")
# Check 4: Answer addresses question keywords
query_words = query.lower().split()
query_keywords = {w for w in query_words if w not in self.STOP_WORDS and len(w) > 2}
if len(query_keywords) > 2:
answer_lower = answer.lower()
keyword_hits = sum(1 for kw in query_keywords if kw in answer_lower)
if keyword_hits == 0:
return False, "answer_does_not_address_query_keywords"
checks.append("addresses_query_keywords")
return True, "; ".join(checks)
def make_fallback_answer(self, query: str, chunks: list[dict],
classification: dict = None,
validation_reason: str = "") -> str:
"""Generate an informative fallback when validation fails."""
celex_list = list(dict.fromkeys(c.get("celex") for c in chunks if c.get("celex")))
titles = {}
for c in chunks:
celex = c.get("celex")
if celex and celex not in titles:
titles[celex] = c.get("title", "EU legislation")
fallback_parts = [
"I found documents related to your question, but couldn't generate a complete answer from the retrieved text.",
]
if validation_reason == "obligation_query_without_obligation_language":
fallback_parts.append(
"The documents mention this topic but the specific employer "
"obligation language was not found in the retrieved passages."
)
elif validation_reason == "insufficient_citation_of_retrieved_sources":
fallback_parts.append(
"The generated answer did not properly cite the specific "
"legal provisions found in the retrieved documents."
)
if celex_list:
fallback_parts.append("\nRelevant documents found:")
for celex in celex_list[:5]:
title = titles.get(celex, "EU legislation")
fallback_parts.append(f"- {title} (CELEX: {celex})")
fallback_parts.append(
"\nTry asking a more specific question about one of these documents."
)
if classification and classification.get("obligation_seeking"):
fallback_parts.append(
"If you're looking for employer responsibilities, try including "
"terms like 'obligations', 'duties', or 'requirements' in your question."
)
return "\n".join(fallback_parts)
def estimate_confidence(chunks: list[dict], classification: dict = None) -> dict:
"""Estimate confidence level for the generated answer.
Returns a dict with:
- level: 'high', 'medium', 'low'
- overall_score: 0.0-1.0
- factors: dict of contributing factors
"""
if not chunks:
return {"level": "low", "overall_score": 0.0, "factors": {"no_chunks": True}}
factors = {}
# Factor 1: Average chunk relevance score
top_scores = [c.get("score", 0.5) for c in chunks[:5]]
avg_score = sum(top_scores) / max(len(top_scores), 1)
factors["relevance_score"] = max(0.0, min(1.0, 1.0 - avg_score))
# Factor 2: Operative articles vs recitals
article_count = sum(1 for c in chunks if (c.get("article") or "").startswith("art_"))
recital_count = sum(1 for c in chunks if (c.get("article") or "").startswith("rct_"))
total = article_count + recital_count
factors["operative_ratio"] = article_count / total if total > 0 else 0.5
# Factor 3: Deontic language presence for obligation queries
if classification and classification.get("obligation_seeking"):
deontic_count = 0
deontic_words = {"shall", "must", "required", "obliged", "duty", "obligation"}
for c in chunks[:5]:
text_lower = c.get("text", "").lower()
if any(w in text_lower for w in deontic_words):
deontic_count += 1
factors["deontic_presence"] = deontic_count / min(5, max(len(chunks), 1))
# Calculate overall score
weights = {"relevance_score": 0.5, "operative_ratio": 0.3, "deontic_presence": 0.2}
present_factors = {k: v for k, v in factors.items() if k in weights}
if present_factors:
total_weight = sum(weights[k] for k in present_factors)
overall = sum(factors[k] * weights[k] for k in present_factors) / total_weight
else:
overall = 0.5
overall = max(0.0, min(1.0, overall))
if overall >= 0.7:
level = "high"
elif overall >= 0.4:
level = "medium"
else:
level = "low"
return {"level": level, "overall_score": round(overall, 3), "factors": factors}
def get_response_prefix(confidence: dict) -> str:
"""Get appropriate hedging prefix based on confidence level."""
if confidence["level"] == "high":
return "Based on the retrieved EU law documents, "
elif confidence["level"] == "medium":
return "Based on the available legal texts, "
else:
return "Based on partial information from related documents, "
|