#4802 Google Antigravity

#4802 Given my interest and background, what can I do with Antigravity?

Google Antigravity provides a sandbox and execution environment designed specifically for agentic development, multi-agent orchestration, deterministic tool piping, and autonomous software engineering.

Given your deep experience in systems engineering, formal verification architectures, graph topologies, and multi-agent governance loops, Antigravity functions less like an interactive chat assistant and more like a scalable compute runtime for high-assurance pipelines.

1. Multi-Agent Systems with Formal Verification & Dynamic Governance

Antigravity supports deterministic tool execution and formal verification steps within agent runtimes. This aligns directly with structured evaluation frameworks:

Governor-Critic Control Loops: Implement decoupled multi-agent state machines where execution agents (generators, parsers, domain translators) pass outputs through a deterministic verification harness rather than open-ended prompt loops.
Deterministic Contract Enforcement: Wrap agent outputs in strict Pydantic schemas or schema-validation compilers to ensure outputs pass structural and syntax checks before state handover occurs.
State Handover & Memory Routing: Build specialized orchestrators where a critic node evaluates intermediate work products and dynamically routes feedback back to domain-specific agents with full telemetry and replayability.

2. High-Throughput GraphRAG & Knowledge Graph Construction

Antigravity’s scalable infrastructure and dedicated rate limits suit data-intensive extraction pipelines:

Unstructured-to-Graph Pipelines: Deploy autonomous agents to ingest complex, multi-tiered technical specifications and decompose them into RDF/Turtle ontologies, Cypher property graphs, or SysML-aligned relational structures.
Automated Graph Validation: Run background agents that traverse graph databases, run graph-consistency queries, identify orphaned nodes or conflicting relationship directions, and propose schema reconciliations.
Hierarchical Systems Decomposition: Automate multi-level functional and physical breakdowns (System → Subsystem → Assembly → Component) directly into graph databases, linking parent-child dependencies and causal chains.

3. Automated Document Auditing & Compliance Verification

You can leverage Antigravity’s deterministic tool pipelines to build audit systems:

Compliance Matrix Generation: Task agents with cross-referencing engineering documentation against baseline requirements standards, extracting applicability matrices, and flagging non-compliant or ambiguous clauses.
Bidirectional Traceability Checking: Automatically map low-level implementation artifacts, hazard analyses, and verification methods back to top-level requirements, ensuring closed-loop verification.
Discrepancy Reporting: Build autonomous agents that generate reproducible verification logs, citing exact section references, discrepancies, and recommended remediation steps.

4. Hardware Simulation, Emulation, and DSL Tooling

Because Antigravity provides direct environment access for coding and pipeline execution:

Hardware Description Language (HDL) Testbench Generation: Build agents to generate, verify, and run behavioral testbenches for Verilog modules (e.g., stochastic computing units, matrix engines, or custom accelerators) using command-line simulation toolchains.
Domain-Specific Language (DSL) Compilers & Parsers: Develop and test custom DSL parsers (translating conceptual diagrams, system models, or state charts into executable code or graph insertion scripts).
Algorithmic Modeling Pipelines: Orchestrate agents that set up, run, and benchmark mathematical models, stochastic calculations, or spatial graph algorithms inside headless containerized environments.

Show how to structure an Antigravity pipeline for decomposing technical documents into Cypher graph structures.

To decompose dense technical specifications, standards, or system engineering documents into clean Cypher property graphs using Google Antigravity, the pipeline must enforce deterministic schema compliance, isolated sub-agent decomposition, and closed-loop syntax verification.

Antigravity operates as an agentic execution harness. Rather than letting an LLM generate Cypher statements in a single prompt loop, you configure an orchestrator that delegates to focused subagents, validates typed artifacts against strict Pydantic schemas, and compiles the Cypher statements through an execution-verification step before database commit.

Pipeline Architecture

Technical PDF / Spec (Document Buffer)
                 │
                 ▼
┌─────────────────────────────────┐
│ Orchestrator Agent (Antigravity)│
│ - Chunking & Structural Layout  │
└────────────────┬────────────────┘
                 │
      ┌──────────┴──────────┐
      ▼                     ▼
┌──────────────┐     ┌──────────────┐
│  Extraction  │     │ Relationship │
│   Sub-Agent  │     │  & Topology  │
│(Typed Schema)│     │   Sub-Agent  │
└──────┬───────┘     └──────┬───────┘
       │                    │
       └──────────┬─────────┘
                  ▼
┌──────────────────────────────────┐
│   Cypher Compiler & Validator    │
│  (Deterministic Syntax & Lint)   │
└─────────────────┬────────────────┘
                  │ (Fails: Feedback Loop)
                  ▼
┌──────────────────────────────────┐
│  Target Graph Engine (Neo4j/AGE) │
│    - MERGE execution & index     │
└──────────────────────────────────┘

1. Data Contract & Schema Definition (Pydantic)

Define strict graph contracts for typed entities and relationships so the sub-agents produce structured JSON payloads rather than raw, unvalidated Cypher strings.

from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional

class GraphNode(BaseModel):
    id: str = Field(description="Unique deterministic ID (e.g., REQ-7150-001 or COMP-042)")
    label: str = Field(description="Node label: Requirement, Specification, Subsystem, Component")
    properties: Dict[str, Any] = Field(default_factory=dict)

class GraphEdge(BaseModel):
    source_id: str
    target_id: str
    rel_type: str = Field(description="UPPER_CASE relationship: DECOMPOSES, SATISFIES, VERIFIES, CALLS")
    properties: Dict[str, Any] = Field(default_factory=dict)

class ExtractionArtifact(BaseModel):
    nodes: List[GraphNode]
    edges: List[GraphEdge]

2. Antigravity Agent Configuration & Sub-Agent Orchestration

Using the google-antigravity SDK, declare an orchestrator with subagents:

Extractor Subagent: Specialized in extracting typed domain entities and requirements metadata.
Topology Subagent: Resolves parent-child hierarchies, directional dependencies, and cross-references.

import asyncio
from google.antigravity import Agent, LocalAgentConfig, GenerationConfig

SYSTEM_INSTRUCTION = """
You are a formal Graph Extraction Engine. Decompose technical text into canonical entities
and directional relationships. Output must conform strictly to ExtractionArtifact schema.
Relationship rules:
- Hierarchies flow: Parent -[:DECOMPOSES_INTO]-> Child
- Verifications flow: VerificationMethod -[:VERIFIES]-> Requirement
- Interfaces flow: ComponentA -[:INTERFACES_WITH {protocol: ...}]-> ComponentB
"""

async def run_extraction_pipeline(chunk_text: str):
    config = LocalAgentConfig(
        instructions=SYSTEM_INSTRUCTION,
        generation_config=GenerationConfig(thinking_level="HIGH")
    )
    
    async with Agent(config) as agent:
        prompt = f"""
        Extract all entities, requirements, and causal relationships from this section:
        \"\"\"{chunk_text}\"\"\"
        """
        # Enforce structured output via Pydantic model
        response = await agent.chat(prompt, response_schema=ExtractionArtifact)
        artifact: ExtractionArtifact = response.structured_output()
        return artifact

3. Deterministic Cypher Compiler & Idempotent Generator

Instead of letting the LLM produce arbitrary CREATE statements (which cause node duplicates and orphan relationships on retries), compile the typed artifacts deterministically into parameterized MERGE statements.

