File size: 4,291 Bytes
db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b db71234 9605d8b | 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 | from typing import Dict, List
import re
from src.retrieval import assess_evidence
def clean_markdown(text: str) -> str:
text = text.replace("\n", " ")
text = re.sub(r"```.*?```", " ", text, flags=re.DOTALL)
text = re.sub(r"#+\s*", "", text)
text = re.sub(r"[-*+]\s+", "", text)
text = re.sub(r"\[(.*?)\]\(.*?\)", r"\1", text)
text = re.sub(r"\*\*(.*?)\*\*", r"\1", text)
text = re.sub(r"\*(.*?)\*", r"\1", text)
text = re.sub(r"\s{2,}", " ", text)
return text.strip()
def question_terms(question: str) -> List[str]:
cleaned = re.sub(r"[^a-zA-Z0-9 ]", " ", question.lower())
tokens = [token for token in cleaned.split() if len(token) > 2]
stop_words = {
"what", "when", "where", "which", "who", "why", "how",
"the", "and", "for", "that", "this", "from", "with", "about",
"can", "use", "uses", "using", "are", "is", "of",
}
return [token for token in tokens if token not in stop_words]
def extract_sentences(text: str) -> List[str]:
cleaned = clean_markdown(text)
parts = re.split(r"(?<=[.!?])\s+", cleaned)
sentences = [part.strip() for part in parts if len(part.strip()) > 20]
return sentences
def score_sentence(sentence: str, terms: List[str]) -> float:
lower = sentence.lower()
term_matches = sum(1 for term in terms if term in lower)
length_bonus = min(len(sentence.split()) / 25, 1)
return term_matches * 2 + length_bonus
def extract_relevant_sentences(text: str, question: str, max_sentences: int = 2) -> List[str]:
sentences = extract_sentences(text)
if not sentences:
return []
terms = question_terms(question)
scored = [(score_sentence(sentence, terms), sentence) for sentence in sentences]
scored.sort(key=lambda pair: (pair[0], len(pair[1].split())), reverse=True)
chosen = []
for _, sentence in scored:
if len(chosen) >= max_sentences:
break
if sentence not in chosen:
chosen.append(sentence)
return chosen or sentences[:max_sentences]
def estimate_confidence(results: List[Dict]) -> str:
evidence = assess_evidence(results)
if evidence["status"] == "sufficient":
return "High enough to answer from the approved corpus"
if evidence["status"] == "limited":
return "Limited — answer should be qualified"
return "Insufficient — abstain"
def build_evidence_answer(question: str, results: List[Dict]) -> str:
evidence = assess_evidence(results)
if evidence["status"] == "insufficient":
return (
"**Evidence status: Insufficient**\n\n"
"I could not find a strong enough match in the approved document collection to answer this question reliably. "
"I will not fill the gap from model memory or outside knowledge."
)
sorted_results = sorted(results, key=lambda item: item.get("hybrid_score", 0), reverse=True)
top_results = sorted_results[:3]
if evidence["status"] == "limited":
lead = (
"**Evidence status: Limited**\n\n"
"The corpus contains potentially relevant material, but the retrieval match is weak. Treat this as a qualified evidence extract rather than a confident answer."
)
else:
lead = "**Evidence status: Sufficient retrieval match**"
answer_lines = [
lead,
"",
"Based only on the retrieved evidence:",
"",
]
cited_sentences = 0
for item in top_results:
chunk_num = item.get("chunk_number", "?")
source = item.get("source", "unknown source")
sentences = extract_relevant_sentences(item.get("text", ""), question, max_sentences=1)
for sentence in sentences:
answer_lines.append(f"- {sentence} **[{source} · chunk {chunk_num}]**")
cited_sentences += 1
if cited_sentences == 0:
answer_lines.append("- No concise answer sentence could be extracted from the retrieved chunks.")
answer_lines.extend(["", "**Evidence used**", ""])
for item in top_results:
answer_lines.append(
f"- {item.get('source', 'unknown source')} · chunk {item.get('chunk_number', '?')} · hybrid score {item.get('hybrid_score', 0):.3f}"
)
return "\n".join(answer_lines)
|