Air-Gapped Document Intelligence & GraphRAG Pipeline

import numpy as np
from typing import List, Dict

class DomainGraphRAG:
    def __init__(self, qdrant_client, neo4j_driver, local_embedder, local_reranker):
        self.qdrant = qdrant_client
        self.neo4j = neo4j_driver
        self.embedder = local_embedder
        self.reranker = local_reranker

    def query(self, record_id: str, question: str) -> Dict:
        # 1. Generate local dense embedding (zero cloud calls)
        query_vector = self.embedder.embed(question)

        # 2. Vector search with strict tenancy filter
        vector_results = self.qdrant.search(
            collection_name="domain_records",
            query_vector=query_vector,
            query_filter={"must": [{"key": "record_id", "match": {"value": record_id}}]},
            limit=25
        )

        # 3. Graph traversal: extract linked domain entities & relationships
        with self.neo4j.session() as session:
            graph_facts = session.run(
                """
                MATCH (e:Entity {id: $rid})-[:RELATED_TO]->(c:Concept)-[:LINKS_TO]->(r:Reference)
                RETURN c.name AS concept, collect(r.name) AS references
                """,
                rid=record_id
            ).data()

        # 4. Cross-encoder reranking
        candidate_docs = [hit.payload["text"] for hit in vector_results]
        ranked_docs = self.reranker.rank(question, candidate_docs, top_k=5)

        return {
            "grounded_context": ranked_docs,
            "graph_ontology": graph_facts,
            "provenance_hashes": [hit.payload["sha256"] for hit in vector_results[:5]]
        }