Spaces:
Sleeping
Sleeping
| 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 | |
| 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)(?<![A-Za-z])(?:mathrm|mathbf|mathit|operatorname)" | |
| r"(?=[A-Z][a-z]?)", | |
| "", | |
| content, | |
| ) | |
| content = re.sub(r'\\alpha', '\u03b1', content) | |
| content = re.sub(r'\\beta', '\u03b2', content) | |
| content = re.sub(r'\\gamma', '\u03b3', content) | |
| content = re.sub(r'_\{?(\d+)\}?', r'\1', content) | |
| # Language models occasionally wrap a chemical token in bare TeX | |
| # grouping braces without a command, e.g. `{CuS}` or `{H₂S}`. These | |
| # braces have no semantic value in prose and render visibly in Gradio. | |
| content = re.sub( | |
| r"\{\s*([A-Z][A-Za-z0-9₀₁₂₃₄₅₆₇₈₉⁰¹²³⁴⁵⁶⁷⁸⁹()\[\]·.\-+^]*)\s*\}", | |
| r"\1", | |
| content, | |
| ) | |
| species = path_species(path_str) | |
| # Rebuild the subject of common generated transition phrases from the | |
| # selected graph endpoints. This catches severely malformed model | |
| # output such as missing hydrate dots/parentheses, where ordinary | |
| # formula-token normalization cannot safely recover the identity. | |
| if len(species) >= 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"(?<!\w){re.escape(alias)}(?!\w)", | |
| generated_segment, | |
| flags=re.IGNORECASE, | |
| ) | |
| ] | |
| alias = max(matched_aliases, key=len) if matched_aliases else "" | |
| formula = plain_formula(entity) | |
| return f"{formula} ({alias})" if alias else formula | |
| transition_subject = re.compile( | |
| r"(?P<prefix>\b(?:reaction\s+pathway|conversion|transformation)\s+" | |
| r"from\s+)(?P<source>[^.;\n]{1,140}?)(?P<connector>\s+to\s+)" | |
| r"(?P<product>[^.;\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"(?<!\w)({re.escape(alias)})(\s*)\(" | |
| r"([^().;\n]{1,40}?)" | |
| r"(?=\s+(?:to|under|upon|when|after|before|was|were|is|are|" | |
| r"occurs?|converts?|converted|degrades?|changes?|forms?)\b|[.;])", | |
| flags=re.IGNORECASE, | |
| ) | |
| alias_pattern = re.compile( | |
| rf"(?<!\w)({re.escape(alias)})(\s*)" | |
| r"\(((?:[^()]|\([^()]*\))*)\)", | |
| flags=re.IGNORECASE, | |
| ) | |
| def _correct_alias_formula(match, expected=graph_species): | |
| inner = match.group(3) | |
| chemical_like = ( | |
| bool(parse_formula(inner)) | |
| or ( | |
| bool(re.search(r"\d", inner)) | |
| and bool( | |
| re.fullmatch( | |
| r"[\sA-Za-z0-9()\[\]{}·.\-+]+", | |
| inner, | |
| ) | |
| ) | |
| ) | |
| ) | |
| if not chemical_like: | |
| return match.group(0) | |
| return f"{match.group(1)} ({plain_formula(expected)})" | |
| content = tolerant_alias_pattern.sub(_correct_alias_formula, content) | |
| content = alias_pattern.sub(_correct_alias_formula, content) | |
| targets = [] | |
| for sp in species: | |
| if not re.search(r'\d', sp): | |
| continue | |
| plain = plain_formula(sp) | |
| if plain: | |
| targets.append(plain) | |
| if not targets: | |
| for i, block in enumerate(preserved): | |
| content = content.replace(f"__MATHBLOCKX{i}__", block) | |
| return self._repair_math_artifacts(content) | |
| replacements = [] | |
| for plain in sorted(set(targets), key=len, reverse=True): | |
| pattern = self._build_loose_formula_pattern(plain) | |
| replacements.append((pattern, self._latexify_inline_species(plain))) | |
| parts = re.split(r'(__MATHBLOCKX\d+__)', content) | |
| for idx, part in enumerate(parts): | |
| if not part or part.startswith("__MATHBLOCKX"): | |
| continue | |
| for pattern, dst in replacements: | |
| if not pattern: | |
| continue | |
| part = re.sub(pattern, lambda m, d=dst: d, part) | |
| parts[idx] = part | |
| content = "".join(parts) | |
| for i, block in enumerate(preserved): | |
| content = content.replace(f"__MATHBLOCKX{i}__", block) | |
| return self._repair_math_artifacts(content) | |
| def _repair_math_artifacts(self, text: str) -> 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)(?<!\\xrightarrow\{)\\\s*t\s*e\s*x\s*t\s*', '', text) | |
| text = re.sub(r'(?<!\\xrightarrow\{)\\ext(?=[A-Za-z])', '', text) | |
| text = re.sub( | |
| r'\b(?:[A-Za-z]\s+){2,}[A-Za-z]\b', | |
| lambda match: match.group(0).replace(" ", ""), | |
| text, | |
| ) | |
| for i, block in enumerate(preserved_blocks): | |
| text = text.replace(f"__LATEXBLOCK{i}__", block) | |
| for i, block in enumerate(inline_blocks): | |
| text = text.replace(f"__INLINEMATHBLOCK{i}__", block) | |
| return text | |
| def _build_latex_path(self, path_str): | |
| edges = parse_path(path_str) | |
| if not edges: | |
| return "" | |
| def _latex_species(text): | |
| t = str(text).strip() | |
| if not t: | |
| return "" | |
| plain = plain_formula(t) | |
| return self._latexify_inline_species(plain).strip("$") | |
| latex = _latex_species(edges[0].reactant) | |
| for edge in edges: | |
| cond = _display_condition_casing(edge.condition) | |
| if cond: | |
| arrow = f"\\xrightarrow{{{cond}}}" | |
| else: | |
| arrow = "\\rightarrow" | |
| latex += f" {arrow} {_latex_species(edge.product)}" | |
| return latex.strip() | |
| def _generate_unverified_path_report( | |
| self, | |
| root_material, | |
| path_str, | |
| retrieved_context=False, | |
| ): | |
| species = path_species(path_str) | |
| conditions = path_conditions(path_str) | |
| conditions = list(dict.fromkeys(conditions)) | |
| start_material = species[0] if species else root_material | |
| final_product = species[-1] if len(species) >= 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: <full citation>)" 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'}", | |
| ], | |
| } | |
| 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 | |
| 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, | |
| ) | |