#4385 System Breakdown to JSON and Python Application to Add to Knowledge Graph
#4385 System Breakdown to JSON and Python Application to Add to Knowledge Graph
#4385 Provide a JSON string for system domain, system within the domain, subsystems of the system, components of each subsystem, parts of each component, each having a function field
{
"system_domain": {
"name": "Automotive Engineering",
"function": "To provide safe, efficient, and controlled personal mobility.",
"system": {
"name": "Electric Passenger Vehicle",
"function": "To transport occupants using electrical energy conversion.",
"subsystems": [
{
"name": "Powertrain Subsystem",
"function": "To convert stored electrical energy into mechanical kinetic energy.",
"components": [
{
"name": "Electric Motor",
"function": "To generate rotational torque via electromagnetism.",
"parts": [
{
"name": "Rotor",
"function": "The rotating part that delivers mechanical work to the shaft."
},
{
"name": "Stator",
"function": "The stationary part that creates a magnetic field using copper windings."
}
]
},
{
"name": "Inverter",
"function": "To convert DC power from the battery to AC power for the motor.",
"parts": [
{
"name": "IGBT Modules",
"function": "High-speed switches that modulate power frequency and voltage."
},
{
"name": "Control Board",
"function": "Processes driver inputs to manage torque commands."
}
]
}
]
},
{
"name": "Energy Storage Subsystem",
"function": "To store and manage high-voltage electrical energy.",
"components": [
{
"name": "Battery Pack",
"function": "To provide a reservoir of chemical energy.",
"parts": [
{
"name": "Lithium-Ion Cells",
"function": "The fundamental units that store charge via ion movement."
},
{
"name": "Busbars",
"function": "Conductive plates that connect cells in series or parallel."
}
]
},
{
"name": "Battery Management System (BMS)",
"function": "To monitor and protect the battery cells.",
"parts": [
{
"name": "Voltage Sensors",
"function": "To measure the electrical potential of individual cell groups."
},
{
"name": "Thermal Sensors",
"function": "To monitor temperature and prevent thermal runaway."
}
]
}
]
}
]
}
}
}
Provide Python snippet to extract the values from the JSON string and create the Cypher Merge statements to add them to a knowledge graph
To process this hierarchical JSON and transform it into Cypher MERGE statements, we need to traverse the nested structure. Using MERGE ensures that we don't create duplicate nodes if the script is run multiple times.
I have included a Python snippet that uses a recursive-style approach to generate the relationships (e.g., HAS_SUBSYSTEM, HAS_COMPONENT) and maintains the function property for every node.
import json
# The JSON data from the previous step
json_data = '''
{
"system_domain": {
"name": "Automotive Engineering",
"function": "To provide safe, efficient, and controlled personal mobility.",
"system": {
"name": "Electric Passenger Vehicle",
"function": "To transport occupants using electrical energy conversion.",
"subsystems": [
{
"name": "Powertrain Subsystem",
"function": "To convert stored electrical energy into mechanical kinetic energy.",
"components": [
{
"name": "Electric Motor",
"function": "To generate rotational torque via electromagnetism.",
"parts": [
{
"name": "Rotor",
"function": "The rotating part that delivers mechanical work to the shaft."
},
{
"name": "Stator",
"function": "The stationary part that creates a magnetic field using copper windings."
}
]
}
]
}
]
}
}
}
'''
data = json.loads(json_data)
statements = []
def create_merge(label, name, function):
"""Formats a Cypher MERGE statement."""
# Escaping single quotes for Cypher safety
safe_name = name.replace("'", "\\'")
safe_func = function.replace("'", "\\'")
return f"MERGE (n:{label} {{name: '{safe_name}'}}) ON CREATE SET n.function = '{safe_func}'"
def create_rel(parent_label, parent_name, child_label, child_name, rel_type):
"""Formats a Cypher Relationship statement."""
p_name = parent_name.replace("'", "\\'")
c_name = child_name.replace("'", "\\'")
return (f"MATCH (p:{parent_label} {{name: '{p_name}'}}), (c:{child_label} {{name: '{c_name}'}}) "
f"MERGE (p)-[:{rel_type}]->(c)")
# Extract Domain
domain = data['system_domain']
statements.append(create_merge("Domain", domain['name'], domain['function']))
# Extract System
system = domain['system']
statements.append(create_merge("System", system['name'], system['function']))
statements.append(create_rel("Domain", domain['name'], "System", system['name'], "HAS_SYSTEM"))
# Extract Subsystems, Components, and Parts
for sub in system.get('subsystems', []):
statements.append(create_merge("Subsystem", sub['name'], sub['function']))
statements.append(create_rel("System", system['name'], "Subsystem", sub['name'], "HAS_SUBSYSTEM"))
for comp in sub.get('components', []):
statements.append(create_merge("Component", comp['name'], comp['function']))
statements.append(create_rel("Subsystem", sub['name'], "Component", comp['name'], "HAS_COMPONENT"))
for part in comp.get('parts', []):
statements.append(create_merge("Part", part['name'], part['function']))
statements.append(create_rel("Component", comp['name'], "Part", part['name'], "HAS_PART"))
# Output the result
for stmt in statements:
print(stmt + ";")
How this works:
Nodes: It creates nodes with specific labels (
Domain,System,Subsystem,Component,Part).Properties: It uses
ON CREATE SETfor thefunctionfield so that if the node already exists, we don't accidentally overwrite data unless you want to.Relationships: It uses
MATCHto find the existing parent and child nodes, thenMERGEto link them with a directional relationship (e.g.,HAS_PART).Formatting: It handles basic string escaping to prevent syntax errors if your component names contain apostrophes.
:::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::
Skip some prompts to get this right ...
:::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::::
I need these to work together EVERY TIME, no errors allowed
To guarantee 100% compatibility between the AI's output and your Python script, you must use JSON Schema constraints. AI models perform significantly better when they are told exactly which keys are required and that additional properties are forbidden.
Here is the "Bulletproof" package to ensure they work together every time.
1. The Prompt (Copy-Paste this)
Task: Generate a system hierarchy JSON. Rules: > 1. You must return ONLY raw JSON. 2. You must follow the schema provided below exactly. 3. Do not omit any levels. 4. Do not add keys outside of "name" and "function".
JSON Structure:
{ "system_domain": { "name": "string", "function": "string", "system": { "name": "string", "function": "string", "subsystems": [ { "name": "string", "function": "string", "components": [ { "name": "string", "function": "string", "parts": [ { "name": "string", "function": "string" } ] } ] } ] } } }
2. The Robust Python Script
I have updated the script with error handling. Even if the AI misses a level, this script will use .get() to prevent a crash, ensuring your pipeline stays alive.
import json
def generate_cypher(json_input):
try:
data = json.loads(json_input)
except json.JSONDecodeError:
return "Error: Invalid JSON provided."
statements = []
def escape(text):
return str(text).replace("'", "\\'")
# 1. Domain
dom = data.get('system_domain', {})
d_name, d_func = escape(dom.get('name', 'Unknown')), escape(dom.get('function', 'N/A'))
statements.append(f"MERGE (d:Domain {{name: '{d_name}'}}) ON CREATE SET d.function = '{d_func}';")
# 2. System
sys = dom.get('system', {})
s_name, s_func = escape(sys.get('name', 'Unknown')), escape(sys.get('function', 'N/A'))
statements.append(f"MERGE (s:System {{name: '{s_name}'}}) ON CREATE SET s.function = '{s_func}';")
statements.append(f"MATCH (d:Domain {{name: '{d_name}'}}), (s:System {{name: '{s_name}'}}) MERGE (d)-[:HAS_SYSTEM]->(s);")
# 3. Subsystems
for sub in sys.get('subsystems', []):
sub_name, sub_func = escape(sub.get('name', 'N/A')), escape(sub.get('function', 'N/A'))
statements.append(f"MERGE (ss:Subsystem {{name: '{sub_name}'}}) ON CREATE SET ss.function = '{sub_func}';")
statements.append(f"MATCH (s:System {{name: '{s_name}'}}), (ss:Subsystem {{name: '{sub_name}'}}) MERGE (s)-[:HAS_SUBSYSTEM]->(ss);")
# 4. Components
for comp in sub.get('components', []):
c_name, c_func = escape(comp.get('name', 'N/A')), escape(comp.get('function', 'N/A'))
statements.append(f"MERGE (c:Component {{name: '{c_name}'}}) ON CREATE SET c.function = '{c_func}';")
statements.append(f"MATCH (ss:Subsystem {{name: '{sub_name}'}}), (c:Component {{name: '{c_name}'}}) MERGE (ss)-[:HAS_COMPONENT]->(c);")
# 5. Parts
for part in comp.get('parts', []):
p_name, p_func = escape(part.get('name', 'N/A')), escape(part.get('function', 'N/A'))
statements.append(f"MERGE (p:Part {{name: '{p_name}'}}) ON CREATE SET p.function = '{p_func}';")
statements.append(f"MATCH (c:Component {{name: '{c_name}'}}), (p:Part {{name: '{p_name}'}}) MERGE (c)-[:HAS_PART]->(p);")
return "\n".join(statements)
# Example Usage:
# cypher_output = generate_cypher(ai_response_string)
# print(cypher_output)
Why this works "Every Time":
Escape Character Handling: The
escape()function prevents Cypher from breaking if the AI uses names like "Driver's Seat" (the single quote would normally crash the query).Null Safety: By using
.get('key', []), the script won't crash if the AI accidentally provides a component without any parts.Strict Prompting: The prompt defines the exact nesting. In the AI world, "Template" is a stronger word than "Example"—it forces the model to fill in the blanks of your specific structure.
From <https://gemini.google.com/app/> Google Gemini 3.0 (Fast)
Comments
Post a Comment