import json import logging import math import os import re import sqlite3 import sys import urllib.parse import urllib.request from collections import Counter from tracernet.config import apply_env_defaults apply_env_defaults() try: import posthog posthog.capture = lambda *args, **kwargs: None except Exception: pass try: from langchain_chroma import Chroma from langchain_huggingface import HuggingFaceEmbeddings LANGCHAIN_AVAILABLE = True except Exception: Chroma = None HuggingFaceEmbeddings = None LANGCHAIN_AVAILABLE = False from rag_query_generator import TracePathQueryGenerator from chemical_formatter import ( display_condition_label as _display_condition_casing, format_chemical_formula, plain_formula, ) from tracernet.crn.repository import CRNRepository, clean_cell from tracernet.services.pathway_evidence import ( build_default_matcher, canonical_formula, formula_phase, normalize_text, parse_formula, parse_path, path_conditions, path_species, ) from tracernet.services.bibliography import ( BibliographicRecord, BibliographyRegistry, extract_dois, first_doi, format_gbt7714, normalize_doi, ) try: from tracernet.services.doi_metadata import resolve_crossref_metadata except ImportError: # Keep the RAG service operational when an older deployment omits the # optional helper module. This fallback deliberately uses the same public # Crossref API and returns only normalized bibliographic metadata. def resolve_crossref_metadata(doi, timeout=8.0): normalized = normalize_doi(doi) if not normalized: return None url = "https://api.crossref.org/works/" + urllib.parse.quote( normalized, safe="" ) request = urllib.request.Request( url, headers={ "Accept": "application/json", "User-Agent": "TRACER-Net/1.0 (bibliographic verification)", }, ) try: with urllib.request.urlopen(request, timeout=timeout) as response: message = json.loads(response.read().decode("utf-8")).get( "message", {} ) except Exception as exc: logger = logging.getLogger(__name__) logger.warning("Crossref DOI lookup failed for %s: %s", normalized, exc) return None def first(value): if isinstance(value, list): return str(value[0]).strip() if value else "" return str(value or "").strip() date_parts = ( message.get("published-print", {}).get("date-parts") or message.get("published-online", {}).get("date-parts") or message.get("issued", {}).get("date-parts") or [] ) year = date_parts[0][0] if date_parts and date_parts[0] else None authors = [] for author in message.get("author") or []: name = " ".join( part for part in (author.get("given", ""), author.get("family", "")) if part ).strip() if name: authors.append(name) return { "title": first(message.get("title")), "authors": authors, "journal": first(message.get("container-title")), "year": year, "volume": first(message.get("volume")), "issue": first(message.get("issue")), "pages": first(message.get("page")), "article_number": first(message.get("article-number")), "doi": normalize_doi(message.get("DOI") or normalized), "url": first(message.get("URL")), } from tracernet.services.report_text import ( compose_graph_only_note, is_bad_generation_text, ) from tracernet.utils.assets import ensure_gzip_asset, find_asset try: from rag_evaluation import evaluate_rag_report except Exception: evaluate_rag_report = None logger = logging.getLogger(__name__) def _resolve_db_path(): override = os.getenv("RAG_DB_PATH", "").strip() if override: candidate = os.path.abspath(os.path.expanduser(override)) if os.path.isfile(candidate): return os.path.dirname(candidate) if os.path.isfile(os.path.join(candidate, "chroma.sqlite3")): return candidate search_roots = ( os.path.dirname(os.path.abspath(__file__)), os.getcwd(), ) database = find_asset("chroma.sqlite3", search_roots, max_depth=2) compressed = find_asset("chroma.sqlite3.gz", search_roots, max_depth=2) if compressed is not None: extracted = ensure_gzip_asset(compressed, compressed.with_suffix("")) if extracted is not None: database = extracted return str(database.parent) if database is not None else "" def _patch_chroma_schema(db_dir: str) -> None: if not db_dir: return db_path = db_dir if os.path.isdir(db_dir): db_path = os.path.join(db_dir, "chroma.sqlite3") if not os.path.exists(db_path): return try: # Ensure the file is writable before attempting UPDATE. # Common failure: extracted from .gz with read-only permissions. if not os.access(db_path, os.W_OK): try: import stat current = os.stat(db_path).st_mode os.chmod(db_path, current | stat.S_IWUSR | stat.S_IWGRP) except Exception: return con = sqlite3.connect(db_path) default_config = { "_type": "CollectionConfigurationInternal", "hnsw_configuration": { "_type": "HNSWConfigurationInternal", "space": "l2", "ef_construction": 100, "ef_search": 100, "num_threads": os.cpu_count() or 4, "M": 16, "resize_factor": 1.2, "batch_size": 100, "sync_threshold": 1000, }, } cur = con.execute("SELECT id, schema_str, config_json_str FROM collections") rows = cur.fetchall() changed = False for cid, schema_str, config_str in rows: if not schema_str: schema = None else: try: schema = json.loads(schema_str) except Exception: schema = None if schema is not None: updated = False def _fix(obj): nonlocal updated if isinstance(obj, dict): if "embedding_function" in obj and isinstance(obj["embedding_function"], dict): ef = obj["embedding_function"] if "_type" not in ef and "type" in ef: ef["_type"] = ef["type"] updated = True for v in obj.values(): _fix(v) elif isinstance(obj, list): for v in obj: _fix(v) _fix(schema) if updated: con.execute( "UPDATE collections SET schema_str = ? WHERE id = ?", (json.dumps(schema), cid), ) changed = True config_obj = None if config_str: try: config_obj = json.loads(config_str) except Exception: config_obj = None if not config_obj or "_type" not in config_obj: con.execute( "UPDATE collections SET config_json_str = ? WHERE id = ?", (json.dumps(default_config), cid), ) changed = True if changed: con.commit() con.close() except Exception: return DB_PATH = _resolve_db_path() _default_embed = "BAAI/bge-m3" EMBEDDING_MODEL = os.getenv("RAG_EMBEDDING_MODEL", _default_embed) RAG_COLLECTION = os.getenv("RAG_COLLECTION") def _load_key_from_file() -> str: """Fallback: read the DeepSeek key from a local file when the env var is unset. Looks next to this module and in the current working directory so the key can simply live in ``DEEPSEEK_API_KEY.txt`` without exporting it.""" candidates = [ os.path.join(os.path.dirname(os.path.abspath(__file__)), "DEEPSEEK_API_KEY.txt"), os.path.join(os.getcwd(), "DEEPSEEK_API_KEY.txt"), ] for path in candidates: try: if os.path.isfile(path): with open(path, "r", encoding="utf-8") as fh: value = fh.read().strip() if value: return value except OSError: continue return "" _raw_deepseek_key = os.getenv("DEEPSEEK_API_KEY", "").strip() or _load_key_from_file() _raw_deepseek_base_url = os.getenv("DEEPSEEK_BASE_URL", "").strip() _raw_deepseek_model = os.getenv("DEEPSEEK_MODEL", "").strip() DEEPSEEK_API_KEY = _raw_deepseek_key DEEPSEEK_BASE_URL = _raw_deepseek_base_url or "https://api.deepseek.com" DEEPSEEK_MODEL = _raw_deepseek_model or "deepseek-chat" # Boot-time confirmation of LLM configuration (never logs the key value). logger.info( "[RAG][LLM] config model=%s base=%s key_set=%s", DEEPSEEK_MODEL, DEEPSEEK_BASE_URL, "yes" if DEEPSEEK_API_KEY else "no", ) # Recent transformers refuses to call torch.load on a .bin checkpoint unless # torch >= 2.6 (CVE-2025-32434); the pinned torch is 2.2.2. Patching only # transformers.utils.import_utils is NOT enough: transformers.modeling_utils # does `from .utils.import_utils import check_torch_load_is_safe` at import # time and therefore holds its own reference. Both must be replaced or the # embedding model fails to load and the whole RAG service is unavailable. def _allow_local_torch_load(*_args, **_kwargs): return None try: import transformers.utils.import_utils # Patching import_utils is enough *provided it happens before # transformers.modeling_utils is first imported*: modeling_utils binds this # name at its own import time and so picks up the patched function. transformers.utils.import_utils.check_torch_load_is_safe = _allow_local_torch_load # Safety net for the reverse order (modeling_utils already loaded, e.g. in # an offline build script). Never import it here — that pulls in the whole # generation stack and stalls application startup. _loaded_modeling_utils = sys.modules.get("transformers.modeling_utils") if _loaded_modeling_utils is not None and hasattr( _loaded_modeling_utils, "check_torch_load_is_safe" ): _loaded_modeling_utils.check_torch_load_is_safe = _allow_local_torch_load except ImportError: pass # A reference-list entry reads as evidence to the matcher: "Douglass DL, Shing # CC, Wang G (1992) The light-induced alteration of realgar to pararealgar" # names both species and a conversion, so it was being classified "direct" and # printed as a supporting passage. It states nothing -- it is a pointer to a # paper that is not in the corpus -- and two such entries were inflating one # edge's evidence count from five real passages to seven. _BIBLIO_AUTHOR_YEAR_RE = re.compile( r"[A-Z][A-Za-z'’-]+\s+[A-Z]{1,3}[,;]?\s*(?:\(\d{4}[a-z]?\)|\d{4}[a-z]?\b)" ) _BIBLIO_LOCATOR_RE = re.compile( r"\b(?:doi:|https?://|\d+\s*:\s*\d+\s*[-–]\s*\d+|\bpp?\.\s*\d+)", re.IGNORECASE ) # "[33] Douglass DL" -- a citation marker in front of a name with initials. # In prose the marker trails the claim it supports; only a reference list puts # one before an author. _BIBLIO_MARKER_AUTHOR_RE = re.compile( r"\[\d{1,3}\]\s*[A-Z][A-Za-z'’-]{2,}\s+[A-Z]{1,3}\b" ) def _is_bibliography_text(text: object) -> bool: """True when the passage is a reference-list entry rather than prose. The decisive shape is a citation marker followed immediately by an author name and initials -- "[33] Douglass DL, Shing CC". Prose also carries [n], but at the end of the sentence it supports, never in front of a name, so this separates the two without needing a locator: the entries seen in this corpus often carry no DOI or page range at all. A verb test cannot stand in for it. "Can Mineral 18:525-527 [33] Douglass DL ..." opens with a journal abbreviation that any modal-verb pattern reads as the word "can". """ s = re.sub(r"\s+", " ", str(text or "")).strip() if not s: return False if _BIBLIO_MARKER_AUTHOR_RE.search(s): return True # A full entry without a bracketed marker: several author-initial groups # and a locator, and short enough to be one entry rather than a paragraph # that happens to cite several works. authors = len(_BIBLIO_AUTHOR_YEAR_RE.findall(s)) if authors >= 2 and _BIBLIO_LOCATOR_RE.search(s) and len(s) < 400: return True return authors >= 1 and _BIBLIO_LOCATOR_RE.search(s) is not None and len(s) < 220 def _compose_direct_evidence_intro(match, ref_index): """Canonical evidence-anchored introduction sentence. A single generic template shared by every direct-evidence fallback so the Introduction reads identically regardless of which code path produced it (generation fallback or evidence-guard synthesis). Driven only by the matched reactant/product/condition and the pathway-vs-edge scope; contains no pigment- or example-specific wording. """ match = match if isinstance(match, dict) else {} def _label(name, formula): name = str(name or "").strip() formula = str(formula or "").strip() if formula and formula.casefold() != name.casefold(): return f"{name} ({formula})" return name or formula reactant_label = _label( match.get("reactant_span") or match.get("reactant") or "the source material", match.get("reactant"), ) product_label = _label( match.get("product_span") or match.get("product") or "the product", match.get("product"), ) condition = re.sub( r"\s*\+\s*", " + ", _display_condition_casing(str(match.get("condition") or "").strip()), ) scope_text = ( "overall conversion" if match.get("evidence_scope") == "pathway_endpoint" else "conversion" ) if condition: return ( f"Under {condition} conditions, {reactant_label} is documented to " f"undergo the {scope_text} to {product_label}, as directly supported " f"by the retrieved literature [{ref_index}]." ) lead = reactant_label[:1].upper() + reactant_label[1:] return ( f"{lead} is documented to undergo the {scope_text} to {product_label}, " f"as directly supported by the retrieved literature [{ref_index}]." ) # How strongly a passage supports a pathway step, weakest to strongest. Used # both to pick the best of several matches and to describe, in the report, how # far a source got when it did not reach "direct". VERDICT_PRIORITY = { "direct": 8, "qualified": 7, "inferred": 6, "related": 5, "condition_mismatch": 4, "contradicted": 3, "phase_conflict": 2, "unsupported": 1, } class _DenseDoc: """Minimal stand-in for a langchain Document. The dense route can now query the Chroma collection directly with a pre-batched query vector (see _embed_queries_batch), which returns plain strings/dicts rather than Document objects. This wrapper gives that raw result the ``.page_content`` / ``.metadata`` shape the downstream code already expects, so nothing after retrieval had to change. """ __slots__ = ("page_content", "metadata") def __init__(self, page_content, metadata): self.page_content = page_content self.metadata = metadata class RAGService: def __init__(self, crn_repository: CRNRepository | None = None): self.llm_available = bool(DEEPSEEK_API_KEY) self.llm_error = ( "" if self.llm_available else "DeepSeek is not configured. Please set DEEPSEEK_API_KEY." ) logging.info("[RAG 1/4] Loading CRN and bibliography metadata") self.query_generator = TracePathQueryGenerator() self.pathway_reference_records = [] self.crn_repository = ( crn_repository if crn_repository is not None else CRNRepository.discover( ( os.path.dirname(os.path.abspath(__file__)), os.getcwd(), ), max_depth=1, ) ) self.bibliography = BibliographyRegistry.load( os.path.dirname(os.path.abspath(__file__)) ) self.doi_metadata_resolver = resolve_crossref_metadata self._doi_metadata_cache = {} self.evidence_matcher = build_default_matcher( os.path.dirname(os.path.abspath(__file__)) ) self.vector_db = None self.chroma_collection = None self.embedder = None self.collection_count = None self.retrieval_mode = "none" self.retrieval_error = "" self.last_query_error = "" db_dir = DB_PATH if not db_dir: self.retrieval_error = "Literature vector database was not found." self._load_pathway_references() return if os.path.isfile(db_dir): db_dir = os.path.dirname(db_dir) # The on-disk Chroma format is version sensitive: an index written by a # newer chromadb than the runtime can fail to load its collection config # ("'dict' object has no attribute 'dimensionality'"). Log the versions # so a rebuilt index can be produced with a matching one. try: import chromadb as _chromadb _chroma_version = getattr(_chromadb, "__version__", "?") except Exception: _chroma_version = "unavailable" try: from importlib.metadata import version as _pkg_version _lc_chroma_version = _pkg_version("langchain-chroma") except Exception: _lc_chroma_version = "?" logging.info( "[RAG][CHROMA] chromadb=%s langchain-chroma=%s", _chroma_version, _lc_chroma_version, ) logging.info("[RAG 2/4] Opening Chroma database at %s", db_dir) available_collections = [] if os.path.exists(os.path.join(db_dir, "chroma.sqlite3")): _patch_chroma_schema(db_dir) try: import chromadb client_probe = chromadb.PersistentClient(path=db_dir) available_collections = [c.name for c in client_probe.list_collections()] except Exception: available_collections = [] try: if not LANGCHAIN_AVAILABLE: raise RuntimeError("langchain_chroma unavailable") collection_name = RAG_COLLECTION if not collection_name: if "literature" in available_collections: collection_name = "literature" elif available_collections: collection_name = available_collections[0] else: collection_name = "langchain" os.environ["RAG_COLLECTION"] = collection_name logging.info( "[RAG 3/4] Loading embedding model %s on CPU", EMBEDDING_MODEL, ) embeddings = HuggingFaceEmbeddings( model_name=EMBEDDING_MODEL, model_kwargs={"device": "cpu"}, encode_kwargs={"normalize_embeddings": True}, ) self.vector_db = Chroma( persist_directory=db_dir, embedding_function=embeddings, collection_name=collection_name, ) try: self.collection_count = self.vector_db._collection.count() if self.vector_db else None except Exception: self.collection_count = None if self.collection_count == 0: try: client = getattr(self.vector_db, "_client", None) if client and hasattr(client, "list_collections"): for col in client.list_collections(): try: if col.count() > 0 and col.name != self.vector_db._collection.name: self.vector_db = Chroma( persist_directory=db_dir, embedding_function=embeddings, collection_name=col.name, ) self.collection_count = self.vector_db._collection.count() os.environ["RAG_COLLECTION"] = col.name break except Exception: continue except Exception: pass self.retrieval_mode = "langchain" if self.vector_db else "none" if self.retrieval_mode != "none": self.retrieval_error = "" logging.info( "[RAG 4/4] LangChain vector store ready: collection=%s, documents=%s", collection_name, self.collection_count, ) except Exception as exc: logging.exception( "LangChain RAG initialization failed; trying direct Chroma fallback: %s", exc, ) self.vector_db = None self.collection_count = None try: import chromadb from chromadb.utils import embedding_functions client = chromadb.PersistentClient(path=db_dir) collection_name = os.getenv("RAG_COLLECTION") if not collection_name: if "literature" in available_collections: collection_name = "literature" elif available_collections: collection_name = available_collections[0] else: collection_name = "langchain" try: self.chroma_collection = client.get_collection(name=collection_name) except Exception: cols = client.list_collections() self.chroma_collection = client.get_collection(name=cols[0].name) if cols else None self.collection_count = ( self.chroma_collection.count() if self.chroma_collection else None ) logging.info( "[RAG 3/4] Loading fallback embedding model %s", EMBEDDING_MODEL, ) try: self.embedder = ( embedding_functions.SentenceTransformerEmbeddingFunction( model_name=EMBEDDING_MODEL ) ) self.retrieval_mode = ( "chroma" if self.chroma_collection else "none" ) except Exception as embedding_exc: # A readable Chroma collection is still useful without # the query embedding model: CRN DOI lookups and the # strict entity/direction/condition scan below operate # directly on stored documents. Keep that deterministic # evidence path available instead of disabling RAG # completely. self.embedder = None self.retrieval_mode = ( "chroma_lexical" if self.chroma_collection else "none" ) self.retrieval_error = ( "Dense embedding unavailable; using CRN provenance " "and strict collection scan only. " f"{type(embedding_exc).__name__}: " f"{str(embedding_exc)[:180]}" ) logging.warning( "[RAG 4/4] Dense embedding unavailable; direct " "collection evidence fallback enabled: %s", embedding_exc, ) if self.retrieval_mode != "none": if self.retrieval_mode != "chroma_lexical": self.retrieval_error = "" logging.info( "[RAG 4/4] Direct Chroma store ready: mode=%s, " "collection=%s, documents=%s", self.retrieval_mode, collection_name, self.collection_count, ) except Exception as fallback_exc: logging.exception( "Direct Chroma fallback initialization failed: %s", fallback_exc, ) self.chroma_collection = None self.embedder = None self.collection_count = None self.retrieval_error = ( f"{type(fallback_exc).__name__}: " f"{str(fallback_exc)[:240]}" ) if self.retrieval_mode == "none" and not self.retrieval_error: self.retrieval_error = "No readable literature vector collection was initialized." self._load_pathway_references() def _smart_truncate(self, text, max_chars=1500): if len(text) <= max_chars: return text return text[:max_chars] + "\n...[Refs truncated]..." def _dedupe_results(self, results): if not results: return [] seen = {} for item in results: key = (item.get("content", "").strip(), item.get("source", "").strip()) score = item.get("score") if key not in seen: seen[key] = item continue prev = seen[key] if score is not None and (prev.get("score") is None or score > prev.get("score")): seen[key] = item return list(seen.values()) def _compact_query(self, query: str, max_len: int = 180) -> str: text = re.sub(r"\s+", " ", str(query or "")).strip() if len(text) > max_len: return text[:max_len - 3] + "..." return text def _log_retrieval(self, query: str, mode: str, scores: list, kept: int): q = self._compact_query(query) top_score = max(scores) if scores else None top_str = f"{top_score:.4f}" if isinstance(top_score, (int, float)) else "n/a" logger.info("[RAG][RETRIEVE] query=%s", q) logger.info("[RAG][RETRIEVE] mode=%s top_score=%s kept=%d", mode, top_str, kept) def _embed_query_text(self, text: str): if not self.embedder: return None if hasattr(self.embedder, "embed_query"): return self.embedder.embed_query(text) if callable(self.embedder): try: return self.embedder([text])[0] except Exception: return None return None def _embed_queries_batch(self, queries): """Embed every path query in one batched forward pass. Returns ``{query_text: vector}``. On CPU, encoding the ~20 query variants of a path together is several times faster than embedding them one at a time inside the retrieval loop, which is the dominant cost of dense retrieval. The vectors are identical to per-query embedding, so retrieval results are unchanged -- only faster. Returns ``{}`` on any failure so callers transparently fall back to per-query embedding. """ texts = [ text for text in dict.fromkeys(queries) if str(text or "").strip() ] if not texts: return {} embed_documents = None langchain_embed = getattr(self.vector_db, "_embedding_function", None) if langchain_embed is not None and hasattr( langchain_embed, "embed_documents" ): embed_documents = langchain_embed.embed_documents elif hasattr(self.embedder, "embed_documents"): embed_documents = self.embedder.embed_documents try: if embed_documents is not None: vectors = embed_documents(texts) elif callable(self.embedder): vectors = self.embedder(texts) else: return {} except Exception: logger.exception( "[RAG][RETRIEVE] batch query embedding failed; falling back " "to per-query embedding" ) return {} if vectors is None or len(vectors) != len(texts): return {} return {text: vector for text, vector in zip(texts, vectors)} def _load_pathway_references(self): self.pathway_reference_records = [ { "reactant": row.reactant, "condition": row.condition, "product": row.final_product, "source": row.source, "doi": first_doi(row.source), } for row in self.crn_repository.rows if row.reactant and row.source ] def _formula_key(self, value): text = normalize_text(value).strip() if not text: return "" formula = canonical_formula(text) if formula: return formula.casefold() return re.sub(r"[^\w]+", "", text, flags=re.UNICODE).casefold() def _text_key(self, value): return re.sub( r"\s+", " ", normalize_text(value).replace("_", " ").strip(), ).casefold() def _rank_pathway_reference_records(self, root_material, path_str): species = path_species(path_str) conditions = path_conditions(path_str) species_keys = {self._formula_key(item) for item in ([root_material] + species) if item} condition_keys = {self._text_key(item) for item in conditions if item} step_pairs = set() for idx in range(max(0, len(species) - 1)): step_pairs.add((self._formula_key(species[idx]), self._formula_key(species[idx + 1]))) path_start_key = self._formula_key(species[0]) if species else self._formula_key(root_material) path_end_key = self._formula_key(species[-1]) if species else "" ranked = [] for record in self.pathway_reference_records: source = record.get("source") or "" if not source: continue reactant_key = self._formula_key(record.get("reactant")) product_key = self._formula_key(record.get("product")) condition_key = self._text_key(record.get("condition")) score = 0 if reactant_key and product_key and reactant_key == path_start_key and product_key == path_end_key: score += 8 if (reactant_key, product_key) in step_pairs: score += 5 if reactant_key and reactant_key in species_keys: score += 2 if product_key and product_key in species_keys: score += 2 if condition_key and condition_key in condition_keys: score += 2 if score >= 3: ranked.append((score, record)) ranked.sort(key=lambda item: item[0], reverse=True) return ranked def _get_pathway_provenance_hints(self, root_material, path_str): ranked = self._rank_pathway_reference_records(root_material, path_str) if not ranked: return [], [] top_score = ranked[0][0] sources = [] dois = [] for score, record in ranked: if score < top_score: break source = clean_cell(record.get("source")) if source and source not in sources: sources.append(source) doi = record.get("doi") or first_doi(source) if doi and doi not in dois: dois.append(doi) return sources[:5], dois[:5] def answer(self, prompt: str, max_tokens: int = 1200) -> str: try: content = self._chat_completion( messages=[ {"role": "system", "content": "You are a professional heritage conservation expert."}, {"role": "user", "content": prompt}, ], temperature=0.3, max_tokens=max_tokens, ) return content or "" except Exception as e: logger.warning("Language-model answer failed: %s", e.__class__.__name__) return "Model service is temporarily unavailable. Please try again later." def _chat_completion(self, messages, temperature=0.3, max_tokens=1200, purpose="llm"): url = DEEPSEEK_BASE_URL.rstrip("/") + "/chat/completions" payload = { "model": DEEPSEEK_MODEL, "messages": messages, "temperature": temperature, "max_tokens": max_tokens, } headers = { "Content-Type": "application/json", } if DEEPSEEK_API_KEY: headers["Authorization"] = f"Bearer {DEEPSEEK_API_KEY}" # Diagnostic: shows in the backend whether DeepSeek is actually invoked. # Never logs the key value, only whether one is configured. logger.info( "[RAG][LLM] call purpose=%s model=%s base=%s key_set=%s", purpose, DEEPSEEK_MODEL, DEEPSEEK_BASE_URL, "yes" if DEEPSEEK_API_KEY else "no", ) req = urllib.request.Request( url, data=json.dumps(payload).encode("utf-8"), headers=headers, method="POST", ) try: with urllib.request.urlopen(req, timeout=60) as resp: status = getattr(resp, "status", None) or resp.getcode() data = json.loads(resp.read().decode("utf-8")) except Exception as exc: # Surface HTTP body for auth/quota errors when available. detail = "" body = getattr(exc, "read", None) if callable(body): try: detail = body().decode("utf-8", "ignore")[:300] except Exception: detail = "" logger.warning( "[RAG][LLM] call FAILED purpose=%s error=%s %s", purpose, exc.__class__.__name__, detail or str(exc), ) raise content = "" message = {} finish_reason = "" try: choice = data["choices"][0] message = choice.get("message") or {} content = (message.get("content") or "").strip() finish_reason = choice.get("finish_reason") or "" except Exception: content = "" logger.info( "[RAG][LLM] response purpose=%s http=%s chars=%d finish=%s", purpose, status, len(content), finish_reason or "?", ) if not content: # Diagnose an empty answer: wrong model name, content filter, or the # whole token budget consumed by reasoning show up here. try: raw = json.dumps(data)[:500] except Exception: raw = str(data)[:500] logger.warning( "[RAG][LLM] EMPTY content purpose=%s finish=%s reasoning_chars=%d usage=%s msg_keys=%s raw=%s", purpose, finish_reason or "?", len(str((message or {}).get("reasoning_content") or "")), data.get("usage") if isinstance(data, dict) else None, sorted(message.keys()) if isinstance(message, dict) else None, raw, ) return content def query(self, query: str, k: int = 3, threshold: float = 0.2): self.last_query_error = "" if not self.vector_db and not self.chroma_collection: self.last_query_error = self.retrieval_error or "Vector database is unavailable." return [] try: score_threshold = float(os.getenv("RAG_QUERY_THRESHOLD", str(threshold))) output = [] vector_ready = self.vector_db and (self.collection_count is None or self.collection_count > 0) chroma_ready = self.chroma_collection and self.embedder and (self.collection_count is None or self.collection_count > 0) if vector_ready: results = self.vector_db.similarity_search_with_score(query, k=k) raw_scores = [] for doc, dist in results: score = 1.0 / (1.0 + dist) if isinstance(dist, (int, float)) and dist >= 0 else 0.0 raw_scores.append(score) if score < score_threshold: continue metadata = doc.metadata or {} source = self._format_reference_source(metadata) output.append({ "content": (doc.page_content or "")[:500], "source": source, "score": score, "metadata": metadata, }) self._log_retrieval(query, "langchain", raw_scores, len(output)) return self._dedupe_results(output) if chroma_ready: vec = self._embed_query_text(query) if vec is None: self._log_retrieval(query, "chroma", [], 0) return [] results = self.chroma_collection.query( query_embeddings=[vec], n_results=k, include=["documents", "metadatas", "distances"], ) docs = results.get("documents", [[]])[0] metas = results.get("metadatas", [[]])[0] dists = results.get("distances", [[]])[0] raw_scores = [] for doc, meta, dist in zip(docs, metas, dists): score = 1.0 / (1.0 + dist) if isinstance(dist, (int, float)) and dist >= 0 else 0.0 raw_scores.append(score) if score < score_threshold: continue metadata = meta or {} source = self._format_reference_source(metadata) output.append({ "content": (doc or "")[:500], "source": source, "score": score, "metadata": metadata, }) self._log_retrieval(query, "chroma", raw_scores, len(output)) return self._dedupe_results(output) except Exception as exc: self.last_query_error = f"{type(exc).__name__}: {str(exc)[:240]}" logger.exception("[RAG][RETRIEVE] query failed: %s", query[:120]) return [] def _latexify_inline_species(self, text): if not text: return "" plain = plain_formula(text) leading_coefficient = "" coefficient_match = re.match(r"^(\d+)(?=[A-Z])", plain) if coefficient_match: leading_coefficient = coefficient_match.group(1) plain = plain[coefficient_match.end():] tokens = re.findall(r'[A-Z][a-z]?|[a-z]+|\d+|[^A-Za-z0-9]+', plain) latex = leading_coefficient buf = "" def _flush_buf(): nonlocal latex, buf if buf: latex += f"\\text{{{buf}}}" buf = "" for tok in tokens: if tok.isdigit(): _flush_buf() latex += f"_{{{tok}}}" continue if re.match(r'^[A-Z][a-z]?$', tok): _flush_buf() latex += f"\\mathrm{{{tok}}}" continue if re.match(r'^[a-z]+$', tok): buf += tok continue buf += tok _flush_buf() return f"${latex}$" def _build_loose_formula_pattern(self, plain): if not plain: return "" chars = [re.escape(ch) for ch in plain] return r'\b' + r'\s*'.join(chars) + r'\b' def _extract_graph_refs(self, path_str, full_path=None): node_ids = [] edge_ids = [] if full_path and isinstance(full_path, (list, tuple)): for idx, node in enumerate(full_path): if idx % 2 == 0: if node and node not in node_ids: node_ids.append(node) else: if node and node not in edge_ids: edge_ids.append(node) if not node_ids: species = path_species(path_str) for sp in species: if sp not in node_ids: node_ids.append(sp) return {"node_ids": node_ids, "edge_ids": edge_ids} def _build_evidence_blocks(self, ref_list, graph_refs): blocks = [] for ref in ref_list or []: dois = extract_dois(ref) blocks.append({ "source": ref, "doi": dois[0] if dois else None, "snippet": None, "node_ids": graph_refs.get("node_ids", []), "edge_ids": graph_refs.get("edge_ids", []), }) return blocks def _get_evidence_matcher(self): matcher = getattr(self, "evidence_matcher", None) if matcher is None: matcher = build_default_matcher(os.path.dirname(os.path.abspath(__file__))) self.evidence_matcher = matcher return matcher def _authored_step_types(self): """Steps the network author marked as something other than literature. Keyed by (reactant, condition, product). A step marked "extended" is one the author derived chemically and knows the source does not state, so a report should say the gap is deliberate rather than implying the evidence search came up short. """ cache = getattr(self, "_authored_step_type_cache", None) if cache is not None: return cache marked = {} try: from tracernet.crn.repository import CRNRepository repo = CRNRepository.from_xlsx( os.path.join(os.path.dirname(os.path.abspath(__file__)), "CRNs.xlsx") ) for row in repo.rows: species = [row.reactant, *row.products] for step in range(1, len(row.products) + 1): kind = row.evidence_type_for(step) if kind and kind != "literature": marked[(species[step - 1], row.condition, species[step])] = kind except Exception: marked = {} self._authored_step_type_cache = marked return marked def _document_text(self, doi): """Every chunk of one source, joined, for document-scoped conditions. A paper states its conditions once in the methods section and does not repeat them beside each result, so the condition for an edge is usually outside the passage that describes the conversion. Read straight from the metadata table rather than through the vector store: this needs no embeddings, so it still works when the vector backend is unavailable. """ key = str(doi or "").strip().lower() if not key: return "" cache = getattr(self, "_document_text_cache", None) if cache is None: cache = self._document_text_cache = {} if key in cache: return cache[key] text = "" try: import sqlite3 db_root = DB_PATH or os.path.dirname(os.path.abspath(__file__)) db = os.path.join(db_root, "chroma.sqlite3") with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn: rows = conn.execute( "select d.string_value from embedding_metadata d " "join embedding_metadata m on m.id = d.id " "where d.key='chroma:document' and m.key='doi' " "and lower(trim(m.string_value)) = ?", (key,), ).fetchall() text = " ".join(r[0] for r in rows if r and r[0]) except Exception: text = "" cache[key] = text return text def _adjacent_document_text(self, doi, chunk_index, current_text): """Join only neighbouring chunks from one DOI for composite evidence. Papers commonly separate the experimental setup from the sentence that reports the product. A bounded +/-1 chunk window recovers that local context without treating unrelated experiments elsewhere in the same paper as one claim. """ key = str(doi or "").strip().lower() try: center = int(chunk_index) except (TypeError, ValueError): return str(current_text or "") if not key: return str(current_text or "") cache = getattr(self, "_adjacent_document_text_cache", None) if cache is None: cache = self._adjacent_document_text_cache = {} cache_key = (key, center) if cache_key in cache: return cache[cache_key] combined = str(current_text or "") try: import sqlite3 db_root = DB_PATH or os.path.dirname(os.path.abspath(__file__)) db = os.path.join(db_root, "chroma.sqlite3") with sqlite3.connect(f"file:{db}?mode=ro", uri=True) as conn: rows = conn.execute( "select ci.int_value, d.string_value " "from embedding_metadata d " "join embedding_metadata m on m.id=d.id " "join embedding_metadata ci on ci.id=d.id " "where d.key='chroma:document' and m.key='doi' " "and ci.key='chunk_index' " "and lower(trim(m.string_value))=? " "and ci.int_value between ? and ? " "order by ci.int_value", (key, center - 1, center + 1), ).fetchall() texts = [str(row[1] or "").strip() for row in rows if row and row[1]] if texts: combined = "\n".join(dict.fromkeys(texts)) except Exception: combined = str(current_text or "") cache[cache_key] = combined return combined def _path_evidence_matches( self, content, path_str, provenance_level="unspecified", doi=None ): if not content or not path_str: return [] # One path analysis issues the same query in several spellings, and the # variants retrieve overlapping documents, so the same chunk is matched # against the same path many times over -- eight times for some sources # in a single run. Matching is a pure function of these four inputs, so # the repeats can be served from memory. Scoped to the service instance, # which lives for one process. key = ( str(path_str), str(provenance_level), str(doi or ""), len(content), hash(content), ) cache = getattr(self, "_edge_match_cache", None) if cache is None: cache = self._edge_match_cache = {} if key not in cache: cache[key] = self._get_evidence_matcher().match_path( path_str, content, provenance_level=provenance_level, condition_context=self._document_text(doi), ) return cache[key] def _path_endpoint_evidence_match( self, content, path_str, provenance_level="unspecified", ): if not content or not path_str: return None matcher = self._get_evidence_matcher() endpoint_matcher = getattr(matcher, "best_endpoint_match", None) if not callable(endpoint_matcher): return None # Same rationale as _path_evidence_matches: a single path analysis issues # the query in several spellings, and each retrieved chunk is offered to # both the resolver and the lexical route, so this endpoint match was # being recomputed many times over for identical inputs. It is a pure # function of these three inputs, so serve the repeats from memory. key = ( str(path_str), str(provenance_level), len(content), hash(content), ) cache = getattr(self, "_endpoint_match_cache", None) if cache is None: cache = self._endpoint_match_cache = {} if key not in cache: cache[key] = endpoint_matcher( path_str, content, provenance_level=provenance_level, ) return cache[key] def _evidence_corpus(self, collection, batch_size, max_documents): """Read the local corpus once and cache text, folded form and BM25 stats. The four-route scan reads every chunk from Chroma, normalises each one several times (resolver prefilter, lexical prefilter, BM25 tokeniser) and recomputes BM25 document frequencies -- all of it identical from path to path because the corpus is static. Doing it once per process turns a multi-second-per-path cost into a one-off. Only raw text and metadata are cached; per-path metadata enrichment still happens at the use sites, so provenance stays path-specific. """ try: signature = collection.count() except Exception: signature = None cache = getattr(self, "_evidence_corpus_cache", None) if ( cache is not None and cache.get("signature") == signature and cache.get("max_documents") == max_documents ): return cache docs = [] offset = 0 while offset < max_documents: payload = collection.get( limit=min(batch_size, max_documents - offset), offset=offset, include=["documents", "metadatas"], ) documents = list(payload.get("documents") or []) metadatas = list(payload.get("metadatas") or []) if not documents: break if len(metadatas) < len(documents): metadatas.extend( {} for _ in range(len(documents) - len(metadatas)) ) for document, metadata in zip(documents, metadatas): docs.append((document, metadata or {})) offset += len(documents) if len(documents) < batch_size: break # ``folded`` is exactly what resolver_may_match/document_may_match used # to compute per document per route; ``tokenized`` matches bm25_tokens # applied to the same folded text, so scoring is byte-for-byte identical. folded = [normalize_text(document).casefold() for document, _ in docs] tokenized = [ re.findall(r"[a-z0-9]+(?:[-_][a-z0-9]+)*", text) for text in folded ] doc_freq = Counter() for tokens in tokenized: doc_freq.update(set(tokens)) corpus_size = len(tokenized) average_length = ( sum(len(tokens) for tokens in tokenized) / corpus_size if corpus_size else 0.0 ) cache = { "signature": signature, "max_documents": max_documents, "docs": docs, "folded": folded, "tokenized": tokenized, "doc_freq": doc_freq, "corpus_size": corpus_size, "average_length": average_length, } self._evidence_corpus_cache = cache return cache def _get_bibliography(self): registry = getattr(self, "bibliography", None) if registry is None: registry = BibliographyRegistry.load( os.path.dirname(os.path.abspath(__file__)) ) self.bibliography = registry return registry def _format_doi_reference(self, doi, metadata=None): normalized_doi = normalize_doi(doi) values = dict(self._resolve_doi_metadata(normalized_doi) or {}) values.update(dict(metadata or {})) values["doi"] = normalized_doi record = self._get_bibliography().resolve(values) if record is None: record = BibliographicRecord.from_mapping(values) return format_gbt7714(record) def _resolve_doi_metadata(self, doi): """Resolve and cache generic DOI metadata without paper-specific rules.""" normalized_doi = normalize_doi(doi) if not normalized_doi: return None cache = getattr(self, "_doi_metadata_cache", None) if cache is None: cache = {} self._doi_metadata_cache = cache if normalized_doi in cache: return cache[normalized_doi] resolver = getattr(self, "doi_metadata_resolver", None) if not callable(resolver): cache[normalized_doi] = None return None try: resolved = resolver(normalized_doi) except Exception: resolved = None if not isinstance(resolved, dict): resolved = None elif normalize_doi(resolved.get("doi")) != normalized_doi: resolved = None cache[normalized_doi] = resolved return resolved @staticmethod def _reference_title_key(value): """Return a conservative key for matching a title or document filename.""" text = re.sub(r"\s+", " ", str(value or "")).strip() if not text: return "" text = text.replace("\\", "/").rsplit("/", 1)[-1] text = re.sub(r"\.(?:md|txt|pdf|docx?|html?)$", "", text, flags=re.I) return re.sub(r"[^a-z0-9]+", " ", text.casefold()).strip() def _normalize_reference_metadata(self, metadata): """Normalize citation metadata without paper-specific aliases.""" values = dict(metadata or {}) if isinstance(metadata, dict) else {} doi = normalize_doi( values.get("doi") or values.get("DOI") or values.get("doi_id") ) if not doi: for key in ( "citation", "reference", "full_citation", "formatted_reference", "source", "source_file", "filename", "file_name", "url", "URL", ): candidates = extract_dois(values.get(key)) if candidates: doi = candidates[0] break if doi: values["doi"] = doi title = ( values.get("title") or values.get("paper_title") or values.get("document_title") ) if not str(title or "").strip(): source = ( values.get("source_file") or values.get("filename") or values.get("file_name") ) if isinstance(source, str) and source.strip(): source = source.replace("\\", "/").rsplit("/", 1)[-1] title = re.sub( r"\.(?:md|txt|pdf|docx?|html?)$", "", source, flags=re.I ).strip() if str(title or "").strip(): values["title"] = re.sub(r"\s+", " ", str(title)).strip() return values def _provenance_metadata_records(self, path_sources, path_dois): """Build generic CRN provenance records for DOI/title reconciliation.""" records = [] seen = set() registry = self._get_bibliography() for raw_source in list(path_sources or []) + list(path_dois or []): source = clean_cell(raw_source) doi = first_doi(source) record = registry.resolve({"doi": doi}) if doi else None resolved_metadata = self._resolve_doi_metadata(doi) if doi else None if record is None and resolved_metadata: record = BibliographicRecord.from_mapping(resolved_metadata) title = record.title if record is not None else "" if not title and source: title = re.sub( r"(?:https?://(?:dx\.)?doi\.org/|\bdoi\s*:\s*)?" r"10\.\d{4,9}/[^\s,;]+", "", source, flags=re.I, ) title = re.sub( r"\.(?:md|txt|pdf|docx?|html?)$", "", title.replace("\\", "/").rsplit("/", 1)[-1], flags=re.I, ).strip(" .;,:-") item = { "source": source, "doi": normalize_doi(doi), "title": title, "metadata": dict(resolved_metadata or {}), } key = ( item["doi"], self._reference_title_key(item["title"]), source.casefold(), ) if key not in seen and any(item.values()): seen.add(key) records.append(item) return records def _enrich_reference_metadata(self, metadata, provenance_records): """Backfill DOI/title only when generic identifiers match exactly.""" values = self._normalize_reference_metadata(metadata) doi = normalize_doi(values.get("doi")) title_key = self._reference_title_key(values.get("title")) for candidate in provenance_records or []: candidate_doi = normalize_doi(candidate.get("doi")) candidate_title = str(candidate.get("title") or "").strip() candidate_title_key = self._reference_title_key(candidate_title) doi_match = bool(doi and candidate_doi and doi == candidate_doi) title_match = bool( title_key and candidate_title_key and title_key == candidate_title_key ) if not (doi_match or title_match): continue if not doi and candidate_doi: values["doi"] = candidate_doi doi = candidate_doi if not title_key and candidate_title: values["title"] = candidate_title title_key = candidate_title_key for key, value in (candidate.get("metadata") or {}).items(): if value not in (None, "", [], {}) and not values.get(key): values[key] = value break return values def _format_reference_source(self, metadata): if not isinstance(metadata, dict): return "Unknown" metadata = self._normalize_reference_metadata(metadata) record = self._get_bibliography().resolve(metadata) if record is not None: return format_gbt7714(record) inline = BibliographicRecord.from_mapping(metadata) has_publication_metadata = bool( inline.authors or inline.journal or inline.year or inline.volume or inline.issue or inline.pages or inline.article_number ) if inline.title and inline.doi and not has_publication_metadata: return f"{inline.title}. DOI: {inline.doi}." has_structured_metadata = bool( has_publication_metadata or inline.doi or inline.url ) if inline.title and not has_structured_metadata: return inline.title if inline.title or has_structured_metadata: return format_gbt7714(inline, mark_incomplete=True) for key in ("citation", "reference", "full_citation", "formatted_reference"): value = metadata.get(key) if isinstance(value, str) and value.strip(): return re.sub(r"\s+", " ", value).strip() source = ( metadata.get("source_file") or metadata.get("source") or metadata.get("filename") or metadata.get("file_name") ) if isinstance(source, str) and source.strip(): source = source.replace("\\", "/").rsplit("/", 1)[-1] return re.sub(r"\.[A-Za-z0-9]+$", "", source).strip() doi = metadata.get("doi") or metadata.get("DOI") or metadata.get("doi_id") return self._format_doi_reference(doi) if doi else "Unknown" def _normalize_inline_formulas(self, content, path_str): if not content: return content preserved = [] def _capture(match): preserved.append(match.group(0)) return f"__MATHBLOCKX{len(preserved) - 1}__" content = re.sub(r'\$\$.*?\$\$', _capture, content, flags=re.DOTALL) content = re.sub(r'\\\[.*?\\\]', _capture, content, flags=re.DOTALL) content = re.sub(r'\$(?!\$)(.*?)\$', r'\1', content) content = re.sub(r'\\\((.*?)\\\)', r'\1', content) # Strip only genuine LaTeX "\text" artifacts (real backslash required). # The previous optional-backslash / case-insensitive [A-Z] patterns also # deleted the substrings text/ext/tex from ordinary words # (contextual->conual, external->ernal, texture->ture). content = re.sub(r'(?i)\\\s*t\s*e\s*x\s*t\s*\{([^}]*)\}', r'\1', content) content = re.sub(r'(?i)\\\s*t\s*e\s*x\s*t\b\s*', '', content) for _ in range(3): normalized_wrappers = re.sub( r"(?i)\\(?:mathrm|mathbf|mathit|operatorname)\s*\{([^{}]*)\}", r"\1", content, ) if normalized_wrappers == content: break content = normalized_wrappers content = re.sub( r"(?i)(?= 2: matcher = self._get_evidence_matcher() def _endpoint_label(entity, generated_segment): canonical = canonical_formula(entity) entity_phase = formula_phase(entity) # Restrict to names registered for this exact polymorph. Without # the phase filter, realgar and pararealgar (both As4S4) would be # treated as labels for the same node. aliases = { record.name for (phase, formula), records in matcher.resolver.by_formula.items() if formula == canonical and phase == entity_phase for record in records if str(record.name or "").strip() } matched_aliases = [ alias for alias in aliases if re.search( rf"(?\b(?:reaction\s+pathway|conversion|transformation)\s+" r"from\s+)(?P[^.;\n]{1,140}?)(?P\s+to\s+)" r"(?P[^.;\n]{1,100}?)(?=\s+(?:is|was|occurs?|occurred|" r"takes?|proceeds?|under|upon|when|where|through)\b)", flags=re.IGNORECASE, ) def _repair_transition_subject(match): return ( match.group("prefix") + _endpoint_label(species[0], match.group("source")) + match.group("connector") + _endpoint_label(species[-1], match.group("product")) ) content = transition_subject.sub(_repair_transition_subject, content) # When generated prose names a database alias and follows it with a # chemical-looking parenthetical formula, enforce the exact formula # from the selected graph path. Ordinary descriptive parentheticals # remain unchanged. matcher = self._get_evidence_matcher() for graph_species in species: canonical = canonical_formula(graph_species) if not canonical: continue species_phase = formula_phase(graph_species) # Only names registered for this exact polymorph may have their # parenthetical formula rewritten to this node; otherwise # "realgar (alpha-As4S4)" would be rewritten to the pararealgar # (p-As4S4) node, which shares the same formula. aliases = { record.name for (phase, formula), records in matcher.resolver.by_formula.items() if formula == canonical and phase == species_phase for record in records if str(record.name or "").strip() } for alias in sorted(aliases, key=len, reverse=True): # Only for a genuinely unclosed parenthetical. Excluding "()" # from the inner class stops this from running past a closing # paren and deleting following prose. A comma must NOT terminate # it either: "arsenolite (As2O3, with As in the +3 state)" is a # well-formed parenthetical and was being turned into # "arsenolite (As2O3), with As in the +3 state)". tolerant_alias_pattern = re.compile( rf"(? str: if not text: return text preserved_blocks = [] def _capture_block(match): preserved_blocks.append(match.group(0)) return f"__LATEXBLOCK{len(preserved_blocks) - 1}__" inline_blocks = [] def _capture_inline(match): inline_blocks.append(match.group(0)) return f"__INLINEMATHBLOCK{len(inline_blocks) - 1}__" text = re.sub(r'\$\$.*?\$\$', _capture_block, text, flags=re.DOTALL) text = re.sub(r'\\\[.*?\\\]', _capture_block, text, flags=re.DOTALL) text = re.sub(r'\$[^$]+\$', _capture_inline, text) text = re.sub(r'\\mathrm\{\s*([^}]+?)\s*\)', r'\\mathrm{\1}', text) def _fix_unclosed(match): inner = match.group(1) tail = match.group(2) if "}" in inner: return match.group(0) return f"\\mathrm{{{inner}}}{tail}" text = re.sub(r'\\mathrm\{([^}\n]{1,50})([)\].,;:])', _fix_unclosed, text) # Strip only genuine LaTeX artifacts (the "\text" command with a real # backslash). The previous patterns used an optional backslash and a # case-insensitive [A-Z], so they also deleted the substrings "text"/ # "ext" from ordinary words (contextual->conual, external->ernal). text = re.sub(r'(?i)(?= 2 else "the mapped product" condition_text = ( ", ".join( _display_condition_casing(format_chemical_formula(item)) for item in conditions ) if conditions else "unspecified graph conditions" ) relation = " -> ".join( format_chemical_formula(item) for item in species ) or ( f"{format_chemical_formula(start_material)} -> " f"{format_chemical_formula(final_product)}" ) retrieval_available = bool( getattr(self, "vector_db", None) or getattr(self, "chroma_collection", None) ) return compose_graph_only_note( relation, condition_text, retrieved_context=bool(retrieved_context), retrieval_available=retrieval_available, ) def _authored_steps_for_path(self, path_str): """The author's notes on the steps of one path, as (step, kind) pairs.""" marked = self._authored_step_types() out = [] for edge in parse_path(path_str): kind = marked.get((edge.reactant, edge.condition, edge.product)) if kind: out.append((f"{edge.reactant} -> {edge.product}", kind)) return out def _generate_contextual_narrative( self, root_material, path_str, safe_context, related_context=None, authored_steps=None, ): """Synthesize a grounded context paragraph when no edge is verified. When retrieval returns literature that is topically relevant but not classified as direct evidence for a specific graph edge, a bare "no evidence" statement wastes the retrieved content. Instead, summarize what the studies actually report (samples, conditions, characterization, observed substances, aging behavior) as *related context*, explicitly without claiming the graph pathway or its steps are verified. """ # Number the sources the way the evidence-backed branch does. Passing # "(Source: )" left the model with no numbers to use, so # it fell back to author-year while every other report cites [n], and # the renderer had no numbering to build a reference list from -- which # is why those citations resolved to nothing. used = [] lines = [] for item in (related_context or []): if not isinstance(item, dict) or not str(item.get("snippet") or "").strip(): continue source = str(item.get("source") or "").strip() if source and source not in used: used.append(source) index = (used.index(source) + 1) if source in used else len(used) + 1 lines.append(f"[{index}] {item.get('snippet')}") if len(lines) >= 4: break related_text = "\n\n".join(lines) self._contextual_sources = tuple(used) primary = (safe_context or "").strip() # Never feed the "nothing retrieved" placeholder to the model: it makes # the LLM echo "no literature was retrieved" and ignore the related # paragraphs that actually are available. if primary.startswith("No relevant literature evidence was retrieved"): primary = "" parts = [chunk for chunk in (primary, related_text) if chunk] if not parts: return "" context = "\n\nAdditional related paragraphs:\n".join(parts) if len(parts) == 2 else parts[0] system_prompt = ( "You are a heritage conservation scientist. The reaction graph proposes " "a degradation pathway, but the retrieved literature does not contain a " "snippet that directly verifies its specific steps. Using ONLY the " "retrieved literature provided, write one cohesive paragraph of RELATED " "CONTEXT that helps a reader interpret the proposed pathway. Describe, " "where the sources state them: the samples or experiments studied, the " "environmental or experimental conditions, the characterization " "techniques used, the substances or phases observed, and how these " "changed with aging or exposure over time. Attribute observations to " "their citation [n]. Never claim the literature proves the graph " "pathway, its intermediates, or its elementary steps; frame it as " "supporting context. If the retrieved text does not actually concern " "this pathway, say so plainly instead of inventing a connection." ) # This branch runs whenever entity matching found no direct evidence, # which is not the same as the source being silent: it also fires when # the paper writes a species differently from the network. Without this, # the prompt's blanket "the literature does not verify its steps" was # asserted in the prose about steps the paper does state. ionic = [step for step, kind in (authored_steps or []) if kind == "ionic_form"] extended = [step for step, kind in (authored_steps or []) if kind == "extended"] if ionic: system_prompt += ( " The cited source DOES state these steps, written as ionic " f"half-reactions rather than as the salts the graph names: " f"{'; '.join(ionic)}. Describe them as reported by the source. " "Do not say the literature fails to verify them." ) if extended: system_prompt += ( " These steps are the network author's own chemical " f"extrapolation and the source does not state them: " f"{'; '.join(extended)}. Say so plainly." ) # The closing sentence is what the reader takes away, so it must not # deny steps the source states. Where the author has recorded that the # source gives them in ionic form, the caveat is scoped to the rest. if ionic: closing = ( "End with one sentence distinguishing the steps the source " "states in ionic form from the remaining steps, which these " "observations do not by themselves verify." ) heading = "Proposed graph pathway:" else: closing = ( "End with one sentence making clear these are contextual " "observations that do not by themselves verify the proposed " "pathway steps." ) heading = "Proposed graph pathway (NOT yet verified):" user_prompt = f""" {heading} {path_str} Starting material: {root_material} Retrieved literature (with evidence classification labels): {context} Write one paragraph (about 4-7 sentences) of related context following the rules. Do not add a title, headings, equations, or a references list. Do not include a "Validation note"; it is added automatically outside your paragraph. {closing} """ try: content = self._chat_completion( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=0.3, max_tokens=1000, purpose="contextual_narrative", ) or "" except Exception: content = "" content = (content or "").strip() if is_bad_generation_text(content): content = "" # Reject a degenerate "nothing was retrieved" echo: real related context # was supplied, so an answer that *opens* by denying it is a generation # failure and should defer to the terse note instead. (A narrative that # mentions "no direct evidence" mid-sentence is fine and kept.) if content and re.match( r"\s*no\s+(?:relevant\s+)?(?:literature|evidence|context|" r"contextual|retrieved|snippet|pathway-matching)\b", content, flags=re.IGNORECASE, ): content = "" if content and len(content) < 40: content = "" logger.info( "[RAG][LLM] contextual_narrative %s", f"used LLM output ({len(content)} chars)" if content else "empty/rejected -> terse note fallback", ) return content def _build_report_payload( self, content, root_material, path_str, ref_list, ref_snippets, full_path=None, evidence_is_weak=False, retrieval_trace=None, related_context=None, contextual_narrative=False, ): latex_path = self._build_latex_path(path_str) latex_block = latex_path.strip() if latex_path else "" if latex_block and not ( latex_block.startswith("$$") or latex_block.startswith("\\[") ): latex_block = f"$$\n{latex_block}\n$$" species = path_species(path_str) start_material = species[0] if species else root_material final_product = species[-1] if len(species) >= 2 else "the mapped product" header = ( "Provenance Analysis Report: Pathway from " f"{format_chemical_formula(start_material)} to " f"{format_chemical_formula(final_product)} in Mural Pigments" ) body = (content or "").strip() report = ( body if body.lower().startswith("provenance analysis report:") else f"{header}\n\n{body}\n\n{latex_block}" ) report = self._normalize_inline_formulas(report, path_str) graph_refs = self._extract_graph_refs(path_str, full_path=full_path) path_terms = ( path_species(path_str) + path_conditions(path_str) ) evidence_blocks = self._build_evidence_blocks(ref_list, graph_refs) evaluation_blocks = [] verdict_priority = VERDICT_PRIORITY for ref_offset, evidence in enumerate(evidence_blocks): bundle = ( ref_snippets[ref_offset] if ref_offset < len(ref_snippets) else {} ) if not isinstance(bundle, dict): bundle = {"text": str(bundle or ""), "snippets": []} snippet_items = [ dict(item) for item in bundle.get("snippets", []) if isinstance(item, dict) and str(item.get("text") or "").strip() ] if not snippet_items and str(bundle.get("text") or "").strip(): snippet_items = [{ "text": str(bundle.get("text") or "").strip(), "score": bundle.get("score"), "edge_matches": [], }] all_matches = [] snippet_statuses = [] for item in snippet_items: matches = [ dict(match) for match in item.get("edge_matches", []) if isinstance(match, dict) ] all_matches.extend(matches) direct = any( match.get("verdict") == "direct" for match in matches ) if matches: status = max( (str(match.get("verdict") or "unsupported") for match in matches), key=lambda value: verdict_priority.get(value, 0), ) else: status = "unclassified" snippet_statuses.append(status) evaluation_blocks.append({ "reference_index": ref_offset + 1, "source": evidence.get("source") or ( ref_list[ref_offset] if ref_offset < len(ref_list) else "" ), "snippet": str(item.get("text") or "").strip(), "score": item.get("score"), "retrieval_origin": item.get("retrieval_origin", "unknown"), "edge_matches": matches, "evidence_status": "direct" if direct else status, }) has_direct = any( match.get("verdict") == "direct" for match in all_matches ) has_non_direct = any( status not in {"direct", "unclassified"} for status in snippet_statuses ) if has_direct and has_non_direct: aggregate_status = "mixed" elif has_direct: aggregate_status = "direct" elif snippet_statuses: aggregate_status = max( snippet_statuses, key=lambda value: verdict_priority.get(value, 0), ) elif str(bundle.get("text") or "").strip(): aggregate_status = "unclassified" else: aggregate_status = "metadata_only" evidence.update({ "snippet": str(bundle.get("text") or "").strip(), "score": bundle.get("score"), "snippets": snippet_items, "provenance_only": bool(bundle.get("provenance_only")), "best_verdict": bundle.get("best_verdict"), "edge_matches": all_matches, "evidence_status": aggregate_status, "has_direct_evidence": has_direct, }) all_edge_matches = [ match for block in evaluation_blocks for match in block.get("edge_matches", []) ] best_by_edge = {} for match in all_edge_matches: edge_index = match.get("edge_index") if ( str(match.get("evidence_scope") or "edge") != "edge" or not isinstance(edge_index, int) or edge_index < 1 ): continue current = best_by_edge.get(edge_index) if current is None or verdict_priority.get( match.get("verdict"), 0 ) > verdict_priority.get(current.get("verdict"), 0): best_by_edge[edge_index] = match edge_count = len(parse_path(path_str)) direct_edges = sum( match.get("verdict") == "direct" for match in best_by_edge.values() ) pathway_endpoint_matches = [ match for match in all_edge_matches if ( match.get("verdict") == "direct" and str(match.get("evidence_scope") or "") == "pathway_endpoint" ) ] # For a one-edge path the directly supported edge is itself the # start-to-final conversion. Multi-edge paths remain stricter: endpoint # evidence never implies support for unmentioned intermediate edges. single_edge_endpoint_supported = edge_count == 1 and direct_edges == 1 endpoint_conversion_supported = bool( pathway_endpoint_matches or single_edge_endpoint_supported ) pathway_endpoint_snippet_count = sum( any( match.get("verdict") == "direct" and str(match.get("evidence_scope") or "") == "pathway_endpoint" for match in block.get("edge_matches", []) ) for block in evaluation_blocks ) if single_edge_endpoint_supported: pathway_endpoint_snippet_count = sum( any( match.get("verdict") == "direct" and str(match.get("evidence_scope") or "edge") == "edge" for match in block.get("edge_matches", []) ) for block in evaluation_blocks ) metrics = { "ref_count": len(ref_list), "retrieval_available": bool( getattr(self, "vector_db", None) or getattr(self, "chroma_collection", None) ), "retrieval_mode": getattr(self, "retrieval_mode", "none"), "retrieval_error": getattr(self, "retrieval_error", ""), "retrieval_trace": dict(retrieval_trace or {}), "retrieval_strategy": str( (retrieval_trace or {}).get("strategy") or "dense vector retrieval" ), "edge_evidence_coverage": ( round(direct_edges / edge_count, 3) if edge_count else 0.0 ), "edge_evidence": [ best_by_edge.get( index, {"edge_index": index, "verdict": "unsupported"}, ) for index in range(1, edge_count + 1) ], "endpoint_conversion_supported": endpoint_conversion_supported, "pathway_endpoint_evidence_coverage": ( 1.0 if endpoint_conversion_supported else 0.0 ), "pathway_endpoint_snippet_count": pathway_endpoint_snippet_count, "related_context_count": len(related_context or []), } if endpoint_conversion_supported and direct_edges < edge_count: scope_note = ( "Validation note: the cited pathway-level evidence supports " "the overall start-to-final conversion only; " f"{edge_count - direct_edges} of {edge_count} individual graph " "edges remain graph-derived and unverified." ) if scope_note.casefold() not in report.casefold(): if latex_block and report.rstrip().endswith(latex_block): prefix = report.rstrip()[:-len(latex_block)].rstrip() report = f"{prefix}\n\n{scope_note}\n\n{latex_block}" else: report = f"{report.rstrip()}\n\n{scope_note}" rag_evaluation = None if evaluate_rag_report: try: rag_evaluation = evaluate_rag_report( report=report, retrieved_snippets=evaluation_blocks, pathway_graph={ "path_str": path_str, "path_terms": path_terms, "graph_refs": graph_refs, }, references=ref_list, ) except Exception as exc: logger.warning( "RAG report evaluation failed: %s", exc, exc_info=True, ) if isinstance(rag_evaluation, dict) and not evidence_is_weak: evaluated_rows = [ item for item in rag_evaluation.get("sentence_level_results", []) if isinstance(item, dict) ] # Keep two kinds of sentence, both requiring a verified citation: # * edge-level direct evidence ("A converts to B"), and # * a close restatement of a cited snippet (mechanism, conditions, # instrumentation, observed phases). # Without the second kind the Introduction collapsed to a single # synthesized line, discarding detail that came straight from the # retrieved literature shown to the reader below it. accepted_statements = [ str(item.get("statement") or "").strip() for item in evaluated_rows if ( item.get("direct_evidence_status") or item.get("restated_evidence_status") ) and item.get("citation_status") is True ] rejected_count = max( 0, len(evaluated_rows) - len(accepted_statements), ) if not accepted_statements: for block in evaluation_blocks: direct_match = next(( match for match in block.get("edge_matches", []) if isinstance(match, dict) and match.get("verdict") == "direct" ), None) if not direct_match: continue ref_number = int(block.get("reference_index") or 1) accepted_statements.append( _compose_direct_evidence_intro(direct_match, ref_number) ) break if rejected_count: if edge_count > direct_edges: endpoint_note = ( "The overall start-to-final conversion has direct " "pathway-level evidence; " if endpoint_conversion_supported else "" ) # Name the steps rather than counting them. "1 of 2 edges" # makes the reader work out which half of the pathway is # unsupported, and the answer is only inferable from the # "edge 1: direct" labels further down the report. authored = self._authored_step_types() unestablished, extrapolated, ionic = [], [], [] for edge in parse_path(path_str): if (best_by_edge.get(edge.index) or {}).get("verdict") == "direct": continue name = f"{edge.reactant} -> {edge.product}" key = (edge.reactant, edge.condition, edge.product) kind = authored.get(key) if kind == "extended": extrapolated.append(name) elif kind == "ionic_form": ionic.append(name) else: unestablished.append(name) def _phrase(names, verb_single, verb_plural): if len(names) == 1: return f"the step {names[0]} {verb_single}" return ( "the steps " + ", ".join(names[:-1]) + f" and {names[-1]} {verb_plural}" ) parts = [] if unestablished: parts.append( f"{_phrase(unestablished, 'is', 'are')} not established " "by direct literature snippets; generated claims for " "those gaps were omitted" ) # An author-marked extrapolation is a different statement # from a failed search, and reporting them alike understates # one and overstates the other. if extrapolated: parts.append( f"{_phrase(extrapolated, 'is', 'are')} marked by the " "network author as a chemical extrapolation rather " "than a literature-stated conversion" ) # The source does state this conversion; it writes the # species as an aqueous ion where the network writes the # salt, and the two have different compositions, so entity # matching cannot join them. Reporting it as unestablished # would understate evidence the paper actually gives. if ionic: parts.append( f"{_phrase(ionic, 'is', 'are')} stated in the cited " "source as an ionic half-reaction, which this " "network represents as the corresponding salt" ) if not parts: parts.append( f"{edge_count - direct_edges} of {edge_count} individual " "graph edges are not established by direct literature " "snippets" ) validation_note = ( f"Validation note: {endpoint_note}" + "; ".join(parts) + "." ) else: # Every edge is supported and only individual sentences were # dropped. Saying so on every report states the gating # policy, not a result of this analysis, so it is left to # the exported evaluation detail. A note appears here only # when part of the pathway is genuinely unestablished. validation_note = "" guarded_body = " ".join( statement for statement in accepted_statements + [validation_note] if statement ).strip() report = self._normalize_inline_formulas( f"{header}\n\n{guarded_body}\n\n{latex_block}", path_str, ) try: rag_evaluation = evaluate_rag_report( report=report, retrieved_snippets=evaluation_blocks, pathway_graph={ "path_str": path_str, "path_terms": path_terms, "graph_refs": graph_refs, }, references=ref_list, ) rag_evaluation[ "generation_guard_removed_statements" ] = rejected_count except Exception as exc: logger.warning( "Guarded RAG report evaluation failed: %s", exc, exc_info=True, ) sentence_traces = [] if isinstance(rag_evaluation, dict): if evidence_is_weak or direct_edges < edge_count: rag_evaluation["manual_verification"] = True rag_evaluation["manual_verification_required"] = True rag_evaluation["manual_verification_status"] = "required" if direct_edges < edge_count: gap_reason = ( f"{edge_count - direct_edges} of {edge_count} graph " "edges lack direct snippet evidence." ) if endpoint_conversion_supported: gap_reason += ( " Endpoint conversion evidence does not verify the " "intermediate graph steps." ) current_reason = str( rag_evaluation.get("manual_verification_reason") or "" ).strip() if gap_reason not in current_reason: rag_evaluation["manual_verification_reason"] = ( f"{current_reason} {gap_reason}".strip() ) metrics.update({ "total_sentences": rag_evaluation.get("total_statements", 0), "supported_sentences": rag_evaluation.get( "supported_statements", 0 ), "total_statements": rag_evaluation.get("total_statements", 0), "supported_statements": rag_evaluation.get( "supported_statements", 0 ), "direct_evidence_statements": rag_evaluation.get( "direct_evidence_statements", 0 ), "restated_evidence_statements": rag_evaluation.get( "restated_evidence_statements", 0 ), "grounded_statements": rag_evaluation.get( "grounded_statements", 0 ), "unsupported_statements": rag_evaluation.get( "unsupported_statements", 0 ), "statements_without_direct_evidence": rag_evaluation.get( "statements_without_direct_evidence", 0 ), "unsupported_by_both_statements": rag_evaluation.get( "unsupported_by_both_statements", 0 ), "any_supported_statements": rag_evaluation.get( "any_supported_statements", 0 ), "evidence_coverage": rag_evaluation.get( "evidence_coverage", 0.0 ), "direct_evidence_coverage": rag_evaluation.get( "direct_evidence_coverage", 0.0 ), "statement_evidence_coverage": rag_evaluation.get( "statement_evidence_coverage", rag_evaluation.get("evidence_coverage", 0.0), ), "citation_accuracy": rag_evaluation.get("citation_accuracy"), "citation_verification_status": rag_evaluation.get( "citation_verification_status", "not_applicable" ), "citation_checked_statements": rag_evaluation.get( "citation_checked_statements", 0 ), "citation_unverifiable_statements": rag_evaluation.get( "citation_unverifiable_statements", 0 ), "graph_alignment": rag_evaluation.get( "graph_alignment", 0.0 ), "graph_alignment_applicable_statements": rag_evaluation.get( "graph_alignment_applicable_statements", 0 ), "graph_alignment_status": rag_evaluation.get( "graph_alignment_status", "not_applicable" ), "graph_derived_statements": rag_evaluation.get( "graph_derived_statements", 0 ), "inference_only_statements": rag_evaluation.get( "inference_only_statements", 0 ), "retrieved_snippet_count": rag_evaluation.get( "retrieved_snippet_count", 0 ), "source_quality_score": rag_evaluation.get( "source_quality_score", 0.0 ), "source_quality_level": rag_evaluation.get( "source_quality_level", "unavailable" ), "source_quality_details": rag_evaluation.get( "source_quality_details", [] ), "source_quality_basis": rag_evaluation.get( "source_quality_basis", "" ), "overall_reliability_score": rag_evaluation.get( "overall_reliability_score", 0.0 ), "overall_reliability_level": rag_evaluation.get( "overall_reliability_level", "not_evaluable" ), "overall_reliability_summary": rag_evaluation.get( "overall_reliability_summary", "" ), "overall_reliability_reasons": rag_evaluation.get( "overall_reliability_reasons", [] ), "manual_verification": rag_evaluation.get( "manual_verification" ), "manual_verification_reason": rag_evaluation.get( "manual_verification_reason" ), "generation_guard_removed_statements": rag_evaluation.get( "generation_guard_removed_statements", 0, ), }) for item in rag_evaluation.get("sentence_level_results", []): if not isinstance(item, dict): continue supporting = item.get("supporting_evidence") sentence_traces.append({ "sentence_id": item.get("sentence_id"), "text": item.get("statement") or "", "support": ( "evidence" if item.get("direct_evidence_status") else "inference" ), "relevance": ( supporting.get("match_score") if isinstance(supporting, dict) else None ), "evidence": [supporting] if supporting else [], "matched_evidence": supporting, "direct_evidence_status": item.get( "direct_evidence_status" ), "restated_evidence_status": item.get( "restated_evidence_status" ), "grounded_evidence_status": item.get( "grounded_evidence_status" ), "support_status": item.get("support_status"), "support_basis": item.get("support_basis"), "citation_status": item.get("citation_status"), "graph_alignment": item.get("graph_alignment"), "graph_alignment_applicable": item.get( "graph_alignment_applicable" ), "graph_alignment_status": item.get( "graph_alignment_status" ), "statement_type": item.get("statement_type"), "cited_references": item.get("cited_references", []), "cited_dois": item.get("cited_dois", []), }) if evidence_is_weak: metrics["evidence_is_weak"] = True metrics["manual_verification"] = True context_sources = () if contextual_narrative: # Marks report bodies that are a grounded synthesis of retrieved # literature (framed as related context), so the display layer may # keep them instead of substituting the terse "no evidence" text. metrics["has_contextual_narrative"] = True # The [n] the narrative cites are only resolvable if the numbering # travels with the report. Kept separate from "references", which # means "sources that establish an edge" and must stay empty here. context_sources = tuple(getattr(self, "_contextual_sources", ()) or ()) return { "report": report, "root_material": root_material, "path_str": path_str, "references": ref_list, "evidence": evidence_blocks, "evaluation_evidence": evaluation_blocks, "reference_snippets": ref_snippets, "graph_refs": graph_refs, "context_sources": list(context_sources), "metrics": metrics, "sentence_traces": sentence_traces, "rag_evaluation": rag_evaluation, "related_context": list(related_context or []), "rag_debug": [ f"retrieval_mode: {self.retrieval_mode}", ( "collection_count: " f"{self.collection_count if self.collection_count is not None else 'n/a'}" ), f"ref_snippet_count: {len(evaluation_blocks)}", f"deepseek_base_url: {DEEPSEEK_BASE_URL}", f"deepseek_model: {DEEPSEEK_MODEL}", ( "deepseek_api_key_set: " f"{'yes' if DEEPSEEK_API_KEY else 'no'}" ), f"evidence_is_weak: {'yes' if evidence_is_weak else 'no'}", ], } @staticmethod def _dense_depth_config(): """Resolve the dense-retrieval depths (ANN_FETCH_K, RAG_OUTPUT_TOP_K). RAG_OUTPUT_TOP_K is how many distance-sorted candidates per query survive into the threshold filter and the cross-query union. RAG_ANN_FETCH_K is the HNSW over-fetch depth: Chroma's effective search_ef = max(index search_ef, n_results), and the persisted index ships search_ef=10, so we over-fetch to raise the effective ef before truncating. Depth 200 is the pre-registered ANN-calibration choice (eval_spec.md 20). ANN_FETCH_K is never allowed below RAG_OUTPUT_TOP_K. """ top_k = int(os.getenv("RAG_TOP_K", "6")) output_top_k = max(top_k, int(os.getenv("RAG_FALLBACK_TOP_K", "20"))) ann_fetch_k = max(output_top_k, int(os.getenv("RAG_ANN_FETCH_K", "200"))) return ann_fetch_k, output_top_k @staticmethod def _limit_dense_candidates(candidates, output_top_k): """Keep only the nearest ``output_top_k`` over-fetched ANN candidates. Candidates arrive distance-sorted (ascending) from the HNSW query; this truncation is what enforces the per-query cap so at most RAG_OUTPUT_TOP_K items reach the threshold filter and the union, regardless of how deep we over-fetched to raise the effective search_ef. """ return list(candidates)[:max(0, int(output_top_k))] def _report_cache_dir(self): return os.getenv("RAG_REPORT_CACHE_DIR") or os.path.join( os.path.dirname(os.path.abspath(__file__)), "rag_report_cache" ) def _report_cache_signature(self): """Corpus fingerprint, so a rebuilt knowledge base invalidates the cache. A cached report is only valid for the corpus it was computed from. The document count changes whenever papers are added or removed, which is exactly when the stored evidence stops being reproducible. Note: this sees the corpus, not the gating rules. If retrieval or evidence gating is ever changed, the cache must be cleared by hand -- the stored evidence was computed under the previous rules. """ count = self.collection_count if count is None: try: collection = getattr(self.vector_db, "_collection", None) or self.chroma_collection count = collection.count() if collection is not None else -1 except Exception: count = -1 return f"docs={count}" def _report_cache_key(self, root_material, path_str): import hashlib payload = "|".join([ str(root_material or ""), str(path_str or ""), self._report_cache_signature(), ]) return hashlib.sha1(payload.encode("utf-8")).hexdigest() def generate_report_for_path(self, root_material, path_str, full_path=None): """Return the pathway report, reusing cached retrieval evidence. Retrieval and evidence gating are deterministic in (path, corpus) yet cost minutes of CPU, so recomputing them per request spends that time reproducing an identical answer; they are cached on disk, keyed by path and corpus fingerprint, so a changed knowledge base recomputes rather than serving stale evidence. The narrative is deliberately NOT cached -- it is regenerated on every call so the wording is never frozen, which costs seconds rather than minutes. Set RAG_REPORT_CACHE=0 to bypass. """ cache_enabled = os.getenv("RAG_REPORT_CACHE", "1").strip().lower() not in ( "0", "false", "no" ) evidence = self.cached_path_evidence(root_material, path_str, full_path) return self._narrate_path_report( evidence, root_material, path_str, full_path ) def cached_path_evidence(self, root_material, path_str, full_path=None): """Retrieval evidence for one path, computed once and cached on disk. Exposed separately so the offline warm-up can fill the cache without generating a narrative: narration is a per-request concern and, with an API key configured, warming 88 paths through the full report path would spend 88 LLM calls on text that is thrown away. """ cache_enabled = os.getenv("RAG_REPORT_CACHE", "1").strip().lower() not in ( "0", "false", "no" ) cache_file = None evidence = None if cache_enabled: cache_file = os.path.join( self._report_cache_dir(), f"{self._report_cache_key(root_material, path_str)}.json", ) try: if os.path.isfile(cache_file): with open(cache_file, "r", encoding="utf-8") as handle: cached = json.load(handle) stored = cached.get("evidence") # Every field the narrative half reads must be present, or a # cache written by an older layout would fail mid-report. if isinstance(stored, dict) and all( key in stored for key in ( "context_str", "ref_list", "ref_snippets", "related_context", "retrieval_trace", ) ): evidence = stored logger.info( "[RAG][CACHE] evidence hit path=%s", str(path_str)[:120] ) except (OSError, ValueError): logger.warning( "[RAG][CACHE] unreadable entry ignored: %s", cache_file ) if evidence is None: evidence = self._retrieve_path_evidence( root_material, path_str, full_path ) if cache_enabled and cache_file: try: os.makedirs(os.path.dirname(cache_file), exist_ok=True) # Write to a temporary file first so a crash mid-write cannot # leave a truncated entry that later reads as valid evidence. temporary = cache_file + ".tmp" with open(temporary, "w", encoding="utf-8") as handle: json.dump( { "root_material": root_material, "path_str": path_str, "signature": self._report_cache_signature(), "evidence": evidence, }, handle, ensure_ascii=False, default=str, ) os.replace(temporary, cache_file) logger.info( "[RAG][CACHE] stored evidence path=%s", str(path_str)[:120] ) except (OSError, TypeError, ValueError): logger.warning("[RAG][CACHE] could not store %s", cache_file) return evidence def _retrieve_path_evidence(self, root_material, path_str, full_path=None): edges = parse_path(path_str) mini_report = {root_material: [path_str]} generated_queries = self.query_generator.generate_queries(mini_report) endpoint_query = "" elemental_endpoint_query = "" if len(edges) >= 2: endpoint_conditions = list(dict.fromkeys( edge.condition for edge in edges if str(edge.condition or "").strip() )) endpoint_query = " ".join( [edges[0].reactant] + endpoint_conditions + [edges[-1].product] ) if edges: endpoint_composition = parse_formula(edges[-1].product) if ( endpoint_composition and len(endpoint_composition) == 1 and endpoint_composition[0][1] == 1 and canonical_formula(edges[-1].product) == endpoint_composition[0][0] ): conditions = list(dict.fromkeys( edge.condition for edge in edges if edge.condition )) elemental_endpoint_query = " ".join( [edges[0].reactant] + conditions + [ "metallic", edges[-1].product, "elemental", edges[-1].product, "zero-valent", ] ) queries = list(dict.fromkeys( [edge.query for edge in edges if edge.query] + ([endpoint_query] if endpoint_query else []) + ([elemental_endpoint_query] if elemental_endpoint_query else []) + [query for query in generated_queries if str(query or "").strip()] )) ref_list = [] ref_snippets = [] reference_index = {} seen_snippets = set() related_context = [] seen_related = set() retrieval_trace = { "query_count": len(queries), "dense_candidates": 0, "dense_kept": 0, "resolver_scanned": 0, "resolver_candidates": 0, "resolver_kept": 0, "bm25_scanned": 0, "bm25_candidates": 0, "bm25_kept": 0, "lexical_scanned": 0, "lexical_candidates": 0, "lexical_kept": 0, "strategy": ( "four-route retrieval (resolver + BM25 + lexical + BGE-M3) " "with chemical evidence gating" ), } provenance_only = [] # Best verdict any passage of a source reached, including passages the # strict gate rejected. A source that matched at "related" is not the # same as one nothing matched in, and the report must not say it is. provenance_best_verdict: dict[str, str] = {} path_sources, path_dois = self._get_pathway_provenance_hints( root_material, path_str, ) provenance_records = self._provenance_metadata_records( path_sources, path_dois, ) score_threshold = float(os.getenv("RAG_SCORE_THRESHOLD", "2.0")) ann_fetch_k, output_top_k = self._dense_depth_config() normalized_path_dois = { normalize_doi(value) for value in path_dois if normalize_doi(value) } provenance_title_keys = { self._reference_title_key(record.get("title")) for record in provenance_records if self._reference_title_key(record.get("title")) } def _is_path_provenance_source(source, metadata): values = metadata or {} doi = normalize_doi( values.get("doi") or values.get("DOI") or values.get("doi_id") or first_doi(source) ) if doi and doi in normalized_path_dois: return True title_candidates = [ values.get("title"), values.get("source_file"), source, ] return any( self._reference_title_key(value) in provenance_title_keys for value in title_candidates if self._reference_title_key(value) ) def _match_dict(match, evidence_scope="edge"): return { "edge_index": match.edge.index, "reactant": match.edge.reactant, "product": match.edge.product, "condition": match.edge.condition, "evidence_scope": evidence_scope, "verdict": match.verdict, "score": match.score, "window": match.window, "reactant_match": match.reactant.match_type, "product_match": match.product.match_type, "reactant_span": match.reactant.span, "product_span": match.product.span, "relation_basis": match.relation_basis, "condition_status": match.condition_status, "provenance_level": match.provenance_level, "reasons": list(match.reasons), } def _record_best_verdict(source, edge_matches, endpoint_match): """Remember the strongest verdict a source reached, kept or rejected. The strict gate publishes only "direct" matches, so without this the report cannot tell a source whose passages matched the species but not the recorded condition from one that matched nothing at all. """ key = str(source or "").strip().casefold() if not key: return candidates = list(edge_matches or []) if endpoint_match is not None: candidates.append(endpoint_match) if not candidates: return best = max( (str(match.verdict or "unsupported") for match in candidates), key=lambda verdict: VERDICT_PRIORITY.get(verdict, 0), ) current = provenance_best_verdict.get(key) if current is None or VERDICT_PRIORITY.get( best, 0 ) > VERDICT_PRIORITY.get(current, 0): provenance_best_verdict[key] = best def _add_evidence( source, content, score=None, metadata=None, provenance_level="unspecified", diagnose=False, retrieval_origin="dense_vector", ): raw_content = str(content or "") composite_used = False # Preserve line breaks while matching so Markdown section headings # can be removed by split_evidence_windows. Collapse only the # final published snippet. matching_content = re.sub(r"[ \t]+", " ", raw_content).strip() clean = re.sub(r"\s+", " ", raw_content).strip()[:1600] if not clean: return False low_quality_related_context = bool( re.search(r"\[MISSING_PAGE_FAIL", clean, flags=re.IGNORECASE) or re.search(r"\\begin\{(?:table|tabular)\}", clean, flags=re.IGNORECASE) ) if _is_bibliography_text(clean): return False edge_matches = self._path_evidence_matches( matching_content, path_str, provenance_level=provenance_level, doi=(metadata or {}).get("doi") or (metadata or {}).get("DOI") or (metadata or {}).get("doi_id"), ) relevant = [ (match, "edge") for match in edge_matches if match.verdict == "direct" ] endpoint_match = self._path_endpoint_evidence_match( matching_content, path_str, provenance_level=provenance_level, ) if endpoint_match is not None and endpoint_match.verdict == "direct": relevant.append((endpoint_match, "pathway_endpoint")) if not relevant: values = metadata or {} doi_value = ( values.get("doi") or values.get("DOI") or values.get("doi_id") or first_doi(source) ) composite_content = self._adjacent_document_text( doi_value, values.get("chunk_index"), raw_content, ) if composite_content.strip() and composite_content != raw_content: composite_matches = self._path_evidence_matches( composite_content, path_str, provenance_level=provenance_level, doi=doi_value, ) composite_endpoint = self._path_endpoint_evidence_match( composite_content, path_str, provenance_level=provenance_level, ) composite_relevant = [ (match, "edge") for match in composite_matches if match.verdict == "direct" ] if ( composite_endpoint is not None and composite_endpoint.verdict == "direct" ): composite_relevant.append( (composite_endpoint, "pathway_endpoint") ) _record_best_verdict( source, composite_matches, composite_endpoint ) if composite_relevant: matching_content = composite_content clean = re.sub( r"\s+", " ", composite_content ).strip()[:1600] edge_matches = composite_matches endpoint_match = composite_endpoint relevant = composite_relevant composite_used = True _record_best_verdict(source, edge_matches, endpoint_match) if not relevant: # Preserve a small, explicitly non-evidential audit trail for # chemically related candidates rejected by the strict gate. candidates = list(edge_matches) if endpoint_match is not None: candidates.append(endpoint_match) candidates.sort(key=lambda item: item.score, reverse=True) best = candidates[0] if candidates else None provenance_source = _is_path_provenance_source( source, metadata, ) materially_related = bool( best # Endpoint-only matches from another starting material are # out of scope. Preserve a missing-entity paragraph only # when it comes from this path's own provenance source. and (best.reactant.matched or provenance_source) and best.relation_basis != "cooccurrence" ) related_key = ( str(source or "").strip().casefold(), str(best.window if best else clean).strip().casefold(), ) if ( materially_related and not low_quality_related_context and related_key not in seen_related ): seen_related.add(related_key) related_context.append({ "source": str(source or "").strip() or self._format_reference_source(metadata or {}), # Related context is not sent to claim generation, so # retain the surrounding paragraph for human review. "snippet": clean, "retrieval_origin": retrieval_origin, "match": _match_dict(best), "reason": ( "Retrieved as chemically related context, but it " "does not directly establish this graph edge." ), }) related_context.sort( key=lambda item: float( (item.get("match") or {}).get("score") or 0 ), reverse=True, ) del related_context[5:] if diagnose: decisions = [ { "edge": match.edge.index, "verdict": match.verdict, "reactant": match.reactant.match_type, "product": match.product.match_type, "relation": match.relation_basis, "condition": match.condition_status, } for match in edge_matches ] logger.info( "[RAG][FILTER] source=%s rejected=%s", str(source or "Unknown")[:180], decisions, ) return False source = str(source or "").strip() or self._format_reference_source(metadata or {}) if not source or source == "Unknown": return False # Publish only the sentence window that actually matched. Keeping # the whole retrieval chunk would leak unrelated nearby examples # into the prompt, report, and PDF. window_matches = {} for match, evidence_scope in relevant: window = re.sub(r"\s+", " ", str(match.window or "")).strip()[:1600] if not window: continue window_matches.setdefault(window, []).append( _match_dict(match, evidence_scope=evidence_scope) ) if not window_matches: return False if source not in reference_index: reference_index[source] = len(ref_list) ref_list.append(source) ref_snippets.append({"text": "", "score": score, "snippets": []}) ref_index = reference_index[source] bundle = ref_snippets[ref_index] added = False for window, match_records in window_matches.items(): snippet_key = (source.casefold(), window.casefold()) if snippet_key in seen_snippets: # The same evidence window may be recalled by several # routes. Preserve all contributing routes instead of # silently attributing it only to the first route. for existing_bundle in ref_snippets: for existing in existing_bundle.get("snippets", []): existing_key = ( str(source).casefold(), str(existing.get("text") or "").casefold(), ) if existing_key == snippet_key: routes = existing.setdefault( "retrieval_routes", [existing.get("retrieval_origin", "unknown")], ) if retrieval_origin not in routes: routes.append(retrieval_origin) break continue seen_snippets.add(snippet_key) evidence_level = ( "edge_direct" if any(item.get("evidence_scope") == "edge" for item in match_records) else "pathway_endpoint" ) bundle["snippets"].append({ "text": window, "score": score, "metadata": dict(metadata or {}), "evidence_level": evidence_level, "retrieval_origin": retrieval_origin, "retrieval_routes": [retrieval_origin], "evidence_aggregation": ( "adjacent_chunks_same_doi" if composite_used else "single_fragment" ), "edge_matches": match_records, }) added = True if not added: if not bundle.get("snippets"): ref_list.pop() ref_snippets.pop() reference_index.pop(source, None) return False bundle["text"] = "\n".join( snippet["text"] for snippet in bundle["snippets"] ) numeric_scores = [ snippet.get("score") for snippet in bundle["snippets"] if isinstance(snippet.get("score"), (int, float)) ] bundle["score"] = min(numeric_scores) if numeric_scores else None return True for source in path_sources: doi = first_doi(source) if doi: continue provenance_only.append(f"CRN source: {source}") for doi in path_dois: normalized_doi = normalize_doi(doi) provenance_only.append(self._format_doi_reference(normalized_doi)) # Chroma keeps vectors in a separate on-disk HNSW segment. When that # segment is missing or stale every similarity search returns nothing, # and embedding each of ~25 path queries just to discard the result # costs seconds per path. Detect it once and fall straight through to # the lexical evidence scan. dense_probe_queries = max(1, int(os.getenv("RAG_DENSE_PROBE_QUERIES", "3"))) dense_disabled = False dense_attempts = 0 # Embed every path query once, batched, instead of embedding each variant # separately inside the loop below -- on CPU that per-query embedding is # the dominant cost of dense retrieval. The batched vectors are identical # to per-query embedding, so results are unchanged. Querying the Chroma # collection directly by vector then reuses these instead of re-embedding. query_vectors = self._embed_queries_batch(queries) dense_collection = ( getattr(self.vector_db, "_collection", None) if self.vector_db else None ) for query in queries: if not self.vector_db and not self.chroma_collection: break vector_ready = self.vector_db and not dense_disabled and ( self.collection_count is None or self.collection_count > 0 ) chroma_ready = ( self.chroma_collection and self.embedder and not dense_disabled and ( self.collection_count is None or self.collection_count > 0 ) ) raw_scores = [] before = len(seen_snippets) if vector_ready: qv = query_vectors.get(query) if qv is not None and dense_collection is not None: # Reuse the pre-batched vector: query the collection directly # instead of letting similarity_search_with_score embed the # query again. Distances are the same cosine distances that # method returns, so downstream scoring is unchanged. raw = dense_collection.query( query_embeddings=[qv], n_results=ann_fetch_k, include=["documents", "metadatas", "distances"], ) r_docs = (raw.get("documents") or [[]])[0] r_metas = (raw.get("metadatas") or [[]])[0] r_dists = (raw.get("distances") or [[]])[0] results = [ (_DenseDoc(doc, meta or {}), dist) for doc, meta, dist in zip(r_docs, r_metas, r_dists) ] else: results = self.vector_db.similarity_search_with_score( query, k=ann_fetch_k, ) # ANN candidates come back distance-sorted (ascending); keep the # nearest output_top_k before threshold filtering and union. results = self._limit_dense_candidates(results, output_top_k) retrieval_trace["dense_candidates"] += len(results) dense_attempts += 1 if ( not retrieval_trace["dense_candidates"] and dense_attempts >= dense_probe_queries ): dense_disabled = True retrieval_trace["dense_index_unusable"] = True logger.warning( "[RAG][RETRIEVE] dense index returned no candidates in %d " "queries; skipping dense retrieval for this path. Rebuild " "the Chroma HNSW segment to restore semantic search.", dense_attempts, ) for rank, (document, score) in enumerate(results): similarity = ( 1.0 / (1.0 + score) if isinstance(score, (int, float)) and score >= 0 else 0.0 ) raw_scores.append(similarity) if not isinstance(score, (int, float)) or score > score_threshold: continue metadata = self._enrich_reference_metadata( document.metadata or {}, provenance_records, ) reference = self._format_reference_source(metadata) doi = metadata.get("doi") or metadata.get("DOI") or metadata.get("doi_id") if doi and "doi" not in str(reference).casefold(): reference = f"{reference}. DOI: {normalize_doi(doi)}" added = _add_evidence( reference, document.page_content, score=score, metadata=metadata, provenance_level=str( metadata.get("provenance_level") or "unspecified" ), diagnose=rank < 3, retrieval_origin="bge_m3", ) if added: retrieval_trace["dense_kept"] += 1 self._log_retrieval( query, "langchain_path", raw_scores, len(seen_snippets) - before, ) elif chroma_ready: vector = query_vectors.get(query) if vector is None: vector = self._embed_query_text(query) if vector is None: self._log_retrieval(query, "chroma_path", [], 0) continue results = self.chroma_collection.query( query_embeddings=[vector], n_results=ann_fetch_k, include=["documents", "metadatas", "distances"], ) # over-fetched, distance-sorted; truncate to output_top_k before # threshold filtering and union. documents = self._limit_dense_candidates( results.get("documents", [[]])[0], output_top_k) metadatas = self._limit_dense_candidates( results.get("metadatas", [[]])[0], output_top_k) distances = self._limit_dense_candidates( results.get("distances", [[]])[0], output_top_k) retrieval_trace["dense_candidates"] += len(documents) for rank, (document, metadata, score) in enumerate( zip(documents, metadatas, distances) ): similarity = ( 1.0 / (1.0 + score) if isinstance(score, (int, float)) and score >= 0 else 0.0 ) raw_scores.append(similarity) if not isinstance(score, (int, float)) or score > score_threshold: continue metadata = self._enrich_reference_metadata( metadata or {}, provenance_records, ) reference = self._format_reference_source(metadata) doi = metadata.get("doi") or metadata.get("DOI") or metadata.get("doi_id") if doi and "doi" not in str(reference).casefold(): reference = f"{reference}. DOI: {normalize_doi(doi)}" added = _add_evidence( reference, document, score=score, metadata=metadata, provenance_level=str( metadata.get("provenance_level") or "unspecified" ), diagnose=rank < 3, retrieval_origin="bge_m3", ) if added: retrieval_trace["dense_kept"] += 1 self._log_retrieval( query, "chroma_path", raw_scores, len(seen_snippets) - before, ) # CRN-provenance targeted retrieval. When the reaction network records # the DOI an edge was derived from, that paper is the intended source of # truth for this pathway, yet general similarity search may rank it # below a topically adjacent paper about a different pigment. Look for # evidence inside that paper directly instead of hoping it surfaces. # The same strict entity/direction/condition gate still applies, so a # cited paper that does not actually state the conversion is not # promoted to evidence. if normalized_path_dois: provenance_collection = getattr( getattr(self, "vector_db", None), "_collection", None ) if provenance_collection is None: provenance_collection = getattr(self, "chroma_collection", None) if provenance_collection is not None and hasattr( provenance_collection, "get" ): for provenance_doi in sorted(normalized_path_dois): try: payload = provenance_collection.get( where={"doi": provenance_doi}, include=["documents", "metadatas"], ) except Exception: logger.exception( "[RAG][PROVENANCE] lookup failed for doi=%s", provenance_doi, ) continue documents = list(payload.get("documents") or []) metadatas = list(payload.get("metadatas") or []) if len(metadatas) < len(documents): metadatas.extend( {} for _ in range(len(documents) - len(metadatas)) ) kept_before = len(seen_snippets) for document, metadata in zip(documents, metadatas): metadata = self._enrich_reference_metadata( metadata or {}, provenance_records ) reference = self._format_reference_source(metadata) doi_value = ( metadata.get("doi") or metadata.get("DOI") or metadata.get("doi_id") ) if doi_value and "doi" not in str(reference).casefold(): reference = f"{reference}. DOI: {normalize_doi(doi_value)}" _add_evidence( reference, document, metadata=metadata, provenance_level=str( metadata.get("provenance_level") or "unspecified" ), retrieval_origin="crn_provenance", ) logger.info( "[RAG][PROVENANCE] doi=%s chunks=%d kept=%d", provenance_doi, len(documents), len(seen_snippets) - kept_before, ) # Build the remaining three retrieval routes over the local corpus on # every run. Dense BGE-M3 retrieval above is one candidate generator; # resolver aliases, BM25 and lexical matching add complementary # candidates. All four routes still pass through _add_evidence, so # co-occurrence or a wrong direction/condition never becomes evidence. if True: collection = getattr(getattr(self, "vector_db", None), "_collection", None) if collection is None: collection = getattr(self, "chroma_collection", None) if collection is not None and hasattr(collection, "get"): scan_before = len(seen_snippets) scanned = 0 prefiltered = 0 batch_size = max(100, int(os.getenv("RAG_EVIDENCE_SCAN_BATCH", "500"))) max_documents = max( batch_size, int(os.getenv("RAG_EVIDENCE_SCAN_MAX_DOCUMENTS", "10000")), ) try: matcher = self._get_evidence_matcher() def entity_terms(entity): terms = {normalize_text(entity).casefold()} formula = canonical_formula(entity) if formula: terms.add(normalize_text(formula).casefold()) for (_phase, known_formula), records in ( matcher.resolver.by_formula.items() ): if known_formula == formula: terms.update( normalize_text(record.name).casefold() for record in records if str(record.name or "").strip() ) return {term for term in terms if len(term) >= 2} def condition_term_groups(condition): groups = [] for component in re.split( r"[+;/]", normalize_text(condition), ): key = component.strip().casefold() if not key: continue aliases = {key} aliases.update( normalize_text(alias).casefold() for alias in matcher.condition_resolver.aliases.get( key, set() ) ) groups.append({alias for alias in aliases if len(alias) >= 2}) return groups scan_specs = [ ( entity_terms(edge.reactant), entity_terms(edge.product), condition_term_groups(edge.condition), ) for edge in edges ] def resolver_may_match(folded): """Entity-resolver route: require both path endpoints. This route deliberately uses the resolver-expanded names/formulae rather than free-text relevance. The strict matcher later decides direction and conditions. ``folded`` is the pre-normalised, case-folded document text supplied by the corpus cache, so normalisation is never repeated per document per route. """ for reactants, products, _condition_groups in scan_specs: if ( any(term in folded for term in reactants) and any(term in folded for term in products) ): return True return False def document_may_match(folded): relation_hit = bool(re.search( r"\b(?:convert|conversion|transform|transition|degrad|" r"oxid|react|form|produc|yield|result)\w*\b", folded, flags=re.IGNORECASE, )) for reactants, products, condition_groups in scan_specs: reactant_hit = bool( reactants and any(term in folded for term in reactants) ) product_hit = bool( products and any(term in folded for term in products) ) condition_hit = any( any(term in folded for term in group) for group in condition_groups if group ) # Require chemical-path scope, not a condition word # alone. Endpoint-only recall is allowed only when # both a condition and transformation relation occur. if reactant_hit and ( product_hit or condition_hit or relation_hit ): return True if product_hit and condition_hit and relation_hit: return True return False def bm25_tokens(value): # Keep formula-like tokens (Pb3O4, beta-PbO2) intact # while also indexing ordinary words. return re.findall( r"[a-z0-9]+(?:[-_][a-z0-9]+)*", normalize_text(value).casefold(), ) def route_kept_count(route): return sum( 1 for bundle in ref_snippets for snippet in bundle.get("snippets", []) if route in snippet.get( "retrieval_routes", [snippet.get("retrieval_origin")], ) ) # The corpus is static, so its text, folded form and BM25 # statistics are read and computed once per process and # reused across every path rather than rebuilt each call. corpus_cache = self._evidence_corpus( collection, batch_size, max_documents ) corpus = corpus_cache["docs"] folded_texts = corpus_cache["folded"] for (document, raw_metadata), folded in zip(corpus, folded_texts): scanned += 1 hit_resolver = resolver_may_match(folded) hit_lexical = document_may_match(folded) if not (hit_resolver or hit_lexical): continue # Metadata enrichment and reference formatting are only # needed for candidates that clear a prefilter, so they # no longer run for every document in the corpus. metadata = self._enrich_reference_metadata( raw_metadata or {}, provenance_records, ) reference = self._format_reference_source(metadata) doi = ( metadata.get("doi") or metadata.get("DOI") or metadata.get("doi_id") ) if doi and "doi" not in str(reference).casefold(): reference = f"{reference}. DOI: {normalize_doi(doi)}" provenance_level = str( metadata.get("provenance_level") or "unspecified" ) if hit_resolver: retrieval_trace["resolver_candidates"] += 1 _add_evidence( reference, document, metadata=metadata, provenance_level=provenance_level, retrieval_origin="resolver", ) if hit_lexical: prefiltered += 1 _add_evidence( reference, document, metadata=metadata, provenance_level=provenance_level, retrieval_origin="lexical", ) # Dependency-free Okapi BM25 keeps the HF package small and # makes the scoring definition explicit and reproducible. # Corpus tokenisation and document frequencies come from the # cache; only the query side is recomputed per path. tokenized = corpus_cache["tokenized"] doc_freq = corpus_cache["doc_freq"] corpus_size = corpus_cache["corpus_size"] average_length = corpus_cache["average_length"] query_tokens = set( token for query in queries for token in bm25_tokens(query) ) k1 = float(os.getenv("RAG_BM25_K1", "1.5")) b = float(os.getenv("RAG_BM25_B", "0.75")) bm25_top_k = max( 1, int(os.getenv("RAG_BM25_TOP_K", "100")), ) ranked = [] for index, tokens in enumerate(tokenized): if not tokens or not query_tokens: continue frequencies = Counter(tokens) length_norm = ( len(tokens) / average_length if average_length else 1.0 ) score = 0.0 for token in query_tokens: frequency = frequencies.get(token, 0) if not frequency: continue frequency_docs = doc_freq.get(token, 0) inverse_document_frequency = math.log( 1.0 + ( corpus_size - frequency_docs + 0.5 ) / (frequency_docs + 0.5) ) score += inverse_document_frequency * ( frequency * (k1 + 1.0) ) / ( frequency + k1 * (1.0 - b + b * length_norm) ) if score > 0: ranked.append((score, index)) ranked.sort(reverse=True) for bm25_score, index in ranked[:bm25_top_k]: document, raw_metadata = corpus[index] metadata = self._enrich_reference_metadata( raw_metadata or {}, provenance_records, ) reference = self._format_reference_source(metadata) doi = ( metadata.get("doi") or metadata.get("DOI") or metadata.get("doi_id") ) if doi and "doi" not in str(reference).casefold(): reference = ( f"{reference}. DOI: {normalize_doi(doi)}" ) _add_evidence( reference, document, # BM25 scores are not on the same scale as Chroma # distances. Keep the route score as metadata rather # than mixing it into the generic distance field. metadata={ **metadata, "bm25_score": round(bm25_score, 6), }, provenance_level=str( metadata.get("provenance_level") or "unspecified" ), retrieval_origin="bm25", ) logger.info( "[RAG][FOUR_ROUTE] path=%s scanned=%d " "resolver=%d bm25=%d lexical=%d", path_str, scanned, retrieval_trace["resolver_candidates"], min(len(ranked), bm25_top_k), prefiltered, ) retrieval_trace["resolver_scanned"] = scanned retrieval_trace["resolver_kept"] = route_kept_count( "resolver" ) retrieval_trace["bm25_scanned"] = scanned retrieval_trace["bm25_candidates"] = min( len(ranked), bm25_top_k ) retrieval_trace["bm25_kept"] = route_kept_count("bm25") retrieval_trace["lexical_scanned"] = scanned retrieval_trace["lexical_candidates"] = prefiltered retrieval_trace["lexical_kept"] = route_kept_count( "lexical" ) except Exception: logger.exception( "[RAG][FOUR_ROUTE] collection scan failed for path=%s", path_str, ) # Related context is an audit aid for evidence gaps, not an appendix # to a successful answer. When direct evidence exists, suppress it to # avoid repeating the accepted paper and surfacing weaker distractors. # Otherwise retain the strongest paragraph from each source so a weak # pathway can still be explained from the retrieved literature. _has_direct_match = any( isinstance(match, dict) and match.get("verdict") == "direct" for bundle in ref_snippets if isinstance(bundle, dict) for snippet in bundle.get("snippets", []) for match in snippet.get("edge_matches", []) ) if _has_direct_match: related_context.clear() elif related_context: strongest_by_source = {} for item in related_context: source_key = re.sub( r"\s+", " ", str(item.get("source") or "Unknown source") ).strip().casefold() current = strongest_by_source.get(source_key) item_score = float((item.get("match") or {}).get("score") or 0) current_score = float( ((current or {}).get("match") or {}).get("score") or 0 ) if current is None or item_score > current_score: strongest_by_source[source_key] = item related_context[:] = sorted( strongest_by_source.values(), key=lambda item: float( (item.get("match") or {}).get("score") or 0 ), reverse=True, )[:3] for reference in dict.fromkeys(provenance_only): if reference in reference_index: continue reference_index[reference] = len(ref_list) ref_list.append(reference) ref_snippets.append({ "text": "", "score": None, "snippets": [], "provenance_only": True, "best_verdict": provenance_best_verdict.get( str(reference or "").strip().casefold() ), }) context_records = [] for ref_index, bundle in enumerate(ref_snippets, 1): for snippet in bundle.get("snippets", []): matches = snippet.get("edge_matches", []) labels = ", ".join( ( "pathway endpoints " f"{item.get('reactant', '')} -> {item.get('product', '')}: " f"{item.get('verdict', 'unsupported')}" if item.get("evidence_scope") == "pathway_endpoint" else f"edge {item['edge_index']}: {item['verdict']}" ) for item in matches ) direct_priority = int(any( item.get("verdict") == "direct" for item in matches if isinstance(item, dict) )) context_records.append(( direct_priority, ref_index, f"[{ref_index}] Evidence classification: {labels}\n" f"{snippet.get('text', '')}", )) context_records.sort(key=lambda item: (-item[0], item[1])) context_parts = [item[2] for item in context_records] context_str = "\n\n".join(context_parts) if not context_str: context_str = "No relevant literature evidence was retrieved from the vector database." return { "context_str": context_str, "ref_list": ref_list, "ref_snippets": ref_snippets, "related_context": related_context, "retrieval_trace": retrieval_trace, } def _narrate_path_report( self, evidence, root_material, path_str, full_path=None ): """Write the report narrative from already-retrieved evidence. Split from retrieval so the expensive, deterministic half can be cached while this half runs fresh on every request. Retrieval and evidence gating depend only on (path, corpus) and cost minutes; the narrative is the one part that should not be frozen in a cache, and it costs seconds. """ context_str = evidence["context_str"] ref_list = evidence["ref_list"] ref_snippets = evidence["ref_snippets"] related_context = evidence["related_context"] retrieval_trace = evidence["retrieval_trace"] safe_context = self._smart_truncate( context_str, max_chars=int(os.getenv("RAG_CONTEXT_MAX_CHARS", "8000")), ) has_snippet_evidence = any( isinstance(item, dict) and str(item.get("text") or "").strip() for item in ref_snippets ) has_direct_evidence = any( match.get("verdict") == "direct" for bundle in ref_snippets if isinstance(bundle, dict) for snippet in bundle.get("snippets", []) for match in snippet.get("edge_matches", []) ) if not has_direct_evidence: content = "" is_contextual = False # If topically relevant literature was retrieved (even without an # edge-verified match), synthesize a grounded related-context # paragraph instead of a bare "no evidence" statement. if has_snippet_evidence or related_context: content = self._generate_contextual_narrative( root_material=root_material, path_str=path_str, safe_context=safe_context, related_context=related_context, authored_steps=self._authored_steps_for_path(path_str), ) is_contextual = bool(content) if not content: content = self._generate_unverified_path_report( root_material=root_material, path_str=path_str, retrieved_context=bool(has_snippet_evidence or related_context), ) return self._build_report_payload( content=content, root_material=root_material, path_str=path_str, ref_list=ref_list, ref_snippets=ref_snippets, full_path=full_path, evidence_is_weak=True, retrieval_trace=retrieval_trace, related_context=related_context, contextual_narrative=is_contextual, ) system_prompt = ( "You are a heritage conservation scientist writing an evidence-grounded " "reaction-mechanism introduction for a provenance report. Open by " "identifying the starting pigment (common name and formula) in one " "clause, then explain the pathway in reaction order: starting material, " "operative condition or reagent, transformation/mechanism, and product, " "naming the observed products when the snippets report them. Fold the " "pigment identification into the first evidence-cited sentence so it is " "not a free-standing uncited claim. Treat the reaction graph and " "literature snippets as different evidence types. Never turn graph " "connectivity into a documented molecular mechanism or observation." ) user_prompt = f""" Write one concise English Introduction paragraph explaining the reaction pathway and the evidence-supported level of mechanism. Pigment: {root_material} Graph pathway: {path_str} Numbered literature snippets: {safe_context} Rules: - Open with a concise identification of the starting pigment (common name and chemical formula), folded into the first evidence-cited sentence; do not add a generic cultural-heritage or degradation preamble, and do not make it a separate uncited sentence. - Describe supported transformations in graph order, stating reactant, experimental condition or reagent, transformation, and product; when a snippet reports the reaction mechanism (e.g. photo-oxidation) or names the observed products, state them. - Use "reaction pathway" or "observed conversion" when the snippets establish only conversion. Use "molecular mechanism" or "elementary step" only when the snippet explicitly reports that level of detail. - Do not include a "Validation note" or scope caveat; that note is added automatically outside your paragraph. - End with one concise scope sentence when the observed conversion is supported but molecular intermediates, by-products, or elementary steps are not given by the retrieved evidence. - A claim that a graph edge is established may use only a snippet classified as direct for that edge, and must end with its matching citation [n]. - A snippet classified as direct for "pathway endpoints" supports only the overall start-to-final conversion and facts explicitly stated in that snippet. It does not establish any intermediate, individual graph edge, elementary mechanism, or complete step sequence. - Do not discuss, cite, or list retrieval candidates that are not classified as direct evidence for either a current graph edge or the current pathway endpoints. - Pathway-endpoint evidence may be cited for its scoped overall conversion even when individual graph edges remain unverified. - Do not cite a reference entry that has no snippet text. - Write the paragraph in your own words as a connected explanation. Do NOT stitch quoted fragments together; synthesise the snippets into flowing scientific prose. - Ground the substance, not the wording: every chemical species, environmental condition and characterisation technique you name must appear in a snippet, but you are expected to rephrase, condense and connect them. - Never introduce a species, reagent, technique or mechanism that no snippet mentions, and never negate or reverse what a snippet reports. - End each sentence with the citation [n] of the snippet it draws on. A sentence citing the wrong snippet is discarded, so cite the source the content actually came from. - Report the specific experimental reagent or condition stated in the snippet; never generalize it to every member of a broader graph condition label. - Do not turn "compared with/to" into "matched", "identical", or "equivalent" unless the snippet explicitly uses a matching or equivalence statement. - Distinguish an observed conversion/product identified by the paper from an elementary reaction mechanism. Do not say that the conversion is unconfirmed when the snippet reports that it occurred. - If the paper reports multiple products or phases, name the relevant co-products and do not present the selected graph product as the sole product. - For every graph step with NO snippet classified as direct for that edge, explicitly call it "graph-derived and unverified"; do not fill the gap by inference. This applies only to steps with no direct snippet of their own: a step whose evidence label reads "edge N: direct" is established, and calling it unverified contradicts the evidence list printed beside the report. A pathway-endpoint snippet does not make a step unverified either -- it simply adds nothing to it, so judge each step by its own edge label alone. - NEVER write that a step is unverified, unsupported, graph-derived, or lacking direct evidence when that step's evidence label reads "edge N: direct". The report prints the labels next to your paragraph, so such a sentence contradicts the evidence in the same document. For a step that IS labelled direct but whose snippets only report that the conversion occurred, the correct statement is that the transformation is supported while the molecular intermediates, by-products and elementary steps are not resolved by the retrieved evidence -- not that the step is unverified. - Do not add a closing sentence about unresolved mechanism or evidence scope: a separate "Evidence Limitation" section is generated from the evidence records and appended automatically. Writing your own version risks stating a different boundary than the one the records support. - Do not invent color or appearance changes, prevention advice, treatment advice, kinetics, or causal explanations. - Preserve the supplied citation numbering. - Use inline LaTeX ($...$) for chemical formulas. - Output only the paragraph, without a title, headings, equation, or references list. """ try: content = self._chat_completion( messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}, ], temperature=0.3, max_tokens=1200, purpose="path_intro", ) or "" except Exception as e: content = f"Error: {str(e)}" if is_bad_generation_text(content): logger.info( "[RAG][LLM] path_intro rejected (is_bad_generation_text); " "using template fallback. preview=%r", content[:120], ) content = "" elif content: logger.info("[RAG][LLM] path_intro used LLM output (%d chars)", len(content)) if not content: # Unified fallback: reuse the single canonical intro template so this # path reads identically to the evidence-guard synthesis. The scope / # validation caveat is appended downstream, not duplicated here. fallback = None for ref_index, bundle in enumerate(ref_snippets, 1): for snippet in bundle.get("snippets", []): for match in snippet.get("edge_matches", []): if match.get("verdict") != "direct": continue fallback = _compose_direct_evidence_intro(match, ref_index) break if fallback: break if fallback: break content = fallback or ( f"The graph records {path_str}; a direct-evidence narrative could not " "be generated, so unsupported graph steps remain graph-derived and " "unverified." ) return self._build_report_payload( content=content, root_material=root_material, path_str=path_str, ref_list=ref_list, ref_snippets=ref_snippets, full_path=full_path, evidence_is_weak=not has_direct_evidence, retrieval_trace=retrieval_trace, related_context=related_context, )