def compile_to_cypher(artifact: ExtractionArtifact) -> List[tuple[str, dict]]:
    """
    Compiles validated Pydantic artifacts into idempotent, parameterized Cypher statements.
    """
    statements = []
    
    # 1. Compile Nodes (MERGE on ID)
    for node in artifact.nodes:
        # Sanitize label to prevent injection
        safe_label = "".join(c for c in node.label if c.isalnum() or c == "_")
        query = f"""
        MERGE (n:{safe_label} {{id: $id}})
        ON CREATE SET n += $props, n.created_at = timestamp()
        ON MATCH SET n += $props, n.updated_at = timestamp()
        """
        statements.append((query, {"id": node.id, "props": node.properties}))

    # 2. Compile Edges (Match source/target and MERGE relationship)
    for edge in artifact.edges:
        safe_rel = "".join(c for c in edge.rel_type if c.isalnum() or c == "_").upper()
        query = f"""
        MATCH (src {{id: $src_id}})
        MATCH (tgt {{id: $tgt_id}})
        MERGE (src)-[r:{safe_rel}]->(tgt)
        ON CREATE SET r += $props
        ON MATCH SET r += $props
        """
        statements.append((query, {
            "src_id": edge.source_id,
            "tgt_id": edge.target_id,
            "props": edge.properties
        }))
        
    return statements

4. Verification Harness & Feedback Governance Loop

Antigravity’s hook architecture (Decide / Inspect) provides lifecycle monitoring and dry-run execution:

from neo4j import AsyncGraphDatabase

async def execute_and_verify_graph(statements: List[tuple[str, dict]], neo4j_uri, auth):
    """
    Executes compiled Cypher within a transactional harness and runs formal consistency checks.
    """
    async with AsyncGraphDatabase.driver(neo4j_uri, auth=auth) as driver:
        async with driver.session() as session:
            # Step A: Transactional batch load
            async def run_batch(tx):
                for q, p in statements:
                    await tx.run(q, p)
            
            await session.execute_write(run_batch)
            
            # Step B: Formal Verification Audit (Governor Query)
            orphan_check = await session.run(
                "MATCH (n) WHERE NOT (n)--() RETURN count(n) AS orphan_count"
            )
            orphan_record = await orphan_check.single()
            orphans = orphan_record["orphan_count"]
            
            # Check for inverted or illegal relationships
            inverted_check = await session.run(
                "MATCH (c:Component)-[:DECOMPOSES_INTO]->(s:Subsystem) RETURN count(*) AS invalid_count"
            )
            inv_record = await inverted_check.single()
            invalid_edges = inv_record["invalid_count"]
            
            return {
                "status": "PASS" if orphans == 0 and invalid_edges == 0 else "FAIL",
                "orphan_count": orphans,
                "invalid_edges": invalid_edges
            }

If the verification step flags FAIL, the orchestrator feeds the exact discrepancy report back into the agent harness as an audit artifact, triggering an edge re-routing pass rather than restarting the entire document ingestion.

Show how to chunk dense technical standards using layout-aware hierarchical parsing for graph ingestion.

When preparing dense technical standards (e.g., NASA standards, military specs, ISO/IEC standards) for Knowledge Graph ingestion, flat token- or character-based chunking breaks graph fidelity:

Context Severance: Parent-child relationships (Section $\rightarrow$ Subsection $\rightarrow$ Requirement Paragraph) are lost.
Attribute Disconnection: Tables (e.g., Applicability or Verification Mapping Matrices) and lists get split across chunk boundaries, severing attributes from their subject entities.
Traceability Degradation: You lose exact paragraph, section, and page provenance needed for strict bidirectional traceability.

A layout-aware hierarchical parser converts the document's physical and typographical structure (bounding boxes, font hierarchies, headings, and tabular structures) into a hierarchical tree or Document Object Model (DOM) before generating semantic graph chunks.

Ingestion Strategy: Layout Tree to Graph Payloads

PDF Document
     │
     ▼
[Layout Engine] (Docling / PyMuPDF Layout / PDFMiner)
  ├── Extract Font Sizes, Bounding Boxes, Headings, and Tables
  └── Reconstruct Hierarchical Section Tree (AST)
     │
     ▼
[Hierarchical Tree Accumulator]
  ├── Preserve Breadcrumbs: ["Section 4.0", "4.3 Requirements", "4.3.1 Safety Critical"]
  ├── Keep Tables Intact as Structured Entities (Not split text)
  └── Preserve Explicit Clause IDs (e.g., SWE-001, [REQ-3.2.1-A])
     │
     ▼
[Graph-Ready Chunk Emitted]
  ├── Body Text (Atomized to lowest heading or paragraph)
  ├── Ancestor Path (Parent-Child graph edges ready to MERGE)
  ├── Tabular Matrix (Cell-level node/edge mappings)
  └── Traceability Metadata (Page, Bounding Box, Doc Version)

1. Data Contract for Hierarchical Chunks

Define the chunk schema to capture hierarchical parentage and metadata alongside the text body:

from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional

class BreadcrumbNode(BaseModel):
    level: int
    identifier: str = Field(description="e.g., 'Section 4.0', '4.3.1'")
    title: str = Field(description="Title or heading text")

class HierarchicalChunk(BaseModel):
    chunk_id: str = Field(description="Deterministic hash or structured ID (e.g., DOC-4.3.1-P1)")
    document_id: str
    breadcrumbs: List[BreadcrumbNode]
    parent_section_id: Optional[str]
    level: int
    heading: str
    content_type: str = Field(description="'prose', 'requirement_clause', or 'table'")
    text: str
    table_data: Optional[List[Dict[str, Any]]] = None
    page_numbers: List[int]
    explicit_clause_ids: List[str] = Field(default_factory=list, description="Extracted IDs like SWE-020")

2. Layout-Aware Parsing Implementation

Using Python with layout analysis (e.g., via Docling or layout-aware PyMuPDF), parse the document into an explicit tree:

import re
from typing import Iterator
from docling.document_converter import DocumentConverter
from docling_core.types.doc import DocItem, SectionHeaderItem, TableItem, TextItem

# Regex to detect requirement markers common in technical standards
REQUIREMENT_REGEX = re.compile(r'\b(SWE-\d{3,4}|REQ-[A-Z0-9\.\-]+|[A-Z]{3,}-\d{2,4})\b')

