Spaces:
Sleeping
Sleeping
File size: 4,390 Bytes
81f3119 2680899 81f3119 2680899 81f3119 | 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 | import subprocess
import re
def get_confidence_label(score):
if score >= 0.60:
return "High Confidence"
elif score >= 0.35:
return "Medium Confidence"
else:
return "Low Confidence"
def build_prompt(question, contexts):
context_text = "\n\n".join(
f"[Source: {c['filename']} | Page {c['page']}]\n{c['content']}"
for c in contexts
)
prompt = f"""
You are an AI assistant. Answer the question strictly using the context below.
If the answer is not present, say "Answer not found in the provided documents."
Context:
{context_text}
Question:
{question}
Answer:
"""
return prompt.strip()
def ask_llm(prompt, model="llama3"):
process = subprocess.Popen(
["ollama", "run", model],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
output, _ = process.communicate(prompt)
return output.strip()
def is_entity_query(query):
words = query.strip().split()
return len(words) <= 3 and not query.strip().endswith("?")
def extract_answer_from_chunks(chunks, max_chars=600):
seen = set()
answer_parts = []
for c in chunks:
text = c["content"].strip()
if text and text not in seen:
answer_parts.append(text)
seen.add(text)
if sum(len(a) for a in answer_parts) > max_chars:
break
return "\n\n".join(answer_parts)
def clean_chunk_text(text, max_chars=400):
# remove extra whitespace
text = re.sub(r"\s+", " ", text).strip()
# remove URLs
text = re.sub(r"http\S+|www\S+", "", text)
# remove emails
text = re.sub(r"\S+@\S+", "", text)
# limit length
if len(text) > max_chars:
text = text[:max_chars].rsplit(" ", 1)[0] + "..."
return text
def keyword_present(text, keyword):
return keyword.lower() in text.lower()
def looks_like_definition(text):
patterns = [
" is the ",
" refers to ",
" is defined as ",
" means "
]
text = text.lower()
return any(p in text for p in patterns)
def extract_best_sentence(text, keyword):
sentences = re.split(r'[.?!]', text)
keyword = keyword.lower()
for s in sentences:
if keyword in s.lower():
return s.strip()
# fallback: return first sentence
if sentences:
return sentences[0].strip()
return text
def build_extractive_answer(
ranked_chunks,
query_word,
question_type,
max_items=3
):
answer_points = []
source_info = []
query_word_lower = query_word.lower()
count = 1
for chunk, score in ranked_chunks:
raw_text = chunk["content"]
# ⛔ keyword must exist
if not keyword_present(raw_text, query_word):
continue
# ⛔ definition questions must look like definitions
if question_type == "definition":
if not looks_like_definition(raw_text):
continue
# Clean the chunk
clean_text = clean_chunk_text(raw_text)
if not clean_text:
continue
# Extract best sentence
best_sentence = extract_best_sentence(clean_text, query_word)
if not best_sentence:
continue
# Highlight keyword
highlighted = re.sub(
rf"\b({re.escape(query_word_lower)})\b",
r"**\1**",
best_sentence,
flags=re.IGNORECASE
)
answer_points.append(f"{count}. {highlighted}")
page_value = chunk.get("page")
safe_page = int(page_value) if page_value is not None else 1
source_info.append({
"filename": chunk["filename"],
"page": safe_page,
"score": float(round(score * 100, 1))
})
count += 1
if count > max_items:
break
return answer_points, source_info
def build_fallback_answer(query):
return (
f"The topic '{query}' was not found in the provided documents.\n\n"
f"Basic explanation:\n"
f"{query} refers to a general concept commonly used in its domain."
)
def summarize_chunks(chunks, max_sentences=3):
text = " ".join(c["content"] for c in chunks[:5])
sentences = re.split(r'(?<=[.!?]) +', text)
summary = " ".join(sentences[:max_sentences])
return summary |