"""DocDoe retrieval engine. The goal is not generic chatbot search. Retrieval should prefer source chunks that help a student score marks: definitions, formulas, diagrams, process steps, difference/advantage answers, and likely exam phrasing. """ from __future__ import annotations import asyncio import json import logging import math import re import time from collections import Counter from dataclasses import dataclass, replace from typing import Any from sqlalchemy import select from sqlalchemy.orm import Session from app.models.document import Document from app.models.document_chunk import DocumentChunk logger = logging.getLogger(__name__) @dataclass(frozen=True) class RetrievedChunk: chunk: DocumentChunk score: float confidence: float = 0.0 exam_score: float = 0.0 signals: tuple[str, ...] = () matched_terms: tuple[str, ...] = () debug_info: dict[str, Any] | None = None citation_quality: float = 0.0 # evidence/citation strength for d) (0-1) _STOPWORDS: frozenset[str] = frozenset( { "a", "an", "the", "and", "or", "but", "in", "on", "at", "to", "for", "of", "with", "by", "from", "as", "is", "was", "are", "were", "be", "been", "being", "have", "has", "had", "do", "does", "did", "will", "would", "could", "should", "may", "might", "shall", "can", "not", "no", "nor", "so", "if", "then", "than", "too", "very", "just", "about", "above", "below", "between", "into", "through", "during", "before", "after", "up", "down", "out", "off", "over", "under", "again", "further", "once", "here", "there", "when", "where", "why", "how", "all", "each", "every", "both", "few", "more", "most", "other", "some", "such", "only", "own", "same", "also", "any", "that", "this", "these", "those", "what", "which", "who", "whom", "its", "his", "her", "their", "our", "your", "my", "it", "he", "she", "they", "we", "you", "me", "him", "them", "us", "chapter", "following", "given", "using", "used", "one", "two", "three", "new", "old", "first", "last", }, ) _WORD_RE = re.compile(r"[a-z0-9]+") _FORMULA_RE = re.compile(r"\b[a-z][a-z0-9_]*\s*=\s*[-+*/a-z0-9().\s]+", re.IGNORECASE) _YEAR_RE = re.compile(r"\b(?:19|20)\d{2}\b") _MARKS_RE = re.compile(r"\b\d+\s*(?:mark|marks|m)\b", re.IGNORECASE) _EXAM_TERMS: frozenset[str] = frozenset( { "definition", "define", "law", "principle", "formula", "equation", "unit", "diagram", "label", "derive", "derivation", "explain", "reason", "advantage", "disadvantage", "difference", "distinguish", "compare", "process", "steps", "function", "cause", "effect", "application", "example", "keyword", "summary", "important", "pyq", "exam", "marks", "question", }, ) _LOW_CONFIDENCE_PREFIX = "LOW_CONFIDENCE_CONTEXT" def _tokenize(text: str) -> list[str]: return [ word for word in _WORD_RE.findall(text.lower()) if len(word) >= 2 and word not in _STOPWORDS ] def _terms(text: str) -> set[str]: return set(_tokenize(text)) def _bigrams(tokens: list[str]) -> set[tuple[str, str]]: if len(tokens) < 2: return set() return {(tokens[i], tokens[i + 1]) for i in range(len(tokens) - 1)} def _trigrams(tokens: list[str]) -> set[tuple[str, str, str]]: if len(tokens) < 3: return set() return {(tokens[i], tokens[i + 1], tokens[i + 2]) for i in range(len(tokens) - 2)} def _build_idf(chunks: list[DocumentChunk]) -> dict[str, float]: n = len(chunks) if n == 0: return {} doc_freq: Counter[str] = Counter() for chunk in chunks: chunk_unique_terms = _terms(chunk.chunk_text) if chunk.heading: chunk_unique_terms.update(_terms(chunk.heading)) doc_freq.update(chunk_unique_terms) return {term: math.log((n + 1) / (1 + df)) + 1.0 for term, df in doc_freq.items()} def _length_normalization(text: str) -> float: token_count = len(_tokenize(text)) if token_count < 8: return 0.45 if token_count < 18: return 0.75 if token_count <= 260: return 1.0 if token_count <= 420: return 0.92 return 0.82 def _exam_signals(text: str, heading: str | None = None, query: str | None = None) -> tuple[str, ...]: haystack = f"{heading or ''}\n{text}".lower() query_terms = _terms(query or "") signals: list[str] = [] if re.search(r"\b(?:is|are|means|refers to|is defined as|can be defined as)\b", haystack): signals.append("definition") if _FORMULA_RE.search(haystack) or any(term in haystack for term in ("formula", "equation", "unit", "si unit")): signals.append("formula") if any(term in haystack for term in ("diagram", "labelled", "label", "draw")): signals.append("diagram") if any(term in haystack for term in ("advantage", "disadvantage", "difference between", "distinguish", "compare")): signals.append("comparison") if any(term in haystack for term in ("step", "process", "first", "second", "then", "finally", "mechanism")): signals.append("process") if _YEAR_RE.search(haystack) or _MARKS_RE.search(haystack) or any(term in haystack for term in ("pyq", "exam", "marks")): signals.append("exam_pattern") if len(_EXAM_TERMS & _terms(haystack)) >= 3: signals.append("exam_keywords") if heading and query_terms and query_terms & _terms(heading): signals.append("heading_match") # c) high-yield from PYQ/curriculum phrasing if any(term in haystack for term in ("high yield", "high-yield", "frequent in exam", "repeated in", "pyq", "important question", "must know", "board exam")): signals.append("high_yield") return tuple(dict.fromkeys(signals)) def exam_relevance_score(text: str, heading: str | None = None, query: str | None = None) -> float: """Return a 0..1 score for how exam-useful a chunk looks.""" signals = _exam_signals(text, heading=heading, query=query) if not text.strip(): return 0.0 score = 0.0 weights = { "definition": 0.18, "formula": 0.16, "diagram": 0.12, "comparison": 0.12, "process": 0.10, "exam_pattern": 0.12, "exam_keywords": 0.14, "heading_match": 0.08, "high_yield": 0.15, # c) boost for PYQ/curriculum high-yield signals } for signal in signals: score += weights.get(signal, 0.0) token_count = len(_tokenize(text)) if 18 <= token_count <= 260: score += 0.06 elif token_count < 8: score -= 0.06 return round(max(0.0, min(score, 1.0)), 3) def _score_chunk_tfidf( chunk: DocumentChunk, query_terms: set[str], query_tokens: list[str], query_bigrams: set[tuple[str, str]], query_phrase: str, idf: dict[str, float], chunk_count: int, chapter_terms: set[str] | None = None, document_title_terms: set[str] | None = None, ) -> tuple[float, dict[str, float]]: text = chunk.chunk_text.lower() chunk_tokens = _tokenize(text) chunk_term_counts = Counter(chunk_tokens) chunk_unique = set(chunk_tokens) overlap = query_terms & chunk_unique tf_idf_score = 0.0 for term in overlap: tf = chunk_term_counts[term] tf_weight = 1.0 + math.log(tf) if tf > 0 else 0.0 tf_idf_score += tf_weight * idf.get(term, 1.0) exact_phrase_boost = 0.0 if query_phrase and query_phrase in text: exact_phrase_boost = 1.25 # calibrated: was 8.0 (too dominant over tfidf) bigram_boost = 0.0 if query_bigrams: bigram_hits = len(query_bigrams & _bigrams(chunk_tokens)) bigram_boost = bigram_hits * 0.65 # calibrated: was *1.5 heading_boost = 0.0 if chunk.heading: heading_terms = _terms(chunk.heading) heading_hits = len(query_terms & heading_terms) heading_boost = heading_hits * 0.90 # calibrated: was *2.7 title_boost = 0.0 if document_title_terms: title_hits = len(query_terms & document_title_terms) title_boost = title_hits * 0.90 # calibrated: was *2.7 chapter_boost = 0.0 if chapter_terms: chapter_hits = len(chapter_terms & (chunk_unique | _terms(chunk.heading or ""))) chapter_boost = min(0.55, chapter_hits * 0.35) # calibrated position_base_boost = 0.0 if chunk_count > 1: position_ratio = chunk.chunk_index / chunk_count position_base_boost = max(0.0, 0.55 * (1.0 - position_ratio)) # calibrated intro_summary_boost = 0.0 intro_summary_terms = {"summary", "recap", "introduction"} chunk_all_text = (chunk.chunk_text + " " + (chunk.heading or "")).lower() if any(term in chunk_all_text for term in intro_summary_terms): intro_summary_boost = 0.65 # calibrated: was 1.5 important_term_boost = 0.0 exam_term_overlap = query_terms & _EXAM_TERMS & chunk_unique important_term_boost = len(exam_term_overlap) * 0.45 # calibrated: was *1.0 # d) tiny citation quality additive (provenance: heading + page + exam signals) citation_quality = 0.0 if chunk.heading: citation_quality += 0.22 if getattr(chunk, "page_number", None) is not None: citation_quality += 0.18 # will be enriched with signals count later; base here citation_quality = round(min(1.0, citation_quality), 3) raw_score = ( tf_idf_score + exact_phrase_boost + bigram_boost + heading_boost + title_boost + chapter_boost + position_base_boost + intro_summary_boost + important_term_boost + (citation_quality * 0.08) # never dominate; small evidence boost ) len_norm = _length_normalization(chunk.chunk_text) final_score = round(raw_score * len_norm, 4) debug_factors = { "tf_idf_overlap_score": tf_idf_score, "exact_phrase_boost": exact_phrase_boost, "bigram_boost": bigram_boost, "heading_boost": heading_boost, "title_boost": title_boost, "chapter_boost": chapter_boost, "position_base_boost": position_base_boost, "intro_summary_boost": intro_summary_boost, "important_term_boost": important_term_boost, "citation_quality_base": citation_quality, "length_normalization": len_norm, } return final_score, debug_factors, citation_quality def _cosine_similarity(vec_a: list[float], vec_b: list[float]) -> float: if len(vec_a) != len(vec_b) or not vec_a: return 0.0 dot = sum(a * b for a, b in zip(vec_a, vec_b)) norm_a = math.sqrt(sum(a * a for a in vec_a)) norm_b = math.sqrt(sum(b * b for b in vec_b)) if norm_a == 0.0 or norm_b == 0.0: return 0.0 return dot / (norm_a * norm_b) def _get_query_embedding(query: str) -> list[float] | None: try: from app.services.embedding_service import generate_embeddings except Exception as exc: # pragma: no cover - defensive logger.debug("Embedding service unavailable: %s", exc) return None coro_factory = lambda: generate_embeddings([query]) # noqa: E731 try: try: asyncio.get_running_loop() except RuntimeError: embeddings = asyncio.run(coro_factory()) else: import concurrent.futures def _run_in_thread() -> list[list[float]]: loop = asyncio.new_event_loop() try: return loop.run_until_complete(coro_factory()) finally: loop.close() with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: embeddings = pool.submit(_run_in_thread).result(timeout=30) return embeddings[0] if embeddings else None except Exception as exc: logger.debug("Query embedding generation failed: %s", exc) return None def _parse_embedding(chunk: DocumentChunk) -> list[float] | None: raw = getattr(chunk, "embedding", None) if not raw: return None try: return json.loads(raw) except (json.JSONDecodeError, TypeError): return None def _reciprocal_rank_fusion( tfidf_ranked: list[tuple[int, float]], vector_ranked: list[tuple[int, float]], k: int = 60, tfidf_weight: float = 0.5, vector_weight: float = 0.5, ) -> dict[int, float]: scores: dict[int, float] = {} for rank, (chunk_idx, _) in enumerate(tfidf_ranked): scores[chunk_idx] = scores.get(chunk_idx, 0.0) + tfidf_weight / (k + rank + 1) for rank, (chunk_idx, _) in enumerate(vector_ranked): scores[chunk_idx] = scores.get(chunk_idx, 0.0) + vector_weight / (k + rank + 1) return scores def _normal_text(text: str) -> str: return " ".join(_tokenize(text)) def _fingerprint(text: str) -> str: tokens = _tokenize(text) return " ".join(tokens[:90]) def _near_duplicate(a: str, b: str, threshold: float = 0.86) -> bool: if not a or not b: return False if _fingerprint(a) == _fingerprint(b): return True a_tokens = _tokenize(a) b_tokens = _tokenize(b) if not a_tokens or not b_tokens: return False if abs(len(a_tokens) - len(b_tokens)) <= 6 and set(a_tokens) == set(b_tokens): return True a_shingles = _trigrams(a_tokens) b_shingles = _trigrams(b_tokens) if not a_shingles or not b_shingles: overlap = len(set(a_tokens) & set(b_tokens)) / max(1, len(set(a_tokens) | set(b_tokens))) return overlap >= threshold return len(a_shingles & b_shingles) / len(a_shingles | b_shingles) >= threshold def _suppress_duplicate_chunks(scored: list[RetrievedChunk]) -> list[RetrievedChunk]: kept: list[RetrievedChunk] = [] seen_exact: set[str] = set() for item in scored: normalized = _normal_text(item.chunk.chunk_text) if not normalized: continue exact = normalized[:500] if exact in seen_exact: continue if any(_near_duplicate(item.chunk.chunk_text, kept_item.chunk.chunk_text) for kept_item in kept): continue seen_exact.add(exact) kept.append(item) return kept def _mmr_diversity_rerank( candidates: list[RetrievedChunk], limit: int, lambda_param: float = 0.72 ) -> list[RetrievedChunk]: """b) MMR (Maximal Marginal Relevance) for diversity on top of score. Balances relevance (score) vs redundancy (proxy sim via _near_duplicate + trigram overlap). lambda high = more relevance, low = more diversity. """ if not candidates or limit <= 0: return [] selected: list[RetrievedChunk] = [] remaining = list(candidates) # assume pre-sorted by score desc while remaining and len(selected) < limit: if not selected: best = remaining.pop(0) selected.append(best) continue # score each remaining best_item = None best_mmr = -1e9 for cand in list(remaining): # relevance rel = cand.score # max sim to selected (redundancy) max_sim = 0.0 for sel in selected: if _near_duplicate(cand.chunk.chunk_text, sel.chunk.chunk_text, threshold=0.65): max_sim = max(max_sim, 0.92) else: # trigram proxy ct = _tokenize(cand.chunk.chunk_text) st = _tokenize(sel.chunk.chunk_text) if ct and st: ctg = _trigrams(ct) stg = _trigrams(st) if ctg and stg: j = len(ctg & stg) / max(1, len(ctg | stg)) max_sim = max(max_sim, j) mmr = lambda_param * rel - (1.0 - lambda_param) * max_sim if mmr > best_mmr: best_mmr = mmr best_item = cand if best_item is None: break selected.append(best_item) remaining.remove(best_item) return selected def _select_diverse_chunks(scored: list[RetrievedChunk], limit: int) -> list[RetrievedChunk]: """Existing diverse + b) MMR on top for stronger diversity (exam chunks often cluster by topic).""" if limit <= 0: return [] candidates = _suppress_duplicate_chunks(scored) # first pass with heading aware selected: list[RetrievedChunk] = [] heading_counts: Counter[str] = Counter() for item in candidates: if len(selected) >= limit: break heading_key = _normal_text(item.chunk.heading or "")[:80] if heading_key and heading_counts[heading_key] >= 2 and len(candidates) > limit: continue if any( _near_duplicate(item.chunk.chunk_text, existing.chunk.chunk_text, threshold=0.74) and item.score < existing.score * 0.98 for existing in selected ): continue selected.append(item) if heading_key: heading_counts[heading_key] += 1 if len(selected) < limit: selected_ids = {item.chunk.id for item in selected} for item in candidates: if item.chunk.id in selected_ids: continue selected.append(item) if len(selected) >= limit: break selected = selected[:limit] # b) apply MMR re-rank for final diversity (uses original score order as candidates) if len(selected) > 1: # re-apply MMR using the scored order as input pool mmr_pool = sorted(scored, key=lambda x: x.score, reverse=True)[: max(limit * 2, 8)] mmr_selected = _mmr_diversity_rerank(mmr_pool, limit) if mmr_selected: selected = mmr_selected return selected[:limit] def retrieval_confidence_score(chunks: list[RetrievedChunk], query: str | None = None) -> float: if not chunks: return 0.0 top = chunks[0] score_component = min(0.42, top.score * 16) if top.score < 0.08 else min(0.42, top.score / (top.score + 3.0)) count_component = min(0.18, len([item for item in chunks if item.score > 0]) * 0.045) signal_component = min(0.22, len(set(top.signals)) * 0.055) exam_component = min(0.12, top.exam_score * 0.12) coverage_component = 0.0 if query: query_terms = _terms(query) if query_terms: top_terms = _terms(top.chunk.chunk_text + " " + (top.chunk.heading or "")) coverage_component = min(0.18, (len(query_terms & top_terms) / len(query_terms)) * 0.18) confidence = score_component + count_component + signal_component + exam_component + coverage_component return round(max(0.0, min(confidence, 0.99)), 3) def is_context_answerable(chunks: list[RetrievedChunk], query: str | None = None, min_confidence: float = 0.22) -> bool: if not chunks: return False if chunks[0].score <= 0: return False return retrieval_confidence_score(chunks, query=query) >= min_confidence def _weak_topic_names(db: Session, user_id: str, subject: str | None) -> set[str]: try: from app.services.weak_topic_service import get_user_weak_topics return { topic.lower() for topic in get_user_weak_topics(db, user_id, subject=subject, limit=15) if topic and len(topic) >= 3 } except Exception: logger.debug("weak-topic lookup failed", exc_info=True) return set() def retrieve_relevant_chunks( db: Session, document_id: str, query: str, limit: int = 5, user_id: str | None = None, debug: bool = False, ) -> list[RetrievedChunk]: started_at = time.perf_counter() ranking_started_at = started_at safe_limit = max(1, min(limit, 20)) document = db.get(Document, document_id) if document is None: return [] chunks = list( db.scalars( select(DocumentChunk) .where(DocumentChunk.document_id == document_id) .order_by(DocumentChunk.chunk_index.asc()), ).all(), ) if not chunks: return [] document_title_terms = _terms(document.title) enriched_query = " ".join(filter(None, [query, document.subject, document.chapter])) query_tokens = _tokenize(enriched_query) query_term_set = set(query_tokens) query_bigrams_set = _bigrams(query_tokens) query_phrase = query.lower().strip() chapter_terms = _terms(" ".join(filter(None, [document.subject, document.chapter]))) if not query_term_set: fallback = [ RetrievedChunk( chunk=c, score=0.0, confidence=0.0, matched_terms=(), debug_info=None, citation_quality=0.0, ) for c in chunks[:safe_limit] ] logger.info( "retrieval timing document_id=%s retrieval_ms=%s ranking_ms=0 method=no_query returned=%s", document_id, int((time.perf_counter() - started_at) * 1000), len(fallback), ) return fallback idf = _build_idf(chunks) chunk_count = len(chunks) tfidf_scores: list[tuple[int, float]] = [] exam_scores: dict[int, float] = {} signal_map: dict[int, tuple[str, ...]] = {} debug_info_map: dict[int, dict[str, Any] | None] = {} matched_terms_map: dict[int, list[str]] = {} cit_qual_map: dict[int, float] = {} from app.core.config import get_settings settings = get_settings() for i, chunk in enumerate(chunks): lexical_score, debug_factors, cit_qual = _score_chunk_tfidf( chunk, query_term_set, query_tokens, query_bigrams_set, query_phrase, idf, chunk_count, chapter_terms=chapter_terms, document_title_terms=document_title_terms, ) exam_score = exam_relevance_score(chunk.chunk_text, heading=chunk.heading, query=query) exam_scores[i] = exam_score signal_map[i] = _exam_signals(chunk.chunk_text, heading=chunk.heading, query=query) # Populate matched terms chunk_tokens = _tokenize(chunk.chunk_text) chunk_unique = set(chunk_tokens) if chunk.heading: chunk_unique.update(_tokenize(chunk.heading)) matched_terms_map[i] = sorted(list(query_term_set & chunk_unique)) # enrich citation quality with signals count (d) sig_count = len(signal_map[i]) cit_qual = round(min(1.0, cit_qual + min(0.35, sig_count * 0.07)), 3) cit_qual_map[i] = cit_qual # Check debug mode if debug: # improved debug output: always rich when debug=True (for Ask/ generate tracing) debug_info = dict(debug_factors) if debug_factors else {} debug_info["exam_relevance_score"] = exam_score debug_info["citation_quality"] = cit_qual debug_info["signals"] = signal_map[i] debug_info["matched_terms_count"] = len(matched_terms_map.get(i, [])) debug_info_map[i] = debug_info else: debug_info_map[i] = None tfidf_scores.append((i, lexical_score + exam_score * 1.35)) tfidf_scores.sort(key=lambda x: (x[1], -chunks[x[0]].chunk_index), reverse=True) has_embeddings = any(getattr(c, "embedding", None) for c in chunks) vector_scores: list[tuple[int, float]] = [] if has_embeddings: query_embedding = _get_query_embedding(query) if query_embedding: for i, chunk in enumerate(chunks): chunk_emb = _parse_embedding(chunk) sim = _cosine_similarity(query_embedding, chunk_emb) if chunk_emb else 0.0 vector_scores.append((i, sim)) vector_scores.sort(key=lambda x: (x[1], -chunks[x[0]].chunk_index), reverse=True) if vector_scores: base_scores = _reciprocal_rank_fusion(tfidf_scores, vector_scores) method = "hybrid_rrf" else: base_scores = {idx: score for idx, score in tfidf_scores} method = "tfidf_exam" top_base_score = max((score for score in base_scores.values() if score > 0), default=0.0) # a) tiny calibrated weak topic additive (never dominate; applied post-RRF) weak_topic_boost = 0.004 # e.g. +0.004 * exam if match weak_topics = _weak_topic_names(db, user_id, document.subject) if user_id else set() final_scored: list[RetrievedChunk] = [] for i, chunk in enumerate(chunks): score = base_scores.get(i, 0.0) signals = list(signal_map.get(i, ())) if method == "hybrid_rrf" and score > 0: score += exam_scores.get(i, 0.0) * 0.0015 weak_matched = False if weak_topics and score > 0: chunk_lower = chunk.chunk_text.lower() if any(topic in chunk_lower for topic in weak_topics): # tiny additive: 0.004 * (exam + 0.5) to keep << lexical/RRF w_add = 0.004 * (exam_scores.get(i, 0.0) + 0.5) score += w_add signals.append("weak_topic") weak_matched = True debug_info = debug_info_map.get(i) if debug_info is not None: debug_info["final_score"] = score debug_info["retrieval_method"] = method debug_info["weak_topic_matched"] = weak_matched debug_info["weak_topic_boost_applied"] = round(0.004 * (exam_scores.get(i, 0.0) + 0.5), 5) if weak_matched else 0.0 cit_q = cit_qual_map.get(i, 0.0) final_scored.append( RetrievedChunk( chunk=chunk, score=round(score, 4), exam_score=exam_scores.get(i, 0.0), signals=tuple(dict.fromkeys(signals)), matched_terms=tuple(matched_terms_map.get(i, [])), debug_info=debug_info, citation_quality=cit_q, ), ) ranking_ms = int((time.perf_counter() - ranking_started_at) * 1000) final_scored.sort(key=lambda item: (item.score, -item.chunk.chunk_index), reverse=True) positive = [item for item in final_scored if item.score > 0] selected = _select_diverse_chunks(positive, safe_limit) if positive else final_scored[:safe_limit] confidence = retrieval_confidence_score(selected, query=query) result = [replace(item, confidence=confidence) for item in selected] logger.info( "retrieval timing document_id=%s method=%s chunks=%s returned=%s top_score=%.4f confidence=%.3f weak_topics=%s retrieval_ms=%s ranking_ms=%s", document_id, method, chunk_count, len(result), result[0].score if result else 0.0, confidence, len(weak_topics), int((time.perf_counter() - started_at) * 1000), ranking_ms, ) return result def chunks_to_context( chunks: list[RetrievedChunk], fallback_text: str | None = None, max_chars: int = 6000, ) -> str: if chunks: # e) prioritize exam_signals + formula + diagram for mark-worthy context first def _priority(item: RetrievedChunk) -> float: pri = item.exam_score * 0.65 + item.score * 0.20 + (item.citation_quality * 0.10) sigs = set(item.signals or ()) if "formula" in sigs or "diagram" in sigs: pri += 0.22 if "definition" in sigs or "high_yield" in sigs: pri += 0.12 if "process" in sigs or "comparison" in sigs: pri += 0.08 return pri prioritized = sorted(chunks, key=_priority, reverse=True) parts: list[str] = [] total = 0 for item in prioritized: part = ( f"[Section {item.chunk.chunk_index + 1} | syllabus_fit={item.score:.4f} " f"| board_grounding={item.confidence:.2f} | exam_importance={item.exam_score:.2f} | cit={item.citation_quality:.2f}]" ) if item.chunk.heading: part += f" ({item.chunk.heading})" part += f"\n{item.chunk.chunk_text}" if total + len(part) > max_chars: remaining = max_chars - total if remaining > 100: parts.append(part[:remaining]) break parts.append(part) total += len(part) + 2 return "\n\n".join(parts) if not fallback_text: return "" budget = min(max_chars - 300, 1500) excerpt = fallback_text[:budget] return ( f"{_LOW_CONFIDENCE_PREFIX}: Syllabus grounding found no strong direct textbook match. " "Use this textbook excerpt only if it directly supports the answer; otherwise state that the syllabus evidence is insufficient.\n\n" f"[Textbook Excerpt]\n{excerpt}" )