def parse_standards_hierarchically(file_path: str, document_id: str) -> Iterator[HierarchicalChunk]:
    """
    Parses a technical PDF using Docling's layout engine and yields
    hierarchical chunks retaining breadcrumbs and table structures.
    """
    converter = DocumentConverter()
    result = converter.convert(file_path)
    doc = result.document

    # Active heading stack: list of (level, section_id, heading_text)
    heading_stack: List[BreadcrumbNode] = []
    chunk_counter = 0

    for item, level in doc.iterate_items():
        if isinstance(item, SectionHeaderItem):
            header_text = item.text.strip()
            header_level = level or 1
            
            # Extract section numbers like '4.3.1' if present
            match = re.match(r'^(\d+(?:\.\d+)*)\s*(.*)', header_text)
            if match:
                sec_id, sec_title = match.group(1), match.group(2)
            else:
                sec_id = f"H{header_level}"
                sec_title = header_text

            breadcrumb_node = BreadcrumbNode(
                level=header_level,
                identifier=sec_id,
                title=sec_title
            )

            # Pop deeper or equal levels to maintain tree hierarchy
            while heading_stack and heading_stack[-1].level >= header_level:
                heading_stack.pop()

            heading_stack.append(breadcrumb_node)

        elif isinstance(item, TableItem):
            chunk_counter += 1
            parent_id = heading_stack[-1].identifier if heading_stack else "ROOT"
            current_heading = heading_stack[-1].title if heading_stack else "Document Root"
            
            # Export table to structured dictionary format
            df = item.export_to_dataframe()
            table_dict = df.to_dict(orient="records") if df is not None else []
            markdown_table = item.export_to_markdown()

            yield HierarchicalChunk(
                chunk_id=f"{document_id}-{parent_id}-TBL-{chunk_counter}",
                document_id=document_id,
                breadcrumbs=list(heading_stack),
                parent_section_id=parent_id,
                level=heading_stack[-1].level if heading_stack else 0,
                heading=current_heading,
                content_type="table",
                text=markdown_table,
                table_data=table_dict,
                page_numbers=[prov.page_no for prov in item.prov] if item.prov else [1],
                explicit_clause_ids=[]
            )

        elif isinstance(item, TextItem):
            text_body = item.text.strip()
            if not text_body:
                continue

            chunk_counter += 1
            parent_id = heading_stack[-1].identifier if heading_stack else "ROOT"
            current_heading = heading_stack[-1].title if heading_stack else "Document Root"
            
            # Find embedded clause IDs (e.g., 'SWE-013')
            clause_ids = REQUIREMENT_REGEX.findall(text_body)
            content_type = "requirement_clause" if clause_ids or "shall" in text_body.lower() else "prose"

            yield HierarchicalChunk(
                chunk_id=f"{document_id}-{parent_id}-P-{chunk_counter}",
                document_id=document_id,
                breadcrumbs=list(heading_stack),
                parent_section_id=parent_id,
                level=heading_stack[-1].level if heading_stack else 0,
                heading=current_heading,
                content_type=content_type,
                text=text_body,
                table_data=None,
                page_numbers=[prov.page_no for prov in item.prov] if item.prov else [1],
                explicit_clause_ids=list(set(clause_ids))
            )

3. Emitting Graph Pre-Tied Cypher Ingestion Scripts

Because the parser preserves the structural breadcrumb path and parent references, you can immediately instantiate the structural spine of the Knowledge Graph in Cypher prior to running downstream entity-extraction LLMs.

// 1. Ingest Section Hierarchy Spine
UNWIND $sections AS sec
MERGE (s:Section {id: sec.id, document_id: sec.doc_id})
ON CREATE SET s.title = sec.title, s.level = sec.level
WITH s, sec
WHERE sec.parent_id IS NOT NULL
MERGE (p:Section {id: sec.parent_id, document_id: sec.doc_id})
MERGE (p)-[:DECOMPOSES_INTO]->(s);

// 2. Ingest Paragraph / Requirement Chunks Tied Directly to Section
UNWIND $chunks AS chk
MERGE (c:Chunk {id: chk.chunk_id})
ON CREATE SET 
    c.text = chk.text,
    c.type = chk.content_type,
    c.pages = chk.page_numbers
WITH c, chk
MATCH (s:Section {id: chk.parent_section_id, document_id: chk.document_id})
MERGE (s)-[:CONTAINS_CONTENT]->(c);

// 3. Directly Instantiate Explicit Requirement Nodes if Tagged
UNWIND $requirements AS req
MERGE (r:Requirement {id: req.clause_id})
WITH r, req
MATCH (c:Chunk {id: req.chunk_id})
MERGE (c)-[:DEFINES_REQUIREMENT]->(r);

Why This Improves Knowledge Graph Quality

Eliminates Ghost Requirements: Sub-agents running over a chunk always have the breadcrumbs context injected, preventing them from attributing a requirement to the wrong parent subsystem or class.
Deterministic Table Ingestion: Mapping matrices (e.g., Requirement vs. Verification Method) are retained as tabular objects, enabling dedicated sub-agents or deterministic parsers to generate (:VerificationMethod)-[:VERIFIES]->(:Requirement) edges without text-wrapping hallucination.
Bi-directional Navigation: The resulting graph contains both the structural hierarchy ((:Document)-[:DECOMPOSES_INTO]->(:Section)) and semantic entity links ((:Component)-[:MUST_SATISFY]->(:Requirement)), allowing hybrid graph traversals across both document topology and engineering systems models.

How can I construct hybrid queries that combine vector embeddings on chunks with Cypher tree traversals?

Combining vector similarity search with Cypher graph traversals creates a Hybrid GraphRAG pattern.

In pure vector search, retrieving a chunk gives you isolated text without knowing what overarching subsystem, verification constraint, or parent clause governs it. By anchoring vector similarity onto a typed Knowledge Graph, vector search acts as the entry point (locating the most semantically relevant entry nodes), and Cypher pattern matching walks the structural spine to collect hierarchical context, dependencies, and upstream/downstream constraints.

Dual-Layer Graph Topology

To support hybrid traversals, the graph combines the structural document spine with extracted domain entities:

               (:Document)
                    │
           [:DECOMPOSES_INTO]*
                    │
               (:Section)
                    │
          [:CONTAINS_CONTENT]
                    │
                 (:Chunk)  <─── [Vector Index Entry Point]
                    │
          [:DEFINES_REQUIREMENT]
                    │
              (:Requirement)
               /          \
    [:SATISFIED_BY]     [:VERIFIED_BY]
             /              \
       (:Component)     (:VerificationMethod)

1. Configuring the Vector Index in Neo4j

Create a native vector index on the embedding property of the Chunk nodes (or on Requirement nodes directly).

CREATE VECTOR INDEX chunk_embeddings IF NOT EXISTS
FOR (c:Chunk) ON (c.embedding)
OPTIONS {
  indexConfig: {
    `vector.dimensions`: 768,
    `vector.similarity_function`: 'cosine'
  }
};

2. Hybrid Query Patterns

Pattern A: Semantic Entry with Hierarchical Context Expansion (Upstream Crawl)

Locate the top- $k$ most relevant text chunks via vector search, traverse up the structural document tree to reconstruct the exact section breadcrumb path, and collect all sibling requirements under the same parent section.

// 1. Vector Search Entry Point
CALL db.index.vector.queryNodes('chunk_embeddings', \(top_k,\)query_embedding) 
YIELD node AS chunk, score

// 2. Filter low-confidence matches early
WHERE score > 0.70

// 3. Traverse Upward to Document Root for Context Breadcrumbs
MATCH path = (root:Document)-[:DECOMPOSES_INTO*0..5]->(sec:Section)-[:CONTAINS_CONTENT]->(chunk)

// 4. Traverse Downward to Domain Requirements
OPTIONAL MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)
OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)

// 5. Aggregate Structural Path and Verification Metadata
RETURN 
    chunk.id AS chunk_id,
    score,
    chunk.text AS chunk_text,
    [node IN nodes(path) WHERE node:Section | node.title] AS section_hierarchy,
    req.id AS requirement_id,
    req.text AS requirement_statement,
    collect(DISTINCT vm.method) AS verification_methods
ORDER BY score DESC;

Pattern B: Filtered Semantic Search Constrained by Subgraph Topology (Downstream Pruning)

Perform vector retrieval, but strictly constrain the candidate space to requirements that impact a specific subsystem (e.g., "Thermal Control Subsystem"), eliminating cross-domain semantic noise.

