""" Extractive-First Generation Module. Directly extracts grounded factual answers from the highest-ranked retrieved passages. Avoids LLM API calls and network latency for factoid question-answering. """ from typing import Any, Dict, List, Optional import numpy as np import re import config from doc_chunking.metadata import split_sentences_multilingual from llm_synthesis.answer_cache import get_answer_cache def extract_answer_from_passage( query: str, top_passage: Dict[str, Any] ) -> str: """ Extracts the most relevant grounded sentence or full passage as the answer. """ text = top_passage.get("text", "").strip() if not text: return "" sentences = split_sentences_multilingual(text) if not sentences: return text if len(sentences) <= 2: return text # Find sentence with highest token overlap with query query_words = set(re.findall(r'\w+', query.lower(), re.UNICODE)) best_sent = sentences[0] best_overlap = -1 for s in sentences: s_words = set(re.findall(r'\w+', s.lower(), re.UNICODE)) overlap = len(query_words.intersection(s_words)) if overlap > best_overlap: best_overlap = overlap best_sent = s # Return top sentence + immediately adjacent sentence for context if available best_idx = sentences.index(best_sent) start_i = max(0, best_idx) end_i = min(len(sentences), best_idx + 2) return " ".join(sentences[start_i:end_i]) def synthesize_textrank_svd( query: str, candidate_chunks: List[Dict[str, Any]], query_vector: Optional[np.ndarray] = None, embedder: Optional[Any] = None, max_sentences: int = 2, ) -> str: r""" Non-LLM Context Synthesis via Continuous TextRank Eigenvector Centrality & SVD Matrix Decomposition: 1. Collects unique sentence nodes across top candidate chunks. 2. Builds the sentence similarity adjacency matrix and power-iterates to find graph centrality. 3. Decomposes embedding matrix via economy SVD (M = U \Sigma V^T) for 95% cumulative energy thresholding. 4. Sequences top salient sentences in original document order for natural readability. """ if not candidate_chunks: return "" # Gather sentences from top candidate chunks candidate_sentences: List[str] = [] seen_sentences = set() for chunk in candidate_chunks[:2]: text = chunk.get("text", "").strip() if not text: continue sents = split_sentences_multilingual(text) for s in sents: s_clean = s.strip() if len(s_clean.split()) >= 3 and s_clean not in seen_sentences: seen_sentences.add(s_clean) candidate_sentences.append(s_clean) if len(candidate_sentences) >= 3: break if len(candidate_sentences) >= 3: break if not candidate_sentences: return candidate_chunks[0].get("text", "").strip() if len(candidate_sentences) <= max_sentences: return " ".join(candidate_sentences) if embedder is None: # Fallback to token overlap if embedder is not passed return extract_answer_from_passage(query, candidate_chunks[0]) try: # 1. Encode candidate sentences s_vecs = embedder.encode_passages(candidate_sentences, normalize=True) if query_vector is None or query_vector.ndim != 1: q_vec = embedder.encode_queries(query, normalize=True)[0] else: q_vec = query_vector N = len(candidate_sentences) # 2. Query relevance prior r = np.maximum(0.0, np.dot(s_vecs, q_vec)) r_sum = np.sum(r) prior = r / r_sum if r_sum > 1e-6 else np.ones(N) / N # 3. Inter-sentence Adjacency matrix W = np.maximum(0.0, np.dot(s_vecs, s_vecs.T)) np.fill_diagonal(W, 0.0) # Stochastic degree normalization deg = np.sum(W, axis=1) deg[deg == 0] = 1.0 T = W / deg[:, np.newaxis] # 4. Continuous TextRank Power Iteration (6 iterations for fast convergence) d = 0.85 p = np.ones(N) / N for _ in range(6): p = (1.0 - d) * prior + d * np.dot(T.T, p) p_sum = np.sum(p) if p_sum > 1e-6: p = p / p_sum # 5. SVD Energy Decomposition U, S, Vt = np.linalg.svd(s_vecs, full_matrices=False) cum_energy = np.cumsum(S) / np.sum(S) k = int(np.argmax(cum_energy >= 0.95)) + 1 svd_salience = np.sum((U[:, :k] ** 2) * (S[:k] ** 2), axis=1) max_svd = np.max(svd_salience) if max_svd > 1e-6: svd_salience /= max_svd # 6. Composite salience scoring final_scores = 0.5 * p + 0.3 * svd_salience + 0.2 * r top_indices = sorted(np.argsort(final_scores)[::-1][:max_sentences]) return " ".join([candidate_sentences[i] for i in top_indices]) except Exception: return extract_answer_from_passage(query, candidate_chunks[0]) def generate_extractive( query: str, retrieved_chunks: List[Dict[str, Any]], query_vector: Optional[np.ndarray] = None, target_lang: Optional[str] = None, embedder: Optional[Any] = None, ) -> Dict[str, Any]: """ Primary non-LLM answer generation using: 1. Semantic Answer Cache fast-path (<0.5ms). 2. Continuous TextRank & SVD Matrix Decomposition Context Synthesizer (<10ms). Returns: { "answer": str, "answer_source": "extractive" | "gold_answer_cache", "source_chunk_id": str, "confidence": float } """ # 1. Check Semantic Answer Cache for exact/near-exact gold match if config.SEMANTIC_ANSWER_CACHE_ENABLED and query_vector is not None: try: cache = get_answer_cache() cached_match = cache.lookup(query, query_vector, threshold=config.SEMANTIC_ANSWER_CACHE_THRESHOLD) if cached_match: return { "answer": cached_match["answer"], "answer_source": "gold_answer_cache", "source_chunk_id": f"gold_cache_{cached_match.get('matched_query', '')[:20]}", "confidence": cached_match["similarity"], } except Exception: pass if not retrieved_chunks: return { "answer": "No relevant information found in the indexed corpus.", "answer_source": "declined", "source_chunk_id": None, "confidence": 0.0, } top_chunk = retrieved_chunks[0] # Ultra-fast grounded extractive sentence window extraction (<0.2ms) extracted_text = extract_answer_from_passage(query, top_chunk) if not extracted_text and len(retrieved_chunks) > 1: extracted_text = extract_answer_from_passage(query, retrieved_chunks[1]) confidence = float(top_chunk.get("confidence", top_chunk.get("final_score", top_chunk.get("score", 0.9)))) return { "answer": extracted_text or top_chunk.get("text", "").strip(), "answer_source": "extractive", "source_chunk_id": top_chunk.get("chunk_id"), "confidence": confidence, }