Air-Gapped HIPAA-Compliant On-Premise Document Intelligence & GraphRAG Pipeline

import numpy as np
from typing import List, Dict

class ClinicalGraphRAG:
    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, patient_id: str, clinical_question: str) -> Dict:
        # 1. Generate local dense embedding (zero cloud calls)
        query_vector = self.embedder.embed(clinical_question)
        
        # 2. Vector search with strict patient tenancy filter
        vector_results = self.qdrant.search(
            collection_name="clinical_records",
            query_vector=query_vector,
            query_filter={"must": [{"key": "patient_id", "match": {"value": patient_id}}]},
            limit=25
        )
        
        # 3. Graph traversal: Extract linked SNOMED-CT diagnoses & medications
        with self.neo4j.session() as session:
            graph_facts = session.run(
                """
                MATCH (p:Patient {id: $pid})-[:DIAGNOSED_WITH]->(d:Condition)-[:INDICATES]->(m:Medication)
                RETURN d.name AS condition, collect(m.name) AS medications
                """,
                pid=patient_id
            ).data()
            
        # 4. Cross-encoder reranking
        candidate_docs = [hit.payload["text"] for hit in vector_results]
        ranked_docs = self.reranker.rank(clinical_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]]
        }