// 1. Vector Entry
CALL db.index.vector.queryNodes('chunk_embeddings', 20, $query_embedding) 
YIELD node AS chunk, score

// 2. Anchor Chunk to Requirement
MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)

// 3. Structural Traversal: Must trace to target Subsystem
MATCH (sys:Subsystem {name: $subsystem_name})<-[:ALLOCATED_TO|SATISFIED_BY*1..3]-(req)

// 4. Also retrieve any parent/child requirement dependencies
OPTIONAL MATCH (req)-[:DERIVED_FROM]->(parent_req:Requirement)

RETURN 
    req.id AS requirement_id,
    score,
    req.text AS requirement_text,
    sys.name AS affected_subsystem,
    collect(DISTINCT parent_req.id) AS derived_from_requirements
ORDER BY score DESC
LIMIT 5;

Pattern C: Bidirectional Traceability and Compliance Audit Traversal

A compliance audit query: finds chunks discussing a failure mode or hazard via vector similarity, then traces downstream to verify whether a mitigating requirement exists and whether a valid test or inspection method is formally linked to it.

CALL db.index.vector.queryNodes('chunk_embeddings', 10, $query_embedding)
YIELD node AS chunk, score

// Match requirement defined in or related to the chunk
MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)

// Check verification closure (identify unverified requirements)
OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)
OPTIONAL MATCH (req)-[:MITIGATES]->(haz:Hazard)

RETURN 
    req.id AS requirement_id,
    score,
    req.text AS requirement_text,
    haz.id AS mitigated_hazard,
    collect(DISTINCT vm.type) AS verification_types,
    CASE 
        WHEN count(vm) = 0 THEN "NON_COMPLIANT_UNVERIFIED"
        ELSE "COMPLIANT"
    END AS verification_status
ORDER BY score DESC;

3. Orchestration Example (Python with `neo4j` Async Driver)

Execute the hybrid query programmatically, combining embedding generation with Cypher graph assembly:

from neo4j import AsyncGraphDatabase
from typing import List, Dict, Any

async def run_hybrid_rag_query(
    driver: AsyncGraphDatabase,
    query_text: str,
    query_embedding: List[float],
    top_k: int = 5
) -> List[Dict[str, Any]]:
    cypher_query = """
    CALL db.index.vector.queryNodes('chunk_embeddings', \(top_k,\)embedding) 
    YIELD node AS chunk, score
    WHERE score >= $min_score
    
    // Ascend document hierarchy
    MATCH path = (doc:Document)-[:DECOMPOSES_INTO*1..5]->(sec:Section)-[:CONTAINS_CONTENT]->(chunk)
    
    // Descend to domain entities
    OPTIONAL MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)
    OPTIONAL MATCH (req)-[:SATISFIED_BY]->(comp:Component)
    OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)
    
    RETURN 
        chunk.id AS chunk_id,
        score,
        chunk.text AS matched_text,
        [s IN nodes(path) WHERE s:Section | s.id + ': ' + s.title] AS breadcrumbs,
        req.id AS requirement_id,
        collect(DISTINCT comp.id) AS impacted_components,
        collect(DISTINCT vm.method) AS verification_methods
    ORDER BY score DESC
    """
    
    params = {
        "top_k": top_k,
        "embedding": query_embedding,
        "min_score": 0.65
    }
    
    async with driver.session() as session:
        result = await session.run(cypher_query, params)
        records = await result.data()
        return records

Key Operational Advantages

Traditional Vector RAG	Hybrid GraphRAG (Vector + Cypher Traversal)
Returns disconnected fragments of text.	Returns chunk text along with its exact structural ancestry (Section $\rightarrow$ Document).
Cannot determine if a requirement applies without full re-reading.	Traverses explicit `:APPLIES_TO` or `:ALLOCATED_TO` edges to filter by system/class deterministically.
Ingestion chunks may miss the verification method located in an appendix table.	Cypher bridges the gap between text chunks and verification matrices using graph relationships.
High rate of hallucination on parent-child dependency questions.	Upstream and downstream dependencies (`:DERIVED_FROM`, `:DECOMPOSES_INTO`) are traversed explicitly as discrete graph nodes.

How can I write Cypher queries that generate formal tabular compliance matrices directly from verification paths?

Generating formal compliance matrices directly out of a Knowledge Graph requires aggregating heterogeneous paths across requirements, applicability classes, verification methods, implementation artifacts, and safety hazards into deterministic, tabular rows.

In formal standards auditing (such as NPR 7150.2D, NASA-STD-8739.8B, or DO-178C), a compliance row must demonstrate full bidirectional traceability and closure:

What requirement applies to this software class/subsystem?
What specific verification method (Test, Analysis, Inspection, Demonstration) proves compliance?
What verification artifact/document delivers that proof?
What safety hazard does this control mitigate?
Is the requirement Compliant (Closed), Partially Compliant, or Non-Compliant (Open)?

Graph Schema Assumptions

(:Standard)-[:DEFINES_REQUIREMENT]->(:Requirement)
(:Requirement)-[:APPLIES_TO {software_class: "Class A", tailorable: false}]->(:SoftwareClassification)
(:Requirement)-[:VERIFIED_BY]->(:VerificationMethod)
(:VerificationMethod)-[:EVIDENCED_BY]->(:Artifact)
(:Requirement)-[:MITIGATES]->(:Hazard)
(:DesignElement|:Component)-[:IMPLEMENTS]->(:Requirement)

1. The Core Compliance Matrix Query

This query unrolls the verification paths for a target baseline document and software class, determines compliance closure status deterministically, and formats the output ready for export to CSV, Pandas, or Markdown tables.

MATCH (doc:Document {id: $doc_id})-[:DECOMPOSES_INTO*1..4]->(sec:Section)-[:CONTAINS_CONTENT]->(chunk:Chunk)
MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)

// 1. Evaluate Applicability (e.g., Class A, B, C, etc.)
OPTIONAL MATCH (req)-[app:APPLIES_TO]->(target:SoftwareClassification {name: $software_class})

// 2. Traverse Verification Methods and Evidence Artifacts
OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)
OPTIONAL MATCH (vm)-[:EVIDENCED_BY]->(art:Artifact)

// 3. Traverse Implementation Design/Code Components
OPTIONAL MATCH (comp:Component)-[:IMPLEMENTS]->(req)

// 4. Traverse Hazards Mitigated by this Requirement
OPTIONAL MATCH (req)-[:MITIGATES]->(haz:Hazard)

// 5. Aggregate Paths per Requirement Clause
WITH 
    sec.id AS section_id,
    req.id AS req_id,
    req.title AS req_title,
    req.text AS requirement_statement,
    app.tailorable AS is_tailorable,
    CASE 
        WHEN app IS NULL THEN "NOT_APPLICABLE"
        ELSE "APPLICABLE"
    END AS applicability,
    collect(DISTINCT vm.method) AS methods,
    collect(DISTINCT art.id + ' (' + art.status + ')') AS evidence_artifacts,
    collect(DISTINCT comp.id) AS implementing_components,
    collect(DISTINCT haz.id) AS mitigated_hazards

