feat(qa): normalized 图 schema 支持(收编容器 normalized 后端)

- src/rag/normalized_release.py:Phase 2B normalized 只读适配器(32 seed/200 result)
- backend_guard:normalized 后端身份校验(Qdrant 12963472 点/Neo4j 精确计数)
- bounded_canonical/direct_flow:normalized one-hop 分支
- pipeline/api/config:graph_schema ∈ {canonical, normalized}

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}

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,

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}"

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

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,

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
         )

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