| """ |
| graphrag_core.py |
| ================= |
| Shared, importable definitions for the Bridge-Symptom GraphRAG pipeline. |
| |
| This module contains every CLASS and FUNCTION definition from the original |
| research notebook (ontology, entity/relation extraction, the Clinical Knowledge |
| Graph wrapper, embedding + vector-index + BM25 + hybrid retrieval, the adaptive |
| query router, external evidence retrieval, evidence fusion + the Clinical |
| Evidence Validator, bridge-symptom graph reasoning, LLM answer generation, the |
| safety/crisis layer, the 5-route intent classifier, and the final |
| BridgeSymptomGraphRAG pipeline class) -- verbatim from the notebook, with only |
| the one-time BUILD EXECUTION (PDF parsing, corpus embedding, corpus-wide entity |
| extraction, KG-file loading, index construction, demo prints) removed. Nothing |
| about *how* the pipeline reasons, retrieves, validates evidence, or generates |
| answers has been changed. |
| |
| Both notebooks import this module: |
| * Notebook 1 (offline build) uses it to build the corpus/graph/indexes once |
| and save every reusable artifact. |
| * Notebook 2 (interactive chatbot) uses it to reconstruct lightweight runtime |
| objects from those saved artifacts in seconds, with NO rebuilding. |
| |
| Do not edit class/function LOGIC here without mirroring the change in the |
| original research notebook (Notebook 1), and vice versa -- this file is the |
| single source of truth for pipeline code going forward. |
| """ |
| import os, sys, re, json, hashlib, dataclasses, warnings, importlib, subprocess |
| from pathlib import Path |
| from datetime import datetime, timezone |
| from typing import Dict, List, Tuple, Optional |
| from collections import Counter |
|
|
| import numpy as np |
| import networkx as nx |
| import torch |
|
|
| def get_logger(name: str): |
| import logging |
| return logging.getLogger(name) |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| DIRS: Dict[str, str] = {} |
| DEVICE = "cpu" |
| has_pkg = lambda name: False |
| dependency_report: Dict = {} |
| GLOBAL_SEED = 42 |
| logger = get_logger("graphrag_core") |
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| CHUNK_LOOKUP: Dict[str, Dict] = {} |
| KG_NAME_LOOKUP: Dict[str, str] = {} |
| DEFAULT_DISEASE_PAIR: Tuple[str, str] = ("", "") |
| EMBEDDING_MODEL = None |
| INTENT_CLASSIFIER = None |
| ANSWER_GENERATOR = None |
|
|
|
|
| def bind_environment(dirs, device, has_pkg_fn, dependency_report_val, global_seed, log): |
| """ |
| Wire this module's globals to values decided by the notebook's own Chapter-1 |
| environment setup (DIRS, DEVICE, has_pkg, dependency_report, GLOBAL_SEED, logger), |
| so every class/function below behaves exactly as it did inline in the notebook. |
| Call this once, immediately after importing graphrag_core, before using anything else. |
| Also triggers the (deferred) optional spaCy/SciSpacy NER backend initialization, |
| which needs a real `has_pkg`/`logger` to run correctly. |
| """ |
| global DIRS, DEVICE, has_pkg, dependency_report, GLOBAL_SEED, logger |
| DIRS, DEVICE, has_pkg, dependency_report, GLOBAL_SEED, logger = ( |
| dirs, device, has_pkg_fn, dependency_report_val, global_seed, log) |
| _init_spacy_matcher() |
| _init_scispacy() |
|
|
|
|
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| _EVIDENCE_LEVEL_PATTERNS = [ |
| (re.compile(r"\bsystematic review\b", re.IGNORECASE), "Systematic Review"), |
| (re.compile(r"\bmeta-analysis\b", re.IGNORECASE), "Meta-Analysis"), |
| (re.compile(r"\b(randomi[sz]ed controlled trial|\bRCT\b)\b", re.IGNORECASE), "RCT"), |
| (re.compile(r"\bcohort stud(y|ies)\b", re.IGNORECASE), "Cohort Study"), |
| (re.compile(r"\bcase-control\b", re.IGNORECASE), "Case-Control Study"), |
| (re.compile(r"\bcase report\b", re.IGNORECASE), "Case Report"), |
| (re.compile(r"\b(guideline|practice parameter|clinical recommendation)\b", re.IGNORECASE), "Guideline"), |
| ] |
|
|
|
|
| def classify_text_evidence_level(text: str) -> str: |
| """ |
| Heuristically classify a text's position in the evidence hierarchy |
| (Systematic Review > Meta-Analysis > RCT > Cohort > Case-Control > Case |
| Report > Guideline), based on keyword patterns. |
| |
| Returns: |
| str: One of `ClinicalKnowledgeGraph.EVIDENCE_LEVEL_WEIGHTS` keys, or |
| "Not Classified" if no pattern matches. |
| """ |
| for pattern, level in _EVIDENCE_LEVEL_PATTERNS: |
| if pattern.search(text): |
| return level |
| return "Not Classified" |
|
|
|
|
| _CITATION_LIST_PATTERN = re.compile(r"([A-Z][a-z]+\s+[A-Z](\.|,)\s*,?\s*){2,}") |
| _COMMON_STOPWORDS = {"the", "a", "an", "is", "are", "was", "were", "and", "of", "in", "to", |
| "that", "this", "with", "for", "as", "by", "on", "be", "it", "have", "has"} |
|
|
|
|
| def _is_junk_sentence(sentence: str) -> bool: |
| """ |
| Identify sentences that are PDF extraction artifacts rather than genuine |
| body text (also reused by Chapter 11's extractive-fallback generator to |
| keep obviously-garbled retrieved text out of a synthesized answer). |
| """ |
| s = sentence.strip() |
| if len(s.split()) < 4: |
| return True |
| lower = s.lower() |
| junk_markers = ("manuscript", "nih-pa", "issn", "doi", "\u00a9", "all rights reserved", |
| "page ", "et al.,", "vol.", "pp.") |
| if any(marker in lower for marker in junk_markers): |
| return True |
| if _CITATION_LIST_PATTERN.search(s): |
| return True |
| letters = [c for c in s if c.isalpha()] |
| if letters and sum(1 for c in letters if c.isupper()) / len(letters) > 0.6: |
| return True |
| if sum(c.isdigit() for c in s) / max(len(s), 1) > 0.3: |
| return True |
| word_set = set(re.findall(r"[a-z]+", lower)) |
| if len(s.split()) >= 8 and not (word_set & _COMMON_STOPWORDS): |
| return True |
| return False |
|
|
|
|
| |
| |
| |
| |
| |
| ANXIETY_SYMPTOMS = { |
| "excessive worry", "restlessness", "fatigue", "difficulty concentrating", |
| "irritability", "muscle tension", "sleep disturbance", "panic attacks", |
| "racing heart", "shortness of breath", "sweating", "trembling", |
| "avoidance behavior", "hypervigilance", "rumination", "emotional dysregulation", |
| } |
| DEPRESSION_SYMPTOMS = { |
| "depressed mood", "anhedonia", "fatigue", "difficulty concentrating", |
| "irritability", "sleep disturbance", "appetite change", "weight change", |
| "psychomotor agitation", "psychomotor retardation", "feelings of worthlessness", |
| "suicidal ideation", "guilt", "restlessness", "rumination", "emotional dysregulation", |
| } |
| |
| |
| |
| BRIDGE_SYMPTOMS = ANXIETY_SYMPTOMS & DEPRESSION_SYMPTOMS |
|
|
| DISEASES = {"generalized anxiety disorder", "major depressive disorder", "panic disorder", "social anxiety disorder"} |
| MEDICATIONS = {"ssri", "snri", "benzodiazepine", "maoi", "tricyclic antidepressant", "buspirone", "mirtazapine"} |
| TREATMENTS = {"cognitive behavioral therapy", "behavioral activation", "exposure therapy", |
| "mindfulness-based stress reduction", "psychodynamic therapy", "interpersonal therapy"} |
| RISK_FACTORS = {"chronic stress", "childhood trauma", "family history", "substance use", |
| "social isolation", "chronic illness"} |
| BIOMARKERS = {"cortisol dysregulation", "hpa axis dysregulation", "elevated inflammatory markers", |
| "reduced hippocampal volume", "serotonin dysregulation"} |
| LIFESTYLE_FACTORS = {"physical inactivity", "poor sleep hygiene", "poor diet", "alcohol use", "smoking"} |
|
|
| ONTOLOGY_MAP = { |
| |
| "generalized anxiety disorder": {"mesh": "D001008", "umls": "C0270549", "snomed": "21897009"}, |
| "major depressive disorder": {"mesh": "D003865", "umls": "C1269683", "snomed": "370143000"}, |
| "fatigue": {"mesh": "D005221", "umls": "C0015672", "snomed": "84229001"}, |
| "difficulty concentrating": {"mesh": "D003221", "umls": "C0392197", "snomed": "451231000124102"}, |
| "irritability": {"mesh": "D057090", "umls": "C0022107", "snomed": "44077006"}, |
| "sleep disturbance": {"mesh": "D012893", "umls": "C0037317", "snomed": "39898005"}, |
| "psychomotor agitation": {"mesh": "D011595", "umls": "C0700078", "snomed": "57718002"}, |
| "cognitive behavioral therapy": {"mesh": "D015928", "umls": "C0009244", "snomed": "228557008"}, |
| "ssri": {"mesh": "D017367", "umls": "C0074554", "snomed": "372754003"}, |
| } |
|
|
| print(f"Bridge symptoms detected in ontology overlap ({len(BRIDGE_SYMPTOMS)}):") |
| for s in sorted(BRIDGE_SYMPTOMS): |
| print(" -", s) |
|
|
|
|
| |
| |
| |
| ALL_ENTITY_SETS = { |
| "Disease": DISEASES, "Symptom": (ANXIETY_SYMPTOMS | DEPRESSION_SYMPTOMS) - BRIDGE_SYMPTOMS, |
| "BridgeSymptom": BRIDGE_SYMPTOMS, "Medication": MEDICATIONS, "Treatment": TREATMENTS, |
| "RiskFactor": RISK_FACTORS, "Biomarker": BIOMARKERS, "Lifestyle": LIFESTYLE_FACTORS, |
| } |
|
|
| |
| |
| |
| |
| |
| |
| _NLP = None |
| _MATCHER = None |
|
|
|
|
| def _init_spacy_matcher() -> None: |
| global _NLP, _MATCHER |
| if not has_pkg("spacy"): |
| return |
| try: |
| import spacy |
| from spacy.matcher import PhraseMatcher |
| try: |
| _NLP = spacy.load("en_core_web_sm") |
| except OSError: |
| _NLP = spacy.blank("en") |
| _MATCHER = PhraseMatcher(_NLP.vocab, attr="LOWER") |
| for label, terms in ALL_ENTITY_SETS.items(): |
| _MATCHER.add(label, [_NLP.make_doc(t) for t in terms]) |
| logger.info("spaCy PhraseMatcher initialized for entity extraction.") |
| except Exception as exc: |
| logger.warning("spaCy setup failed (%s); falling back to regex extraction.", exc) |
| _NLP, _MATCHER = None, None |
|
|
|
|
| def extract_entities_regex(text: str) -> List[Dict]: |
| """Fallback entity extractor: regex word-boundary scan over curated term sets.""" |
| text_lower = text.lower() |
| found = [] |
| for label, terms in ALL_ENTITY_SETS.items(): |
| for term in terms: |
| for match in re.finditer(r"(?<!\w)" + re.escape(term) + r"(?!\w)", text_lower): |
| found.append({"text": term, "label": label, "start": match.start(), "end": match.end()}) |
| return found |
|
|
|
|
| def extract_entities(text: str) -> List[Dict]: |
| """ |
| Extract clinical entities from a text chunk using spaCy PhraseMatcher if |
| available, otherwise a regex fallback. Both paths return the same schema. |
| |
| Args: |
| text (str): Input text (a corpus chunk or external abstract). |
| |
| Returns: |
| List[Dict]: [{"text": str, "label": str, "start": int, "end": int}, ...] |
| """ |
| if _NLP is not None and _MATCHER is not None: |
| doc = _NLP.make_doc(text) |
| matches = _MATCHER(doc) |
| curated = [{"text": doc[s:e].text.lower(), "label": _NLP.vocab.strings[m_id], |
| "start": doc[s].idx, "end": doc[e - 1].idx + len(doc[e - 1])} |
| for m_id, s, e in matches] |
| else: |
| curated = extract_entities_regex(text) |
| |
| |
| |
| |
| combined = curated + _extract_entities_scispacy(text) |
| return _dedup_entity_spans(combined) |
|
|
|
|
| _RELATION_PATTERNS = [ |
| (re.compile(r"\btreat(?:s|ed|ment)?\b"), "TREATS"), |
| (re.compile(r"\bcauses?\b"), "CAUSES"), |
| (re.compile(r"\bincreases? (the )?risk\b"), "INCREASES_RISK"), |
| (re.compile(r"\bco-?occurs? with\b"), "CO_OCCURS"), |
| (re.compile(r"\bcontraindicated\b"), "CONTRAINDICATED_WITH"), |
| (re.compile(r"\brecommend(?:s|ed|ation)?\b"), "RECOMMENDED_BY"), |
| ] |
|
|
|
|
| def extract_relations(text: str, entities: List[Dict]) -> List[Dict]: |
| """ |
| Extract relations between co-occurring entities within the same sentence, |
| using cue-phrase pattern matching to assign a relation type; defaults to |
| RELATED_TO when entities co-occur without a recognized cue phrase. |
| |
| Args: |
| text (str): The source text the entities were extracted from. |
| entities (List[Dict]): Output of `extract_entities`. |
| |
| Returns: |
| List[Dict]: [{"source": str, "target": str, "relation": str, "sentence": str}, ...] |
| """ |
| sentences = re.split(r"(?<=[.!?])\s+", text) |
| relations = [] |
| offset = 0 |
| for sentence in sentences: |
| sent_start, sent_end = offset, offset + len(sentence) |
| offset = sent_end + 1 |
| sent_entities = [e for e in entities if sent_start <= e["start"] < sent_end] |
| if len(sent_entities) < 2: |
| continue |
| relation_type = "RELATED_TO" |
| for pattern, rel_label in _RELATION_PATTERNS: |
| if pattern.search(sentence.lower()): |
| relation_type = rel_label |
| break |
| for i in range(len(sent_entities)): |
| for j in range(i + 1, len(sent_entities)): |
| relations.append({ |
| "source": sent_entities[i]["text"], "target": sent_entities[j]["text"], |
| "source_label": sent_entities[i]["label"], "target_label": sent_entities[j]["label"], |
| "relation": relation_type, "sentence": sentence.strip(), |
| }) |
| return relations |
|
|
|
|
| def detect_bridge_symptoms(entities: List[Dict]) -> List[str]: |
| """Return the subset of extracted entities that are BridgeSymptom-labeled.""" |
| return sorted({e["text"] for e in entities if e["label"] == "BridgeSymptom"}) |
|
|
|
|
| def normalize_entity(entity_text: str) -> Dict: |
| """ |
| Map a curated entity string onto ontology codes. |
| |
| This is the explicit integration point for a licensed UMLS/SNOMED CT API: |
| replace the `ONTOLOGY_MAP.get(...)` lookup with a call to your terminology |
| server if you have credentials, keeping the same return schema. |
| |
| Args: |
| entity_text (str): Canonical lowercase entity string. |
| |
| Returns: |
| Dict: {"mesh": str|None, "umls": str|None, "snomed": str|None} |
| """ |
| return ONTOLOGY_MAP.get(entity_text, {"mesh": None, "umls": None, "snomed": None}) |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| _SCISPACY_NLP = None |
| _SCISPACY_LABEL_MAP = { |
| "DISEASE": "Disease", "CHEMICAL": "Medication", "DRUG": "Medication", |
| "ENTITY": "Symptom", "SIGN_OR_SYMPTOM": "Symptom", "NEOPLASTIC_PROCESS": "Disease", |
| "PHARMACOLOGIC_SUBSTANCE": "Medication", "THERAPEUTIC_OR_PREVENTIVE_PROCEDURE": "Treatment", |
| } |
| def _init_scispacy() -> None: |
| """Deferred (see `_init_spacy_matcher` note above) -- called once from `bind_environment()`.""" |
| global _SCISPACY_NLP |
| for _sci_model in ("en_ner_bc5cdr_md", "en_core_sci_sm"): |
| if has_pkg("scispacy") and has_pkg("spacy"): |
| try: |
| import spacy as _spacy_sci |
| _SCISPACY_NLP = _spacy_sci.load(_sci_model) |
| logger.info("SciSpacy biomedical NER model loaded: %s", _sci_model) |
| return |
| except Exception: |
| _SCISPACY_NLP = None |
| if _SCISPACY_NLP is None: |
| logger.info("SciSpacy not available; hybrid NER uses curated PhraseMatcher/regex only.") |
|
|
|
|
| def _extract_entities_scispacy(text: str) -> List[Dict]: |
| """Return SciSpacy biomedical entity spans mapped onto this project's label schema, |
| or [] if no SciSpacy model is loaded. Labels not in the map default to 'Symptom'.""" |
| if _SCISPACY_NLP is None or not text: |
| return [] |
| try: |
| doc = _SCISPACY_NLP(text[:100000]) |
| except Exception: |
| return [] |
| out = [] |
| for ent in doc.ents: |
| out.append({"text": ent.text.lower(), "label": _SCISPACY_LABEL_MAP.get(ent.label_.upper(), "Symptom"), |
| "start": ent.start_char, "end": ent.end_char, "extractor": "scispacy"}) |
| return out |
|
|
|
|
| def _dedup_entity_spans(spans: List[Dict]) -> List[Dict]: |
| """Drop duplicate entity spans (same lowercased text + overlapping character range), |
| preferring the curated-label span (which carries the precise project schema) over a |
| SciSpacy span when both cover the same mention.""" |
| spans_sorted = sorted(spans, key=lambda s: (s.get("extractor") == "scispacy", s["start"])) |
| kept = [] |
| for s in spans_sorted: |
| clash = False |
| for k in kept: |
| if s["text"] == k["text"] and not (s["end"] <= k["start"] or s["start"] >= k["end"]): |
| clash = True |
| break |
| if not clash: |
| kept.append(s) |
| return kept |
|
|
|
|
|
|
| |
| |
| |
| def process_corpus_for_entities(corpus: List[Dict]) -> Tuple[List[Dict], List[Dict]]: |
| """ |
| Run entity + relation extraction across every chunk of a corpus. |
| |
| Args: |
| corpus (List[Dict]): Output of Chapter 2's `build_local_corpus` (or any |
| corpus with the same `{chunk_id, text}` schema, e.g. external results). |
| |
| Returns: |
| Tuple[List[Dict], List[Dict]]: (all_entity_records, all_relation_records), |
| each annotated with the originating chunk_id for provenance. |
| """ |
| all_entities, all_relations = [], [] |
| for item in corpus: |
| chunk_id, text = item["chunk_id"], item["text"] |
| entities = extract_entities(text) |
| for e in entities: |
| e["chunk_id"] = chunk_id |
| e["ontology"] = normalize_entity(e["text"]) |
| relations = extract_relations(text, entities) |
| for r in relations: |
| r["chunk_id"] = chunk_id |
| all_entities.extend(entities) |
| all_relations.extend(relations) |
| return all_entities, all_relations |
|
|
|
|
|
|
|
|
| |
| |
| |
| import networkx as nx |
| from datetime import datetime, timezone |
|
|
|
|
| def _dkg_noisy_or(c1: float, c2: float) -> float: |
| """Combine two independent confidence estimates so repeated corroboration pushes |
| confidence up (never below either individual estimate), capped at 0.99. Used by |
| ClinicalKnowledgeGraph's node/edge merge logic -- see the Dynamic KG merge comments below.""" |
| c1, c2 = max(0.0, min(0.99, c1)), max(0.0, min(0.99, c2)) |
| return round(min(1 - (1 - c1) * (1 - c2), 0.99), 4) |
|
|
|
|
| class ClinicalKnowledgeGraph: |
| """ |
| Wraps a networkx.MultiDiGraph with clinical-schema-aware node/edge insertion, |
| noisy-OR confidence-weighted merging when the SAME node/edge is added again -- |
| whether that's two seed calls touching the same concept (Section 4.2) or, now, |
| a dynamic-ingestion mention of a concept the literature seed already has (Section |
| 4.3) -- plus graph versioning (.snapshot() / .changelog) so the graph's growth over |
| time stays auditable. |
| |
| Node schema: node_id, canonical_name, node_type, description, ontology_id, |
| synonyms, confidence, source, timestamp, graph_version_added |
| Edge schema: relation, confidence, relation_confidence, evidence, evidence_count, |
| evidence_level, pubmed_id, publication_year, guideline_id, weight, |
| extraction_method, source, paper_id, timestamp, graph_version_added |
| """ |
|
|
| VALID_NODE_TYPES = {"Disease", "BridgeSymptom", "Symptom", "Medication", "DrugClass", |
| "Treatment", "RiskFactor", "Biomarker", "Lifestyle", "Guideline", |
| "ResearchPaper", "ClinicalTrial", "Drug", "Gene", "PatientGroup", |
| "Evidence", "DiagnosticCriteria", |
| "ComorbidityProfile", "RiskConsequence", "Complication", "ClinicalManifestation", |
| |
| |
| |
| "ProtectiveFactor", "BrainRegion", "Neurotransmitter", "SideEffect", |
| "AssessmentScale", "ClinicalOutcome"} |
| VALID_EDGE_TYPES = {"HAS_SYMPTOM", "HAS_BRIDGE_SYMPTOM", "BRIDGE_TO", "CO_OCCURS", "TREATS", |
| "CAUSES", "SUPPORTED_BY", "VALIDATED_BY", "RECOMMENDED_BY", "MENTIONED_IN", |
| "INCREASES_RISK", "RELATED_TO", "CONTRAINDICATED_WITH", "INTERACTS_WITH", |
| "ASSOCIATED_WITH", |
| "SHARES_RISK_FACTOR", "CO_OCCURS_WITH", "INCREASES_RISK_OF", |
| "COMPLICATED_BY", "SHARES_MECHANISM", "TREATED_WITH", "MANIFESTS_AS", |
| |
| |
| "DECREASES_RISK_OF", "AFFECTS", "ACTIVATES", "INHIBITS", "IMPROVES", |
| "WORSENS", "PREDICTS", "DIAGNOSED_BY", "MEASURED_BY"} |
|
|
| |
| |
| EVIDENCE_LEVEL_WEIGHTS = { |
| "Systematic Review": 1.0, "Meta-Analysis": 0.95, "RCT": 0.85, |
| "Cohort Study": 0.7, "Case-Control Study": 0.6, "Case Report": 0.4, |
| "Guideline": 0.9, "Expert Opinion": 0.3, "Not Classified": 0.5, |
| } |
|
|
| def __init__(self): |
| self.graph = nx.MultiDiGraph() |
| |
| |
| |
| |
| |
| self.version = 0 |
| self.changelog = [] |
| self._log(f"Graph initialized (version {self.version})") |
|
|
| def _log(self, message: str) -> None: |
| self.changelog.append({"timestamp": datetime.now(timezone.utc).isoformat(), |
| "version": self.version, "message": message}) |
|
|
| def snapshot(self, note: str = "") -> int: |
| """Bump the graph version and record a changelog entry. Call this after each |
| distinct ingestion batch (the literature seed, then each dynamic-ingestion run) |
| so the graph's history stays auditable.""" |
| self.version += 1 |
| self._log(f"Snapshot v{self.version}: {note}" if note else f"Snapshot v{self.version}") |
| logger.info("Knowledge graph snapshot v%d: %s", self.version, note) |
| return self.version |
|
|
| def add_node(self, node_id: str, canonical_name: str, node_type: str, |
| description: str = "", confidence: float = 1.0, source: str = "manual", |
| ontology_id: str = "", synonyms: List[str] = None, timestamp: str = None) -> None: |
| """ |
| Insert a node, or merge into an existing one (confidence-weighted average, |
| source list union, synonym union) if `node_id` already exists -- implements |
| the required 'entity merging' and 'confidence update' behaviors. |
| |
| Args: |
| ontology_id (str): External terminology code (MeSH/UMLS/SNOMED-style; |
| see Chapter 3's honesty note on ontology mapping provenance). |
| synonyms (List[str]): Alternate surface forms for this concept. |
| """ |
| if node_type not in self.VALID_NODE_TYPES: |
| raise ValueError(f"Invalid node_type '{node_type}'. Must be one of {self.VALID_NODE_TYPES}") |
| timestamp = timestamp or datetime.now(timezone.utc).isoformat() |
| synonyms_str = ";".join(synonyms) if synonyms else "" |
|
|
| if self.graph.has_node(node_id): |
| existing = self.graph.nodes[node_id] |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| existing["confidence"] = _dkg_noisy_or(existing["confidence"], confidence) |
| sources = set(existing.get("source", "").split(";")) | {source} |
| existing["source"] = ";".join(sorted(s for s in sources if s)) |
| existing_syn = set(existing.get("synonyms", "").split(";")) | set(synonyms_str.split(";")) |
| existing["synonyms"] = ";".join(sorted(s for s in existing_syn if s)) |
| if not existing.get("ontology_id") and ontology_id: |
| existing["ontology_id"] = ontology_id |
| existing["timestamp"] = timestamp |
| else: |
| self.graph.add_node(node_id, canonical_name=canonical_name, node_type=node_type, |
| description=description, ontology_id=ontology_id, |
| synonyms=synonyms_str, confidence=confidence, |
| source=source, timestamp=timestamp, |
| graph_version_added=self.version) |
| self._log(f"Added node {node_id} ({node_type}) from source '{source}'") |
|
|
| def add_edge(self, source_id: str, target_id: str, relation: str, confidence: float = 1.0, |
| evidence: str = "", source: str = "manual", paper_id: str = "", |
| evidence_level: str = "Not Classified", pubmed_id: str = "", |
| guideline_id: str = "", extraction_method: str = "manual", |
| publication_year: str = "", timestamp: str = None) -> None: |
| """ |
| Insert an edge, deduplicating exact (source, target, relation) triples by |
| merging confidence and evidence instead of creating parallel duplicates. |
| `weight` is derived automatically from confidence and evidence-level |
| strength, for use by Chapter 10's weighted/guideline-aware traversal. |
| |
| Args: |
| evidence_level (str): One of `EVIDENCE_LEVEL_WEIGHTS` keys. |
| pubmed_id (str): PMID backing this edge, if derived from PubMed evidence. |
| guideline_id (str): Identifier of a clinical guideline backing this edge. |
| extraction_method (str): "manual_seed" | "regex_ner" | "spacy_ner" | "external_retrieval". |
| """ |
| if relation not in self.VALID_EDGE_TYPES: |
| raise ValueError(f"Invalid relation '{relation}'. Must be one of {self.VALID_EDGE_TYPES}") |
| if not (self.graph.has_node(source_id) and self.graph.has_node(target_id)): |
| logger.debug("Skipping edge %s->%s: endpoint node missing.", source_id, target_id) |
| return |
| timestamp = timestamp or datetime.now(timezone.utc).isoformat() |
| level_weight = self.EVIDENCE_LEVEL_WEIGHTS.get(evidence_level, 0.5) |
| weight = round(confidence * level_weight, 4) |
|
|
| for _, tgt, key, data in self.graph.out_edges(source_id, keys=True, data=True): |
| if tgt == target_id and data.get("relation") == relation: |
| |
| data["confidence"] = _dkg_noisy_or(data["confidence"], confidence) |
| data["relation_confidence"] = data["confidence"] |
| |
| data["evidence_count"] = data.get("evidence_count", 1) + 1 |
| if publication_year and not data.get("publication_year"): |
| data["publication_year"] = publication_year |
| |
| |
| if level_weight > self.EVIDENCE_LEVEL_WEIGHTS.get(data.get("evidence_level", "Not Classified"), 0.5): |
| data["evidence_level"] = evidence_level |
| data["weight"] = round(data["confidence"] * |
| self.EVIDENCE_LEVEL_WEIGHTS.get(data["evidence_level"], 0.5), 4) |
| if evidence and evidence not in data.get("evidence", ""): |
| data["evidence"] = (data.get("evidence", "") + " | " + evidence).strip(" |") |
| if pubmed_id and pubmed_id not in data.get("pubmed_id", ""): |
| data["pubmed_id"] = (data.get("pubmed_id", "") + ";" + pubmed_id).strip(";") |
| return |
|
|
| self.graph.add_edge(source_id, target_id, relation=relation, confidence=confidence, |
| relation_confidence=confidence, evidence=evidence, evidence_count=1, |
| evidence_level=evidence_level, pubmed_id=pubmed_id, |
| publication_year=publication_year, |
| guideline_id=guideline_id, weight=weight, extraction_method=extraction_method, |
| source=source, paper_id=paper_id, timestamp=timestamp, |
| graph_version_added=self.version) |
|
|
| def node_count_by_type(self) -> Dict[str, int]: |
| """Return a histogram of node counts per node_type.""" |
| from collections import Counter |
| return dict(Counter(nx.get_node_attributes(self.graph, "node_type").values())) |
|
|
| def export_all(self, export_dir: str) -> Dict[str, str]: |
| """Export the graph to GraphML, GEXF, and JSON (node-link) formats.""" |
| os.makedirs(export_dir, exist_ok=True) |
| paths = {} |
| graphml_path = os.path.join(export_dir, "clinical_kg.graphml") |
| gexf_path = os.path.join(export_dir, "clinical_kg.gexf") |
| json_path = os.path.join(export_dir, "clinical_kg.json") |
|
|
| g_copy = nx.MultiDiGraph() |
| g_copy.add_nodes_from(self.graph.nodes(data=True)) |
| g_copy.add_edges_from(self.graph.edges(data=True)) |
| nx.write_graphml(g_copy, graphml_path) |
| nx.write_gexf(g_copy, gexf_path) |
| with open(json_path, "w", encoding="utf-8") as f: |
| json.dump(nx.node_link_data(self.graph), f, indent=2, default=str) |
|
|
| paths.update({"graphml": graphml_path, "gexf": gexf_path, "json": json_path}) |
| logger.info("Knowledge graph exported: %s", paths) |
| return paths |
|
|
|
|
| |
| |
| |
| |
| _PUB_YEAR_RE = re.compile(r"\b(19[5-9]\d|20[0-4]\d)\b") |
|
|
|
|
| def _extract_publication_year(text: str) -> str: |
| """Return the most recent plausible publication year found in `text`, or "" if none.""" |
| if not text: |
| return "" |
| years = _PUB_YEAR_RE.findall(text) |
| return max(years) if years else "" |
|
|
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| def build_kg_name_lookup(kg: ClinicalKnowledgeGraph) -> Dict[str, str]: |
| """ |
| Map every node's canonical_name and synonyms (all lowercased) to its node_id, so a raw |
| query string -- or, here, a node from the loaded KG file -- can be matched against the |
| ACTUAL graph contents rather than guessed via a hardcoded ID scheme. Also indexes any |
| trailing "(...)" abbreviation both attached and stripped (e.g. "Cognitive Behavioral |
| Therapy (CBT)" indexes as itself, "cognitive behavioral therapy", AND "cbt"), so all common |
| surface forms of the same concept resolve to one node. |
| """ |
| lookup = {} |
| for node_id, data in kg.graph.nodes(data=True): |
| name = data.get("canonical_name", "").strip().lower() |
| if name: |
| lookup[name] = node_id |
| stripped = re.sub(r"\s*\([^)]*\)\s*$", "", name).strip() |
| if stripped and stripped != name: |
| lookup[stripped] = node_id |
| abbrev_match = re.search(r"\(([^)]+)\)\s*$", name) |
| if abbrev_match: |
| abbrev = abbrev_match.group(1).strip() |
| if abbrev: |
| lookup[abbrev] = node_id |
| for syn in data.get("synonyms", "").split(";"): |
| syn = syn.strip().lower() |
| if syn: |
| lookup[syn] = node_id |
| return lookup |
|
|
|
|
| def _kg_make_node_id(node_type: str, canonical_name: str) -> str: |
| """Deterministic node ID for a NEW node loaded from the KG file (never used for a node that |
| already exists in the graph -- those are resolved by name via `resolve_or_create_node`).""" |
| slug = re.sub(r"[^a-z0-9]+", "_", canonical_name.strip().lower()).strip("_") |
| return f"kg:{node_type.lower()}:{slug}" |
|
|
|
|
| def resolve_or_create_node(kg: ClinicalKnowledgeGraph, name_lookup: Dict[str, str], canonical_name: str, |
| node_type: str, aliases: List[str], confidence: float, source: str) -> str: |
| """ |
| Resolve `canonical_name` against the CURRENT graph's name lookup (exact name or any known |
| synonym, case-insensitive); if found, merge into that existing node via `kg.add_node` |
| (confidence/synonyms/source merge automatically -- see Section 4.1's noisy-OR upgrade). The |
| existing node's node_type and node_id are NEVER changed by a merge. If no match exists, |
| create a new node with a fresh deterministic ID and register it in `name_lookup` |
| immediately, so later nodes in the SAME load resolve to it instead of creating a duplicate. |
| |
| Returns: |
| str: The resolved or newly-created node_id. |
| """ |
| key = canonical_name.strip().lower() |
| node_id = name_lookup.get(key) |
| if node_id is None: |
| for alias in aliases: |
| node_id = name_lookup.get(alias.strip().lower()) |
| if node_id: |
| break |
|
|
| if node_id is not None: |
| kg.add_node(node_id, canonical_name, kg.graph.nodes[node_id]["node_type"], |
| confidence=confidence, source=source, synonyms=aliases) |
| return node_id |
|
|
| node_id = _kg_make_node_id(node_type, canonical_name) |
| kg.add_node(node_id, canonical_name, node_type, confidence=confidence, source=source, synonyms=aliases) |
| name_lookup[key] = node_id |
| for alias in aliases: |
| name_lookup.setdefault(alias.strip().lower(), node_id) |
| return node_id |
|
|
|
|
| print("Entity resolution helpers ready: build_kg_name_lookup, resolve_or_create_node.") |
|
|
|
|
| |
| |
| |
| import warnings |
| import numpy as np |
|
|
| |
| |
| |
| |
| |
| |
| |
| EMBEDDING_MODEL_CANDIDATES = [ |
| "pritamdeka/S-PubMedBert-MS-MARCO", |
| "BAAI/bge-small-en-v1.5", |
| "sentence-transformers/all-MiniLM-L6-v2", |
| ] |
|
|
|
|
| class EmbeddingModel: |
| """ |
| Unified embedding interface. Tries each model in `EMBEDDING_MODEL_CANDIDATES` |
| in order (best retrieval quality first); falls back to a TF-IDF + TruncatedSVD |
| projection if no sentence-transformers model can be loaded at all (graceful |
| degradation, never a placeholder no-op). |
| """ |
|
|
| def __init__(self, model_name: str = None, svd_dim: int = 256): |
| self.backend, self.model, self.model_name = None, None, None |
| self.svd_dim = svd_dim |
| self._tfidf, self._svd = None, None |
|
|
| candidates = [model_name] if model_name else EMBEDDING_MODEL_CANDIDATES |
| if has_pkg("sentence_transformers"): |
| from sentence_transformers import SentenceTransformer |
| for candidate in candidates: |
| try: |
| self.model = SentenceTransformer(candidate, device=DEVICE) |
| self.backend, self.model_name = "sentence_transformers", candidate |
| logger.info("EmbeddingModel backend: sentence_transformers (%s)", candidate) |
| break |
| except Exception as exc: |
| logger.warning("Could not load embedding model '%s' (%s); trying next candidate.", |
| candidate, exc) |
|
|
| if self.backend is None: |
| self.backend = "tfidf_svd" |
| logger.info("EmbeddingModel backend: TF-IDF + TruncatedSVD fallback") |
|
|
| def fit_fallback(self, corpus_texts: List[str]) -> None: |
| """ |
| Fit the TF-IDF+SVD fallback pipeline on the corpus (fallback backend only). |
| |
| Robustness note: fitting TruncatedSVD on a degenerate input (very few |
| documents, e.g. during an isolated single-query encode() call before |
| any real corpus was ever fit) can produce a harmless |
| "invalid value encountered in divide" RuntimeWarning from sklearn's |
| internal explained-variance bookkeeping -- it doesn't affect the |
| actual output (verified: output remains a valid, correctly-shaped |
| vector), but it's noisy. Suppressed narrowly here rather than globally, |
| so other genuine warnings elsewhere in the notebook still surface. |
| """ |
| if self.backend != "tfidf_svd": |
| return |
| from sklearn.feature_extraction.text import TfidfVectorizer |
| from sklearn.decomposition import TruncatedSVD |
| self._tfidf = TfidfVectorizer(max_features=20000, stop_words="english", ngram_range=(1, 2)) |
| tfidf_matrix = self._tfidf.fit_transform(corpus_texts) |
| n_components = min(self.svd_dim, tfidf_matrix.shape[1] - 1, tfidf_matrix.shape[0] - 1) |
| self._svd = TruncatedSVD(n_components=max(n_components, 2), random_state=GLOBAL_SEED) |
| with warnings.catch_warnings(): |
| warnings.filterwarnings("ignore", category=RuntimeWarning, module="sklearn.*") |
| self._svd.fit(tfidf_matrix) |
|
|
| def encode(self, texts: List[str], batch_size: int = 32) -> np.ndarray: |
| """ |
| Encode a list of texts into a 2D numpy array of dense embeddings, |
| L2-normalized so cosine similarity reduces to a dot product. |
| |
| Args: |
| texts (List[str]): Input texts. |
| batch_size (int): Batch size for the transformer backend. |
| |
| Returns: |
| np.ndarray: shape (len(texts), embedding_dim) |
| """ |
| if self.backend == "sentence_transformers": |
| embeddings = self.model.encode(texts, batch_size=batch_size, show_progress_bar=False, |
| convert_to_numpy=True, normalize_embeddings=True) |
| return embeddings.astype(np.float32) |
|
|
| if self._tfidf is None: |
| self.fit_fallback(texts) |
| tfidf_matrix = self._tfidf.transform(texts) |
| dense = self._svd.transform(tfidf_matrix) |
|
|
| |
| |
| target_dim = self._svd.n_components |
| if dense.shape[1] < target_dim: |
| pad = np.zeros((dense.shape[0], target_dim - dense.shape[1]), dtype=dense.dtype) |
| dense = np.hstack([dense, pad]) |
| elif dense.shape[1] > target_dim: |
| dense = dense[:, :target_dim] |
|
|
| norms = np.linalg.norm(dense, axis=1, keepdims=True) |
| norms[norms == 0] = 1.0 |
| return (dense / norms).astype(np.float32) |
|
|
|
|
|
|
|
|
| |
| |
| |
| class VectorIndex: |
| """ |
| Unified nearest-neighbor index interface. Uses FAISS (IndexFlatIP over |
| normalized vectors, i.e. cosine similarity) when available, otherwise falls |
| back to scikit-learn's NearestNeighbors with cosine metric. |
| """ |
|
|
| def __init__(self, embeddings: np.ndarray, chunk_ids: List[str]): |
| self.chunk_ids = chunk_ids |
| self.dim = embeddings.shape[1] if embeddings.size else 0 |
| self.backend = None |
| self._sk_index = None |
| self._faiss_index = None |
|
|
| if embeddings.shape[0] == 0: |
| logger.warning("VectorIndex initialized with 0 vectors.") |
| return |
|
|
| if has_pkg("faiss"): |
| try: |
| import faiss |
| self._faiss_index = faiss.IndexFlatIP(self.dim) |
| self._faiss_index.add(embeddings) |
| self.backend = "faiss" |
| except Exception as exc: |
| logger.warning("FAISS index build failed (%s); using sklearn fallback.", exc) |
|
|
| if self.backend is None: |
| from sklearn.neighbors import NearestNeighbors |
| self._sk_index = NearestNeighbors(metric="cosine", n_neighbors=min(50, len(chunk_ids))) |
| self._sk_index.fit(embeddings) |
| self.backend = "sklearn" |
|
|
| logger.info("VectorIndex backend: %s (%d vectors, dim=%d)", self.backend, len(chunk_ids), self.dim) |
|
|
| def search(self, query_embedding: np.ndarray, top_k: int = 10) -> List[Tuple[str, float]]: |
| """ |
| Return the top_k (chunk_id, similarity_score) pairs for a query embedding. |
| |
| Args: |
| query_embedding (np.ndarray): shape (dim,) or (1, dim). |
| top_k (int): Number of results to return. |
| |
| Returns: |
| List[Tuple[str, float]]: Sorted by descending similarity. |
| """ |
| if self.backend is None or not self.chunk_ids: |
| return [] |
| query_embedding = query_embedding.reshape(1, -1) |
| top_k = min(top_k, len(self.chunk_ids)) |
|
|
| if self.backend == "faiss": |
| scores, indices = self._faiss_index.search(query_embedding, top_k) |
| return [(self.chunk_ids[i], float(s)) for i, s in zip(indices[0], scores[0]) if i != -1] |
|
|
| distances, indices = self._sk_index.kneighbors(query_embedding, n_neighbors=top_k) |
| return [(self.chunk_ids[i], float(1 - d)) for i, d in zip(indices[0], distances[0])] |
|
|
|
|
|
|
|
|
| |
| |
| |
| class BM25RetrieverWrapper: |
| """ |
| Sparse lexical retriever. Uses rank_bm25.BM25Okapi when available; otherwise |
| falls back to a TF-IDF-cosine retriever with an identical `.search()` interface. |
| """ |
|
|
| def __init__(self, corpus: List[Dict]): |
| self.corpus = corpus |
| self.chunk_ids = [c["chunk_id"] for c in corpus] |
| self.backend = None |
| self._bm25 = None |
| self._tfidf_vectorizer = None |
| self._tfidf_matrix = None |
|
|
| if not corpus: |
| logger.warning("BM25RetrieverWrapper initialized with empty corpus.") |
| return |
|
|
| tokenized = [c["text"].lower().split() for c in corpus] |
| if has_pkg("rank_bm25"): |
| try: |
| from rank_bm25 import BM25Okapi |
| self._bm25 = BM25Okapi(tokenized) |
| self.backend = "bm25" |
| except Exception as exc: |
| logger.warning("rank_bm25 init failed (%s); using TF-IDF fallback.", exc) |
|
|
| if self.backend is None: |
| from sklearn.feature_extraction.text import TfidfVectorizer |
| self._tfidf_vectorizer = TfidfVectorizer(stop_words="english") |
| self._tfidf_matrix = self._tfidf_vectorizer.fit_transform([c["text"] for c in corpus]) |
| self.backend = "tfidf" |
|
|
| logger.info("BM25RetrieverWrapper backend: %s", self.backend) |
|
|
| def search(self, query: str, top_k: int = 10) -> List[Tuple[str, float]]: |
| """Return top_k (chunk_id, score) pairs by lexical relevance to `query`.""" |
| if not self.corpus: |
| return [] |
| if self.backend == "bm25": |
| scores = self._bm25.get_scores(query.lower().split()) |
| else: |
| from sklearn.metrics.pairwise import cosine_similarity |
| query_vec = self._tfidf_vectorizer.transform([query]) |
| scores = cosine_similarity(query_vec, self._tfidf_matrix)[0] |
| ranked = sorted(zip(self.chunk_ids, scores), key=lambda x: x[1], reverse=True) |
| return ranked[:top_k] |
|
|
|
|
|
|
|
|
| |
| |
| |
| def _minmax_normalize(scored: List[Tuple[str, float]]) -> Dict[str, float]: |
| """Min-max normalize a list of (id, score) pairs into a {id: score in [0,1]} dict.""" |
| if not scored: |
| return {} |
| values = [s for _, s in scored] |
| lo, hi = min(values), max(values) |
| span = (hi - lo) or 1.0 |
| return {cid: (s - lo) / span for cid, s in scored} |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| CLINICAL_SYNONYMS = { |
| "depression": ["major depressive disorder", "mdd", "depressive disorder", "low mood"], |
| "anxiety": ["generalized anxiety disorder", "gad", "anxiety disorder"], |
| "insomnia": ["sleep disturbance", "poor sleep", "sleeplessness"], |
| "ssri": ["selective serotonin reuptake inhibitor", "antidepressant"], |
| "snri": ["serotonin norepinephrine reuptake inhibitor", "antidepressant"], |
| "cbt": ["cognitive behavioral therapy", "cognitive behavioural therapy"], |
| "comorbid": ["co-occurring", "comorbidity", "co-morbid"], |
| "suicidal": ["suicidal ideation", "self-harm"], |
| "fatigue": ["tiredness", "exhaustion", "low energy"], |
| "worry": ["excessive worry", "rumination"], |
| } |
|
|
|
|
| def expand_query(query: str, max_extra_terms: int = 6) -> str: |
| """Append a handful of clinical synonyms/abbreviations for terms present in `query`, |
| producing a lexically richer string for BM25. Purely additive: the original query text |
| is preserved verbatim at the front, so it can only help lexical recall, never hurt it.""" |
| q_lower = query.lower() |
| extras = [] |
| for term, syns in CLINICAL_SYNONYMS.items(): |
| if re.search(r"\b" + re.escape(term) + r"\b", q_lower): |
| for s in syns: |
| if s not in q_lower and s not in extras: |
| extras.append(s) |
| return query + (" " + " ".join(extras[:max_extra_terms]) if extras else "") |
|
|
|
|
| def _dedup_candidates(candidates: List[Dict], similarity_threshold: float = 0.92) -> List[Dict]: |
| """Remove near-duplicate chunks from a ranked candidate list (e.g. the same passage |
| surfaced by both BM25 and dense search, or overlapping chunks), keeping the higher-ranked |
| copy. Uses fast lexical (token-set Jaccard) similarity so it adds negligible latency.""" |
| kept, kept_tokens = [], [] |
| for c in candidates: |
| toks = set(re.findall(r"[a-z0-9]+", c.get("text", "").lower())) |
| if any(len(toks & kt) / max(len(toks | kt), 1) > similarity_threshold for kt in kept_tokens): |
| continue |
| kept.append(c) |
| kept_tokens.append(toks) |
| return kept |
|
|
|
|
| def compress_context(query: str, text: str, max_sentences: int = 4, max_chars: int = 600) -> str: |
| """Query-aware extractive compression: keep the `max_sentences` sentences most similar to |
| `query` (by dense embedding cosine), preserving their original order. Falls back to a plain |
| character truncation if the text is already short or embeddings are unavailable.""" |
| sentences = [s.strip() for s in re.split(r"(?<=[.!?])\s+", text.strip()) if len(s.split()) >= 4] |
| if len(sentences) <= max_sentences: |
| return text[:max_chars] |
| try: |
| q_emb = EMBEDDING_MODEL.encode([query])[0] |
| s_embs = EMBEDDING_MODEL.encode(sentences) |
| sims = s_embs @ q_emb |
| top_idx = sorted(sorted(range(len(sentences)), key=lambda i: -sims[i])[:max_sentences]) |
| return " ".join(sentences[i] for i in top_idx)[:max_chars] |
| except Exception: |
| return text[:max_chars] |
|
|
|
|
| |
| |
| |
| |
| CROSS_ENCODER_CANDIDATES = [ |
| "cross-encoder/ms-marco-MiniLM-L-12-v2", |
| "cross-encoder/ms-marco-MiniLM-L-6-v2", |
| ] |
|
|
|
|
| class CrossEncoderReranker: |
| """ |
| Second-stage reranker: scores (query, passage) pairs jointly with a |
| cross-encoder, which captures query-passage interaction that first-stage |
| bi-encoder (dense) and lexical (BM25) scoring cannot -- this is the |
| standard fix for low first-stage precision (BM25 + dense fusion alone |
| over-retrieves topically-related-but-not-directly-relevant chunks). |
| Falls back to a no-op (keeps original ranking) if no cross-encoder in |
| `CROSS_ENCODER_CANDIDATES` can be loaded. |
| """ |
|
|
| def __init__(self, model_name: str = None): |
| self.model, self.model_name = None, None |
| candidates = [model_name] if model_name else CROSS_ENCODER_CANDIDATES |
| if has_pkg("sentence_transformers"): |
| from sentence_transformers import CrossEncoder |
| for candidate in candidates: |
| try: |
| self.model = CrossEncoder(candidate, device=DEVICE) |
| self.model_name = candidate |
| logger.info("CrossEncoderReranker loaded: %s", candidate) |
| break |
| except Exception as exc: |
| logger.warning("Could not load CrossEncoder '%s' (%s); trying next candidate.", |
| candidate, exc) |
| if self.model is None: |
| logger.warning("No cross-encoder available; reranking disabled (hybrid-fusion order used as-is).") |
|
|
| def rerank(self, query: str, candidates: List[Dict], top_k: int) -> List[Dict]: |
| """ |
| Rerank `candidates` (each with a "text" field) by cross-encoder |
| relevance to `query`, returning the top_k. |
| |
| SAFETY NET (bugfix): confirmed on a real run that the cross-encoder can |
| take the #1 first-stage result -- the strongest lexical+dense signal |
| available -- and push it completely out of the top_k. That's a much |
| more aggressive failure than simple reordering: it means the single |
| best pre-rerank candidate can vanish from the final answer entirely. |
| This is more often the cross-encoder misjudging a short/paraphrased |
| query than a genuine signal that the top first-stage result is |
| irrelevant, so the pre-rerank #1 is now guaranteed a slot in the final |
| top_k (replacing the weakest reranked slot if it would otherwise be |
| dropped) -- the reranker still freely reorders everything else. |
| |
| Args: |
| query (str): The search query. |
| candidates (List[Dict]): Candidate chunks from first-stage hybrid |
| retrieval, already sorted by fused hybrid score (candidates[0] |
| is the strongest first-stage result). |
| top_k (int): Number of results to keep after reranking. |
| |
| Returns: |
| List[Dict]: Reranked candidates, each with an added "rerank_score" field. |
| """ |
| if self.model is None or not candidates: |
| return candidates[:top_k] |
| original_top1 = candidates[0] |
| pairs = [(query, c["text"]) for c in candidates] |
| scores = self.model.predict(pairs) |
| for c, s in zip(candidates, scores): |
| c["rerank_score"] = float(s) |
| candidates.sort(key=lambda c: c["rerank_score"], reverse=True) |
| reranked = candidates[:top_k] |
|
|
| if top_k > 0 and not any(c["chunk_id"] == original_top1["chunk_id"] for c in reranked): |
| reranked[-1] = original_top1 |
|
|
| return reranked |
|
|
|
|
|
|
| class HybridRetriever: |
| """ |
| Combines BM25 (sparse) and dense vector search into a candidate pool via |
| weighted, min-max-normalized score fusion, then applies a cross-encoder |
| reranker as a second stage to sharpen precision on the final top_k. |
| """ |
|
|
| def __init__(self, bm25: BM25RetrieverWrapper, vector_index: VectorIndex, |
| embedding_model: EmbeddingModel, reranker: CrossEncoderReranker = None, alpha: float = 0.6): |
| """ |
| Args: |
| alpha (float): Weight given to the dense score; (1 - alpha) goes to BM25. |
| Raised from 0.5 -> 0.6 by default now that Chapter 5 uses a stronger, |
| biomedical-aware embedding model -- dense scores are more trustworthy |
| than before, so they should dominate the fusion slightly more. Tune |
| this directly if you evaluate and find a different split works better. |
| reranker (CrossEncoderReranker): Optional second-stage reranker. |
| """ |
| self.bm25 = bm25 |
| self.vector_index = vector_index |
| self.embedding_model = embedding_model |
| self.reranker = reranker |
| self.alpha = alpha |
| self.use_query_expansion = True |
|
|
| def retrieve(self, query: str, top_k: int = 10, rerank_pool_size: int = 30, |
| metadata_filter: Dict = None) -> List[Dict]: |
| """ |
| Retrieve the top_k chunks for `query`: fuse BM25 + dense scores into a |
| wider candidate pool (`rerank_pool_size`), then rerank that pool with |
| the cross-encoder before truncating to top_k. Optional metadata filtering |
| is applied before reranking. |
| |
| Args: |
| query (str): Natural language query. |
| top_k (int): Number of results to return. |
| rerank_pool_size (int): Size of the first-stage candidate pool handed |
| to the cross-encoder (must be >= top_k; wider pool = better |
| recall for the reranker to work with, at added latency cost). |
| metadata_filter (Dict): Exact-match filter applied to chunk metadata. |
| |
| Returns: |
| List[Dict]: [{"chunk_id", "text", "score", "bm25_score", "dense_score", |
| "rerank_score", ...metadata}] |
| """ |
| pool_size = max(rerank_pool_size, top_k) |
| candidate_pool = min(pool_size, len(self.bm25.chunk_ids)) or top_k |
| bm25_query = expand_query(query) if getattr(self, "use_query_expansion", False) else query |
| bm25_results = self.bm25.search(bm25_query, top_k=candidate_pool) |
| query_embedding = self.embedding_model.encode([query])[0] |
| dense_results = self.vector_index.search(query_embedding, top_k=candidate_pool) |
|
|
| bm25_norm = _minmax_normalize(bm25_results) |
| dense_norm = _minmax_normalize(dense_results) |
| all_ids = set(bm25_norm) | set(dense_norm) |
|
|
| fused = [] |
| for chunk_id in all_ids: |
| b_score, d_score = bm25_norm.get(chunk_id, 0.0), dense_norm.get(chunk_id, 0.0) |
| hybrid_score = self.alpha * d_score + (1 - self.alpha) * b_score |
| chunk = CHUNK_LOOKUP.get(chunk_id) |
| if chunk is None: |
| continue |
| if metadata_filter and any(chunk.get(k) != v for k, v in metadata_filter.items()): |
| continue |
| fused.append({**chunk, "score": hybrid_score, "bm25_score": b_score, "dense_score": d_score}) |
|
|
| fused.sort(key=lambda r: r["score"], reverse=True) |
| fused = _dedup_candidates(fused) |
|
|
| if self.reranker is not None: |
| return self.reranker.rerank(query, fused, top_k) |
| return fused[:top_k] |
|
|
|
|
|
|
|
|
| |
| |
| |
| _EXTERNAL_CUES = re.compile(r"\b(latest|recent|new|current|guideline|clinical trial|study|research|" |
| r"evidence|meta-analysis|systematic review)\b", re.IGNORECASE) |
| |
| |
| |
| |
| |
| |
| _REASONING_CUES = re.compile(r"\b(why|how|link|links|connection|relationship|comorbid|comorbidity|bridge|" |
| r"both anxiety and depression|underlying mechanism|safe|safely|combine|" |
| r"combination|interact|interaction|contraindicat|together|mix)\b", re.IGNORECASE) |
|
|
|
|
| |
| |
| |
| LOCAL_MISS_SCORE_THRESHOLD = 0.35 |
|
|
|
|
| def _llm_assisted_symptom_detection(query: str) -> Dict[str, bool]: |
| """ |
| Fallback symptom-domain detector for query understanding: exact-phrase |
| matching (Chapter 3's extract_entities) misses plain-language symptom |
| descriptions -- e.g. "I constantly worry about everything" never matches |
| the literal curated phrase "excessive worry". When an LLM is loaded, ask |
| it directly whether the query describes anxiety-domain and/or |
| depression-domain symptoms in the patient's own words. |
| |
| CONFIRMED NECESSARY on a real query: "I constantly worry about |
| everything, and lately I also feel hopeless and lose interest in |
| activities I used to enjoy" matched ZERO curated symptom phrases via |
| exact matching, despite clearly describing both GAD (worry) and MDD |
| (anhedonia/hopelessness) symptomatology -- meaning graph-based reasoning |
| never fired for exactly the kind of query this system exists to handle. |
| |
| Args: |
| query (str): The raw user query. |
| |
| Returns: |
| Dict[str, bool]: {"anxiety_domain": bool, "depression_domain": bool}. |
| Both False if no LLM is loaded yet (safe to call from any chapter, |
| including before Chapter 11 has run) or if the call fails -- this |
| must never crash routing. |
| """ |
| generator = globals().get("ANSWER_GENERATOR") |
| if generator is None or generator.backend != "transformers": |
| return {"anxiety_domain": False, "depression_domain": False} |
|
|
| prompt = ( |
| "A patient wrote the following message. Respond with ONLY a JSON object like " |
| '{"anxiety_domain": true, "depression_domain": false}.\n' |
| "\"anxiety_domain\" = true if the message describes anxiety/excessive-worry symptoms, " |
| "even in plain, non-clinical words.\n" |
| "\"depression_domain\" = true if the message describes depression/low-mood/anhedonia " |
| "symptoms, even in plain, non-clinical words.\n\n" |
| f"Message: {query}\nJSON:" |
| ) |
| try: |
| if generator.task == "text-generation": |
| messages = [{"role": "user", "content": prompt}] |
| if generator.tokenizer is not None and getattr(generator.tokenizer, "chat_template", None): |
| prompt_text = generator.tokenizer.apply_chat_template(messages, tokenize=False, |
| add_generation_prompt=True) |
| else: |
| prompt_text = prompt |
| result = generator.pipeline(prompt_text, max_new_tokens=40, do_sample=False, return_full_text=False) |
| else: |
| result = generator.pipeline(prompt, max_new_tokens=40, do_sample=False) |
| text = result[0]["generated_text"] |
| match = re.search(r"\{.*\}", text, re.DOTALL) |
| if match: |
| parsed = json.loads(match.group(0)) |
| return {"anxiety_domain": bool(parsed.get("anxiety_domain", False)), |
| "depression_domain": bool(parsed.get("depression_domain", False))} |
| except Exception as exc: |
| logger.debug("LLM-assisted symptom detection failed (%s); falling back to keyword-only signal.", exc) |
| return {"anxiety_domain": False, "depression_domain": False} |
|
|
|
|
| class QueryRouter: |
| """ |
| Adaptive router: analyzes an incoming natural-language clinical question and |
| decides which retrieval subsystems to invoke. Combines query-level cues |
| (keywords, entity types) with an empirical local-retrieval-confidence check |
| (`is_local_miss`), applied once actual local results are available in the |
| pipeline (Chapter 14), so the external fallback is evidence-based rather |
| than guessed purely from query phrasing. |
| """ |
|
|
| def __init__(self, kg: ClinicalKnowledgeGraph, local_corpus_size: int): |
| self.kg = kg |
| self.local_corpus_size = local_corpus_size |
| self.kg_canonical_names = {d.get("canonical_name", "").lower() for _, d in kg.graph.nodes(data=True)} |
|
|
| @staticmethod |
| def is_local_miss(local_results: List[Dict], threshold: float = LOCAL_MISS_SCORE_THRESHOLD) -> bool: |
| """ |
| Determine whether local retrieval "missed" -- i.e. found nothing, or its |
| best result scored below `threshold` -- meaning the pipeline should |
| fall back to external sources regardless of what the query text alone |
| suggested. |
| |
| Args: |
| local_results (List[Dict]): Output of `HybridRetriever.retrieve`. |
| threshold (float): Minimum acceptable top-1 score to trust local results. |
| |
| Returns: |
| bool: True if local retrieval should be considered insufficient. |
| """ |
| if not local_results: |
| return True |
| top_score = local_results[0].get("rerank_score", local_results[0].get("score", 0.0)) |
| return top_score < threshold |
|
|
| def analyze(self, query: str) -> Dict: |
| """ |
| Produce a routing decision for `query`. |
| |
| Returns: |
| Dict: { |
| "entities": List[Dict], "bridge_symptom_hit": bool, |
| "external_signal": bool, "reasoning_signal": bool, |
| "local_kb_hit_rate": float, "modes": List[str] # ordered, e.g. ["local", "graph"] |
| } |
| """ |
| entities = extract_entities(query) |
| bridge_hit = any(e["label"] == "BridgeSymptom" for e in entities) |
| external_signal = bool(_EXTERNAL_CUES.search(query)) |
|
|
| |
| |
| |
| |
| |
| entity_texts = {e["text"] for e in entities} |
| anxiety_signal = bool(entity_texts & ANXIETY_SYMPTOMS) |
| depression_signal = bool(entity_texts & DEPRESSION_SYMPTOMS) |
| if not (anxiety_signal and depression_signal): |
| llm_signal = _llm_assisted_symptom_detection(query) |
| anxiety_signal = anxiety_signal or llm_signal["anxiety_domain"] |
| depression_signal = depression_signal or llm_signal["depression_domain"] |
| comorbidity_signal = anxiety_signal and depression_signal |
|
|
| reasoning_signal = bool(_REASONING_CUES.search(query)) or bridge_hit or comorbidity_signal |
|
|
| |
| |
| |
| |
| |
| kg_hits = sum(1 for e in entities if e["text"] in self.kg_canonical_names) |
| local_hit_rate = (kg_hits / len(entities)) if entities else 0.0 |
|
|
| modes = [] |
| if self.local_corpus_size > 0: |
| modes.append("local") |
| if reasoning_signal: |
| modes.append("graph") |
| if external_signal or not entities: |
| modes.append("external") |
| if not modes: |
| modes = ["local", "external"] |
|
|
| decision = {"entities": entities, "bridge_symptom_hit": bridge_hit, |
| "external_signal": external_signal, "reasoning_signal": reasoning_signal, |
| "comorbidity_signal": comorbidity_signal, |
| "local_kb_hit_rate": round(local_hit_rate, 3), "modes": modes} |
| logger.info("Router decision for query %r: %s", query, decision["modes"]) |
| return decision |
|
|
|
|
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| import requests |
|
|
|
|
| def _external_cache_path(source: str, key: str) -> str: |
| """Build a deterministic cache file path for an external query result.""" |
| digest = hashlib.md5(f"{source}:{key}".encode("utf-8")).hexdigest() |
| return os.path.join(DIRS["data_external"], f"{source}_{digest}.json") |
|
|
|
|
| def load_cached_external(source: str, key: str): |
| """Return a cached external-source result if present, else None.""" |
| path = _external_cache_path(source, key) |
| if os.path.exists(path): |
| with open(path, "r", encoding="utf-8") as f: |
| return json.load(f) |
| return None |
|
|
|
|
| def save_cached_external(source: str, key: str, results: List[Dict]) -> None: |
| """Persist an external-source result to the cache.""" |
| os.makedirs(DIRS["data_external"], exist_ok=True) |
| with open(_external_cache_path(source, key), "w", encoding="utf-8") as f: |
| json.dump(results, f) |
|
|
|
|
| |
| |
| |
| ENTREZ_EMAIL = "jubayerhr@gmail.com" |
| WHO_FACT_SHEET_URLS = { |
| "depression": "https://www.who.int/news-room/fact-sheets/detail/depression", |
| "anxiety": "https://www.who.int/news-room/fact-sheets/detail/anxiety-disorders", |
| "mental health": "https://www.who.int/news-room/fact-sheets/detail/mental-health-strengthening-our-response", |
| } |
|
|
|
|
| def pubmed_search(query: str, max_results: int = 5) -> List[Dict]: |
| """ |
| Query PubMed via NCBI Entrez esearch + efetch for abstracts relevant to `query`. |
| Results are cached (Chapter 2's cache contract) to avoid repeat network calls. |
| |
| Args: |
| query (str): Search query (typically the user's clinical question or key entities). |
| max_results (int): Maximum number of PubMed records to fetch. |
| |
| Returns: |
| List[Dict]: [{"id", "title", "text", "source_type": "pubmed", "url"}, ...]. |
| Returns [] gracefully if Biopython is unavailable or the network call fails. |
| """ |
| cached = load_cached_external("pubmed", f"{query}:{max_results}") |
| if cached is not None: |
| return cached |
| if not has_pkg("Bio"): |
| logger.warning("Biopython not available; skipping PubMed retrieval.") |
| return [] |
| try: |
| from Bio import Entrez |
| Entrez.email = ENTREZ_EMAIL |
| handle = Entrez.esearch(db="pubmed", term=query, retmax=max_results) |
| record = Entrez.read(handle); handle.close() |
| pmids = record.get("IdList", []) |
| if not pmids: |
| save_cached_external("pubmed", f"{query}:{max_results}", []) |
| return [] |
| fetch_handle = Entrez.efetch(db="pubmed", id=pmids, rettype="abstract", retmode="xml") |
| records = Entrez.read(fetch_handle); fetch_handle.close() |
|
|
| results = [] |
| for article in records.get("PubmedArticle", []): |
| try: |
| medline = article["MedlineCitation"] |
| pmid = str(medline["PMID"]) |
| article_data = medline["Article"] |
| title = str(article_data.get("ArticleTitle", "")) |
| abstract_parts = article_data.get("Abstract", {}).get("AbstractText", []) |
| abstract = " ".join(str(p) for p in abstract_parts) |
| results.append({"id": pmid, "title": title, "text": abstract or title, |
| "source_type": "pubmed", "url": f"https://pubmed.ncbi.nlm.nih.gov/{pmid}/"}) |
| except Exception as inner_exc: |
| logger.debug("Skipping malformed PubMed record: %s", inner_exc) |
| save_cached_external("pubmed", f"{query}:{max_results}", results) |
| logger.info("PubMed retrieval: %d result(s) for query %r", len(results), query) |
| return results |
| except Exception as exc: |
| logger.warning("PubMed retrieval failed (network/Entrez issue): %s", exc) |
| return [] |
|
|
|
|
| def who_search(query: str, max_results: int = 3) -> List[Dict]: |
| """ |
| Fetch relevant WHO fact-sheet content by matching `query` against a curated |
| map of known WHO mental-health fact-sheet URLs, then extracting page text. |
| |
| Args: |
| query (str): Search query; matched against curated WHO topic keys. |
| max_results (int): Maximum number of fact sheets to return. |
| |
| Returns: |
| List[Dict]: [{"id", "title", "text", "source_type": "who", "url"}, ...]. |
| Returns [] gracefully on any network/parsing failure. |
| """ |
| cached = load_cached_external("who", f"{query}:{max_results}") |
| if cached is not None: |
| return cached |
| matched_topics = [t for t in WHO_FACT_SHEET_URLS if t in query.lower()] or list(WHO_FACT_SHEET_URLS)[:1] |
| results = [] |
| for topic in matched_topics[:max_results]: |
| url = WHO_FACT_SHEET_URLS[topic] |
| try: |
| response = requests.get(url, timeout=10, headers={"User-Agent": "Mozilla/5.0 (research-bot)"}) |
| response.raise_for_status() |
| text = response.text |
| |
| |
| |
| |
| |
| try: |
| from bs4 import BeautifulSoup |
| soup = BeautifulSoup(response.content, "html.parser") |
| text = " ".join(p.get_text(" ", strip=True) for p in soup.find_all("p")) |
| except ImportError: |
| text = re.sub(r"<[^>]+>", " ", text) |
| results.append({"id": f"who_{topic}", "title": f"WHO Fact Sheet: {topic.title()}", |
| "text": clean_text(text)[:5000], "source_type": "who", "url": url}) |
| except Exception as exc: |
| logger.warning("WHO fact-sheet fetch failed for '%s': %s", topic, exc) |
| save_cached_external("who", f"{query}:{max_results}", results) |
| return results |
|
|
|
|
| def wikipedia_medical_search(query: str, max_results: int = 2) -> List[Dict]: |
| """ |
| Fetch summaries from Wikipedia's REST API, restricted to pages whose |
| categories indicate medical/mental-health content (a lightweight content |
| filter, not a guarantee of clinical accuracy -- Wikipedia is treated as a |
| lower-confidence source in Chapter 9's evidence ranking). |
| |
| Args: |
| query (str): Search query. |
| max_results (int): Maximum number of pages to return. |
| |
| Returns: |
| List[Dict]: [{"id", "title", "text", "source_type": "wikipedia", "url"}, ...]. |
| """ |
| cached = load_cached_external("wikipedia", f"{query}:{max_results}") |
| if cached is not None: |
| return cached |
| try: |
| search_resp = requests.get("https://en.wikipedia.org/w/api.php", |
| params={"action": "query", "list": "search", "srsearch": query, |
| "format": "json", "srlimit": max_results}, timeout=10) |
| search_resp.raise_for_status() |
| hits = search_resp.json().get("query", {}).get("search", []) |
|
|
| results = [] |
| for hit in hits: |
| title = hit["title"] |
| summary_resp = requests.get(f"https://en.wikipedia.org/api/rest_v1/page/summary/{requests.utils.quote(title)}", |
| timeout=10) |
| if summary_resp.status_code != 200: |
| continue |
| data = summary_resp.json() |
| extract = data.get("extract", "") |
| is_medical = any(kw in (extract + title).lower() for kw in |
| ("disorder", "symptom", "treatment", "therapy", "mental health", |
| "depression", "anxiety", "psychiatric", "clinical")) |
| if is_medical and extract: |
| results.append({"id": f"wiki_{title}", "title": title, "text": extract, |
| "source_type": "wikipedia", |
| "url": data.get("content_urls", {}).get("desktop", {}).get("page", "")}) |
| save_cached_external("wikipedia", f"{query}:{max_results}", results) |
| return results |
| except Exception as exc: |
| logger.warning("Wikipedia retrieval failed: %s", exc) |
| return [] |
|
|
|
|
| def retrieve_external_evidence(query: str, per_source_limit: int = 3) -> List[Dict]: |
| """ |
| Orchestrate all external sources, then deduplicate (by title similarity) and |
| rank by embedding-cosine relevance to the query. |
| |
| Args: |
| query (str): The user's clinical question. |
| per_source_limit (int): Max results requested per external source. |
| |
| Returns: |
| List[Dict]: Deduplicated, ranked external evidence items. |
| """ |
| import difflib |
| raw = (pubmed_search(query, per_source_limit) + who_search(query, per_source_limit) |
| + wikipedia_medical_search(query, per_source_limit)) |
| if not raw: |
| return [] |
|
|
| deduped, seen_titles = [], [] |
| for item in raw: |
| if any(difflib.SequenceMatcher(None, item["title"].lower(), t).ratio() > 0.85 for t in seen_titles): |
| continue |
| seen_titles.append(item["title"].lower()) |
| deduped.append(item) |
|
|
| query_emb = EMBEDDING_MODEL.encode([query])[0] |
| item_embs = EMBEDDING_MODEL.encode([d["text"][:1000] for d in deduped]) if deduped else np.zeros((0, 1)) |
| for item, emb in zip(deduped, item_embs): |
| item["relevance_score"] = float(np.dot(query_emb, emb)) |
| deduped.sort(key=lambda d: d["relevance_score"], reverse=True) |
| return deduped |
|
|
|
|
| print("Chapter 8 external retrieval functions defined: pubmed_search, who_search, " |
| "wikipedia_medical_search, retrieve_external_evidence.") |
| print("NOTE: these require internet access enabled in the Kaggle kernel settings to return results.") |
|
|
|
|
| |
| |
| |
| SOURCE_RELIABILITY_WEIGHT = {"guideline": 1.0, "pubmed": 0.95, "who": 0.9, |
| "local_kb": 0.8, "graph": 0.75, "wikipedia": 0.5} |
| _NEGATION_CUES = re.compile(r"\b(no|not|without|does not|doesn't|isn't|is not|lack of)\b", re.IGNORECASE) |
|
|
|
|
| @dataclasses.dataclass |
| class EvidenceItem: |
| """A single, unified piece of evidence regardless of originating subsystem.""" |
| evidence_id: str |
| text: str |
| source_type: str |
| source_ref: str |
| retrieval_score: float |
| reliability_weight: float |
| confidence: float = 0.0 |
| evidence_level: str = "Not Classified" |
| reliability_score: float = 0.0 |
| source_doc_evidence_level: str = "" |
|
|
|
|
| def merge_evidence(local_results: List[Dict], external_results: List[Dict], |
| graph_reasoning_text: str = "") -> List[EvidenceItem]: |
| """ |
| Merge local hybrid-retrieval results, external retrieval results, and a |
| graph-reasoning narrative into one unified list of EvidenceItem objects. |
| |
| Args: |
| local_results (List[Dict]): Output of `HybridRetriever.retrieve`. |
| external_results (List[Dict]): Output of `retrieve_external_evidence`. |
| graph_reasoning_text (str): Human-readable reasoning-chain narrative |
| from Chapter 10 (treated as a single evidence item if non-empty). |
| |
| Returns: |
| List[EvidenceItem]: Unmerged (deduplication happens in `deduplicate_evidence`). |
| """ |
| items = [] |
| for r in local_results: |
| items.append(EvidenceItem(evidence_id=r["chunk_id"], text=r["text"], source_type="local_kb", |
| source_ref=r["source_file"], retrieval_score=r["score"], |
| reliability_weight=SOURCE_RELIABILITY_WEIGHT["local_kb"], |
| source_doc_evidence_level=r.get("doc_evidence_level", ""))) |
| for r in external_results: |
| items.append(EvidenceItem(evidence_id=r["id"], text=r["text"], source_type=r["source_type"], |
| source_ref=r.get("url", ""), retrieval_score=r.get("relevance_score", 0.5), |
| reliability_weight=SOURCE_RELIABILITY_WEIGHT.get(r["source_type"], 0.5))) |
| if graph_reasoning_text: |
| items.append(EvidenceItem(evidence_id="graph_reasoning", text=graph_reasoning_text, source_type="graph", |
| source_ref="clinical_knowledge_graph", retrieval_score=0.8, |
| reliability_weight=SOURCE_RELIABILITY_WEIGHT["graph"])) |
| return items |
|
|
|
|
| def deduplicate_evidence(items: List[EvidenceItem], similarity_threshold: float = 0.9) -> List[EvidenceItem]: |
| """Remove near-duplicate evidence items using normalized text similarity.""" |
| import difflib |
| unique, seen_texts = [], [] |
| for item in items: |
| norm = re.sub(r"\s+", " ", item.text.strip().lower()) |
| if any(difflib.SequenceMatcher(None, norm, s).ratio() > similarity_threshold for s in seen_texts): |
| continue |
| seen_texts.append(norm) |
| unique.append(item) |
| return unique |
|
|
|
|
| _NEGATION_PROXIMITY_WINDOW = 6 |
|
|
|
|
| def _is_entity_negated(text: str, entity: str) -> bool: |
| """ |
| Check whether `entity` has a negation cue WITHIN a small word window on |
| EITHER side of it in `text` -- e.g. both "not reduce fatigue" and |
| "fatigue was not reduced" should count as negating "fatigue"; English |
| negation can precede or follow the entity depending on sentence structure, |
| so both directions must be checked (an earlier version of this function |
| only checked preceding words and missed cases like "SSRIs do not |
| increase risk", where the negation follows the entity). |
| |
| Args: |
| text (str): The evidence text to search. |
| entity (str): The shared entity string to check for negation around. |
| |
| Returns: |
| bool: True if a negation cue appears within `_NEGATION_PROXIMITY_WINDOW` |
| words before or after this specific entity mention. |
| """ |
| text_lower, entity_lower = text.lower(), entity.lower() |
| idx = text_lower.find(entity_lower) |
| if idx == -1: |
| return False |
| before_words = text_lower[:idx].split()[-_NEGATION_PROXIMITY_WINDOW:] |
| after_words = text_lower[idx + len(entity_lower):].split()[:_NEGATION_PROXIMITY_WINDOW] |
| return bool(_NEGATION_CUES.search(" ".join(before_words + after_words))) |
|
|
|
|
| def detect_contradictions(items: List[EvidenceItem]) -> List[Tuple[str, str]]: |
| """ |
| Flag pairs of evidence items that appear to contradict each other on a |
| SPECIFIC shared entity: both mention the same entity, but one negates it |
| nearby (e.g. "does not increase risk") and the other doesn't. |
| |
| BUGFIX: the previous version checked "does either text contain ANY |
| negation word anywhere" -- with evidence passages up to 500 characters, |
| at least one of two unrelated passages contains a stray "not"/"no" almost |
| by chance, so ~50% of evidence pairs were being flagged as "conflicting" |
| even when they had nothing to do with each other. Confirmed on a real run: |
| a coherent, well-sourced answer was marked "Confidence Level: Low" purely |
| because of this false-positive rate. Now the negation must be near the |
| SAME shared entity in both texts to count as a real conflict. |
| |
| Returns: |
| List[Tuple[str, str]]: Pairs of evidence_id that appear genuinely contradictory. |
| """ |
| contradictions = [] |
| for i in range(len(items)): |
| for j in range(i + 1, len(items)): |
| a, b = items[i], items[j] |
| a_entities = {e["text"] for e in extract_entities(a.text)} |
| b_entities = {e["text"] for e in extract_entities(b.text)} |
| for entity in a_entities & b_entities: |
| if _is_entity_negated(a.text, entity) != _is_entity_negated(b.text, entity): |
| contradictions.append((a.evidence_id, b.evidence_id)) |
| break |
| return contradictions |
|
|
|
|
| def compute_confidence(items: List[EvidenceItem]) -> None: |
| """Assign a final confidence score per item: retrieval_score * reliability_weight. In-place.""" |
| for item in items: |
| item.confidence = round(item.retrieval_score * item.reliability_weight, 4) |
|
|
|
|
| def rank_evidence(items: List[EvidenceItem]) -> List[EvidenceItem]: |
| """Sort evidence items by descending confidence.""" |
| return sorted(items, key=lambda i: i.confidence, reverse=True) |
|
|
|
|
| |
| |
| def build_clinical_context(items: List[EvidenceItem], max_items: int = 10, |
| query: str = "") -> Tuple[str, List[Dict]]: |
| """ |
| Build the unified clinical context string (with numbered citation markers) |
| that Chapter 11's generator consumes, plus a structured citation list. |
| |
| Args: |
| items (List[EvidenceItem]): Ranked evidence items. |
| max_items (int): Maximum number of items to include in the context. |
| |
| Returns: |
| Tuple[str, List[Dict]]: (context_string, citation_list) |
| """ |
| top_items = items[:max_items] |
| context_lines, citations = [], [] |
| for idx, item in enumerate(top_items, start=1): |
| |
| |
| snippet = compress_context(query, item.text) if query else item.text[:500] |
| context_lines.append(f"[{idx}] ({item.source_type}, confidence={item.confidence:.2f}) {snippet}") |
| reliability = item.reliability_score if item.reliability_score > 0 else item.confidence |
| citations.append({"marker": idx, "source_type": item.source_type, "source_ref": item.source_ref, |
| "confidence": item.confidence, "evidence_level": item.evidence_level, |
| "reliability_score": round(reliability, 4)}) |
| return "\n\n".join(context_lines), citations |
|
|
|
|
| print("Chapter 9 evidence fusion pipeline defined: merge_evidence -> deduplicate_evidence -> " |
| "detect_contradictions -> compute_confidence -> rank_evidence -> build_clinical_context.") |
|
|
|
|
| |
| |
| |
| |
| |
| |
|
|
|
|
| def classify_evidence_level(item: "EvidenceItem") -> str: |
| """ |
| Classify an evidence item's position in the evidence hierarchy (Systematic |
| Review > Meta-Analysis > RCT > Cohort > Case-Control > Case Report > |
| Guideline). For local-KB items, PREFERS the document-level classification |
| computed in Chapter 2 from the full source paper (`source_doc_evidence_level`) |
| over re-guessing from the small retrieved snippet -- a snippet rarely |
| repeats a phrase like "randomized controlled trial" even when the source |
| document genuinely is one, so snippet-only classification was |
| systematically under-rating good evidence. External sources (PubMed |
| abstracts, WHO fact sheets) don't have this issue since their retrieved |
| text already IS the whole document, so they still classify from `item.text` |
| directly, with source-type-based defaults as a last resort. |
| |
| Args: |
| item (EvidenceItem): The evidence item to classify. |
| |
| Returns: |
| str: One of `ClinicalKnowledgeGraph.EVIDENCE_LEVEL_WEIGHTS` keys. |
| """ |
| if item.source_doc_evidence_level and item.source_doc_evidence_level != "Not Classified": |
| return item.source_doc_evidence_level |
| snippet_level = classify_text_evidence_level(item.text) |
| if snippet_level != "Not Classified": |
| return snippet_level |
| if item.source_type in ("who", "graph"): |
| return "Guideline" |
| if item.source_type == "wikipedia": |
| return "Expert Opinion" |
| return "Not Classified" |
|
|
|
|
| def estimate_claim_support(items: List["EvidenceItem"], corroboration_threshold: float = 0.45) -> Dict[str, int]: |
| """ |
| Estimate how well each evidence item's claim is corroborated by the REST of |
| the evidence pool, via pairwise embedding similarity -- a claim repeated |
| (in substance) across multiple independent sources is more trustworthy than |
| one appearing in only a single source. |
| |
| Args: |
| items (List[EvidenceItem]): Ranked, deduplicated evidence. |
| corroboration_threshold (float): Minimum cosine similarity to count as corroboration. |
| |
| Returns: |
| Dict[str, int]: evidence_id -> number of OTHER items corroborating it. |
| """ |
| if len(items) < 2: |
| return {item.evidence_id: 0 for item in items} |
| embeddings = EMBEDDING_MODEL.encode([item.text[:500] for item in items]) |
| support_counts = {} |
| for i, item in enumerate(items): |
| sims = embeddings @ embeddings[i] |
| support_counts[item.evidence_id] = int(np.sum(sims > corroboration_threshold) - 1) |
| return support_counts |
|
|
|
|
| |
| |
| |
| def run_clinical_evidence_validation(items: List["EvidenceItem"], |
| contradictions: List[Tuple[str, str]]) -> Dict: |
| """ |
| Full Clinical Evidence Validator pipeline (Task 7): classify evidence level, |
| estimate corroboration, prioritize guideline-backed evidence, compute a |
| combined reliability score per item, and assemble a conflict report. |
| |
| Args: |
| items (List[EvidenceItem]): Ranked, deduplicated evidence (Chapter 9 output). |
| contradictions (List[Tuple[str, str]]): Contradictory evidence-id pairs |
| (Chapter 9's `detect_contradictions` output). |
| |
| Returns: |
| Dict: { |
| "validated_evidence": List[EvidenceItem] (sorted by reliability_score, mutated in place), |
| "evidence_level_summary": Dict[str, int], |
| "support_counts": Dict[str, int], |
| "conflict_report": List[Dict], |
| "mean_reliability_score": float, |
| } |
| """ |
| support_counts = estimate_claim_support(items) |
|
|
| for item in items: |
| item.evidence_level = classify_evidence_level(item) |
| level_weight = ClinicalKnowledgeGraph.EVIDENCE_LEVEL_WEIGHTS.get(item.evidence_level, 0.5) |
| corroboration_bonus = min(support_counts.get(item.evidence_id, 0) * 0.05, 0.15) |
| item.reliability_score = round(min(item.confidence * level_weight + corroboration_bonus, 1.0), 4) |
|
|
| validated = sorted(items, key=lambda i: i.reliability_score, reverse=True) |
|
|
| from collections import Counter |
| evidence_level_summary = dict(Counter(item.evidence_level for item in items)) |
|
|
| item_lookup = {item.evidence_id: item for item in items} |
| conflict_report = [ |
| {"item_a": item_lookup[a].text[:150], "item_b": item_lookup[b].text[:150], |
| "source_a": item_lookup[a].source_type, "source_b": item_lookup[b].source_type} |
| for a, b in contradictions if a in item_lookup and b in item_lookup |
| ] |
|
|
| mean_reliability = float(np.mean([i.reliability_score for i in items])) if items else 0.0 |
|
|
| logger.info("Clinical Evidence Validator: %d item(s) validated, %d conflict(s) found, " |
| "mean reliability=%.3f", len(items), len(conflict_report), mean_reliability) |
|
|
| return {"validated_evidence": validated, "evidence_level_summary": evidence_level_summary, |
| "support_counts": support_counts, "conflict_report": conflict_report, |
| "mean_reliability_score": round(mean_reliability, 4)} |
|
|
|
|
| print("Chapter 9B Clinical Evidence Validator defined: classify_evidence_level -> " |
| "estimate_claim_support -> run_clinical_evidence_validation.") |
|
|
|
|
|
|
| def assess_evidence_sufficiency(validated_evidence: List["EvidenceItem"], reasoning_text: str = "", |
| min_items: int = 1, min_reliability: float = 0.15) -> Dict: |
| """Improvement #8: decide, BEFORE generation, whether there is enough trustworthy evidence |
| to justify a synthesized clinical answer. Returns a decision dict; the pipeline uses it to |
| withhold a confident answer (and say so) when nothing usable was retrieved -- so the model |
| is never asked to synthesize a clinical claim from an empty or near-empty evidence base. |
| |
| Returns: |
| Dict: {"sufficient": bool, "reason": str, "n_items": int, "max_reliability": float} |
| """ |
| n = len(validated_evidence) |
| max_rel = max((e.reliability_score if e.reliability_score > 0 else e.confidence |
| for e in validated_evidence), default=0.0) |
| if reasoning_text: |
| return {"sufficient": True, "reason": "graph reasoning path available", |
| "n_items": n, "max_reliability": round(max_rel, 4)} |
| if n < min_items or max_rel < min_reliability: |
| return {"sufficient": False, |
| "reason": f"only {n} evidence item(s), max reliability {max_rel:.2f} < {min_reliability}", |
| "n_items": n, "max_reliability": round(max_rel, 4)} |
| return {"sufficient": True, "reason": "evidence threshold met", |
| "n_items": n, "max_reliability": round(max_rel, 4)} |
|
|
|
|
|
|
| |
| |
| |
| def _confidence_to_weight(confidence: float) -> float: |
| """Convert an edge confidence in (0,1] to a positive path-length weight (lower = stronger evidence).""" |
| return -np.log(max(confidence, 1e-6)) |
|
|
|
|
| def find_shortest_path(kg: ClinicalKnowledgeGraph, source: str, target: str) -> List[str]: |
| """Find the topologically shortest (fewest-hop) path between two node IDs.""" |
| try: |
| return nx.shortest_path(kg.graph, source, target) |
| except (nx.NetworkXNoPath, nx.NodeNotFound) as exc: |
| logger.debug("No shortest path %s -> %s: %s", source, target, exc) |
| return [] |
|
|
|
|
| def find_highest_confidence_path(kg: ClinicalKnowledgeGraph, source: str, target: str) -> List[str]: |
| """ |
| Find the path maximizing the product of edge confidences, by minimizing the |
| sum of negative-log confidences (Dijkstra over a transformed weight). |
| """ |
| G = kg.graph |
| weighted = nx.DiGraph() |
| for u, v, data in G.edges(data=True): |
| w = _confidence_to_weight(data.get("confidence", 0.5)) |
| if not weighted.has_edge(u, v) or weighted[u][v]["weight"] > w: |
| weighted.add_edge(u, v, weight=w) |
| try: |
| return nx.shortest_path(weighted, source, target, weight="weight") |
| except (nx.NetworkXNoPath, nx.NodeNotFound) as exc: |
| logger.debug("No highest-confidence path %s -> %s: %s", source, target, exc) |
| return [] |
|
|
|
|
| def find_bridge_symptom_chains(kg: ClinicalKnowledgeGraph, disease_a: str, disease_b: str, |
| max_depth: int = 4, max_chains: int = 8) -> List[List[str]]: |
| """ |
| Enumerate reasoning chains connecting `disease_a` and `disease_b` that pass |
| through at least one BridgeSymptom node -- the central explanatory mechanism |
| of this thesis. Implements **adaptive hop selection**: starts at the |
| shortest plausible depth (2 hops) and widens the search only if nothing is |
| found, rather than always searching to the maximum depth. |
| |
| Args: |
| disease_a, disease_b (str): Disease node IDs, e.g. "disease:gad", "disease:mdd". |
| max_depth (int): Upper bound on path length (hops) to search. |
| max_chains (int): Maximum number of candidate chains to return for ranking. |
| |
| Returns: |
| List[List[str]]: Candidate bridge-symptom chains (node-ID sequences), |
| UNRANKED -- use `rank_reasoning_paths` to select the best one(s). |
| """ |
| G = kg.graph.to_undirected() |
| chains = [] |
| for depth in range(2, max_depth + 1): |
| try: |
| for path in nx.all_simple_paths(G, disease_a, disease_b, cutoff=depth): |
| if any(G.nodes[n].get("node_type") == "BridgeSymptom" for n in path) and path not in chains: |
| chains.append(path) |
| except nx.NodeNotFound as exc: |
| logger.warning("Bridge-symptom chain search failed: %s", exc) |
| return [] |
| if chains: |
| break |
| return chains[:max_chains] |
|
|
|
|
| def _path_confidence(kg: ClinicalKnowledgeGraph, path: List[str]) -> float: |
| """Geometric mean of edge confidences along `path` (order-agnostic direction lookup).""" |
| confidences = [] |
| for i in range(len(path) - 1): |
| edge_data = kg.graph.get_edge_data(path[i], path[i + 1]) or kg.graph.get_edge_data(path[i + 1], path[i]) |
| if edge_data: |
| confidences.append(max(d.get("confidence", 0.5) for d in edge_data.values())) |
| return float(np.exp(np.mean(np.log(np.clip(confidences, 1e-6, 1.0))))) if confidences else 0.0 |
|
|
|
|
| def _path_guideline_support(kg: ClinicalKnowledgeGraph, path: List[str]) -> bool: |
| """Check whether any edge along `path` is backed by guideline-level evidence.""" |
| for i in range(len(path) - 1): |
| edge_data = kg.graph.get_edge_data(path[i], path[i + 1]) or kg.graph.get_edge_data(path[i + 1], path[i]) |
| if edge_data and any(d.get("evidence_level") == "Guideline" for d in edge_data.values()): |
| return True |
| return False |
|
|
|
|
| def rank_reasoning_paths(kg: ClinicalKnowledgeGraph, chains: List[List[str]]) -> List[Dict]: |
| """ |
| Rank candidate reasoning chains by a composite clinical score: confidence- |
| weighted, with a bonus for guideline-backed edges and a mild penalty for |
| unnecessarily long chains (shorter, well-supported explanations are more |
| clinically useful than long speculative ones). This is 'Clinical Path |
| Ranking' + 'Evidence Path Ranking' from Task 8 -- not simple shortest-path. |
| |
| Args: |
| chains (List[List[str]]): Candidate node-ID chains. |
| |
| Returns: |
| List[Dict]: [{"path": List[str], "confidence": float, "guideline_supported": bool, |
| "length": int, "score": float}], sorted by score descending. |
| """ |
| ranked = [] |
| for path in chains: |
| confidence = _path_confidence(kg, path) |
| guideline_supported = _path_guideline_support(kg, path) |
| length_penalty = 1.0 / (1.0 + 0.1 * max(len(path) - 3, 0)) |
| score = confidence * (1.15 if guideline_supported else 1.0) * length_penalty |
| ranked.append({"path": path, "confidence": round(confidence, 4), |
| "guideline_supported": guideline_supported, "length": len(path), |
| "score": round(score, 4)}) |
| ranked.sort(key=lambda r: r["score"], reverse=True) |
| return ranked |
|
|
|
|
| def build_reasoning_explanation(kg: ClinicalKnowledgeGraph, chain: List[str]) -> str: |
| """ |
| Turn a graph node-ID chain into a human-readable clinical reasoning sentence. |
| BUGFIX: the arrow is now drawn in the TRUE direction of each edge (not always |
| forward relative to traversal order) -- e.g. if the path walks Symptom -> |
| Disease but the real edge is Disease-[HAS_BRIDGE_SYMPTOM]->Symptom, this |
| renders "Symptom <--[HAS_BRIDGE_SYMPTOM]-- Disease", not a misleading |
| forward arrow that would misstate which node has which relation to which. |
| Used both for the LLM prompt (Ch.11) and the 'Reasoning Path' in Ch.14. |
| """ |
| if not chain: |
| return "No reasoning path found between the specified concepts." |
| node_names = [kg.graph.nodes[n].get("canonical_name", n) for n in chain] |
|
|
| parts = [node_names[0]] |
| for i in range(len(chain) - 1): |
| forward_data = kg.graph.get_edge_data(chain[i], chain[i + 1]) |
| backward_data = kg.graph.get_edge_data(chain[i + 1], chain[i]) |
| if forward_data: |
| relation = next(iter(forward_data.values()))["relation"] |
| arrow = f"--[{relation}]-->" |
| elif backward_data: |
| relation = next(iter(backward_data.values()))["relation"] |
| arrow = f"<--[{relation}]--" |
| else: |
| arrow = "--[RELATED_TO]-->" |
| parts.append(arrow) |
| parts.append(node_names[i + 1]) |
| return " ".join(parts) |
|
|
|
|
| def select_best_reasoning_path(kg: ClinicalKnowledgeGraph, disease_a: str, disease_b: str) -> Dict: |
| """ |
| End-to-end reasoning entry point used by Chapter 14: find candidate |
| bridge-symptom chains, rank them, and return the best one with its |
| human-readable explanation and supporting metadata. |
| |
| Returns: |
| Dict: {"explanation": str, "confidence": float, "guideline_supported": bool} |
| (confidence=0.0 and a "no path found" explanation if none exists). |
| """ |
| chains = find_bridge_symptom_chains(kg, disease_a, disease_b) |
| if not chains: |
| return {"explanation": "", "confidence": 0.0, "guideline_supported": False} |
| ranked = rank_reasoning_paths(kg, chains) |
| best = ranked[0] |
| return {"explanation": build_reasoning_explanation(kg, best["path"]), |
| "confidence": best["confidence"], "guideline_supported": best["guideline_supported"]} |
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
|
|
| |
| |
| |
|
|
|
|
| def resolve_entities_to_kg_nodes(kg: ClinicalKnowledgeGraph, query: str, query_entities: List[Dict], |
| name_lookup: Dict[str, str]) -> List[str]: |
| """ |
| Map a query's extracted entities (Chapter 3's NER) AND a direct |
| substring/word-boundary scan of the query text against every known node |
| name/synonym to real node IDs in the CURRENT graph. This is what lets |
| reasoning respond to what the user actually asked, instead of always |
| defaulting to one hardcoded disease pair. |
| |
| Args: |
| kg (ClinicalKnowledgeGraph): The graph to resolve against. |
| query (str): The raw user query. |
| query_entities (List[Dict]): Chapter 3's extracted entities for this query. |
| name_lookup (Dict[str, str]): Output of `build_kg_name_lookup`. |
| |
| Returns: |
| List[str]: Node IDs actually present in the graph that the query refers to. |
| """ |
| matched = set() |
| query_lower = query.lower() |
| for name, node_id in name_lookup.items(): |
| if name and re.search(r"\b" + re.escape(name) + r"\b", query_lower): |
| matched.add(node_id) |
| for e in query_entities: |
| node_id = name_lookup.get(e["text"].strip().lower()) |
| if node_id: |
| matched.add(node_id) |
| return list(matched) |
|
|
|
|
| def find_query_adaptive_reasoning(kg: ClinicalKnowledgeGraph, matched_node_ids: List[str], |
| default_disease_pair: Tuple[str, str] = None) -> Dict: |
| """ |
| Build a reasoning explanation genuinely anchored to the query's own |
| entities. Tries, in order of specificity: |
| 1. A BridgeSymptom the query mentions -> the disease-symptom-disease |
| chain through THAT specific symptom (verified to actually exist). |
| 2. Two Drug/DrugClass nodes the query mentions -> their direct |
| relationship (e.g. CONTRAINDICATED_WITH, INTERACTS_WITH, TREATS). |
| 3. One Drug/DrugClass/Treatment/Guideline/DiagnosticCriteria node -> its |
| most relevant direct relationship in the graph. |
| 4. Any OTHER node the query matched (AssessmentScale, Gene, BrainRegion, |
| Neurotransmitter, Biomarker, RiskFactor, ProtectiveFactor, SideEffect, |
| ClinicalOutcome, PatientGroup, ...) -> its most relevant direct |
| relationship. Added alongside Section 4.3's dynamic ingestion, which |
| populates many more of these types than the original literature seed |
| alone did -- without this tier, a query naming e.g. "GAD-7" or |
| "amygdala" would always fall through to the generic fallback below |
| even though the graph now has a specific, real relationship for it. |
| 5. Fallback: the generic best bridge-symptom chain between the two core |
| diseases (still genuinely computed, just not anchored to specifics |
| the query didn't provide -- flagged as "general_fallback" so the |
| caller can be transparent about this). |
| |
| Args: |
| kg (ClinicalKnowledgeGraph): The graph to reason over. |
| matched_node_ids (List[str]): Output of `resolve_entities_to_kg_nodes`. |
| default_disease_pair (Tuple[str, str]): Used only for the fallback case. |
| |
| Returns: |
| Dict: {"explanation": str, "confidence": float, "guideline_supported": bool, "anchor": str} |
| """ |
| if default_disease_pair is None: |
| |
| |
| default_disease_pair = DEFAULT_DISEASE_PAIR |
| node_types = nx.get_node_attributes(kg.graph, "node_type") |
|
|
| bridge_matches = [n for n in matched_node_ids if node_types.get(n) == "BridgeSymptom"] |
| if bridge_matches: |
| symptom_id = bridge_matches[0] |
| disease_a, disease_b = default_disease_pair |
| if kg.graph.has_edge(disease_a, symptom_id) and kg.graph.has_edge(disease_b, symptom_id): |
| chain = [disease_a, symptom_id, disease_b] |
| return {"explanation": build_reasoning_explanation(kg, chain), |
| "confidence": _path_confidence(kg, chain), |
| "guideline_supported": _path_guideline_support(kg, chain), "anchor": "bridge_symptom"} |
|
|
| drug_matches = [n for n in matched_node_ids if node_types.get(n) in ("Drug", "DrugClass")] |
| if len(drug_matches) >= 2: |
| a, b = drug_matches[0], drug_matches[1] |
| edge_data = kg.graph.get_edge_data(a, b) or kg.graph.get_edge_data(b, a) |
| if edge_data: |
| best_edge = max(edge_data.values(), key=lambda d: d.get("confidence", 0)) |
| return {"explanation": build_reasoning_explanation(kg, [a, b]), |
| "confidence": best_edge.get("confidence", 0.5), |
| "guideline_supported": best_edge.get("relation") == "RECOMMENDED_BY", "anchor": "drug_interaction"} |
|
|
| anchor_candidates = drug_matches + [n for n in matched_node_ids |
| if node_types.get(n) in ("Treatment", "Guideline", "DiagnosticCriteria")] |
| for node_id in anchor_candidates: |
| neighbors = list(kg.graph.successors(node_id)) or list(kg.graph.predecessors(node_id)) |
| if neighbors: |
| chain = [node_id, neighbors[0]] |
| return {"explanation": build_reasoning_explanation(kg, chain), |
| "confidence": _path_confidence(kg, chain), "guideline_supported": True, "anchor": "treatment_guideline"} |
|
|
| |
| |
| |
| |
| |
| remaining_candidates = [n for n in matched_node_ids if n not in anchor_candidates and n not in bridge_matches] |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| remaining_candidates.sort(key=lambda n: node_types.get(n) == "Disease") |
| for node_id in remaining_candidates: |
| neighbors = list(kg.graph.successors(node_id)) or list(kg.graph.predecessors(node_id)) |
| if neighbors: |
| chain = [node_id, neighbors[0]] |
| return {"explanation": build_reasoning_explanation(kg, chain), |
| "confidence": _path_confidence(kg, chain), |
| "guideline_supported": _path_guideline_support(kg, chain), "anchor": "specific_entity"} |
|
|
| fallback = select_best_reasoning_path(kg, *default_disease_pair) |
| fallback["anchor"] = "general_fallback" |
| return fallback |
|
|
| _DEFAULT_DISEASE_PAIR_NAMES = ("generalized anxiety disorder", "major depressive disorder") |
|
|
|
|
| def resolve_default_disease_pair(kg: ClinicalKnowledgeGraph, name_lookup: Dict[str, str]) -> Tuple[str, str]: |
| """ |
| Resolve the flagship GAD/MDD disease pair used as the final reasoning fallback BY NAME, |
| not a hardcoded node ID. This replaces the old hardcoded ("disease:gad", "disease:mdd") |
| default that made sense when Section 4.2 hand-assigned those exact IDs -- now that the |
| graph is built ENTIRELY by Section 4.3's dynamic ingestion (the static seed has been |
| retired), every node's ID comes from the ingestion pipeline's own deterministic scheme and |
| cannot be predicted in advance. Falls back to the two highest-confidence Disease-type nodes |
| in the graph if the exact names can't be found for any reason (e.g. an unusually degraded |
| ingestion run with no live sources and a tiny/empty local KB), so this never raises even in |
| a worst-case session. |
| """ |
| gad_id = name_lookup.get(_DEFAULT_DISEASE_PAIR_NAMES[0]) |
| mdd_id = name_lookup.get(_DEFAULT_DISEASE_PAIR_NAMES[1]) |
| if gad_id and mdd_id: |
| return (gad_id, mdd_id) |
| logger.warning("Could not resolve the default GAD/MDD disease pair by name (found gad=%s, mdd=%s); " |
| "falling back to the two highest-confidence Disease-type nodes in the graph.", gad_id, mdd_id) |
| disease_nodes = sorted( |
| [(n, d) for n, d in kg.graph.nodes(data=True) if d.get("node_type") == "Disease"], |
| key=lambda nd: nd[1].get("confidence", 0), reverse=True, |
| ) |
| if len(disease_nodes) >= 2: |
| return (disease_nodes[0][0], disease_nodes[1][0]) |
| if len(disease_nodes) == 1: |
| return (disease_nodes[0][0], disease_nodes[0][0]) |
| logger.warning("No Disease-type nodes found in the graph at all -- reasoning fallback will find no paths.") |
| return ("", "") |
|
|
|
|
| def propagate_confidence(kg: ClinicalKnowledgeGraph, path: List[str]) -> Dict: |
| """Improvement #4: propagate confidence along a reasoning path, returning both the |
| cumulative (product) path confidence and a per-hop breakdown. This exposes HOW confident |
| each individual link in a multi-hop chain is -- not just the aggregate -- so an answer can |
| show which hop is the weakest point in the reasoning. |
| |
| Returns: |
| Dict: {"path_confidence": float, "hops": [{"from","to","relation","confidence"}...], |
| "weakest_hop": Dict|None} |
| """ |
| hops, running = [], 1.0 |
| for i in range(len(path) - 1): |
| edge_data = kg.graph.get_edge_data(path[i], path[i + 1]) or kg.graph.get_edge_data(path[i + 1], path[i]) |
| if not edge_data: |
| continue |
| best = max(edge_data.values(), key=lambda d: d.get("confidence", 0.0)) |
| conf = float(best.get("confidence", 0.5)) |
| running *= conf |
| hops.append({"from": kg.graph.nodes[path[i]].get("canonical_name", path[i]), |
| "to": kg.graph.nodes[path[i + 1]].get("canonical_name", path[i + 1]), |
| "relation": best.get("relation", "RELATED_TO"), "confidence": round(conf, 4)}) |
| weakest = min(hops, key=lambda h: h["confidence"]) if hops else None |
| return {"path_confidence": round(running, 4), "hops": hops, "weakest_hop": weakest} |
|
|
|
|
|
|
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| |
| MEDGEMMA_MODEL_NAME = "jubayer009/medgemma-1.5-4b-mental-health-rag" |
| MEDGEMMA_MIN_GPU_GB = 10 |
|
|
| |
| |
| |
| |
| |
| |
| CANDIDATE_MODELS = [ |
| {"name": "Qwen/Qwen2.5-7B-Instruct", "task": "text-generation", "min_gpu_gb": 12}, |
| {"name": "microsoft/Phi-3-mini-4k-instruct", "task": "text-generation", "min_gpu_gb": 6}, |
| {"name": "google/flan-t5-base", "task": "text2text-generation", "min_gpu_gb": 0}, |
| ] |
|
|
|
|
| def _available_gpu_gb() -> float: |
| """Return total memory (GB) of GPU 0, or 0.0 if no CUDA device is available.""" |
| if not torch.cuda.is_available(): |
| return 0.0 |
| return torch.cuda.get_device_properties(0).total_memory / (1024 ** 3) |
|
|
|
|
| class PromptBuilder: |
| """Builds the prompt (chat-message form for instruct/chat models, plain-string form otherwise).""" |
|
|
| _SYSTEM_MESSAGE = ( |
| "You are a warm, experienced clinician talking directly to a patient. Explain things in " |
| "plain, everyday language -- avoid unexplained medical jargon, and briefly define any " |
| "clinical term you do need to use. Using the evidence provided, give a clear, DIRECT, and " |
| "informative answer -- state what the evidence indicates plainly and confidently, and " |
| "include relevant specifics (mechanisms, symptom names, treatment options) rather than " |
| "vague generalities. Only hedge or express uncertainty when the evidence is genuinely " |
| "thin or conflicting -- do not pad a well-supported answer with repeated disclaimers or " |
| "excessive hedging language. Answer as a SINGLE flowing, empathetic paragraph (short " |
| "bullet points only when listing several distinct options, like treatments). Do NOT " |
| "include citation markers, bracket numbers, or reference numbers anywhere in your answer " |
| "-- sources will be listed separately afterward. Do NOT quote the evidence verbatim; " |
| "explain it in your own words, the way a confident, knowledgeable doctor summarizes " |
| "findings for a patient in clinic. If the evidence is genuinely insufficient, say so " |
| "plainly and kindly rather than guessing. End with one brief, natural sentence noting " |
| "this is general information and not a substitute for personalized care from a licensed " |
| "clinician." |
| ) |
|
|
| @staticmethod |
| def build_messages(query: str, clinical_context: str, reasoning_explanation: str) -> List[Dict]: |
| """Build a chat-style messages list for instruct/chat models.""" |
| user_content = ( |
| f"PATIENT'S QUESTION:\n{query}\n\n" |
| f"CLINICAL REASONING CONTEXT (for your understanding only, do not quote):\n{reasoning_explanation}\n\n" |
| f"EVIDENCE (for your understanding only, do not quote or cite by number):\n{clinical_context}\n\n" |
| "Please answer the patient's question now, in the style described." |
| ) |
| return [{"role": "system", "content": PromptBuilder._SYSTEM_MESSAGE}, |
| {"role": "user", "content": user_content}] |
|
|
| @staticmethod |
| def build(query: str, clinical_context: str, reasoning_explanation: str) -> str: |
| """Build a single-string prompt for non-chat (e.g. seq2seq) models.""" |
| return ( |
| f"{PromptBuilder._SYSTEM_MESSAGE}\n\n" |
| f"QUESTION:\n{query}\n\n" |
| f"CLINICAL REASONING CONTEXT (for your understanding only, do not quote):\n{reasoning_explanation}\n\n" |
| f"EVIDENCE (for your understanding only, do not quote or cite by number):\n{clinical_context}\n\n" |
| "ANSWER (one warm, professional paragraph, bullets only if listing multiple items, no citations):" |
| ) |
|
|
|
|
| |
| |
|
|
|
|
| def _clean_sentence(sentence: str) -> str: |
| """Strip stray list markers/numbering and normalize whitespace/terminal punctuation.""" |
| s = re.sub(r"^[\-\*\d\.\)\s]+", "", sentence.strip()) |
| s = re.sub(r"\s+", " ", s).strip() |
| if s and not s.endswith((".", "!", "?")): |
| s += "." |
| return s |
|
|
|
|
| |
| |
| def _select_clean_sentences(evidence_items: List["EvidenceItem"], max_sentences: int = 6) -> List[str]: |
| """Select up to `max_sentences` clean, non-redundant, non-boilerplate sentences from top evidence.""" |
| import difflib |
| selected, seen = [], [] |
| for item in evidence_items: |
| for sentence in re.split(r"(?<=[.!?])\s+", item.text.strip())[:2]: |
| if _is_junk_sentence(sentence): |
| continue |
| cleaned = _clean_sentence(sentence) |
| if any(difflib.SequenceMatcher(None, cleaned.lower(), s).ratio() > 0.8 for s in seen): |
| continue |
| seen.append(cleaned.lower()) |
| selected.append(cleaned) |
| if len(selected) >= max_sentences: |
| return selected |
| return selected |
|
|
|
|
| _EMPATHY_CUES = re.compile(r"\b(scared|worried|afraid|anxious about|struggling|overwhelmed|" |
| r"don't know what to do|helpless)\b", re.IGNORECASE) |
|
|
|
|
| def _opening_frame(query: str) -> str: |
| """Pick a natural, warm clinical framing sentence based on the query's intent.""" |
| q = query.lower() |
| empathy_prefix = "I understand this can feel worrying. " if _EMPATHY_CUES.search(query) else "" |
| if any(w in q for w in ("treat", "manage", "recommend", "therapy", "medication")): |
| return empathy_prefix + "Based on current clinical evidence, here are approaches commonly recommended:" |
| if any(w in q for w in ("safe", "interact", "combine", "combination", "contraindicat")): |
| return empathy_prefix + "Based on the available clinical evidence regarding this combination:" |
| if any(w in q for w in ("why", "how", "link", "relationship", "connection", "comorbid")): |
| return empathy_prefix + "Current clinical evidence suggests the following explanation:" |
| return empathy_prefix + "Based on the available clinical evidence:" |
|
|
|
|
| class AnswerGenerator: |
| """ |
| Generates the final answer as a single warm, patient-friendly paragraph (or |
| a short bulleted list for genuine multi-item answers), with no inline |
| citation markers. Backend priority: the strongest instruct/chat model that |
| fits the available GPU memory (see `CANDIDATE_MODELS`), falling back |
| through progressively smaller models, and finally to a deterministic |
| extractive/template composer that is always available. |
| """ |
|
|
| def __init__(self, model_name: str = None): |
| self.backend, self.pipeline, self.tokenizer, self.task, self.model_name = None, None, None, None, None |
| self.processor = None |
| if not has_pkg("transformers"): |
| logger.warning("transformers not available; using extractive fallback.") |
| self.backend = "extractive_fallback" |
| return |
|
|
| |
| |
| if model_name is None or model_name == MEDGEMMA_MODEL_NAME: |
| gpu_gb = _available_gpu_gb() |
| if gpu_gb == 0.0 or gpu_gb >= MEDGEMMA_MIN_GPU_GB: |
| try: |
| from transformers import AutoProcessor, AutoModelForImageTextToText |
| _hf_token_kwarg = {"token": HF_TOKEN} if globals().get("HF_TOKEN") else {} |
| self.processor = AutoProcessor.from_pretrained(MEDGEMMA_MODEL_NAME, **_hf_token_kwarg) |
| load_kwargs = {"torch_dtype": torch.bfloat16, "device_map": "auto"} if DEVICE == "cuda" \ |
| else {} |
| self.model = AutoModelForImageTextToText.from_pretrained( |
| MEDGEMMA_MODEL_NAME, **load_kwargs, **_hf_token_kwarg |
| ) |
| self.backend, self.model_name = "medgemma", MEDGEMMA_MODEL_NAME |
| logger.info("AnswerGenerator backend: medgemma (%s)", MEDGEMMA_MODEL_NAME) |
| except Exception as exc: |
| logger.warning("Could not load MedGemma generator '%s' (%s) -- this usually means " |
| "either the Hugging Face license for MedGemma hasn't been accepted " |
| "for this account, HF_TOKEN wasn't provided/valid (see Chapter 1B), " |
| "or there wasn't enough GPU memory. Falling back to the general-" |
| "purpose candidates below.", MEDGEMMA_MODEL_NAME, exc) |
| else: |
| logger.info("Skipping MedGemma generator: needs ~%sGB GPU memory, %.1fGB available.", |
| MEDGEMMA_MIN_GPU_GB, gpu_gb) |
|
|
| if self.backend is not None: |
| return |
|
|
| candidates = [c for c in CANDIDATE_MODELS if model_name is None or c["name"] == model_name] or CANDIDATE_MODELS |
| gpu_gb = _available_gpu_gb() |
|
|
| for cand in candidates: |
| if cand["min_gpu_gb"] > 0 and gpu_gb < cand["min_gpu_gb"]: |
| logger.info("Skipping %s: needs ~%sGB GPU memory, %.1fGB available.", |
| cand["name"], cand["min_gpu_gb"], gpu_gb) |
| continue |
| try: |
| from transformers import pipeline as hf_pipeline |
| pipeline_kwargs = {"torch_dtype": torch.float16, "device_map": "auto"} if DEVICE == "cuda" \ |
| else {"device": -1} |
| if globals().get("HF_TOKEN"): |
| pipeline_kwargs["token"] = HF_TOKEN |
| self.pipeline = hf_pipeline(cand["task"], model=cand["name"], **pipeline_kwargs) |
| self.tokenizer = getattr(self.pipeline, "tokenizer", None) |
| self.task, self.model_name, self.backend = cand["task"], cand["name"], "transformers" |
| |
| |
| |
| |
| gen_config = getattr(getattr(self.pipeline, "model", None), "generation_config", None) |
| if gen_config is not None: |
| gen_config.max_length = None |
| logger.info("AnswerGenerator backend: transformers (%s)", cand["name"]) |
| break |
| except Exception as exc: |
| logger.warning("Could not load '%s' (%s); trying next candidate.", cand["name"], exc) |
|
|
| if self.backend is None: |
| self.backend = "extractive_fallback" |
| logger.info("AnswerGenerator backend: extractive_fallback") |
|
|
| def _extractive_fallback(self, query: str, evidence_items: List["EvidenceItem"]) -> str: |
| """Deterministic, fully-implemented answer composer used whenever no LLM backend is loaded.""" |
| opening = _opening_frame(query) |
| sentences = _select_clean_sentences(evidence_items) |
|
|
| if not sentences: |
| return ("There is not enough specific evidence in the current knowledge base or retrieved " |
| "literature to answer this question with confidence. Please consult a licensed " |
| "clinician for guidance tailored to your specific situation.") |
|
|
| use_bullets = opening.rstrip().endswith(":") and len(sentences) >= 2 |
| if use_bullets: |
| bullet_block = "\n".join(f"- {s}" for s in sentences) |
| answer_body = f"{opening}\n\n{bullet_block}" |
| else: |
| answer_body = f"{opening} " + " ".join(sentences) |
|
|
| disclaimer = ("As with any mental health concern, this information should be discussed with a " |
| "qualified clinician for personalized diagnosis and treatment.") |
| return f"{answer_body}\n\n{disclaimer}" |
|
|
| def _generate_with_medgemma(self, query: str, clinical_context: str, reasoning_explanation: str) -> str: |
| """ |
| MedGemma is a multimodal (image+text) architecture (AutoProcessor + |
| AutoModelForImageTextToText), not a plain text-only causal LM -- even though this notebook |
| only ever gives it text turns. Its chat template (chat_template.jinja, shipped with the |
| checkpoint) expects each message's `content` to be a LIST of typed blocks (e.g. |
| `[{"type": "text", "text": "..."}]`), matching how Gemma 3-family multimodal models are |
| used in text-only mode; that's the one real difference from the plain-string chat messages |
| the other ("transformers") backend below uses. |
| """ |
| messages = PromptBuilder.build_messages(query, clinical_context, reasoning_explanation) |
| mm_messages = [{"role": m["role"], "content": [{"type": "text", "text": m["content"]}]} |
| for m in messages] |
| inputs = self.processor.apply_chat_template( |
| mm_messages, add_generation_prompt=True, tokenize=True, |
| return_dict=True, return_tensors="pt", |
| ).to(self.model.device) |
| input_len = inputs["input_ids"].shape[-1] |
| with torch.inference_mode(): |
| output_ids = self.model.generate(**inputs, max_new_tokens=420, do_sample=False) |
| text = self.processor.decode(output_ids[0][input_len:], skip_special_tokens=True).strip() |
| return text |
|
|
| def generate(self, query: str, clinical_context: str, reasoning_explanation: str, |
| citations: List[Dict], evidence_items: List["EvidenceItem"]) -> str: |
| """Produce the final, patient-friendly answer text (no inline citation markers).""" |
| if self.backend == "medgemma": |
| try: |
| text = self._generate_with_medgemma(query, clinical_context, reasoning_explanation) |
| text = re.sub(r"\[\d+\]", "", text) |
| text = re.sub(r"[ \t]{2,}", " ", text).strip() |
| if text: |
| return text |
| logger.warning("MedGemma returned an empty answer; using extractive fallback.") |
| except Exception as exc: |
| logger.warning("MedGemma generation failed at inference time (%s); using extractive " |
| "fallback.", exc) |
| return self._extractive_fallback(query, evidence_items) |
| if self.backend == "transformers": |
| try: |
| if self.task == "text-generation": |
| messages = PromptBuilder.build_messages(query, clinical_context, reasoning_explanation) |
| if self.tokenizer is not None and getattr(self.tokenizer, "chat_template", None): |
| prompt_text = self.tokenizer.apply_chat_template(messages, tokenize=False, |
| add_generation_prompt=True) |
| else: |
| prompt_text = PromptBuilder.build(query, clinical_context, reasoning_explanation) |
| result = self.pipeline(prompt_text, max_new_tokens=420, do_sample=False, |
| return_full_text=False) |
| else: |
| prompt_text = PromptBuilder.build(query, clinical_context, reasoning_explanation) |
| result = self.pipeline(prompt_text, max_new_tokens=380, do_sample=False) |
|
|
| text = result[0]["generated_text"].strip() |
| text = re.sub(r"\[\d+\]", "", text) |
| text = re.sub(r"[ \t]{2,}", " ", text).strip() |
| if text: |
| return text |
| logger.warning("LLM returned an empty answer; using extractive fallback.") |
| except Exception as exc: |
| logger.warning("LLM generation failed at inference time (%s); using extractive fallback.", exc) |
| return self._extractive_fallback(query, evidence_items) |
|
|
|
|
|
|
|
|
| |
| |
| |
| _CRISIS_PATTERN = re.compile( |
| r"\b(suicid\w*|kill myself|end my life|ending my life|want to die|no reason to live|" |
| r"self[- ]?harm|hurt myself|harming myself|cutting myself|overdose on purpose)\b", |
| re.IGNORECASE, |
| ) |
|
|
|
|
| def is_crisis_query(query: str) -> bool: |
| """ |
| Detect whether a query indicates the person may be describing active |
| suicidal ideation or self-harm intent, in which case the pipeline MUST |
| bypass RAG synthesis entirely (Chapter 14 checks this before doing any |
| retrieval/generation) and return a fixed, vetted crisis response instead. |
| Retrieval-and-synthesis is never appropriate for this category of query, |
| regardless of how confident the retrieved evidence appears. |
| |
| Args: |
| query (str): The user's question. |
| |
| Returns: |
| bool: True if the query matches a crisis pattern. |
| """ |
| return bool(_CRISIS_PATTERN.search(query)) |
|
|
|
|
| def build_crisis_response() -> str: |
| """ |
| Return a fixed, vetted crisis-support message. This text is intentionally |
| NOT generated by the LLM or extractive composer -- for safety-critical |
| content like this, a hard-coded, reviewed response is required rather than |
| anything synthesized from retrieved documents. |
| |
| Returns: |
| str: The crisis-support response text. |
| """ |
| return ( |
| "If you are having thoughts of suicide or self-harm, please reach out for support right away. " |
| "In the US, you can call or text 988 (Suicide & Crisis Lifeline) any time, day or night. " |
| "If you are outside the US, please look up a crisis line in your country -- a directory is " |
| "available at https://www.iasp.info/resources/Crisis_Centres/. If you are in immediate danger, " |
| "please contact your local emergency services now.\n\n" |
| "You do not have to go through this alone. Please also consider reaching out to a mental health " |
| "professional or a trusted person in your life as soon as you can." |
| ) |
|
|
|
|
| DRUG_INTERACTION_TABLE = [ |
| ({"ssri", "maoi"}, "Risk of serotonin syndrome when SSRIs and MAOIs are combined."), |
| ({"snri", "maoi"}, "Risk of serotonin syndrome when SNRIs and MAOIs are combined."), |
| ({"ssri", "snri"}, "Combining an SSRI with an SNRI raises serotonergic burden and serotonin-syndrome risk."), |
| ({"benzodiazepine", "alcohol"}, "Combining benzodiazepines with alcohol increases sedation/respiratory-depression risk."), |
| ({"benzodiazepine", "opioid"}, "Combining benzodiazepines with opioids carries a serious respiratory-depression risk."), |
| ({"tricyclic antidepressant", "maoi"}, "Combining TCAs and MAOIs carries a serious hypertensive/serotonergic risk."), |
| ({"ssri", "nsaid"}, "SSRIs combined with NSAIDs increase gastrointestinal bleeding risk."), |
| ] |
|
|
| |
| CONTRAINDICATION_TABLE = [ |
| ({"maoi"}, {"tyramine"}, "MAOIs require dietary tyramine restriction to avoid hypertensive crisis."), |
| ({"benzodiazepine"}, {"substance use", "alcohol use"}, "Benzodiazepines are used cautiously in patients with a substance-use history."), |
| ({"tricyclic antidepressant"}, {"cardiac"}, "TCAs are used cautiously in patients with cardiac conduction disease."), |
| ] |
| _PREGNANCY_CUES = re.compile(r"\b(pregnan\w+|breastfeed\w+|lactat\w+|expecting a baby|prenatal|perinatal)\b", re.IGNORECASE) |
| _PEDIATRIC_CUES = re.compile(r"\b(child|children|infant|toddler|adolescent|teenager|pediatric|paediatric|my son|my daughter|my kid)\b", re.IGNORECASE) |
| _MEDICATION_CUES = re.compile(r"\b(ssri|snri|maoi|benzodiazepine|antidepressant|medication|drug|dose|dosage|prescri\w+)\b", re.IGNORECASE) |
|
|
|
|
| def check_pregnancy_warnings(text: str) -> List[str]: |
| """Flag pregnancy/breastfeeding + medication co-mentions that warrant a clinician check.""" |
| if _PREGNANCY_CUES.search(text) and _MEDICATION_CUES.search(text): |
| return ["This question involves medication during pregnancy or breastfeeding -- medication " |
| "safety in this context is individualized and must be confirmed with a prescribing " |
| "clinician or obstetric pharmacist before any change."] |
| return [] |
|
|
|
|
| def check_pediatric_warnings(text: str) -> List[str]: |
| """Flag pediatric + medication co-mentions that warrant specialist guidance.""" |
| if _PEDIATRIC_CUES.search(text) and _MEDICATION_CUES.search(text): |
| return ["This question involves medication in a child or adolescent -- pediatric dosing and " |
| "safety differ from adults and should be guided by a child/adolescent psychiatrist."] |
| return [] |
|
|
|
|
| def check_contraindications(text: str) -> List[str]: |
| """Scan text for drug/condition combinations flagged in CONTRAINDICATION_TABLE.""" |
| t = text.lower() |
| out = [] |
| for drugs, conditions, message in CONTRAINDICATION_TABLE: |
| if any(d in t for d in drugs) and any(c in t for c in conditions): |
| out.append(message) |
| return out |
|
|
|
|
| def check_unsupported_claims(answer: str, evidence_items: List["EvidenceItem"], |
| overlap_threshold: float = 0.35) -> List[str]: |
| """ |
| Flag sentences in `answer` whose semantic overlap with the evidence pool |
| falls below `overlap_threshold`, indicating a potentially unsupported claim. |
| |
| Args: |
| answer (str): Generated answer text. |
| evidence_items (List[EvidenceItem]): The evidence the answer should be grounded in. |
| overlap_threshold (float): Minimum cosine similarity to the best-matching |
| evidence item required to consider a sentence supported. |
| |
| Returns: |
| List[str]: Sentences flagged as unsupported (possible hallucinations). |
| """ |
| if not evidence_items: |
| return re.split(r"(?<=[.!?])\s+", answer.strip()) |
| evidence_embs = EMBEDDING_MODEL.encode([e.text[:500] for e in evidence_items]) |
| sentences = [s for s in re.split(r"(?<=[.!?])\s+", answer.strip()) if len(s.split()) > 3] |
| flagged = [] |
| for sentence in sentences: |
| sent_emb = EMBEDDING_MODEL.encode([sentence])[0] |
| best_sim = float(np.max(evidence_embs @ sent_emb)) if len(evidence_embs) else 0.0 |
| if best_sim < overlap_threshold: |
| flagged.append(sentence) |
| return flagged |
|
|
|
|
| def check_drug_interactions(text: str) -> List[str]: |
| """Scan text for co-mentioned drug pairs known to carry a clinically significant interaction risk.""" |
| text_lower = text.lower() |
| warnings = [] |
| for drug_pair, message in DRUG_INTERACTION_TABLE: |
| if all(drug in text_lower for drug in drug_pair): |
| warnings.append(message) |
| return warnings |
|
|
|
|
| def compute_confidence_level(evidence_items: List["EvidenceItem"], unsupported_count: int, |
| contradiction_count: int) -> str: |
| """ |
| Classify overall answer confidence as High / Medium / Low. Uses the Clinical |
| Evidence Validator's `reliability_score` (Chapter 9B) when available -- which |
| already factors in evidence-hierarchy strength and cross-source corroboration |
| -- falling back to raw retrieval confidence if the validator wasn't run, |
| penalized further by unsupported-claim and contradiction counts. |
| """ |
| if not evidence_items: |
| return "Low" |
| scores = [e.reliability_score if e.reliability_score > 0 else e.confidence for e in evidence_items] |
| mean_score = float(np.mean(scores)) |
| penalty = 0.15 * unsupported_count + 0.15 * contradiction_count |
| score = max(mean_score - penalty, 0.0) |
| if score >= 0.65: |
| return "High" |
| if score >= 0.35: |
| return "Medium" |
| return "Low" |
|
|
|
|
| def run_safety_verification(answer: str, evidence_items: List["EvidenceItem"], |
| contradictions: List[Tuple[str, str]], |
| conflict_report: List[Dict] = None, query: str = "") -> Dict: |
| """ |
| Run the full safety layer and produce a structured report: hallucination/ |
| unsupported-claim detection, contradiction detection, drug interactions, |
| explicit missing-evidence detection, a clinical-consistency proxy score, |
| and an overall confidence level with low-confidence warning. |
| |
| Returns: |
| Dict: { |
| "unsupported_sentences": List[str], "contradiction_count": int, |
| "drug_interaction_warnings": List[str], "missing_evidence": bool, |
| "clinical_consistency_score": float, "confidence_level": str, |
| "low_confidence_warning": bool |
| } |
| """ |
| unsupported = check_unsupported_claims(answer, evidence_items) |
| combined = f"{query}\n{answer}" |
| drug_warnings = check_drug_interactions(combined) |
| pregnancy_warnings = check_pregnancy_warnings(combined) |
| pediatric_warnings = check_pediatric_warnings(combined) |
| contraindication_warnings = check_contraindications(combined) |
| missing_evidence = len(evidence_items) == 0 |
| conflict_count = len(conflict_report) if conflict_report else len(contradictions) |
| clinical_consistency_score = round(1.0 - min(conflict_count / max(len(evidence_items), 1), 1.0), 4) |
| confidence_level = compute_confidence_level(evidence_items, len(unsupported), len(contradictions)) |
| return { |
| "unsupported_sentences": unsupported, |
| "contradiction_count": len(contradictions), |
| "drug_interaction_warnings": drug_warnings, |
| "pregnancy_warnings": pregnancy_warnings, |
| "pediatric_warnings": pediatric_warnings, |
| "contraindication_warnings": contraindication_warnings, |
| "missing_evidence": missing_evidence, |
| "clinical_consistency_score": clinical_consistency_score, |
| "confidence_level": confidence_level, |
| "low_confidence_warning": confidence_level == "Low", |
| } |
|
|
|
|
| print("Chapter 12 safety layer defined: check_unsupported_claims, check_drug_interactions, " |
| "compute_confidence_level, run_safety_verification.") |
|
|
|
|
| |
| |
| |
| class IntentClassifier: |
| """ |
| Explicit intent classifier implementing the 5-route dispatch from the |
| project's Phase-2 architecture diagram: general mental health, anxiety- |
| only, depression-only, anxiety-depression comorbidity, and other |
| (lifestyle/treatment/crisis/miscellaneous). Retrieval strategy is binary |
| (A vs B) even though the intent label is 5-way: only Route 4 (comorbidity) |
| uses Retrieval Strategy B (adds the comorbidity knowledge graph + multi-hop |
| bridge-symptom reasoning); every other route uses Strategy A (local + |
| online + BM25 hybrid retrieval only, no graph traversal). |
| """ |
|
|
| ROUTES = ("general_mental_health", "anxiety_only", "depression_only", "comorbidity", "other") |
|
|
| def classify(self, query: str, router_decision: Dict) -> Dict: |
| """ |
| Args: |
| query (str): The raw user question. |
| router_decision (Dict): Output of `QueryRouter.analyze(query)` -- |
| reused so entity extraction and LLM-assisted symptom-domain |
| detection aren't computed twice for the same query. |
| |
| Returns: |
| Dict: {"route": one of `ROUTES`, "retrieval_strategy": "A" | "B", |
| "graph_enabled": bool, "rationale": str} |
| """ |
| if is_crisis_query(query): |
| return {"route": "other", "retrieval_strategy": "A", "graph_enabled": False, |
| "rationale": "Crisis/self-harm pattern detected (Route N) -- the pipeline " |
| "bypasses retrieval and generation entirely; see Chapter 12."} |
|
|
| entities = router_decision["entities"] |
| entity_texts = {e["text"] for e in entities} |
| anxiety_disease_names = {"generalized anxiety disorder", "panic disorder", |
| "social anxiety disorder"} |
| depression_disease_names = {"major depressive disorder"} |
| disease_texts = {e["text"] for e in entities if e["label"] == "Disease"} |
|
|
| anxiety_signal = bool(entity_texts & ANXIETY_SYMPTOMS) or bool(disease_texts & anxiety_disease_names) |
| depression_signal = bool(entity_texts & DEPRESSION_SYMPTOMS) or bool(disease_texts & depression_disease_names) |
|
|
| |
| |
| if not (anxiety_signal and depression_signal): |
| llm_signal = _llm_assisted_symptom_detection(query) |
| anxiety_signal = anxiety_signal or llm_signal["anxiety_domain"] |
| depression_signal = depression_signal or llm_signal["depression_domain"] |
|
|
| if anxiety_signal and depression_signal: |
| return {"route": "comorbidity", "retrieval_strategy": "B", "graph_enabled": True, |
| "rationale": "Query describes both anxiety- and depression-domain symptoms " |
| "(Route 4) -- engages the comorbidity KG and multi-hop " |
| "bridge-symptom reasoning."} |
| if anxiety_signal: |
| return {"route": "anxiety_only", "retrieval_strategy": "A", "graph_enabled": False, |
| "rationale": "Query describes anxiety-domain symptoms only (Route 2)."} |
| if depression_signal: |
| return {"route": "depression_only", "retrieval_strategy": "A", "graph_enabled": False, |
| "rationale": "Query describes depression-domain symptoms only (Route 3)."} |
| if entities or router_decision.get("external_signal"): |
| return {"route": "general_mental_health", "retrieval_strategy": "A", "graph_enabled": False, |
| "rationale": "Query is mental-health-related but doesn't specifically name " |
| "anxiety or depression symptoms (Route 1)."} |
| return {"route": "other", "retrieval_strategy": "A", "graph_enabled": False, |
| "rationale": "Query didn't match a specific mental-health domain " |
| "(Route N: lifestyle/treatment/other)."} |
|
|
|
|
| |
| |
| |
| |
| _FINE_INTENT_PATTERNS = [ |
| ("emergency", re.compile(r"\b(suicid\w*|kill myself|self[- ]?harm|overdose|emergency|crisis)\b", re.IGNORECASE)), |
| ("safety", re.compile(r"\b(safe|safety|contraindicat\w+|interact\w+|side effect|adverse|pregnan\w+|breastfeed\w+)\b", re.IGNORECASE)), |
| ("medication", re.compile(r"\b(medication|drug|ssri|snri|maoi|benzodiazepine|antidepressant|dose|dosage|prescri\w+)\b", re.IGNORECASE)), |
| ("treatment", re.compile(r"\b(treatment|therap(?:y|ies)|cbt|psychotherapy|manage\w*|intervention)\b", re.IGNORECASE)), |
| ("diagnosis", re.compile(r"\b(diagnos\w+|criteria|screen\w+|assess\w+|dsm|icd|symptom check)\b", re.IGNORECASE)), |
| ("comorbidity", re.compile(r"\b(comorbid\w*|co-?occur\w*|both anxiety and depression|bridge symptom|link between)\b", re.IGNORECASE)), |
| ] |
|
|
|
|
| def _fine_intent_category(query: str) -> str: |
| """Return a fine-grained intent label (emergency/safety/medication/treatment/diagnosis/ |
| comorbidity/factual). Checked in priority order -- emergency and safety win over topical |
| labels so a medication-safety question is never mislabeled as a plain medication lookup.""" |
| for label, pattern in _FINE_INTENT_PATTERNS: |
| if pattern.search(query): |
| return label |
| return "factual" |
|
|
|
|
| |
| |
| |
| _SOURCE_TYPE_LABELS = {"local_kb": "Local knowledge base", "pubmed": "PubMed", |
| "who": "WHO fact sheet", "wikipedia": "Wikipedia", |
| "graph": "Clinical knowledge graph"} |
|
|
|
|
| def _format_source_label(citation: Dict) -> str: |
| """Render a single citation dict as a human-readable reference line, including its evidence level.""" |
| label = _SOURCE_TYPE_LABELS.get(citation["source_type"], citation["source_type"]) |
| ref = citation.get("source_ref", "") |
| ref_str = f" \u2014 {ref}" if ref else "" |
| level = citation.get("evidence_level", "Not Classified") |
| return f"{label}{ref_str} [{level}] (reliability: {citation.get('reliability_score', citation['confidence']):.2f})" |
|
|
|
|
| @dataclasses.dataclass |
| class PipelineOutput: |
| """The structured final output object matching the system architecture diagram.""" |
| answer: str |
| supporting_sources: List[Dict] |
| reasoning_path: str |
| evidence_ranking: List[Dict] |
| confidence_score: str |
| safety_report: Dict |
| intent_route: str = "unknown" |
| retrieval_strategy: str = "A" |
| intent_category: str = "factual" |
| retrieved_chunks: List[Dict] = dataclasses.field(default_factory=list) |
| graph_nodes: List[Dict] = dataclasses.field(default_factory=list) |
|
|
| def _build_recommendations(self) -> List[str]: |
| """Assemble the 'Recommendations (if any)' section from whatever the |
| safety layer and confidence level actually flagged for THIS answer -- |
| never generic filler, and empty when nothing applies.""" |
| recs = [] |
| for _key in ("drug_interaction_warnings", "pregnancy_warnings", |
| "pediatric_warnings", "contraindication_warnings"): |
| if self.safety_report.get(_key): |
| recs.extend(self.safety_report[_key]) |
| if self.safety_report.get("low_confidence_warning"): |
| recs.append("Confidence in this answer is low -- consider rephrasing your question " |
| "with more specific terms, or discuss it directly with a clinician.") |
| if self.safety_report.get("missing_evidence"): |
| recs.append("No strong supporting evidence was found for this specific question -- " |
| "please verify with a licensed clinician before acting on this answer.") |
| if self.safety_report.get("conflict_report"): |
| recs.append(f"{len(self.safety_report['conflict_report'])} potentially conflicting " |
| "piece(s) of evidence were found; this answer reflects the higher-" |
| "reliability source, but a clinician can help resolve the discrepancy.") |
| return recs |
|
|
| def format_report(self) -> str: |
| """ |
| Render the final, user-facing report as the five Response-Structuring |
| sections from the architecture diagram: Answer Summary, Key |
| Explanations, Recommendations (if any), Sources & References, and |
| Disclaimer. |
| """ |
| lines = ["Answer Summary:", self.answer.strip(), ""] |
|
|
| lines.append("Key Explanations:") |
| if self.reasoning_path and not self.reasoning_path.startswith("No multi-hop"): |
| lines.append(f" Clinical reasoning path (Route 4 / Strategy B -- comorbidity " |
| f"KG, multi-hop bridge-symptom traversal):") |
| lines.append(f" {self.reasoning_path}") |
| else: |
| lines.append(f" Answered via Retrieval Strategy {self.retrieval_strategy} " |
| f"(route: {self.intent_route}) -- hybrid local + online retrieval. " |
| f"Multi-hop graph reasoning applies only to anxiety-depression " |
| f"comorbidity questions (Route 4), so it was not used here.") |
| lines.append("") |
|
|
| recommendations = self._build_recommendations() |
| if recommendations: |
| lines.append("Recommendations:") |
| for rec in recommendations: |
| lines.append(f" - {rec}") |
| lines.append("") |
|
|
| if self.retrieved_chunks: |
| lines.append("Retrieved Context (top chunks):") |
| for ch in self.retrieved_chunks[:5]: |
| sec = f" [{ch['section']}]" if ch.get("section") else "" |
| lines.append(f" - ({ch['source_file']}{sec}, score={ch['score']}) {ch['preview']}...") |
| lines.append("") |
|
|
| if self.graph_nodes: |
| lines.append("Graph Nodes Used:") |
| lines.append(" " + ", ".join(f"{g['name']} ({g['type']})" for g in self.graph_nodes)) |
| lines.append("") |
|
|
| if self.supporting_sources: |
| lines.append("Sources & References:") |
| for citation in self.supporting_sources: |
| lines.append(f" [{citation['marker']}] {_format_source_label(citation)}") |
| else: |
| lines.append("Sources & References: no supporting evidence was retrieved for this query.") |
| lines.append("") |
|
|
| lines.append(f"Confidence Level: {self.confidence_score}") |
| lines.append("") |
| lines.append("Disclaimer: this is general educational information grounded in retrieved " |
| "evidence, not a diagnosis or a substitute for care from a licensed clinician. " |
| "If you are in crisis, contact emergency services or a crisis line immediately.") |
|
|
| return "\n".join(lines) |
|
|
| |
| |
| |
| |
| |
| |
| |
|
|
|
|
| class BridgeSymptomGraphRAG: |
| """ |
| The complete, integrated Bridge-Symptom GraphRAG pipeline, composing every |
| chapter's components -- the explicit 5-route Intent Classifier (Ch.13B), the |
| Clinical Evidence Validator (Ch.9B), guideline-aware ranked reasoning gated to |
| the comorbidity route only (Ch.10), and empirical local-miss -> external |
| fallback (Ch.7) -- behind a single `.answer(query)` interface, reasoning over a |
| knowledge graph that combines a literature-grounded static core with Section |
| 4.3's dynamic, multi-source build-time ingestion. The graph itself never |
| changes AT QUERY TIME -- `.answer()` only ever reads `self.kg`. |
| """ |
|
|
| def __init__(self, kg: ClinicalKnowledgeGraph, retriever: HybridRetriever, |
| router: QueryRouter, generator: AnswerGenerator): |
| self.kg = kg |
| self.retriever = retriever |
| self.router = router |
| self.generator = generator |
|
|
| def answer(self, query: str, top_k: int = 5, |
| disease_pair: Tuple[str, str] = None) -> PipelineOutput: |
| """ |
| Run the full pipeline for a natural-language clinical question. |
| |
| Args: |
| query (str): The user's question. |
| top_k (int): Number of local chunks to retrieve. |
| disease_pair (Tuple[str, str]): Node IDs used for bridge-symptom chain search |
| when the router detects a reasoning-oriented query. Defaults to None, which |
| resolves to the dynamically-determined DEFAULT_DISEASE_PAIR (Chapter 10) -- |
| the graph no longer has fixed, predictable IDs for GAD/MDD to hardcode here. |
| |
| Returns: |
| PipelineOutput: Final structured result (answer, sources, reasoning |
| path, evidence ranking, confidence score, safety report). |
| """ |
| if is_crisis_query(query): |
| logger.warning("Crisis-pattern query detected: bypassing retrieval/generation for safety.") |
| return PipelineOutput( |
| answer=build_crisis_response(), supporting_sources=[], |
| reasoning_path="Safety override: retrieval and generation were bypassed because this query " |
| "matched a suicide/self-harm crisis pattern.", |
| evidence_ranking=[], confidence_score="N/A (Safety Override)", |
| safety_report={"unsupported_sentences": [], "contradiction_count": 0, |
| "drug_interaction_warnings": [], "missing_evidence": True, |
| "clinical_consistency_score": 1.0, "confidence_level": "N/A (Safety Override)", |
| "low_confidence_warning": False, "crisis_override": True}, |
| intent_route="other", retrieval_strategy="A", |
| ) |
|
|
| decision = self.router.analyze(query) |
| |
| |
| |
| |
| |
| intent = INTENT_CLASSIFIER.classify(query, decision) |
| local_results = self.retriever.retrieve(query, top_k=top_k) if "local" in decision["modes"] else [] |
|
|
| |
| |
| if "external" not in decision["modes"] and QueryRouter.is_local_miss(local_results): |
| logger.info("Local retrieval confidence below threshold; falling back to external sources.") |
| decision["modes"].append("external") |
|
|
| logger.info("Pipeline routing decision: modes=%s | intent route=%s (%s)", |
| decision["modes"], intent["route"], intent["retrieval_strategy"]) |
| external_results = retrieve_external_evidence(query) if "external" in decision["modes"] else [] |
| |
| |
| |
|
|
| reasoning_text, guideline_supported, reasoning_anchor = "", False, "" |
| matched_nodes = [] |
| if intent["graph_enabled"]: |
| |
| |
| |
| matched_nodes = resolve_entities_to_kg_nodes(self.kg, query, decision["entities"], KG_NAME_LOOKUP) |
| reasoning_result = find_query_adaptive_reasoning(self.kg, matched_nodes, disease_pair) |
| reasoning_text = reasoning_result["explanation"] |
| guideline_supported = reasoning_result["guideline_supported"] |
| reasoning_anchor = reasoning_result["anchor"] |
| logger.info("Reasoning anchor for query %r: %s (matched %d KG node(s))", |
| query, reasoning_anchor, len(matched_nodes)) |
|
|
| evidence = merge_evidence(local_results, external_results, reasoning_text) |
| evidence = deduplicate_evidence(evidence) |
| contradictions = detect_contradictions(evidence) |
| compute_confidence(evidence) |
| evidence = rank_evidence(evidence) |
|
|
| validation = run_clinical_evidence_validation(evidence, contradictions) |
| validated_evidence = validation["validated_evidence"] |
| context, citations = build_clinical_context(validated_evidence, query=query) |
|
|
| |
| |
| sufficiency = assess_evidence_sufficiency(validated_evidence, reasoning_text) |
| if not sufficiency["sufficient"]: |
| logger.info("Evidence-sufficiency gate: withholding synthesis (%s).", sufficiency["reason"]) |
| answer_text = ("I don't have enough reliable evidence in the available sources to answer this " |
| "specific question confidently. Please consult a licensed clinician, or try " |
| "rephrasing your question with more specific clinical terms.") |
| else: |
| answer_text = self.generator.generate(query, context, reasoning_text, citations, validated_evidence) |
| safety = run_safety_verification(answer_text, validated_evidence, contradictions, |
| conflict_report=validation["conflict_report"], query=query) |
| safety["conflict_report"] = validation["conflict_report"] |
|
|
| reasoning_display = reasoning_text or "No multi-hop reasoning path was required for this query." |
| if reasoning_text and guideline_supported: |
| reasoning_display += " (This pathway is supported by clinical guideline-level evidence.)" |
| if reasoning_text and reasoning_anchor == "general_fallback": |
| reasoning_display += (" (Note: this is the general GAD-MDD bridge-symptom relationship -- " |
| "your question didn't reference a specific symptom, medication, or " |
| "treatment this graph could anchor a more targeted path to.)") |
|
|
| retrieved_chunks_view = [{"chunk_id": r.get("chunk_id"), "source_file": r.get("source_file", ""), |
| "section": r.get("section", ""), |
| "score": round(r.get("rerank_score", r.get("score", 0.0)), 4), |
| "preview": r.get("text", "")[:160]} for r in local_results] |
| graph_nodes_view = [{"node_id": n, "name": self.kg.graph.nodes[n].get("canonical_name", n), |
| "type": self.kg.graph.nodes[n].get("node_type", "")} |
| for n in matched_nodes if self.kg.graph.has_node(n)] |
|
|
| return PipelineOutput( |
| answer=answer_text, |
| supporting_sources=citations, |
| reasoning_path=reasoning_display, |
| evidence_ranking=[{"evidence_id": e.evidence_id, "source_type": e.source_type, |
| "evidence_level": e.evidence_level, "reliability_score": e.reliability_score} |
| for e in validated_evidence], |
| confidence_score=safety["confidence_level"], |
| safety_report=safety, |
| intent_route=intent["route"], |
| retrieval_strategy=intent["retrieval_strategy"], |
| intent_category=_fine_intent_category(query), |
| retrieved_chunks=retrieved_chunks_view, |
| graph_nodes=graph_nodes_view, |
| ) |
|
|
|
|
|
|