// 6. Compute Closure Status Deterministically
WITH 
    section_id,
    req_id,
    req_title,
    requirement_statement,
    applicability,
    is_tailorable,
    methods,
    evidence_artifacts,
    implementing_components,
    mitigated_hazards,
    CASE 
        WHEN applicability = "NOT_APPLICABLE" THEN "N/A"
        WHEN size(implementing_components) = 0 THEN "NON_COMPLIANT_NO_IMPLEMENTATION"
        WHEN size(methods) = 0 THEN "NON_COMPLIANT_NO_VERIFICATION_METHOD"
        WHEN size(evidence_artifacts) = 0 THEN "OPEN_PENDING_ARTIFACTS"
        WHEN any(art IN evidence_artifacts WHERE art CONTAINS "DRAFT" OR art CONTAINS "REJECTED") THEN "PARTIALLY_COMPLIANT"
        ELSE "FULLY_COMPLIANT"
    END AS compliance_status

RETURN 
    section_id AS Section,
    req_id AS `Requirement ID`,
    req_title AS `Title`,
    applicability AS `Applicability`,
    compliance_status AS `Compliance Status`,
    is_tailorable AS `Tailorable`,
    methods AS `Verification Methods`,
    implementing_components AS `Implementing Components`,
    evidence_artifacts AS `Artifact Proofs`,
    mitigated_hazards AS `Mitigated Hazards`
ORDER BY section_id ASC, req_id ASC;

2. Discrepancy & Gap-Analysis Query (Red-Flag Matrix)

To run a fast safety and mission assurance audit that only reports requirements lacking full verification closure or missing linked safety hazards:

MATCH (req:Requirement)-[app:APPLIES_TO]->(:SoftwareClassification {name: $software_class})
WHERE app.status = "MANDATORY"

// Find requirements missing critical links
OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)
OPTIONAL MATCH (comp:Component)-[:IMPLEMENTS]->(req)
OPTIONAL MATCH (req)-[:MITIGATES]->(haz:Hazard)

WITH req, vm, comp, haz
WHERE vm IS NULL OR comp IS NULL

RETURN 
    req.id AS non_compliant_req_id,
    req.text AS statement,
    CASE WHEN comp IS NULL THEN "MISSING_DESIGN_ALLOCATION" ELSE "ALLOCATED" END AS design_state,
    CASE WHEN vm IS NULL THEN "MISSING_VERIFICATION_PATH" ELSE "METHOD_DEFINED" END AS verification_state,
    collect(DISTINCT haz.id) AS unmitigated_hazards
ORDER BY req.id;

3. Traceability Coverage Rollup (Executive Dashboard Matrix)

Produce a high-level summary showing the percentage of compliant requirements broken down by document section:

MATCH (sec:Section)-[:CONTAINS_CONTENT]->(:Chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)
MATCH (req)-[app:APPLIES_TO]->(:SoftwareClassification {name: $software_class})

OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)-[:EVIDENCED_BY]->(art:Artifact {status: "APPROVED"})
OPTIONAL MATCH (:Component)-[:IMPLEMENTS]->(req)

WITH sec, req, 
     CASE 
         WHEN vm IS NOT NULL AND art IS NOT NULL THEN 1 
         ELSE 0 
     END AS is_closed

WITH sec.id AS section_id,
     sec.title AS section_title,
     count(DISTINCT req) AS total_requirements,
     sum(is_closed) AS closed_requirements

RETURN 
    section_id AS `Section ID`,
    section_title AS `Section Title`,
    total_requirements AS `Total Mandated`,
    closed_requirements AS `Verified & Closed`,
    round(100.0 * closed_requirements / total_requirements, 2) AS `Compliance %`
ORDER BY section_id;

4. Direct Export to Pandas / Markdown Matrix via Python

When pulling these results directly into automated reporting pipelines or Docling/PDF generation:

import pandas as pd
from neo4j import GraphDatabase

def export_compliance_matrix(neo4j_uri, auth, doc_id: str, software_class: str) -> pd.DataFrame:
    query = """
    MATCH (doc:Document {id: $doc_id})-[:DECOMPOSES_INTO*1..4]->(sec:Section)-[:CONTAINS_CONTENT]->(chunk:Chunk)
    MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)
    OPTIONAL MATCH (req)-[app:APPLIES_TO]->(target:SoftwareClassification {name: $software_class})
    OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)
    OPTIONAL MATCH (vm)-[:EVIDENCED_BY]->(art:Artifact)
    OPTIONAL MATCH (comp:Component)-[:IMPLEMENTS]->(req)
    
    WITH sec, req, app,
         collect(DISTINCT vm.method) AS methods,
         collect(DISTINCT art.id) AS artifacts,
         collect(DISTINCT comp.id) AS components
         
    RETURN 
        sec.id AS Section,
        req.id AS Requirement_ID,
        CASE WHEN app IS NOT NULL THEN "YES" ELSE "NO" END AS Applicable,
        components AS Implemented_By,
        methods AS Verification_Methods,
        artifacts AS Verification_Artifacts,
        CASE 
            WHEN app IS NULL THEN "N/A"
            WHEN size(components) > 0 AND size(methods) > 0 AND size(artifacts) > 0 THEN "COMPLIANT"
            ELSE "NON_COMPLIANT"
        END AS Status
    ORDER BY sec.id, req.id
    """
    
    with GraphDatabase.driver(neo4j_uri, auth=auth) as driver:
        with driver.session() as session:
            result = session.run(query, doc_id=doc_id, software_class=software_class)
            records = [record.data() for record in result]
            
    df = pd.DataFrame(records)
    # Convert list columns to clean comma-separated strings for tabular rendering
    for col in ["Implemented_By", "Verification_Methods", "Verification_Artifacts"]:
        df[col] = df[col].apply(lambda x: ", ".join(x) if isinstance(x, list) and x else "-")
        
    return df

Add reciprocal rank fusion for full text and vector search

Reciprocal Rank Fusion (RRF) normalizes and combines the rankings from disparate search retrieval systems (dense vector semantic search and BM25/full-text lexical search) without requiring raw score calibration or min-max normalization.

The formula for each item $d$ across retrieval channels $M$ is:

RRF(d) = \sum_{m \in M} \frac{1}{k + r_m(d)}

Where:

$r_m(d)$ is the 1-based rank of item $d$ in retrieval method $m$ .
$k$ is the smoothing constant (standard default is $k = 60$ ). It dampens the penalty difference between adjacent ranks deep in the result set while prioritizing items that appear near the top of both lists.

1. Prerequisites: Indexes in Neo4j

Ensure both the vector index and the native full-text index are active on your target nodes (Chunk or Requirement):

// 1. Vector Index for dense embeddings
CREATE VECTOR INDEX chunk_embeddings IF NOT EXISTS
FOR (c:Chunk) ON (c.embedding)
OPTIONS {
  indexConfig: {
    `vector.dimensions`: 768,
    `vector.similarity_function`: 'cosine'
  }
};

// 2. Full-Text (Lucene/BM25) Index for keyword & clause matching
CREATE FULLTEXT INDEX chunk_fulltext IF NOT EXISTS
FOR (c:Chunk) ON EACH [c.text];

2. Pure Cypher Query: Hybrid Vector + BM25 with RRF & Graph Traversal

This query:

Calls the Vector index to get the top- $N$ semantic candidates with their 1-based ranks.
Calls the Full-Text index to get the top- $N$ lexical candidates with their 1-based ranks.
Computes the RRF score using $k = 60$ .
Re-ranks and picks the top- $K$ chunks.
Traverses the Knowledge Graph spine (Section hierarchy and domain Requirement links) to return context-rich results.

