#4735 Traceability Using NASA/Provider Document Knowledge Graphs
#4735 Traceability Using NASA/Provider Document Knowledge Graphs
#4735 Given NASA NPR 7150.2D as one example, and a NASA launch provider software development plan as another independent example, how feasible and practical is it to generate knowledge graphs for each that are truly beneficial to accessing compliance of one to the other?
Building dual knowledge graphs (KGs) to evaluate compliance between NASA NPR 7150.2D (NASA Software Engineering Requirements) and a commercial launch provider’s Software Development Plan (SDP) is technically feasible and highly practical, provided the architecture uses a hybrid neuro-symbolic approach (ontological graph schema + vector embeddings/LLM evaluation) rather than relying on pure entity-relation triples.
Traditional keyword search and standard RAG struggle with regulatory compliance because compliance is structural, conditional (software class, safety criticality), and hierarchical. A graph-based representation solves the structural traceability challenge.
1. Schema & Ontological Alignment
To make compliance assessment automated and queryable, both graphs must map to a shared upper compliance ontology.
┌──────────────────────────────┐
│ Shared Compliance Schema │
└──────────────┬───────────────┘
│
┌───────────────────────┴───────────────────────┐
▼ ▼
┌──────────────────────────────┐ ┌──────────────────────────────┐
│ NPR 7150.2D Graph │ │ Provider SDP Graph │
├──────────────────────────────┤ ├──────────────────────────────┤
│ (:Requirement {swe_id, text})│ │ (:Section {id, title, text}) │
│ (:ClassApplicability {A..E}) │◄───────│ (:Process {name, scope}) │
│ (:SafetyCriticality {bool}) │ Traces │ (:Artifact {name, cadence}) │
│ (:VerificationMethod) │ Matches│ (:Role {title, authority}) │
└──────────────────────────────┘ └──────────────────────────────┘
NPR 7150.2D Graph Schema (The "Rulebook")
Nodes:
Requirement: ExplicitSWE-xxxprovisions with text, intent, and rationale.Classification: Software Classes (Class A through Class E, Safety-Critical flag).ComplianceMatrix: Applicable, Tailored, or Not Applicable mappings per class.VerificationArtifact: Required work products (e.g., Software Configuration Management Plan, Software Test Plan).
Edges:
(:Requirement)-[:APPLIES_TO]->(:Classification)(:Requirement)-[:REQUIRES_ARTIFACT]->(:VerificationArtifact)(:Requirement)-[:DEPENDS_ON]->(:Requirement)
Provider SDP Graph Schema (The "Implementation")
Nodes:
SDPSection: Paragraphs, clauses, and sub-clauses.ProcessActivity: Specific engineering workflows (e.g., peer reviews, static analysis, CI/CD gates).Role: Responsible engineering authority (e.g., Lead Safety Engineer, Software Quality Assurance).ProviderArtifact: Deliverables, records, and repositories defined in the plan.
Edges:
(:ProcessActivity)-[:PRODUCES]->(:ProviderArtifact)(:Role)-[:OWNS_PROCESS]->(:ProcessActivity)(:SDPSection)-[:DEFINES]->(:ProcessActivity)
2. Ingestion & Graph Construction Feasibility
| Dimension | NPR 7150.2D (Standard) | Provider SDP (Plan) | Feasibility Assessment |
| Structure Quality | High (Structured text, SWE numbering, Appendix Table C matrix). | Variable (Ranges from structured MIL-STD-498 format to agile/commercial prose). | High Feasibility. NPR 7150.2D parses deterministically. SDP requires LLM-assisted schema extraction for processes/artifacts. |
| Extraction Method | Deterministic regex + Markdown/XML parsing. | Chunking with LLM structured output (Pydantic schema extraction) + node consolidation. | Moderate to High. The SDP extraction needs prompt validation to prevent hallucinated entity relations. |
| Dynamic Updates | Static (Changes only on NASA policy baseline release). | Dynamic (Updates across contract revisions, waivers, and sprints). | High Practicality. Graph can version SDP releases as subgraphs or snapshots. |
3. Practical Mechanics of the Compliance Mapping
Once both graphs exist in a property graph store (such as Neo4j) or RDF triple store, compliance evaluation runs through three automated stages:
Stage 1: Structural Scoping (Deterministic Filtering)
Before checking prose compliance, filter the NPR 7150.2D graph based on the project's parameters (e.g., Class A / Flight Software / Safety-Critical).
Cypher/SPARQL queries prune the required
SWE-xxxset down to only the applicable baseline.Requirements marked as tailored or waived are flagged before entering text analysis.
Stage 2: Entity & Semantic Cross-Linking
Establish candidate links across the two graph domains:
(:SDPSection)-[:CLAIMS_COMPLIANCE_TO]->(:Requirement)(:ProviderArtifact)-[:SATISFIES_ARTIFACT]->(:VerificationArtifact)Embeddings generated over SDP sections identify high-similarity matches against specific SWE requirements, creating candidate
[:EVALUATES_AGAINST {similarity_score}]edges.
Stage 3: Verification & Gap Traversal (Neuro-Symbolic Audit)
An orchestrated verification agent traverses candidate compliance edges and evaluates whether the SDP prose genuinely fulfills the SWE-xxx requirement intent:
Direct Match: Process defined, artifact generated, role assigned -->
[:COMPLIANT].Partial / Ambiguous: Process mentioned, but missing cadence, independent QA verification, or required artifact -->
[:PARTIAL_GAP].Complete Absence: No node in the SDP graph links to an applicable
SWE-xxx-->[:DEFICIENCY].
(SWE-021: Peer Reviews) ──[:REQUIRES_ARTIFACT]──> (Review Records)
▲ ▲
│ [:COMPLIANCE_TARGET] │ [:VALIDATES]
│ │
(SDP Sec 4.3: Code Review) ──[:PRODUCES]──> (GitHub PR Logs)
│
└──> Graph Query Result: COMPLIANT (Evidence: PR Logs meet review record criteria)
4. Key Implementation Bottlenecks and How to Mitigate Them
Semantic Drift & Nomenclature Mismatch
Challenge: NPR 7150.2D uses traditional terminology (e.g., Software Configuration Management Plan, Software Design Document), whereas modern launch providers use cloud-native and Agile terminology (e.g., Infrastructure as Code, Confluence Architecture Specs, GitOps pipelines).
Mitigation: Populate the shared ontology with a Synonym / Concept Equivalence Matrix mapping modern DevOps concepts to NASA formal artifact terms.
Prose Intent vs. Literal Presence
Challenge: A plan can state "All code is reviewed prior to merge," but fail to specify independent SQA participation required for Class A safety-critical software.
Mitigation: Graph properties must include verification criteria checklists derived from NASA-HDBK-2203 (NASA Software Engineering Handbook). The audit agent validates each criterion as a distinct property on the compliance edge.
Tailoring and Waiver Context
Challenge: A missing requirement in the SDP graph is not always a non-compliance; it may be an approved deviation.
Mitigation: Ingest the provider's Compliance Matrix and approved waiver documents as a third contextual graph overlay.
5. Summary Verdict
Feasibility: 8.5 / 10 — The schemas are well-bounded, NPR 7150.2D is highly structured, and LLM-driven structured extraction reliably maps technical plans into node/edge networks.
Practical Value: Very High — Replaces manual, weeks-long compliance matrix reconciliation with instant gap-traversal queries, highlights unaddressed SWE IDs, and exposes missing dependencies across engineering artifacts.
Practical Value: Very High. What is that on a 0 to 10 range?
On a 0 to 10 scale, Very High translates to an 8.5 / 10 for this specific use case.
Here is the breakdown of what drives that score and what keeps it from being a 10:
Why it scores an 8.5 (The High-Value Drivers):
Instant Delta Analysis: Pinpoints unmapped
SWE-xxxrequirements across hundreds of pages in seconds rather than weeks of manual matrix checking.Relational Depth: Audits don't just check if a topic is mentioned; graph traversal verifies that required secondary artifacts (e.g., test records, review logs) and assigned roles actually link back to the process.
Re-audit Velocity: When a provider submits Revision B or C of an SDP, re-running the graph diff isolates only what changed, eliminating full-document re-reviews.
What prevents it from being a 9.5 or 10 (The Practical Headwinds):
Engineering Nuance (The Final Mile): The graph can identify where evidence exists, but a human subject matter expert must still make the final judgment call on whether a proprietary commercial workflow genuinely satisfies safety intent.
Upfront Schema Maintenance: You have to invest time up front building the synonym mappings (e.g., translating traditional NASA artifact names to modern commercial CI/CD and GitOps constructs).
Seems like the NPR 7150.2D graph alone could be used as a starter-template to be separately merged with provider documents to document specific-document traceability
That is precisely the most practical and scalable architecture for this problem.
Treating the NPR 7150.2D Knowledge Graph as an immutable "Gold Standard Template" (or Master Reference Graph) turns compliance assessment into an automated, multi-tenant evaluation engine rather than a bespoke one-off data science project for every document.
The "Gold Master + Provider Overlay" Architecture
┌────────────────────────────────────────┐
│ NPR 7150.2D Master Backbone Graph │
│ (Immutable: SWE Nodes, Classes A-E, │
│ Required Artifacts, Verification) │
└───────────────────┬────────────────────┘
│
┌──────────────────────┼──────────────────────┐
│ Instance Merger │ Instance Merger │
▼ ▼ ▼
┌────────────────────┐ ┌────────────────────┐ ┌────────────────────┐
│ Provider A: SDP │ │ Provider A: SCMP │ │ Provider B: SDP │
│ (Rev C Overlay) │ │ (Rev A Overlay) │ │ (Agile Workflow) │
└────────────────────┘ └────────────────────┘ └────────────────────┘
By decoupling the regulatory rulebook from the provider's submissions, you achieve three immediate operational advantages:
1. Zero-Redundancy Regulatory Modeling
NPR 7150.2D changes infrequently. Once you construct the base graph:
Every requirement (
SWE-001throughSWE-142+), its software class applicability matrix (Classes A–E, Safety-Critical flag), handbook guidance linkages, and required evidentiary artifacts are captured once.This base graph acts as a standardized topological "jig" against which any document—Software Development Plans (SDPs), Software Configuration Management Plans (SCMPs), Software Assurance Plans (SSAPs), or Verification & Validation Plans (VVPs)—can be aligned.
2. Multi-Document Aggregation and Cross-Silo Traceability
A single provider SDP rarely satisfies all of NPR 7150.2D on its own. Typically, compliance is distributed across multiple program documents:
The SDP might satisfy
SWE-021(Peer Reviews) andSWE-033(Testing Phases).The SCMP satisfies
SWE-079throughSWE-085(Configuration Management, Version Control).The Hazard Reports & Safety Plans satisfy
SWE-134(Software Safety Analyses).
When individual document graphs are merged into a workspace that references the static NPR 7150.2D backbone, the combined graph reveals program-level coverage:
(SWE-080: Baselines) ◄───[:SATISFIED_BY]─── (SCMP Sec 3.2: Git Branching Strategy)
(SWE-021: Peer Reviews) ◄─[:SATISFIED_BY]─── (SDP Sec 4.1: Pull Request Review Rules)
(SWE-134: Fault Trees) ◄──[:SATISFIED_BY]─── (Hazard Report 04: C&C FMEA)
A simple graph traversal query instantly highlights which SWE nodes still have zero incoming [:SATISFIED_BY] edges across the provider's entire document suite.
3. Native Diffing Across Revisions and Multiple Providers
Provider Re-Submissions (Rev A $\rightarrow$ Rev B):
Instead of re-evaluating 300 pages, merge the Rev B document graph into a branch. A graph delta query isolates specifically which
[:SATISFIED_BY]edges changed, broke, or were newly established.Cross-Provider Benchmarking:
NASA SMA and engineering oversight teams can overlay Provider A’s flight software SDP and Provider B’s flight software SDP onto the identical NPR 7150.2D template to immediately compare maturity, rigor, and tailoring strategies.
Practical Implementation Pattern
To make the template pattern operational in a property graph (like Neo4j):
Namespace Separation: Tag the base graph nodes with a
:Standardor:NASA_NPRlabel and a version property (version: "7150.2D").Dynamic Overlay Edges: Create provider nodes labeled
:ProviderEvidencewith metadata specifying the exact source document, page, section, and hash.Traceability Relational Types:
[:ADDRESSES {status: "FULL" | "PARTIAL" | "NON_COMPLIANT", confidence: 0.92, auditor_notes: "..."}][:TAILORS {rationale_id: "WVD-042", approval_status: "APPROVED"}][:EVIDENCED_BY]linking SWE artifact nodes to concrete provider repository/file outputs.
This transforms compliance auditing from an unstructured prose evaluation into a deterministic subgraph isomorphism and path-coverage problem.
Provide a Python application using Google Gemini API if necessary to create the initial NPR 7150.2D knowledge graph based on a provided PDF file.
Below is a complete, production-ready Python application using the Google GenAI SDK (google-genai), Pydantic, and Neo4j.
It uploads the NPR 7150.2D PDF directly to Gemini via the Files API, extracts the full requirement hierarchy and Class A–E compliance matrix into strict Pydantic schemas, and writes the baseline template graph into Neo4j.
Prerequisites
Install the dependencies:
pip install google-genai pydantic neo4j
Set your environment variables:
export GEMINI_API_KEY="your-api-key"
export NEO4J_URI="bolt://localhost:7687"
export NEO4J_USER="neo4j"
export NEO4J_PASSWORD="your-password"
Python Pipeline (build_npr_graph.py)
import os
import time
from typing import List, Optional
from pydantic import BaseModel, Field
from google import genai
from google.genai import types
from neo4j import GraphDatabase
# ==============================================================================
# 1. Pydantic Extraction Schemas (Domain-Specific for NASA NPR 7150.2D)
# ==============================================================================
class SWERequirement(BaseModel):
swe_id: str = Field(description="The formal SWE identifier, e.g., 'SWE-021'.")
title: str = Field(description="Short title or focus area of the requirement.")
full_text: str = Field(description="Verbatim shall statement of the requirement.")
section: str = Field(description="NPR chapter/section number, e.g., '3.1.2'.")
applicability_classes: List[str] = Field(
description="List of applicable software classes from Table C (e.g., ['A', 'B', 'C', 'Safety-Critical'])."
)
required_artifacts: List[str] = Field(
default_factory=list,
description="Formal artifacts or records mandated (e.g., 'Peer Review Records', 'SCMP')."
)
verification_methods: List[str] = Field(
default_factory=list,
description="Verification approach, e.g., 'Review of documentation', 'Test', 'Audit'."
)
cross_references: List[str] = Field(
default_factory=list,
description="Referenced standards or other SWE IDs (e.g., ['NASA-STD-8739.8', 'SWE-102'])."
)
class NPRKnowledgeGraphPayload(BaseModel):
document_id: str = Field(default="NPR 7150.2D")
revision: str = Field(default="D")
requirements: List[SWERequirement]
# ==============================================================================
# 2. Graph Ingestion Pipeline
# ==============================================================================
class NPRGraphBuilder:
def __init__(self, neo4j_uri: str, neo4j_user: str, neo4j_pass: str):
self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_pass))
self.client = genai.Client()
def close(self):
self.driver.close()
def init_graph_constraints(self):
"""Set up unique constraints and indexes for the NPR standard backbone."""
queries = [
"CREATE CONSTRAINT swe_id_unique IF NOT EXISTS FOR (r:SWERequirement) REQUIRE r.swe_id IS UNIQUE;",
"CREATE CONSTRAINT class_id_unique IF NOT EXISTS FOR (c:SoftwareClass) REQUIRE c.name IS UNIQUE;",
"CREATE CONSTRAINT artifact_name_unique IF NOT EXISTS FOR (a:StandardArtifact) REQUIRE a.name IS UNIQUE;",
"CREATE CONSTRAINT doc_unique IF NOT EXISTS FOR (d:StandardDocument) REQUIRE d.id IS UNIQUE;"
]
with self.driver.session() as session:
for q in queries:
session.run(q)
print("[Neo4j] Graph constraints initialized.")
def extract_from_pdf(self, pdf_path: str) -> NPRKnowledgeGraphPayload:
"""Uploads NPR 7150.2D PDF to Gemini Files API and performs structured extraction."""
print(f"[Gemini] Uploading {pdf_path}...")
uploaded_file = self.client.files.upload(
file=pdf_path,
config=types.UploadFileConfig(
display_name="NPR_7150_2D",
mime_type="application/pdf"
)
)
print(f"[Gemini] Upload complete. URI: {uploaded_file.uri}")
# Wait briefly for server-side processing
time.sleep(3)
extraction_prompt = (
"You are a NASA Software Safety and Mission Assurance engineering expert. "
"Analyze the attached NASA NPR 7150.2D document, specifically focusing on the "
"Chapter requirements (SWE-001 through the end) and Appendix Table C (Compliance Matrix). "
"Extract all SWE requirements, capturing their exact text, applicability to software classes "
"(Classes A, B, C, D, E, and Safety Critical), required evidence/artifacts, verification methods, "
"and cross-referenced requirements."
)
print("[Gemini] Extracting structured compliance ontology (this may take a minute)...")
response = self.client.models.generate_content(
model="gemini-2.5-flash",
contents=[uploaded_file, extraction_prompt],
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=NPRKnowledgeGraphPayload,
temperature=0.0
)
)
# Clean up the uploaded file
try:
self.client.files.delete(name=uploaded_file.name)
print("[Gemini] Temporary file cleaned up from Files API.")
except Exception:
pass
# Parse output into Pydantic model
payload = NPRKnowledgeGraphPayload.model_validate_json(response.text)
print(f"[Gemini] Successfully extracted {len(payload.requirements)} SWE requirement nodes.")
return payload
def populate_neo4j(self, payload: NPRKnowledgeGraphPayload):
"""Executes Cypher batches to instantiate the immutable standard backbone."""
cypher_tx = """
MERGE (doc:StandardDocument {id: $doc_id})
SET doc.revision = $revision,
doc.updated_at = timestamp()
WITH doc
UNWIND $requirements AS req_data
MERGE (r:SWERequirement {swe_id: req_data.swe_id})
SET r.title = req_data.title,
r.full_text = req_data.full_text,
r.section = req_data.section,
r.verification_methods = req_data.verification_methods,
r.updated_at = timestamp()
MERGE (doc)-[:DEFINES_REQUIREMENT]->(r)
// Link Software Classes (Class A, B, C, D, E, Safety-Critical)
FOREACH (cls IN req_data.applicability_classes |
MERGE (c:SoftwareClass {name: trim(cls)})
MERGE (r)-[:MANDATED_FOR_CLASS]->(c)
)
// Link Mandated Verification Artifacts
FOREACH (art IN req_data.required_artifacts |
MERGE (a:StandardArtifact {name: trim(art)})
MERGE (r)-[:MANDATES_ARTIFACT]->(a)
)
// Cross-SWE dependencies
FOREACH (ref IN req_data.cross_references |
MERGE (other:SWERequirement {swe_id: trim(ref)})
MERGE (r)-[:DEPENDS_ON]->(other)
)
"""
req_dicts = [req.model_dump() for req in payload.requirements]
with self.driver.session() as session:
session.run(
cypher_tx,
doc_id=payload.document_id,
revision=payload.revision,
requirements=req_dicts
)
print(f"[Neo4j] Standard backbone successfully written to the graph.")
# ==============================================================================
# 3. Execution Entry Point
# ==============================================================================
if __name__ == "__main__":
PDF_PATH = "NPR_7150_2D.pdf"
NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USER = os.getenv("NEO4J_USER", "neo4j")
NEO4J_PASS = os.getenv("NEO4J_PASSWORD", "password")
if not os.path.exists(PDF_PATH):
print(f"Error: Target file '{PDF_PATH}' not found in current directory.")
exit(1)
builder = NPRGraphBuilder(NEO4J_URI, NEO4J_USER, NEO4J_PASS)
try:
builder.init_graph_constraints()
kg_data = builder.extract_from_pdf(PDF_PATH)
builder.populate_neo4j(kg_data)
finally:
builder.close()
Cypher Verification Queries
Once populated, run these queries in Neo4j Browser or via Cypher to inspect your master template:
1. Trace all requirements and artifacts for Class A Flight Software
MATCH (r:SWERequirement)-[:MANDATED_FOR_CLASS]->(c:SoftwareClass {name: "A"})
OPTIONAL MATCH (r)-[:MANDATES_ARTIFACT]->(a:StandardArtifact)
RETURN r.swe_id, r.title, collect(a.name) AS required_artifacts
ORDER BY r.swe_id;
2. Scope the safety-critical compliance perimeter
MATCH (r:SWERequirement)-[:MANDATED_FOR_CLASS]->(c:SoftwareClass)
WHERE c.name =~ "(?i).*safety.*"
RETURN r.swe_id, r.title, r.full_text;Repeat for generating the knowledge graph for the provider SDP
Below is the companion pipeline that extracts the provider's Software Development Plan (SDP), instantiates the provider-specific nodes/processes/artifacts, and automatically resolves candidate compliance edges (:ADDRESSES_REQUIREMENT, :SATISFIES_ARTIFACT) against the pre-existing NPR 7150.2D Master Backbone Graph in Neo4j.
Python Pipeline (build_sdp_overlay_graph.py)
import os
import time
from typing import List, Optional
from pydantic import BaseModel, Field
from google import genai
from google.genai import types
from neo4j import GraphDatabase
# ==============================================================================
# 1. Pydantic Extraction Schemas (Provider SDP Overlay Domain)
# ==============================================================================
class SDPArtifact(BaseModel):
name: str = Field(description="Name of the deliverable or internal record, e.g., 'GitHub PR Review Log', 'Static Analysis Report'.")
format_or_tool: Optional[str] = Field(None, description="Tooling or repository format, e.g., 'SonarQube', 'Jira', 'GitLab'.")
lifecycle_phase: Optional[str] = Field(None, description="Phase produced, e.g., 'Implementation', 'Integration', 'Release'.")
maps_to_standard_artifact: Optional[str] = Field(
None,
description="Standard NASA artifact name this satisfies (e.g., 'Peer Review Records', 'Software Version Description')."
)
class SDPProcessActivity(BaseModel):
name: str = Field(description="Core engineering activity or gate, e.g., 'Automated Unit & Integration Testing', 'Independent Peer Review'.")
description: str = Field(description="Summary of how the provider executes this process.")
responsible_role: str = Field(description="Role or authority accountable, e.g., 'Flight Software Lead', 'Software Quality Assurance'.")
cadence_or_trigger: Optional[str] = Field(None, description="Trigger condition, e.g., 'On merge to main', 'Prior to CDR'.")
produced_artifacts: List[str] = Field(default_factory=list, description="Names of artifacts generated by this activity.")
class SDPSectionClaim(BaseModel):
section_id: str = Field(description="SDP section number, e.g., '4.2.1'.")
section_title: str = Field(description="Title of the section.")
summary: str = Field(description="Concise description of the technical approach in this section.")
targeted_swe_ids: List[str] = Field(
default_factory=list,
description="List of NPR 7150.2D SWE identifiers explicitly or implicitly addressed (e.g., ['SWE-021', 'SWE-033'])."
)
processes: List[SDPProcessActivity] = Field(default_factory=list)
artifacts: List[SDPArtifact] = Field(default_factory=list)
tailoring_or_waivers: Optional[str] = Field(None, description="Any documented deviations, waivers, or tailoring rationale.")
class ProviderSDPPayload(BaseModel):
provider_name: str = Field(description="Name of the commercial provider or organization.")
project_name: str = Field(description="Target vehicle or mission project, e.g., 'Lunar Lander Flight Software'.")
document_title: str = Field(default="Software Development Plan")
revision: str = Field(description="Document revision, e.g., 'Rev C' or '1.2'.")
sections: List[SDPSectionClaim]
# ==============================================================================
# 2. Provider Graph Ingestion Pipeline
# ==============================================================================
class SDPGraphBuilder:
def __init__(self, neo4j_uri: str, neo4j_user: str, neo4j_pass: str):
self.driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_pass))
self.client = genai.Client()
def close(self):
self.driver.close()
def init_graph_constraints(self):
"""Ensure indexes exist for provider nodes and traceability lookup."""
queries = [
"CREATE CONSTRAINT provider_doc_unique IF NOT EXISTS FOR (d:ProviderDocument) REQUIRE (d.provider, d.project, d.revision) IS UNIQUE;",
"CREATE INDEX sdp_sec_idx IF NOT EXISTS FOR (s:SDPSection) ON (s.doc_id, s.section_id);",
"CREATE INDEX provider_art_idx IF NOT EXISTS FOR (a:ProviderArtifact) ON (a.name, a.provider);"
]
with self.driver.session() as session:
for q in queries:
session.run(q)
print("[Neo4j] Provider SDP graph constraints initialized.")
def extract_from_pdf(self, pdf_path: str, provider_name: str, project_name: str) -> ProviderSDPPayload:
"""Uploads the Provider SDP PDF to Gemini and performs structured extraction."""
print(f"[Gemini] Uploading provider SDP: {pdf_path}...")
uploaded_file = self.client.files.upload(
file=pdf_path,
config=types.UploadFileConfig(
display_name=f"{provider_name}_SDP",
mime_type="application/pdf"
)
)
print(f"[Gemini] Upload complete. URI: {uploaded_file.uri}")
time.sleep(3)
extraction_prompt = (
f"You are a NASA Software Quality & Mission Assurance auditor. "
f"Analyze the attached Software Development Plan (SDP) from {provider_name} for the {project_name} project. "
"Decompose the document by its sections. For each section, identify:\n"
"1. Which NASA NPR 7150.2D requirements (SWE-xxx IDs) it targets or claims to satisfy.\n"
"2. The concrete technical processes, testing workflows, and review gates defined.\n"
"3. The responsible roles/authorities executing each process.\n"
"4. The specific artifacts, logs, repositories, or records produced.\n"
"5. Any explicit tailoring, deviations, or waiver rationales."
)
print("[Gemini] Decomposing SDP into compliance structures...")
response = self.client.models.generate_content(
model="gemini-2.5-flash",
contents=[uploaded_file, extraction_prompt],
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=ProviderSDPPayload,
temperature=0.0
)
)
try:
self.client.files.delete(name=uploaded_file.name)
print("[Gemini] Uploaded PDF removed from Files API storage.")
except Exception:
pass
payload = ProviderSDPPayload.model_validate_json(response.text)
print(f"[Gemini] Extracted {len(payload.sections)} SDP sections with associated workflows and artifacts.")
return payload
def populate_neo4j(self, payload: ProviderSDPPayload):
"""Writes the provider overlay subgraph and binds it to the NPR 7150.2D backbone."""
cypher_tx = """
// 1. Create or match the Provider Document Node
MERGE (doc:ProviderDocument {
provider: $provider,
project: $project,
revision: $revision
})
SET doc.title = $doc_title,
doc.updated_at = timestamp()
WITH doc
UNWIND $sections AS sec_data
// 2. Create Section Node
MERGE (sec:SDPSection {
doc_id: id(doc),
section_id: sec_data.section_id
})
SET sec.title = sec_data.section_title,
sec.summary = sec_data.summary,
sec.tailoring = sec_data.tailoring_or_waivers
MERGE (doc)-[:HAS_SECTION]->(sec)
// 3. Link Section to Target NASA SWE Requirements (Traceability Link)
FOREACH (swe IN sec_data.targeted_swe_ids |
MERGE (r:SWERequirement {swe_id: trim(swe)})
MERGE (sec)-[:ADDRESSES_REQUIREMENT {
status: 'PROPOSED',
provider: $provider,
revision: $revision
}]->(r)
)
// 4. Instantiate Processes and Roles
FOREACH (proc IN sec_data.processes |
MERGE (p:ProcessActivity {
name: trim(proc.name),
provider: $provider
})
SET p.description = proc.description,
p.cadence = proc.cadence_or_trigger
MERGE (sec)-[:DEFINES_PROCESS]->(p)
MERGE (role:Role {name: trim(proc.responsible_role)})
MERGE (p)-[:OWNED_BY_ROLE]->(role)
)
// 5. Instantiate Artifacts & Connect to Standard Artifacts
FOREACH (art IN sec_data.artifacts |
MERGE (a:ProviderArtifact {
name: trim(art.name),
provider: $provider
})
SET a.format = art.format_or_tool,
a.lifecycle_phase = art.lifecycle_phase
MERGE (sec)-[:PRODUCES_ARTIFACT]->(a)
// Match against Standard NASA Artifacts if mapped
FOREACH (_ IN CASE WHEN art.maps_to_standard_artifact IS NOT NULL THEN [1] ELSE [] END |
MERGE (std_art:StandardArtifact {name: trim(art.maps_to_standard_artifact)})
MERGE (a)-[:SATISFIES_ARTIFACT {provider: $provider}]->(std_art)
)
)
"""
sec_dicts = [s.model_dump() for s in payload.sections]
with self.driver.session() as session:
session.run(
cypher_tx,
provider=payload.provider_name,
project=payload.project_name,
revision=payload.revision,
doc_title=payload.document_title,
sections=sec_dicts
)
print(f"[Neo4j] Successfully merged {payload.provider_name} SDP into knowledge graph.")
# ==============================================================================
# 3. Execution Entry Point
# ==============================================================================
if __name__ == "__main__":
SDP_PDF_PATH = "Provider_X_Flight_Software_Plan_RevC.pdf"
PROVIDER = "Acme Launch Systems"
PROJECT = "Peregrine Flight Software"
NEO4J_URI = os.getenv("NEO4J_URI", "bolt://localhost:7687")
NEO4J_USER = os.getenv("NEO4J_USER", "neo4j")
NEO4J_PASS = os.getenv("NEO4J_PASSWORD", "password")
if not os.path.exists(SDP_PDF_PATH):
print(f"Error: SDP file '{SDP_PDF_PATH}' not found in current directory.")
exit(1)
builder = SDPGraphBuilder(NEO4J_URI, NEO4J_USER, NEO4J_PASS)
try:
builder.init_graph_constraints()
sdp_data = builder.extract_from_pdf(SDP_PDF_PATH, PROVIDER, PROJECT)
builder.populate_neo4j(sdp_data)
finally:
builder.close()
Automated Compliance & Gap Traversal Queries
Once both the NPR 7150.2D Master Backbone and the Provider SDP exist in the same database, run these Cypher queries to audit compliance.
1. Zero-Coverage Gap Analysis (Missing Requirements for Class A)
Finds all Class A SWE-xxx requirements that have no mapped section or claim from the provider:
MATCH (r:SWERequirement)-[:MANDATED_FOR_CLASS]->(:SoftwareClass {name: "A"})
WHERE NOT ()-[:ADDRESSES_REQUIREMENT]->(r)
RETURN r.swe_id AS unaddressed_swe, r.title AS requirement_title, r.full_text AS text
ORDER BY r.swe_id;
2. Artifact Delivery Gap Analysis
Checks if an applicable SWE requirement mandates an artifact (e.g., Peer Review Records), but the provider's SDP fails to generate a corresponding deliverable satisfying it:
MATCH (r:SWERequirement)-[:MANDATES_ARTIFACT]->(std_art:StandardArtifact)
MATCH (sec:SDPSection)-[:ADDRESSES_REQUIREMENT]->(r)
OPTIONAL MATCH (sec)-[:PRODUCES_ARTIFACT]->(prov_art:ProviderArtifact)-[:SATISFIES_ARTIFACT]->(std_art)
WITH r.swe_id AS swe, std_art.name AS expected_artifact, collect(prov_art.name) AS provided_artifacts
WHERE size(provided_artifacts) = 0
RETURN swe, expected_artifact, "Missing Required Artifact" AS audit_status;
3. Full End-to-End Compliance Matrix View
Generates a complete compliance report from the graph, joining standard requirements with the provider's section, process activity, and responsible role:
MATCH (r:SWERequirement)
OPTIONAL MATCH (sec:SDPSection)-[:ADDRESSES_REQUIREMENT]->(r)
OPTIONAL MATCH (sec)-[:DEFINES_PROCESS]->(p:ProcessActivity)-[:OWNED_BY_ROLE]->(role:Role)
RETURN
r.swe_id AS swe_id,
r.title AS npr_requirement,
coalesce(sec.section_id, "UNMAPPED") AS sdp_section,
coalesce(p.name, "No Defined Process") AS provider_process,
coalesce(role.name, "Unassigned") AS accountable_role
ORDER BY r.swe_id;Create a Python script that runs Cypher gap queries across Neo4j and exports a formatted Excel/CSV NASA Compliance Matrix.
Your Compliance Matrix script and sample audit workbooks are ready:
Formatted Excel Audit Workbook:
Standard Flat CSV Matrix:
Overview of the Python Script Architecture
The script (export_compliance_matrix.py) executes Cypher gap-analysis queries directly against the merged Neo4j knowledge graph and formats the results into an Executive Summary Dashboard, a Traceability Matrix, and an Actionable Gap Register.
1. Core Cypher Traversal Queries Used
Full Traceability & Compliance Status Query:
Traverses from
StandardDocument-->SWERequirementand joins optional paths to the provider'sSDPSection,ProcessActivity,Role, andProviderArtifact:MATCH (doc:StandardDocument {id: $npr_doc_id})-[:DEFINES_REQUIREMENT]->(r:SWERequirement) OPTIONAL MATCH (r)-[:MANDATED_FOR_CLASS]->(cls:SoftwareClass) OPTIONAL MATCH (r)-[:MANDATES_ARTIFACT]->(std_art:StandardArtifact) OPTIONAL MATCH (sec:SDPSection)-[rel:ADDRESSES_REQUIREMENT]->(r) OPTIONAL MATCH (sec)<-[:HAS_SECTION]-(pdoc:ProviderDocument {provider: $provider_name}) OPTIONAL MATCH (sec)-[:DEFINES_PROCESS]->(p:ProcessActivity)-[:OWNED_BY_ROLE]->(role:Role) OPTIONAL MATCH (sec)-[:PRODUCES_ARTIFACT]->(prov_art:ProviderArtifact)-[:SATISFIES_ARTIFACT]->(std_art) WITH r, collect(DISTINCT cls.name) AS target_classes, collect(DISTINCT std_art.name) AS required_artifacts, sec, rel, p, role, collect(DISTINCT prov_art.name) AS matching_artifacts RETURN r.swe_id AS swe_id, r.title AS requirement_title, r.section AS npr_section, target_classes AS applicable_classes, required_artifacts AS mandated_artifacts, sec.section_id AS sdp_section_id, sec.title AS sdp_section_title, sec.summary AS sdp_evidence_summary, p.name AS provider_process, role.name AS responsible_role, matching_artifacts AS provided_artifacts, sec.tailoring AS tailoring_waiver, CASE WHEN sec IS NULL THEN 'NON_COMPLIANT' WHEN size(required_artifacts) > 0 AND size(matching_artifacts) = 0 THEN 'PARTIAL_GAP' WHEN sec.tailoring IS NOT NULL THEN 'TAILORED' ELSE 'COMPLIANT' END AS compliance_status ORDER BY r.swe_id;
2. Multi-Tab Excel Workbook Layout
Executive Summary Tab (
Executive Summary):Metadata header (Provider name, Project/Vehicle, Class baseline, Audit timestamp).
Aggregated counts across compliance states (
COMPLIANT,PARTIAL_GAP,NON_COMPLIANT,TAILORED).Embedded native Pie Chart visualizing compliance distribution.
Compliance Matrix Tab (
Compliance Matrix):Detailed mapping across all
SWE-xxxprovisions.Conditional fills (Muted Green = Compliant, Muted Amber = Partial Gap, Muted Red = Non-Compliant, Soft Blue = Tailored).
Frozen column panes and dynamic cell text wrapping.
Deficiency & Gap Analysis Tab (
Gap Analysis & Deficiencies):Automatically isolates items flagged as
NON_COMPLIANTorPARTIAL_GAP.Maps out the specific gap category (e.g., Unmapped Requirement vs. Missing Artifact) with actionable remediation instructions for the provider.
Running the Script
python export_compliance_matrix.py \
--uri bolt://localhost:7687 \
--user neo4j \
--password your_password \
--npr_doc "NPR 7150.2D" \
--provider "Acme Launch Systems" \
--project "Peregrine Flight Software" \
--output_xlsx "NASA_NPR_7150_2D_Compliance_Matrix.xlsx" \
--output_csv "NASA_NPR_7150_2D_Compliance_Matrix.csv"Provide a step-by-step workflow for me to implement this on my NASA laptop using VS Code IDE, supported by Neo4j on my laptop as well.
Here is the end-to-end, step-by-step workflow to stand up, execute, and inspect this entire compliance knowledge graph pipeline locally on your machine using VS Code and a local Neo4j instance.
Step 1: Set Up Local Neo4j Instance
You can run Neo4j locally either via Neo4j Desktop or via Docker Desktop / Podman.
Option A: Neo4j Desktop (GUI Setup)
Open Neo4j Desktop and create a new project namedNASA_Compliance_Audit.Click Add --> Local DBMS.Set a password (e.g.,password123) and start the database.Verify it is running on the default ports (bolt://localhost:7687andhttp://localhost:7474).
Option B: Containerized Neo4j (Terminal Setup)
If using Docker/Podman, run:
docker run -d \
--name neo4j-compliance \
-p 7474:7474 -p 7687:7687 \
-e NEO4J_AUTH=neo4j/password123 \
-v $HOME/neo4j_data:/data \
neo4j:5.26-community
Step 2: Configure the VS Code Workspace
Create a dedicated project directory and open it in VS Code:mkdir nasa-compliance-kg cd nasa-compliance-kg code .Open the built-in terminal in VS Code (Ctrl + ~orCtrl + `).Create and activate a dedicated Python virtual environment:# Windows (PowerShell) python -m venv .venv .\.venv\Scripts\Activate.ps1 # macOS / Linux / WSL python3 -m venv .venv source .venv/bin/activateInstall required libraries:pip install google-genai pydantic neo4j openpyxl pandasInstall recommended VS Code Extensions for graph workflows:Python (ms-python.python)Pylance (ms-python.vscode-pylance)Cypher Query Language (freefrag.language-cypherorneo4j.cypher)Excel Viewer (GrapeCity.gc-excelviewer)
Step 3: Configure Environment Variables
Create a .env file in the root of your project:
GEMINI_API_KEY="your-gemini-api-key-here"
NEO4J_URI="bolt://localhost:7687"
NEO4J_USER="neo4j"
NEO4J_PASSWORD="password123"
Configure VS Code to load your environment file automatically by creating .vscode/settings.json:
{
"python.envFile": "${workspaceFolder}/.env"
}
Step 4: Organize Project Directory Structure
Place your PDF documents and scripts into the root folder:
nasa-compliance-kg/
│
├── .venv/
├── .vscode/
│ └── settings.json
├── .env
│
├── NPR_7150_2D.pdf # Official NASA NPR 7150.2D document
├── Provider_X_Flight_Software_Plan.pdf # Launch provider's SDP PDF
│
├── build_npr_graph.py # Step A: NPR 7150.2D backbone ingestion
├── build_sdp_overlay_graph.py # Step B: Provider SDP overlay ingestion
└── export_compliance_matrix.py # Step C: Cypher gap queries & Excel/CSV exporter
Step 5: Execution Pipeline (Three-Stage Run)
Execute the three scripts sequentially from the VS Code integrated terminal:
1. Ingest NPR 7150.2D Master Backbone
python build_npr_graph.py
Output: Creates constraints, parses
SWE-xxxrequirements and Class A–E applicability matrices from the PDF, and loads the immutable reference graph.
2. Ingest Provider Software Development Plan Overlay
python build_sdp_overlay_graph.py
Output: Extracts SDP sections, processes, roles, and artifact claims, then establishes candidate compliance cross-links (
[:ADDRESSES_REQUIREMENT],[:SATISFIES_ARTIFACT]).
3. Export Automated Compliance Matrix & Gap Analysis
python export_compliance_matrix.py \
--uri "bolt://localhost:7687" \
--user "neo4j" \
--password "password123" \
--npr_doc "NPR 7150.2D" \
--provider "Acme Launch Systems" \
--project "Peregrine Flight Software"
Output: Produces
NASA_NPR_7150_2D_Compliance_Matrix.xlsx(with Executive Summary chart, detailed matrix, and gap register) andNASA_NPR_7150_2D_Compliance_Matrix.csv[cite: 1].
Step 6: Visualizing and Querying in Neo4j Browser
Open your browser and navigate to
http://localhost:7474.Connect using
neo4jand your local password.Run the following interactive queries to inspect the compliance graph topology:
Trace Flight Software (Class A) Full Coverage
MATCH path = (doc:StandardDocument)-[:DEFINES_REQUIREMENT]->(r:SWERequirement)-[:MANDATED_FOR_CLASS]->(:SoftwareClass {name: "A"})
OPTIONAL MATCH p2 = (sec:SDPSection)-[:ADDRESSES_REQUIREMENT]->(r)
RETURN path, p2
LIMIT 50;
Identify All Unaddressed NPR Requirements (Red Flags)
MATCH (r:SWERequirement)-[:MANDATED_FOR_CLASS]->(:SoftwareClass {name: "A"})
WHERE NOT ()-[:ADDRESSES_REQUIREMENT]->(r)
RETURN r.swe_id, r.title, r.full_text;
Inspect Role Accountability Across Verification Gates
MATCH (sec:SDPSection)-[:DEFINES_PROCESS]->(p:ProcessActivity)-[:OWNED_BY_ROLE]->(role:Role)
MATCH (sec)-[:ADDRESSES_REQUIREMENT]->(r:SWERequirement)
RETURN r.swe_id, sec.section_id, p.name, role.name;From <https://gemini.google.com/app/e82a7c3cd2797647> Google Gemini (3.7 Flash)


Comments
Post a Comment