diff --git a/src/rag/normalized_release.py b/src/rag/normalized_release.py new file mode 100644 index 0000000..4d2ed42 --- /dev/null +++ b/src/rag/normalized_release.py @@ -0,0 +1,282 @@ +from __future__ import annotations + +import os +from dataclasses import dataclass +from typing import Any, Iterable + + +EXPECTED_RELEASE = "plant_srt_pc2025_v3_20260817_normalized" +MAX_SEEDS = 32 +MAX_RESULTS = 200 + + +def _text(value: Any) -> str: + return "" if value is None else str(value).strip() + + +def _dedupe(values: Iterable[Any], *, normalize: bool = False) -> list[str]: + output: list[str] = [] + seen: set[str] = set() + for value in values: + text = " ".join(_text(value).split()) + if normalize: + text = text.casefold() + if not text or text in seen: + continue + seen.add(text) + output.append(text) + if len(output) >= MAX_SEEDS: + break + return output + + +@dataclass(frozen=True) +class NormalizedReleaseSettings: + release_id: str + qdrant_collection: str + + @classmethod + def from_env(cls) -> "NormalizedReleaseSettings": + release_id = os.environ.get("NORMALIZED_RELEASE_ID", "").strip() + collection = os.environ.get("V15_QDRANT_COLLECTION", "").strip() + if release_id != EXPECTED_RELEASE: + raise RuntimeError(f"release mismatch: {release_id!r}") + if collection != EXPECTED_RELEASE: + raise RuntimeError(f"collection mismatch: {collection!r}") + return cls(release_id=release_id, qdrant_collection=collection) + + +class NormalizedGraphAdapter: + """Read-only, bounded access to the normalized Phase 2B Neo4j schema.""" + + _ONE_HOP = """ + MATCH (seed:Entity) + WHERE seed.canonical_id IN $canonical_ids + OR seed.name_norm IN $name_norms + MATCH (seed)-[r:ASSERTS]-(other:Entity) + WITH DISTINCT startNode(r) AS source, + endNode(r) AS target, + r + OPTIONAL MATCH (source)-[support:SUPPORTED_BY]->(paper:Paper) + WHERE support.assertion_id = r.assertion_id + RETURN r.assertion_id AS assertion_id, + source.name AS source, + source.entity_type AS source_type, + source.canonical_id AS source_canonical_id, + r.relation_norm AS relation, + r.relation_category AS relation_category, + target.name AS target, + target.entity_type AS target_type, + target.canonical_id AS target_canonical_id, + r.layer AS layer, + paper.paper_id AS paper_id, + paper.pmid AS pmid, + paper.pmcid AS pmcid, + paper.doi AS doi, + paper.title AS title + LIMIT $limit + """ + + _IDENTITY = """ + MATCH (entity:Entity) + WITH count(entity) AS entity_count + MATCH (paper:Paper) + RETURN entity_count, + count(paper) AS paper_count + """ + + def __init__(self, driver: Any) -> None: + self.driver = driver + + def identity(self) -> dict[str, Any]: + with self.driver.session() as session: + record = session.run(self._IDENTITY).single() + if record is None: + raise RuntimeError("normalized graph identity query returned no row") + result = dict(record) + result["release_id"] = EXPECTED_RELEASE + result["graph_schema"] = "normalized" + return result + + def resolve_entities( + self, + *, + canonical_ids: Iterable[Any], + names: Iterable[Any], + limit: int = MAX_SEEDS, + ) -> list[dict[str, Any]]: + ids = _dedupe(canonical_ids) + name_norms = _dedupe(names, normalize=True) + if not ids and not name_norms: + return [] + bounded_limit = max(1, min(int(limit or 1), MAX_RESULTS)) + statement = """ + MATCH (entity:Entity) + WHERE entity.canonical_id IN $canonical_ids + OR entity.name_norm IN $name_norms + RETURN entity.canonical_id AS canonical_id, + entity.name AS name, + entity.name_norm AS name_norm, + entity.entity_type AS entity_type, + entity.layer AS layer + LIMIT $limit + """ + with self.driver.session() as session: + records = session.run( + statement, + canonical_ids=ids, + name_norms=name_norms, + limit=bounded_limit, + ) + return [dict(record) for record in records] + + def one_hop( + self, + *, + canonical_ids: Iterable[Any], + names: Iterable[Any], + limit: int, + ) -> list[dict[str, Any]]: + ids = _dedupe(canonical_ids) + name_norms = _dedupe(names, normalize=True) + if not ids and not name_norms: + return [] + bounded_limit = max(1, min(int(limit or 1), MAX_RESULTS)) + with self.driver.session() as session: + records = session.run( + self._ONE_HOP, + canonical_ids=ids, + name_norms=name_norms, + limit=bounded_limit, + ) + output = [dict(record) for record in records] + for row in output: + row.setdefault("relation_id", _text(row.get("assertion_id"))) + row.setdefault("evidence", "") + row.setdefault("source_system", "V15_V16_canonical") + row.setdefault("kb_version", EXPECTED_RELEASE) + row.setdefault("output_layer", _text(row.get("layer"))) + return output + + def entity_type_counts(self) -> list[dict[str, Any]]: + statement = """ + MATCH (entity:Entity) + RETURN entity.entity_type AS entity_type, + count(entity) AS count + ORDER BY count DESC + """ + with self.driver.session() as session: + return [dict(record) for record in session.run(statement)] + + def relationship_counts(self) -> list[dict[str, Any]]: + statement = """ + MATCH ()-[relationship]->() + WHERE type(relationship) IN ['ASSERTS', 'SUPPORTED_BY'] + RETURN type(relationship) AS rel_type, + count(relationship) AS count + ORDER BY count DESC + """ + with self.driver.session() as session: + return [dict(record) for record in session.run(statement)] + + def list_entities( + self, + *, + entity_types: Iterable[Any], + limit: int, + offset: int, + ) -> list[dict[str, Any]]: + types = _dedupe(entity_types) + if not types: + return [] + bounded_limit = max(1, min(int(limit or 1), 1000)) + bounded_offset = max(0, min(int(offset or 0), 1_000_000)) + statement = """ + MATCH (entity:Entity) + WHERE entity.entity_type IN $entity_types + RETURN entity.canonical_id AS canonical_id, + entity.name AS name, + entity.name_norm AS name_norm, + entity.entity_type AS entity_type, + entity.layer AS layer + ORDER BY entity.name_norm, entity.canonical_id + SKIP $offset + LIMIT $limit + """ + with self.driver.session() as session: + records = session.run( + statement, + entity_types=types, + offset=bounded_offset, + limit=bounded_limit, + ) + return [dict(record) for record in records] + + def search_entities( + self, + query: str, + *, + entity_types: Iterable[Any], + limit: int, + ) -> list[dict[str, Any]]: + query_norm = " ".join(_text(query).casefold().split()) + if not query_norm: + return [] + types = _dedupe(entity_types) + bounded_limit = max(1, min(int(limit or 1), 100)) + statement = """ + MATCH (entity:Entity) + WHERE ($all_types OR entity.entity_type IN $entity_types) + AND ( + entity.name_norm = $query_norm + OR entity.name_norm STARTS WITH $query_norm + ) + RETURN entity.canonical_id AS canonical_id, + entity.name AS name, + entity.name_norm AS name_norm, + entity.entity_type AS entity_type, + entity.layer AS layer + ORDER BY CASE WHEN entity.name_norm = $query_norm THEN 0 ELSE 1 END, + entity.name_norm, + entity.canonical_id + LIMIT $limit + """ + with self.driver.session() as session: + records = session.run( + statement, + all_types=not types, + entity_types=types, + query_norm=query_norm, + limit=bounded_limit, + ) + return [dict(record) for record in records] + + def entity_detail( + self, + name: str, + *, + entity_types: Iterable[Any], + neighbor_limit: int, + ) -> dict[str, Any] | None: + matches = self.search_entities( + name, + entity_types=entity_types, + limit=10, + ) + query_norm = " ".join(_text(name).casefold().split()) + entity = next( + ( + row + for row in matches + if _text(row.get("name_norm")).casefold() == query_norm + ), + None, + ) + if entity is None: + return None + relations = self.one_hop( + canonical_ids=[entity.get("canonical_id")], + names=[], + limit=neighbor_limit, + ) + return {"entity": entity, "relations": relations} diff --git a/src/rag/qa_v1_1_73/api.py b/src/rag/qa_v1_1_73/api.py index ca912a2..00cc848 100644 --- a/src/rag/qa_v1_1_73/api.py +++ b/src/rag/qa_v1_1_73/api.py @@ -93,6 +93,13 @@ def info() -> Dict[str, Any]: "second_round": config.enable_second_round, "bge_rerank": config.enable_bge_rerank, "graph_schema": config.graph_schema, + "backend": { + "release_id": os.environ.get("NORMALIZED_RELEASE_ID", ""), + "qdrant_collection": os.environ.get( + "V15_QDRANT_COLLECTION", "" + ), + "graph_schema": config.graph_schema, + }, "knowledge_chain_defaults": { "chain_top_k": 3, "chain_hops": 2, diff --git a/src/rag/qa_v1_1_73/backend_guard.py b/src/rag/qa_v1_1_73/backend_guard.py index 63777ef..3fa5b57 100644 --- a/src/rag/qa_v1_1_73/backend_guard.py +++ b/src/rag/qa_v1_1_73/backend_guard.py @@ -7,10 +7,13 @@ from urllib.request import Request, urlopen from neo4j import GraphDatabase +from src.rag.normalized_release import EXPECTED_RELEASE as NORMALIZED_RELEASE + ALLOWED_QDRANT_COLLECTIONS = { "plant_srt_direct_v1_bge_m3_20260702", "plant_srt_direct_v1_bge_m3_citations_20260717", + NORMALIZED_RELEASE, } REQUIRED_NODE_LABELS = { "SRT2Paper", @@ -44,6 +47,10 @@ EXPECTED_NEO4J_CANONICAL_ENTITIES = 3_820_996 EXPECTED_NEO4J_CANONICAL_ASSERTIONS = 8_246_875 EXPECTED_NEO4J_CANONICAL_ASSERTS = 8_246_875 EXPECTED_NEO4J_SUPPORTED_BY = 8_202_284 +EXPECTED_NORMALIZED_ENTITIES = 6_316_654 +EXPECTED_NORMALIZED_PAPERS = 96_648 +EXPECTED_NORMALIZED_ASSERTS = 12_963_472 +EXPECTED_NORMALIZED_SUPPORTED_BY = 13_012_368 class BackendIdentityError(RuntimeError): @@ -70,8 +77,126 @@ def _qdrant_sample(url: str, collection: str) -> Dict[str, Any]: return dict(points[0].get("payload") or {}) if points else {} +def _validate_normalized_backends() -> Dict[str, Any]: + release_id = os.environ.get("NORMALIZED_RELEASE_ID", "").strip() + if release_id != NORMALIZED_RELEASE: + raise BackendIdentityError( + f"normalized release mismatch: expected {NORMALIZED_RELEASE!r}, " + f"got {release_id!r}" + ) + collection = os.environ.get("V15_QDRANT_COLLECTION", "").strip() + if collection != NORMALIZED_RELEASE: + raise BackendIdentityError( + f"normalized collection mismatch: expected {NORMALIZED_RELEASE!r}, " + f"got {collection!r}" + ) + qdrant_url = os.environ.get( + "V15_QDRANT_URL", os.environ.get("QDRANT_URL", "http://127.0.0.1:6333") + ).rstrip("/") + qdrant_info = _qdrant_json( + f"{qdrant_url}/collections/{collection}" + ).get("result", {}) + qdrant_points = int(qdrant_info.get("points_count") or 0) + if qdrant_points != EXPECTED_NORMALIZED_ASSERTS: + raise BackendIdentityError( + "normalized Qdrant identity mismatch: " + f"expected {EXPECTED_NORMALIZED_ASSERTS} points, got {qdrant_points}" + ) + sample_payload = _qdrant_sample(qdrant_url, collection) + if sample_payload.get("kb_version") != NORMALIZED_RELEASE: + raise BackendIdentityError( + "normalized Qdrant payload has the wrong kb_version" + ) + + neo4j_uri = os.environ.get("NEO4J_URI", "bolt://127.0.0.1:19788") + neo4j_user = os.environ.get("NEO4J_USER", "neo4j") + neo4j_password = os.environ.get("NEO4J_PASSWORD", "") + driver = GraphDatabase.driver( + neo4j_uri, + auth=(neo4j_user, neo4j_password), + connection_timeout=15, + ) + try: + with driver.session() as session: + labels = set( + session.run( + "CALL db.labels() YIELD label RETURN label" + ).value("label") + ) + relationship_types = set( + session.run( + "CALL db.relationshipTypes() YIELD relationshipType " + "RETURN relationshipType" + ).value("relationshipType") + ) + entity_count = int( + session.run( + "MATCH (entity:Entity) RETURN count(entity) AS count" + ).single()["count"] + ) + paper_count = int( + session.run( + "MATCH (paper:Paper) RETURN count(paper) AS count" + ).single()["count"] + ) + asserts_count = int( + session.run( + "MATCH ()-[r:ASSERTS]->() RETURN count(r) AS count" + ).single()["count"] + ) + supported_by_count = int( + session.run( + "MATCH ()-[r:SUPPORTED_BY]->() RETURN count(r) AS count" + ).single()["count"] + ) + finally: + driver.close() + + if labels != {"Entity", "Paper"}: + raise BackendIdentityError( + f"normalized Neo4j labels mismatch: {sorted(labels)}" + ) + if not {"ASSERTS", "SUPPORTED_BY"}.issubset(relationship_types): + raise BackendIdentityError( + "normalized Neo4j relationship types mismatch: " + f"{sorted(relationship_types)}" + ) + counts = ( + entity_count, + paper_count, + asserts_count, + supported_by_count, + ) + expected = ( + EXPECTED_NORMALIZED_ENTITIES, + EXPECTED_NORMALIZED_PAPERS, + EXPECTED_NORMALIZED_ASSERTS, + EXPECTED_NORMALIZED_SUPPORTED_BY, + ) + if counts != expected: + raise BackendIdentityError( + f"normalized Neo4j counts mismatch: expected={expected}, got={counts}" + ) + return { + "backend_policy": "neo4j_qdrant_only", + "release_id": release_id, + "graph_schema": "normalized", + "neo4j_uri": neo4j_uri, + "neo4j_store": NORMALIZED_RELEASE, + "neo4j_entities": entity_count, + "neo4j_papers": paper_count, + "neo4j_asserts": asserts_count, + "neo4j_supported_by": supported_by_count, + "qdrant_url": qdrant_url, + "qdrant_collection": collection, + "qdrant_points": qdrant_points, + } + + def validate_backends() -> Dict[str, Any]: graph_schema = os.environ.get("QA_V1_GRAPH_SCHEMA", "canonical").strip().lower() + if graph_schema == "normalized": + return _validate_normalized_backends() if graph_schema != "canonical": raise BackendIdentityError( f"QA_V1_GRAPH_SCHEMA must be canonical, got {graph_schema!r}" diff --git a/src/rag/qa_v1_1_73/bounded_canonical_retriever.py b/src/rag/qa_v1_1_73/bounded_canonical_retriever.py index 9a69d50..993f1c9 100644 --- a/src/rag/qa_v1_1_73/bounded_canonical_retriever.py +++ b/src/rag/qa_v1_1_73/bounded_canonical_retriever.py @@ -9,6 +9,8 @@ from neo4j import Query from qdrant_client import QdrantClient from qdrant_client.http import models +from src.rag.normalized_release import NormalizedGraphAdapter + from .direct_flow_retriever import DirectFlowRetriever, _norm, _text from .graph_path_chain import NativeGraphPathRetriever from .schemas import EvidenceItem, RetrievalHypothesis @@ -372,6 +374,31 @@ class BoundedCanonicalDirectFlowRetriever( *, method: str, ) -> List[EvidenceItem]: + if self.graph_schema == "normalized": + rows = NormalizedGraphAdapter(self.driver).one_hop( + canonical_ids=canonical_ids, + names=names, + limit=limit, + ) + output: List[EvidenceItem] = [] + for payload in rows: + assertion_id = _text( + payload.get("assertion_id") or payload.get("relation_id") + ) + if assertion_id in self._bounded_seen_assertions: + continue + self._bounded_seen_assertions.add(assertion_id) + item = self.item_from_payload( + payload, + "neo4j_normalized_one_hop", + "normalized_seed", + ) + item.vector_score = float(payload.get("quality_score") or 0.0) + item.rrf_score = max(item.rrf_score, item.vector_score) + output.append(item) + if len(output) >= min(max(1, int(limit or 1)), 200): + break + return output ids = [ value for value in canonical_ids diff --git a/src/rag/qa_v1_1_73/config.py b/src/rag/qa_v1_1_73/config.py index 6317309..6e482c6 100644 --- a/src/rag/qa_v1_1_73/config.py +++ b/src/rag/qa_v1_1_73/config.py @@ -35,9 +35,9 @@ class QAConfig: graph_schema = os.environ.get( "QA_V1_GRAPH_SCHEMA", "canonical" ).strip().lower() - if graph_schema != "canonical": + if graph_schema not in {"canonical", "normalized"}: raise RuntimeError( - "QA V1.1.73 production only permits the canonical graph schema" + "QA V1.1.73 production permits canonical or normalized graph schema" ) return cls( project_root=root, diff --git a/src/rag/qa_v1_1_73/direct_flow_retriever.py b/src/rag/qa_v1_1_73/direct_flow_retriever.py index 83f5ca6..ca793a4 100644 --- a/src/rag/qa_v1_1_73/direct_flow_retriever.py +++ b/src/rag/qa_v1_1_73/direct_flow_retriever.py @@ -10,11 +10,17 @@ import requests from neo4j import GraphDatabase from qdrant_client import QdrantClient +from src.rag.normalized_release import ( + EXPECTED_RELEASE as NORMALIZED_RELEASE, + NormalizedReleaseSettings, +) + from .schemas import EvidenceItem, RetrievalHypothesis ALLOWED_QDRANT_COLLECTIONS = { "plant_srt_direct_v1_bge_m3_20260702", "plant_srt_direct_v1_bge_m3_citations_20260717", + NORMALIZED_RELEASE, } @@ -64,7 +70,7 @@ class DirectFlowRetriever: self.graph_schema = ( graph_schema or os.environ.get("QA_V1_GRAPH_SCHEMA", "canonical") ).strip().lower() - if self.graph_schema != "canonical": + if self.graph_schema not in {"canonical", "normalized"}: raise RuntimeError( "Legacy Entity/RELATION and PCEntity/PC_REL graphs are disabled" ) @@ -75,6 +81,12 @@ class DirectFlowRetriever: raise RuntimeError( f"Qdrant collection {self.collection!r} is not allowed in production" ) + if self.graph_schema == "normalized": + NormalizedReleaseSettings.from_env() + if self.collection != NORMALIZED_RELEASE: + raise RuntimeError( + "normalized graph schema requires the matched normalized collection" + ) self.embed_url = os.environ.get("QA_V112_EMBED_URL", os.environ.get("QA_V1_EMBED_URL", "http://127.0.0.1:11434/api/embed")) self.embed_model = os.environ.get("QA_V112_EMBED_MODEL", os.environ.get("QA_V1_EMBED_MODEL", "bge-m3:567m")) self.neo4j_uri = os.environ.get("NEO4J_URI", "bolt://127.0.0.1:18788") @@ -106,7 +118,10 @@ class DirectFlowRetriever: use_neo4j: bool = True, use_graph_expansion: bool = True, ) -> Tuple[List[EvidenceItem], Dict[str, Any]]: - effective_use_neo4j = bool(use_neo4j) and self.graph_schema == "canonical" + effective_use_neo4j = bool(use_neo4j) and self.graph_schema in { + "canonical", + "normalized", + } effective_use_graph_expansion = ( bool(use_graph_expansion) and effective_use_neo4j ) diff --git a/src/rag/qa_v1_1_73/pipeline.py b/src/rag/qa_v1_1_73/pipeline.py index cb539e0..2081da9 100644 --- a/src/rag/qa_v1_1_73/pipeline.py +++ b/src/rag/qa_v1_1_73/pipeline.py @@ -258,7 +258,7 @@ class QAV1Pipeline: # Anchor-only controls which entities may seed graph traversal. It must # never disable Neo4j retrieval or native graph-chain construction. direct_neo4j = ( - graph_schema == "canonical" + graph_schema in {"canonical", "normalized"} and bool(use_neo4j) and bool(graph_plan.get("enabled", True)) ) @@ -962,12 +962,13 @@ class QAV1Pipeline: direct_flow_switches = { "use_qdrant": bool(options.get("use_qdrant", True)), "use_neo4j": ( - child_use_neo4j and self.config.graph_schema == "canonical" + child_use_neo4j + and self.config.graph_schema in {"canonical", "normalized"} ), "use_graph_expansion": ( bool(options.get("use_graph_expansion", True)) and child_use_neo4j - and self.config.graph_schema == "canonical" + and self.config.graph_schema in {"canonical", "normalized"} ), } if self.agentic_controller is not None: @@ -1057,8 +1058,10 @@ class QAV1Pipeline: chains: List[Dict[str, Any]] = [] native_audit: Dict[str, Any] = {"status": "disabled"} seed_audit: Dict[str, Any] = {} - if self.config.graph_schema == "legacy": - native_audit["status"] = "disabled_for_legacy_schema" + if self.config.graph_schema in {"legacy", "normalized"}: + native_audit["status"] = ( + f"disabled_for_{self.config.graph_schema}_schema" + ) elif bool(options.get("native_chain_enabled", True)) and bool( direct_flow_switches["use_neo4j"] ): @@ -1631,8 +1634,10 @@ class QAV1Pipeline: "status": "not_attempted", "fallback_to_native_seed_expansion": False, } - if self.config.graph_schema == "legacy": - native_graph_chain_audit["status"] = "disabled_for_legacy_schema" + if self.config.graph_schema in {"legacy", "normalized"}: + native_graph_chain_audit["status"] = ( + f"disabled_for_{self.config.graph_schema}_schema" + ) elif native_chain_enabled and ( direct_flow_switches["use_neo4j"] or endpoint_path_neo4j_enabled