// Step 1: Semantic Vector Search
CALL db.index.vector.queryNodes('chunk_embeddings', \(candidate_limit,\)query_embedding) 
YIELD node AS chunk, score AS vector_score
ORDER BY vector_score DESC
WITH collect(chunk) AS vector_results

// Step 2: Lexical Full-Text Search (Lucene BM25)
CALL db.index.fulltext.queryNodes('chunk_fulltext', \(query_text, {limit:\)candidate_limit}) 
YIELD node AS chunk, score AS text_score
ORDER BY text_score DESC
WITH vector_results, collect(chunk) AS text_results

// Step 3: Unroll and Compute Ranks for Both Channels
UNWIND range(0, size(vector_results) - 1) AS v_idx
WITH vector_results, text_results, 
     vector_results[v_idx] AS chunk, 
     1.0 / ($k + (v_idx + 1)) AS v_rrf

// Accumulate unique chunks and add text channel score
WITH vector_results, text_results, collect({chunk: chunk, rrf: v_rrf}) AS vector_scored_nodes

UNWIND range(0, size(text_results) - 1) AS t_idx
WITH vector_scored_nodes, 
     text_results[t_idx] AS chunk, 
     1.0 / ($k + (t_idx + 1)) AS t_rrf

WITH vector_scored_nodes + collect({chunk: chunk, rrf: t_rrf}) AS all_scored_pairs

// Step 4: Aggregate RRF Scores per Unique Chunk
UNWIND all_scored_pairs AS pair
WITH pair.chunk AS chunk, sum(pair.rrf) AS rrf_score
ORDER BY rrf_score DESC
LIMIT $final_top_k

// Step 5: Graph Spine & Requirement Traversal
MATCH path = (doc:Document)-[:DECOMPOSES_INTO*1..5]->(sec:Section)-[:CONTAINS_CONTENT]->(chunk)
OPTIONAL MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)
OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)

RETURN 
    chunk.id AS chunk_id,
    round(rrf_score, 6) AS rrf_score,
    chunk.text AS text_snippet,
    [s IN nodes(path) WHERE s:Section | s.id + ': ' + s.title] AS breadcrumbs,
    req.id AS requirement_id,
    collect(DISTINCT vm.method) AS verification_methods
ORDER BY rrf_score DESC;

3. Modular Python Implementation with Dynamic Parameterization

This implementation abstracts embedding generation, query building, and handles graceful fallbacks if full-text returns zero hits (e.g., highly technical jargon or non-matching acronyms):

from neo4j import AsyncGraphDatabase
from typing import List, Dict, Any

async def run_rrf_hybrid_search(
    driver: AsyncGraphDatabase,
    query_text: str,
    query_embedding: List[float],
    candidate_limit: int = 30,
    final_top_k: int = 5,
    rrf_k: int = 60
) -> List[Dict[str, Any]]:
    """
    Executes Reciprocal Rank Fusion combining BM25 full-text search
    and cosine vector search, then projects graph traversal paths.
    """
    query = """
    // 1. Vector Search
    CALL db.index.vector.queryNodes('chunk_embeddings', \(candidate_limit,\)embedding) 
    YIELD node AS chunk, score AS v_score
    ORDER BY v_score DESC
    WITH collect(chunk) AS v_list

    // 2. Full-Text Search
    CALL db.index.fulltext.queryNodes('chunk_fulltext', \(query_text, {limit:\)candidate_limit}) 
    YIELD node AS chunk, score AS t_score
    ORDER BY t_score DESC
    WITH v_list, collect(chunk) AS t_list

    // 3. Score Vector Ranks
    UNWIND range(0, size(v_list) - 1) AS v_idx
    WITH v_list, t_list, v_list[v_idx] AS chunk, 1.0 / ($k + (v_idx + 1)) AS score
    WITH t_list, collect({chunk: chunk, score: score}) AS v_pairs

    // 4. Score Full-Text Ranks
    UNWIND range(0, size(t_list) - 1) AS t_idx
    WITH v_pairs, t_list[t_idx] AS chunk, 1.0 / ($k + (t_idx + 1)) AS score
    WITH v_pairs + collect({chunk: chunk, score: score}) AS all_pairs

    // 5. Aggregate RRF and Deduplicate
    UNWIND all_pairs AS item
    WITH item.chunk AS chunk, sum(item.score) AS rrf_score
    ORDER BY rrf_score DESC
    LIMIT $top_k

    // 6. Knowledge Graph Expansion
    MATCH (sec:Section)-[:CONTAINS_CONTENT]->(chunk)
    OPTIONAL MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)
    OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)

    RETURN 
        chunk.id AS chunk_id,
        round(rrf_score, 5) AS score,
        chunk.text AS text,
        sec.id AS section_id,
        sec.title AS section_title,
        req.id AS req_id,
        collect(DISTINCT vm.method) AS verification_methods
    ORDER BY score DESC;
    """
    
    params = {
        "query_text": query_text,
        "embedding": query_embedding,
        "candidate_limit": candidate_limit,
        "top_k": final_top_k,
        "k": rrf_k
    }

    async with driver.session() as session:
        result = await session.run(query, params)
        records = await result.data()
        return records

Comparison: Single vs. RRF Retrieval for Technical Specs

Scenario	Vector-Only	BM25 Full-Text Only	RRF Hybrid + Graph
Clause ID lookup (e.g., `SWE-020`)	Often misses or ranks low due to low semantic context of short tokens.	Exact match, ranks #1 immediately.	Ranks at top because BM25 rank contribution dominates rank 1.
Concept query ("redundancy in telemetry uplink")	Retrieves semantically relevant sections even if different terms are used.	Fails if the text uses "dual communication bus" instead.	Vector channel surfaces candidate; RRF keeps it in the top- $K$ .
Ambiguous Acronyms (e.g., `FMEA`, `SMA`)	Can match multiple unrelated semantic clusters.	Restricts to exact matches, but might pull non-essential mentions.	Dual hit across both channels triggers top-tier rank elevation.

Show how to incorporate graph-topology proximity or shortest path length directly into the RRF ranking score.

Incorporating graph-topology proximity or shortest path length directly into Reciprocal Rank Fusion extends classical 2-channel RRF into a 3-channel structural re-ranking engine:

Score(d) = \underbrace{\frac{w_v}{k + r_v(d)}}_{\text{Vector Rank}} + \underbrace{\frac{w_t}{k + r_t(d)}}_{\text{Full-Text Rank}} + \underbrace{\frac{w_g}{k + r_g(d)}}_{\text{Graph Proximity Rank}}

Where:

$r_v(d)$ is the 1-based rank from dense vector search.
$r_t(d)$ is the 1-based rank from BM25/Lucene full-text search.
$r_g(d)$ is the 1-based rank derived from topological distance to a known context anchor (e.g., a specific subsystem node like Subsystem {name: 'Guidance, Navigation, and Control'} or an active hazard Hazard {id: 'HAZ-CRIT-004'}).
$w_v, w_t, w_g$ are channel weights (defaulting to $1.0$ ).

Proximity Scoring Strategies

To rank candidates by graph proximity, measure the distance between each candidate's associated requirement/component and the anchor node:

Shortest Path Length ( $L$ ): Hop count between candidate node and anchor node ( $1 \le L \le L_{max}$ ). Shorter paths receive higher rank. Candidates disconnected from the anchor receive a virtual penalty distance ( $L = \infty$ ).
Path Weight / Relationship Strength: If edges have weights (e.g., :ALLOCATED_TO has higher weight than :REFERENCED_IN), compute Dijkstra weighted cost.

