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