Cypher Implementation: Vector + BM25 + Graph-Topology RRF

This query takes semantic embeddings, full-text tokens, and a target Anchor Entity (e.g., an affected subsystem or parent assembly), computes candidate distances via shortest path, ranks all three modalities, and aggregates the final score.

// -------------------------------------------------------------
// Channel 1: Vector Search (Semantic)
// -------------------------------------------------------------
CALL db.index.vector.queryNodes('chunk_embeddings', \(candidate_limit,\)query_embedding)
YIELD node AS chunk, score AS v_score
ORDER BY v_score DESC
WITH collect(chunk) AS v_list

// -------------------------------------------------------------
// Channel 2: Full-Text Search (Lexical)
// -------------------------------------------------------------
CALL db.index.fulltext.queryNodes('chunk_fulltext', \(query_text, {limit:\)candidate_limit})
YIELD node AS chunk, score AS t_score
ORDER BY t_score DESC
WITH v_list, collect(chunk) AS t_list

// Union candidate set for graph distance evaluation
WITH v_list, t_list,
     apoc.coll.toSet(v_list + t_list) AS candidate_chunks

// -------------------------------------------------------------
// Channel 3: Graph Topology Distance (Anchor Proximity)
// -------------------------------------------------------------
MATCH (anchor:Subsystem {id: $anchor_subsystem_id})
UNWIND candidate_chunks AS chunk

// Optional traversal: Link chunk to Requirement, then trace path to anchor
OPTIONAL MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)
OPTIONAL MATCH p = shortestPath((req)-[:ALLOCATED_TO|SATISFIED_BY|DECOMPOSES*..6]-(anchor))

WITH v_list, t_list, chunk,
     // If no path exists, penalize with an arbitrarily high length
     CASE WHEN p IS NULL THEN 999 ELSE length(p) END AS hop_distance
ORDER BY hop_distance ASC

// Collect topological ordering
WITH v_list, t_list, collect(chunk) AS g_list

// -------------------------------------------------------------
// Compute Reciprocal Rank Fusion Across All 3 Channels
// -------------------------------------------------------------
// 1. Vector RRF Scores
UNWIND range(0, size(v_list) - 1) AS v_idx
WITH v_list, t_list, g_list,
     v_list[v_idx] AS chunk,
     \(w_v * (1.0 / (\)k + (v_idx + 1))) AS v_rrf
WITH t_list, g_list, collect({chunk: chunk, rrf: v_rrf}) AS scored_items

// 2. Full-Text RRF Scores
UNWIND range(0, size(t_list) - 1) AS t_idx
WITH t_list, g_list, scored_items,
     t_list[t_idx] AS chunk,
     \(w_t * (1.0 / (\)k + (t_idx + 1))) AS t_rrf
WITH g_list, scored_items + collect({chunk: chunk, rrf: t_rrf}) AS scored_items

// 3. Graph Proximity RRF Scores
UNWIND range(0, size(g_list) - 1) AS g_idx
WITH scored_items,
     g_list[g_idx] AS chunk,
     \(w_g * (1.0 / (\)k + (g_idx + 1))) AS g_rrf
WITH scored_items + collect({chunk: chunk, rrf: g_rrf}) AS all_channel_scores

// -------------------------------------------------------------
// Aggregate Scores, Deduplicate, and Expand Final Subgraph
// -------------------------------------------------------------
UNWIND all_channel_scores AS item
WITH item.chunk AS chunk, sum(item.rrf) AS combined_rrf
ORDER BY combined_rrf DESC
LIMIT $top_k

// Graph Expansion on Winning Nodes
MATCH (sec:Section)-[:CONTAINS_CONTENT]->(chunk)
OPTIONAL MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)
OPTIONAL MATCH p_anchor = shortestPath((req)-[:ALLOCATED_TO|SATISFIED_BY|DECOMPOSES*..6]-(anchor:Subsystem {id: $anchor_subsystem_id}))

RETURN
    chunk.id AS chunk_id,
    round(combined_rrf, 5) AS final_rrf_score,
    chunk.text AS text_snippet,
    sec.id AS section_id,
    req.id AS requirement_id,
    CASE WHEN p_anchor IS NOT NULL THEN length(p_anchor) ELSE null END AS hops_to_anchor
ORDER BY final_rrf_score DESC;

Python Pipeline Execution

This asynchronous driver function injects weights dynamically. If a user is querying a known subsystem, boosting $w_g$ prioritizes requirements directly attached to that subsystem in the engineering breakdown.

from neo4j import AsyncGraphDatabase
from typing import List, Dict, Any

async def run_topological_rrf_search(
    driver: AsyncGraphDatabase,
    query_text: str,
    query_embedding: List[float],
    anchor_subsystem_id: str,
    weights: Dict[str, float] = None,
    candidate_limit: int = 30,
    top_k: int = 5,
    rrf_k: int = 60
) -> List[Dict[str, Any]]:
    """
    Ranks chunks using Reciprocal Rank Fusion over 3 modalities:
    1. Vector Cosine Similarity
    2. BM25 Lucene Full-text Score
    3. Shortest Path Hop Proximity to Anchor Node
    """
    if weights is None:
        weights = {"w_v": 1.0, "w_t": 1.0, "w_g": 1.2}  # Slight boost to graph proximity

    params = {
        "query_text": query_text,
        "query_embedding": query_embedding,
        "anchor_subsystem_id": anchor_subsystem_id,
        "candidate_limit": candidate_limit,
        "top_k": top_k,
        "k": rrf_k,
        "w_v": weights.get("w_v", 1.0),
        "w_t": weights.get("w_t", 1.0),
        "w_g": weights.get("w_g", 1.0)
    }

    # Execute Cypher query defined above
    async with driver.session() as session:
        result = await session.run(TOPOLOGICAL_RRF_CYPHER, params)
        records = await result.data()
        return records

Why Topological RRF Prevents Semantic Drift

Disambiguating Cross-System Jargon: A query like "pressure transducer failure thresholds" will match both environmental control systems (ECLSS) and propulsion subsystems in raw vector/text search. Providing the propulsion subsystem anchor node boosts propulsion-specific requirements to the top via the graph rank channel.
Preserving Parent Scope: Unrelated chunks sharing identical wording (e.g., standard boilerplate safety verbiage) are penalized in the topological channel if they sit on distant branches of the systems breakdown tree.
Deterministic Tie-Breaking: When vector and BM25 scores are virtually tied across several clauses, topological proximity serves as a domain-grounded tie-breaker based on actual systems engineering allocations.

Show how to replace shortestPath with Dijkstra weighted path traversals in Cypher for graph-distance RRF.

Using unweighted shortestPath treats every relationship as identical. In an engineering knowledge graph, however, traversing an explicit allocation (:ALLOCATED_TO) reflects tighter architectural coupling than a parent-child structural grouping (:DECOMPOSES_INTO) or a casual cross-reference (:REFERENCES).

By assigning explicit traversal impedance (resistance/cost) to relationship types or edge properties, Dijkstra’s algorithm computes the minimum-cost cumulative path between candidate requirement chunks and the active anchor node:

\text{Cost}(p) = \sum_{e \in p} \text{weight}(e)

Lower impedance yields lower total cost, which earns a higher rank in the topological channel of the Reciprocal Rank Fusion calculation:

Score(d) = \frac{w_v}{k + r_v(d)} + \frac{w_t}{k + r_t(d)} + \frac{w_g}{k + r_{\text{dijkstra}}(d)}

1. Relationship Weight (Impedance) Model

Define cost properties on edges (or specify them via APOC Dijkstra configuration):

Relationship	Semantics	Weight / Cost	Rationale
`:IMPLEMENTS` / `:ALLOCATED_TO`	Direct structural assignment	1.0	Tightest semantic binding
`:SATISFIES` / `:VERIFIES`	Closed-loop verification link	1.5	High verification affinity
:DECOMPOSES_INTO	Hierarchical breakdown step	2.0	Traversing system levels
:MITIGATES	Safety/hazard control link	2.5	Risk-domain bridge
:REFERENCES	Informational cross-reference	5.0	Weak associational coupling

Set default edge properties if they are stored in the graph:

MATCH ()-[r:ALLOCATED_TO]->() SET r.cost = 1.0;
MATCH ()-[r:SATISFIES]->() SET r.cost = 1.5;
MATCH ()-[r:DECOMPOSES_INTO]->() SET r.cost = 2.0;
MATCH ()-[r:REFERENCES]->() SET r.cost = 5.0;

2. Cypher Implementation: Hybrid RRF with APOC Dijkstra

Neo4j provides Dijkstra pathfinding via the APOC library function apoc.algo.dijkstra.

// =============================================================
// Channel 1: Semantic Vector Search
// =============================================================
CALL db.index.vector.queryNodes('chunk_embeddings', \(candidate_limit,\)query_embedding)
YIELD node AS chunk, score AS v_score
ORDER BY v_score DESC
WITH collect(chunk) AS v_list

// =============================================================
// Channel 2: Lexical Full-Text Search (BM25)
// =============================================================
CALL db.index.fulltext.queryNodes('chunk_fulltext', \(query_text, {limit:\)candidate_limit})
YIELD node AS chunk, score AS t_score
ORDER BY t_score DESC
WITH v_list, collect(chunk) AS t_list

// Deduplicate candidate pool for graph-distance evaluation
WITH v_list, t_list,
     apoc.coll.toSet(v_list + t_list) AS candidate_chunks

// =============================================================
// Channel 3: Dijkstra Weighted Path to Anchor Entity
// =============================================================
MATCH (anchor:Subsystem {id: $anchor_subsystem_id})
UNWIND candidate_chunks AS chunk

// Anchor requirement to chunk
OPTIONAL MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)

// Run Dijkstra shortest weighted path algorithm
// apoc.algo.dijkstra(startNode, endNode, relTypes, weightPropertyName, defaultWeight)
CALL apoc.do.when(
    req IS NOT NULL,
    'CALL apoc.algo.dijkstra(req, anchor, "ALLOCATED_TO|SATISFIED_BY|DECOMPOSES_INTO|REFERENCES", "cost", 2.0)
     YIELD path, weight RETURN path, weight',
    'RETURN null AS path, 999.0 AS weight',
    {req: req, anchor: anchor}
) YIELD value

WITH v_list, t_list, chunk,
     coalesce(value.weight, 999.0) AS path_cost
ORDER BY path_cost ASC

// Collect topological order ranked by weighted cost
WITH v_list, t_list, collect(chunk) AS g_list

// =============================================================
// Reciprocal Rank Fusion Across All 3 Ranked Lists
// =============================================================
// 1. Vector Channel
UNWIND range(0, size(v_list) - 1) AS v_idx
WITH v_list, t_list, g_list,
     v_list[v_idx] AS chunk,
     \(w_v * (1.0 / (\)k + (v_idx + 1))) AS v_rrf
WITH t_list, g_list, collect({chunk: chunk, rrf: v_rrf}) AS scored_items

// 2. Full-Text Channel
UNWIND range(0, size(t_list) - 1) AS t_idx
WITH t_list, g_list, scored_items,
     t_list[t_idx] AS chunk,
     \(w_t * (1.0 / (\)k + (t_idx + 1))) AS t_rrf
WITH g_list, scored_items + collect({chunk: chunk, rrf: t_rrf}) AS scored_items

// 3. Dijkstra Topological Channel
UNWIND range(0, size(g_list) - 1) AS g_idx
WITH scored_items,
     g_list[g_idx] AS chunk,
     \(w_g * (1.0 / (\)k + (g_idx + 1))) AS g_rrf
WITH scored_items + collect({chunk: chunk, rrf: g_rrf}) AS all_channel_scores

// =============================================================
// Aggregate RRF Scores & Expand Verification Context
// =============================================================
UNWIND all_channel_scores AS item
WITH item.chunk AS chunk, sum(item.rrf) AS combined_rrf
ORDER BY combined_rrf DESC
LIMIT $top_k

MATCH (sec:Section)-[:CONTAINS_CONTENT]->(chunk)
OPTIONAL MATCH (chunk)-[:DEFINES_REQUIREMENT]->(req:Requirement)
OPTIONAL MATCH (req)-[:VERIFIED_BY]->(vm:VerificationMethod)

RETURN
    chunk.id AS chunk_id,
    round(combined_rrf, 5) AS final_rrf_score,
    chunk.text AS text_snippet,
    sec.id AS section_id,
    req.id AS requirement_id,
    collect(DISTINCT vm.method) AS verification_methods
ORDER BY final_rrf_score DESC;

3. Alternative: Neo4j Graph Data Science (GDS) Dijkstra Implementation

If working with very large graphs or requiring in-memory projected graphs, the Neo4j GDS library (gds.shortestPath.dijkstra) provides higher throughput:

// 1. In-memory Projection (executed once)
CALL gds.graph.project(
    'subsystem-requirements-graph',
    ['Requirement', 'Subsystem', 'Component'],
    {
        ALLOCATED_TO: {type: 'ALLOCATED_TO', properties: 'cost', orientation: 'UNDIRECTED'},
        DECOMPOSES_INTO: {type: 'DECOMPOSES_INTO', properties: 'cost', orientation: 'UNDIRECTED'}
    }
);

// 2. Query Call for Individual Candidate Chunks
MATCH (anchor:Subsystem {id: $anchor_subsystem_id})
MATCH (req:Requirement {id: $candidate_req_id})

CALL gds.shortestPath.dijkstra.stream('subsystem-requirements-graph', {
    sourceNode: req,
    targetNode: anchor,
    relationshipWeightProperty: 'cost'
})
YIELD totalCost, nodeIds
RETURN totalCost, [nodeId IN nodeIds | gds.util.asNode(nodeId).id] AS path_nodes;

Unweighted Hop Count vs. Dijkstra Path Cost

Metric	Unweighted shortestPath	Dijkstra Weighted Path
Path Selection	Selects any path with the fewest edges, even if connected through low-confidence references.	Selects the path with the least aggregate architectural resistance.
Edge Semantics	A generic citation (`:REFERENCES`) is treated as equal to a hard allocation (`:ALLOCATED_TO`).	Direct allocations penalize the distance far less than informational links.
Noise Filtering	Candidate requirements from unrelated sections can tie at 3 hops through accidental cross-links.	Cross-cutting references accrue high cumulative cost ( $5.0 \times N$ ), dropping irrelevant candidates out of the top topological ranks.

From <https://gemini.google.com/app/7ccb47437184e141> Google Gemini (3.8 Flash)