#4801 System Breakdown +Application of Ideal Final Result and SCAMPER
#4801 System Breakdown +Application of Ideal Final Result and SCAMPER
#4801 Describe a python generative-AI app to break down a user-specified system (e.g. bicycle) into constituent sub-systems, components and parts, apply TRIZ Ideal Final Result (IFR) and SCAMPER to each, and report all in a markdown file
To build this application, the most reliable architecture couples a structured schema layer (Pydantic) with an LLM execution pipeline to enforce clean hierarchy decomposition before passing each element through TRIZ and SCAMPER creative operators.
Here is the architectural blueprint, data model, prompt design, and complete implementation.
System Architecture & Pipeline
[User Input: e.g. "Bicycle"]
│
▼
[Phase 1: Decomposition Agent] ──▶ Outputs recursive Subsystems & Components (Pydantic)
│
▼
[Phase 2: Innovation Engine] ──▶ Applies TRIZ (IFR) & SCAMPER per element
│
▼
[Phase 3: Markdown Generator] ──▶ Formats hierarchical report & exports `.md`
Hierarchical Decomposition: The LLM decomposes the target system across three tiers:
Subsystem: Primary functional groups (e.g., Drivetrain, Braking System, Frame & Steering).
Component: Operational assemblies within a subsystem (e.g., Derailleur, Caliper assembly).
Part: Discrete mechanical or electrical units (e.g., Roller chain, Brake pad, Cable tension spring).
Systematic Ideation:
TRIZ IFR (Ideal Final Result): Formulates how the part can deliver its primary useful function by itself, with zero cost, zero mass, or by utilizing existing resources.
SCAMPER: Systematically runs Substitute, Combine, Adapt, Modify/Magnify, Put to another use, Eliminate, and Reverse/Rearrange against the part's boundaries.
Structured Output Enforcement: Structured generation ensures the LLM returns well-formed JSON matching the recursive data schema, avoiding unstructured text parsing issues.
Data Models & Prompt Engineering
1. Target Schemas (schemas.py)
from typing import List, Optional
from pydantic import BaseModel, Field
class ScamperTechnique(BaseModel):
substitute: str = Field(description="Alternative materials, energy sources, or mechanisms.")
combine: str = Field(description="Integration with adjacent parts or shared functions.")
adapt: str = Field(description="Borrowing mechanisms from other domains or nature.")
modify_magnify: str = Field(description="Altering shape, scale, motion, or frequency.")
put_to_another_use: str = Field(description="Secondary utilities or lifecycle repurposing.")
eliminate: str = Field(description="Removing the element and shifting its load to the system.")
reverse_rearrange: str = Field(description="Inverting the sequence, geometry, or operational order.")
class IdeationReport(BaseModel):
primary_function: str = Field(description="Core Useful Function of this unit.")
triz_ifr: str = Field(description="Ideal Final Result: function performs itself with no extra cost/weight.")
scamper: ScamperTechnique
class Part(BaseModel):
name: str
description: str
ideation: IdeationReport
class Component(BaseModel):
name: str
description: str
parts: List[Part]
ideation: Optional[IdeationReport] = None
class Subsystem(BaseModel):
name: str
description: str
components: List[Component]
class SystemDecomposition(BaseModel):
system_name: str
summary: str
subsystems: List[Subsystem]
Complete Python Implementation
This script utilizes google-genai (or compatible structured-output SDKs) to decompose the system and compile the Markdown report.
import os
from google import genai
from google.genai import types
from schemas import SystemDecomposition
SYSTEM_PROMPT = """
You are an expert mechanical engineer and systematic innovation practitioner certified in TRIZ and SCAMPER methodologies.
Your task:
1. Deconstruct the user-specified machine or system into its distinct Subsystems -> Components -> Parts.
2. For each element, establish:
- The Primary Useful Function.
- The TRIZ Ideal Final Result (IFR) formulation: "The system/part performs [action] by itself without introducing harm, cost, or complexity."
- Detailed, concrete SCAMPER interventions (avoid generic platitudes; suggest concrete materials, physics effects, or geometric shifts).
"""
def generate_system_analysis(target_system: str, api_key: str | None = None) -> SystemDecomposition:
client = genai.Client(api_key=api_key or os.environ.get("GEMINI_API_KEY"))
response = client.models.generate_content(
model="gemini-2.5-flash",
contents=f"Analyze and break down this system completely: {target_system}",
config=types.GenerateContentConfig(
system_instruction=SYSTEM_PROMPT,
response_mime_type="application/json",
response_schema=SystemDecomposition,
temperature=0.3,
),
)
return SystemDecomposition.model_validate_json(response.text)
def export_to_markdown(data: SystemDecomposition, filename: str) -> None:
lines = [
f"# Systematic Innovation Analysis: {data.system_name.title()}\n",
f"> **System Overview:** {data.summary}\n",
"---\n",
]
for sub in data.subsystems:
lines.append(f"## Subsystem: {sub.name}\n")
lines.append(f"*{sub.description}*\n")
for comp in data.components:
lines.append(f"### Component: {comp.name}\n")
lines.append(f"*{comp.description}*\n")
for part in comp.parts:
ideation = part.ideation
lines.append(f"#### Part: {part.name}\n")
lines.append(f"- **Description:** {part.description}")
lines.append(f"- **Primary Useful Function:** {ideation.primary_function}")
lines.append(f"- **TRIZ Ideal Final Result (IFR):** {ideation.triz_ifr}\n")
lines.append("| SCAMPER Lens | Proposed Innovation |")
lines.append("| :--- | :--- |")
lines.append(f"| **Substitute** | {ideation.scamper.substitute} |")
lines.append(f"| **Combine** | {ideation.scamper.combine} |")
lines.append(f"| **Adapt** | {ideation.scamper.adapt} |")
lines.append(f"| **Modify/Magnify** | {ideation.scamper.modify_magnify} |")
lines.append(f"| **Put to Another Use** | {ideation.scamper.put_to_another_use} |")
lines.append(f"| **Eliminate** | {ideation.scamper.eliminate} |")
lines.append(f"| **Reverse/Rearrange** | {ideation.scamper.reverse_rearrange} |\n")
with open(filename, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
print(f"Analysis saved to {filename}")
if __name__ == "__main__":
target = input("Enter system to analyze (e.g., Bicycle): ") or "Bicycle"
analysis = generate_system_analysis(target)
export_to_markdown(analysis, f"{target.lower().replace(' ', '_')}_innovation_report.md")
Sample Output Snippet (bicycle_innovation_report.md)
## Subsystem: Transmission & Drivetrain
*Transfers mechanical power generated by the rider's legs to the rear drive wheel.*
### Component: Chain Drive Assembly
*Mechanical linkages routing power via tension.*
#### Part: Roller Chain
- **Description:** Series of interconnected inner/outer steel plates, pins, and bushings.
- **Primary Useful Function:** Transmit tensile rotational force across a distance.
- **TRIZ Ideal Final Result (IFR):** The pedal crank directly couples to the wheel rim via contactless magnetic gear induction, eliminating physical chain links, friction, and maintenance entirely.
| SCAMPER Lens | Proposed Innovation |
| :--- | :--- |
| **Substitute** | Replace steel plates with carbon-nanotube continuous cords or self-lubricating polymer belts. |
| **Combine** | Integrate the chain with internal sensor links that read chain tension and cadence directly. |
| **Adapt** | Borrow the fluid-pumping mechanism of hydraulic drives to replace links with enclosed hydrostatic channels. |
| **Modify/Magnify** | Apply an asymmetric link profile that matches cadence stroke dead spots. |
| **Put to Another Use** | Embed piezoelectric elements that harvest vibrational energy from link engagement to power bike lights. |
| **Eliminate** | Remove the chain completely; use a shaft drive, enclosed planetary gear set, or hub-mounted motor-generator drive-by-wire. |
| **Reverse/Rearrange** | Drive the front wheel rather than the rear wheel to shorten the linkage line to the handlebar/pedal center. |Show how to batch the decomposition over multiple LLM calls so complex machines like cars don't hit output token limits.
Breaking down a complex system in a single LLM call inevitably hits output token limits, triggers premature truncation, or forces the model to gloss over parts with shallow responses.
The solution is an asynchronous Map-Reduce tree traversal:
Root Step: Identify high-level subsystems.
Fan-Out 1 (Map): In parallel, break each subsystem into components.
Fan-Out 2 (Map): In parallel, break each component into parts and run TRIZ/SCAMPER.
Reduce Step: Reassemble the tree in memory and write the final report.
Layered Schemas (batch_schemas.py)
Decouple the single massive schema into independent, single-responsibility models for each step.
from typing import List
from pydantic import BaseModel, Field
# --- Step 1: Subsystem Level ---
class SubsystemStub(BaseModel):
name: str
description: str
class RootDecomposition(BaseModel):
system_name: str
summary: str
subsystems: List[SubsystemStub]
# --- Step 2: Component Level ---
class ComponentStub(BaseModel):
name: str
description: str
class SubsystemBreakdown(BaseModel):
components: List[ComponentStub]
# --- Step 3: Part & Ideation Level ---
class ScamperTechnique(BaseModel):
substitute: str
combine: str
adapt: str
modify_magnify: str
put_to_another_use: str
eliminate: str
reverse_rearrange: str
class PartWithIdeation(BaseModel):
name: str
description: str
primary_function: str
triz_ifr: str
scamper: ScamperTechnique
class ComponentBreakdown(BaseModel):
parts: List[PartWithIdeation]
# --- Final Aggregated Models (For Report Generation) ---
class FullComponent(BaseModel):
name: str
description: str
parts: List[PartWithIdeation]
class FullSubsystem(BaseModel):
name: str
description: str
components: List[FullComponent]
class FullSystemReport(BaseModel):
system_name: str
summary: str
subsystems: List[FullSubsystem]
Asynchronous Batched Pipeline (pipeline.py)
Using asyncio.Semaphore bounds the maximum number of concurrent requests to prevent provider rate-limit spikes (HTTP 429).
import os
import asyncio
from google import genai
from google.genai import types
from batch_schemas import (
RootDecomposition,
SubsystemBreakdown,
ComponentBreakdown,
FullComponent,
FullSubsystem,
FullSystemReport,
SubsystemStub,
ComponentStub
)
client = genai.Client(api_key=os.environ.get("GEMINI_API_KEY"))
# Limit concurrent API calls to stay within rate quotas
SEMAPHORE = asyncio.Semaphore(5)
async def call_llm(contents: str, schema, system_prompt: str):
"""Wrapper to run synchronous SDK calls in an async thread pool with rate-limiting."""
async with SEMAPHORE:
return await asyncio.to_thread(
client.models.generate_content,
model="gemini-2.5-flash",
contents=contents,
config=types.GenerateContentConfig(
system_instruction=system_prompt,
response_mime_type="application/json",
response_schema=schema,
temperature=0.2,
)
)
# --- Phase 1: Break into Subsystems ---
async def get_root_subsystems(system_name: str) -> RootDecomposition:
prompt = (
"Identify the major mechanical, electrical, and structural subsystems "
f"of the following machine: {system_name}. Limit to 4-7 primary subsystems."
)
res = await call_llm(
contents=f"Analyze root system: {system_name}",
schema=RootDecomposition,
system_prompt=prompt
)
return RootDecomposition.model_validate_json(res.text)
# --- Phase 2: Break Subsystem into Components ---
async def get_components(system_name: str, sub: SubsystemStub) -> SubsystemBreakdown:
prompt = (
f"You are analyzing the '{sub.name}' subsystem of a {system_name}. "
f"Subsystem context: {sub.description}. "
"Decompose this subsystem into 3-5 critical functional components/assemblies."
)
res = await call_llm(
contents=f"Decompose subsystem: {sub.name}",
schema=SubsystemBreakdown,
system_prompt=prompt
)
return SubsystemBreakdown.model_validate_json(res.text)
# --- Phase 3: Break Component into Parts + Ideation ---
async def get_parts_and_ideation(system_name: str, sub_name: str, comp: ComponentStub) -> FullComponent:
prompt = (
f"Context: Machine = '{system_name}', Subsystem = '{sub_name}', Component = '{comp.name}'.\n"
f"Component description: {comp.description}.\n"
"1. List 2-4 critical discrete parts belonging to this component.\n"
"2. For each part, establish its Primary Useful Function.\n"
"3. Formulate the TRIZ Ideal Final Result (IFR): The part delivers its function by itself "
"without cost, mass, or complexity.\n"
"4. Provide concrete, technical SCAMPER interventions."
)
res = await call_llm(
contents=f"Generate parts and ideation for: {comp.name}",
schema=ComponentBreakdown,
system_prompt=prompt
)
data = ComponentBreakdown.model_validate_json(res.text)
return FullComponent(name=comp.name, description=comp.description, parts=data.parts)
# --- Orchestrator ---
async def process_subsystem(system_name: str, sub: SubsystemStub) -> FullSubsystem:
comp_breakdown = await get_components(system_name, sub)
# Process all components for this subsystem in parallel
comp_tasks = [
get_parts_and_ideation(system_name, sub.name, comp)
for comp in comp_breakdown.components
]
full_components = await asyncio.gather(*comp_tasks)
return FullSubsystem(name=sub.name, description=sub.description, components=full_components)
async def run_pipeline(system_name: str) -> FullSystemReport:
print(f"[*] Step 1: Identifying top-level subsystems for '{system_name}'...")
root = await get_root_subsystems(system_name)
print(f"[*] Step 2 & 3: Fanning out across {len(root.subsystems)} subsystems and nested parts...")
sub_tasks = [process_subsystem(system_name, sub) for sub in root.subsystems]
full_subsystems = await asyncio.gather(*sub_tasks)
return FullSystemReport(
system_name=root.system_name,
summary=root.summary,
subsystems=full_subsystems
)
Markdown Exporter (export.py)
Once the tree is aggregated into FullSystemReport, writing to disk requires no network activity:
from batch_schemas import FullSystemReport
def write_markdown_report(report: FullSystemReport, filepath: str):
out = [
f"# TRIZ & SCAMPER Analysis: {report.system_name.title()}",
f"> **System Scope:** {report.summary}\n",
"---\n"
]
for sub in report.subsystems:
out.append(f"## Subsystem: {sub.name}")
out.append(f"*{sub.description}*\n")
for comp in sub.components:
out.append(f"### Component: {comp.name}")
out.append(f"*{comp.description}*\n")
for part in comp.parts:
out.append(f"#### Part: {part.name}")
out.append(f"- **Description:** {part.description}")
out.append(f"- **Primary Useful Function:** {part.primary_function}")
out.append(f"- **TRIZ IFR:** {part.triz_ifr}\n")
out.append("| SCAMPER Dimension | Engineering Concept |")
out.append("| :--- | :--- |")
out.append(f"| **Substitute** | {part.scamper.substitute} |")
out.append(f"| **Combine** | {part.scamper.combine} |")
out.append(f"| **Adapt** | {part.scamper.adapt} |")
out.append(f"| **Modify/Magnify** | {part.scamper.modify_magnify} |")
out.append(f"| **Put to Other Use** | {part.scamper.put_to_another_use} |")
out.append(f"| **Eliminate** | {part.scamper.eliminate} |")
out.append(f"| **Rearrange/Reverse** | {part.scamper.reverse_rearrange} |\n")
with open(filepath, "w", encoding="utf-8") as f:
f.write("\n".join(out))
print(f"[✔] Report compiled: {filepath}")
if __name__ == "__main__":
target = "Modern Electric Vehicle (Passenger Car)"
report = asyncio.run(run_pipeline(target))
write_markdown_report(report, "ev_car_triz_report.md")
Scaling Properties & Benefits
| Dimension | Monolithic Call | Batched Map-Reduce Tree |
| Output Token Ceiling | Hard ceiling (~4k to 8k tokens depending on provider). Truncates on large systems. | Arbitrary length. Total report can scale to 100k+ tokens across calls without truncation. |
| Detail & Specificity | Broad, generic ("Brake Pad: friction material"). | Narrow context window allows deeply technical, domain-specific TRIZ/SCAMPER proposals. |
| Failure Recovery | If it fails at 90%, the entire run is lost. | Retries can be attached at the individual leaf level (tenacity or custom retry wrappers). |
| Execution Time | Single-threaded output generation (often 30–60s). | Highly parallelized via asyncio.gather; total elapsed time is bounded only by tree depth, not breadth. |
Show how to add SQLite caching with Tenacity retry logic to avoid re-running expensive LLM calls if the script is interrupted.
To make the fan-out pipeline crash-resilient and deterministic across restarts, we wrap the execution layer with two primitives:
Content-Hashed SQLite Cache: Computes a SHA-256 hash over the combined
(model, prompt, schema_name, system_instruction)tuple. If cached, it returns the raw JSON string immediately without consuming API quota.Tenacity Exponential Backoff: Catches transient API hiccups (HTTP 429 rate limits, 503 service unavailabilities, timeouts) and transparently backs off before retrying.
1. The Persistent Cache Engine (cache.py)
Using SQLite in WAL (Write-Ahead Logging) mode allows high-throughput concurrent reads and serialized writes from parallel asyncio worker threads without database lock collisions.
import sqlite3
import hashlib
import json
from typing import Optional
DB_FILE = "llm_cache.db"
def init_cache_db():
"""Initializes the cache table with WAL mode for concurrent async access."""
with sqlite3.connect(DB_FILE) as conn:
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS response_cache (
cache_key TEXT PRIMARY KEY,
model TEXT,
prompt TEXT,
response_text TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
def make_cache_key(model: str, contents: str, system_prompt: str, schema_name: str) -> str:
"""Generates an immutable SHA-256 fingerprint of the prompt inputs."""
payload = f"{model}::{system_prompt}::{contents}::{schema_name}"
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def get_cached_response(cache_key: str) -> Optional[str]:
with sqlite3.connect(DB_FILE) as conn:
cur = conn.cursor()
cur.execute("SELECT response_text FROM response_cache WHERE cache_key = ?", (cache_key,))
row = cur.fetchone()
return row[0] if row else None
def set_cached_response(cache_key: str, model: str, contents: str, response_text: str):
with sqlite3.connect(DB_FILE) as conn:
conn.execute(
"""
INSERT OR REPLACE INTO response_cache (cache_key, model, prompt, response_text)
VALUES (?, ?, ?, ?)
""",
(cache_key, model, contents, response_text)
)
2. Robust Cached Executor with Tenacity (llm_runner.py)
We decorate the raw network call with @retry from tenacity configured for exponential jitter, then layer the cache check around it.
pip install tenacity
App:
import os
import asyncio
from typing import Type, TypeVar
from pydantic import BaseModel
from google import genai
from google.genai import types
from google.genai.errors import APIError
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential,
retry_if_exception_type,
before_sleep_log
)
import logging
from cache import init_cache_db, make_cache_key, get_cached_response, set_cached_response
# Configure logger for visibility during backoff
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("LLMRunner")
T = TypeVar("T", bound=BaseModel)
client = genai.Client(api_key=os.environ.get("GEMINI_API_KEY"))
init_cache_db()
# Bounded concurrency to respect local resource and provider connection pools
SEMAPHORE = asyncio.Semaphore(5)
@retry(
retry=retry_if_exception_type((APIError, TimeoutError, ConnectionError)),
wait=wait_random_exponential(min=2, max=30),
stop=stop_after_attempt(5),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True
)
def _raw_api_call(contents: str, system_prompt: str, schema: Type[T], model: str) -> str:
"""Synchronous network call isolated with robust retry mechanics."""
response = client.models.generate_content(
model=model,
contents=contents,
config=types.GenerateContentConfig(
system_instruction=system_prompt,
response_mime_type="application/json",
response_schema=schema,
temperature=0.2,
)
)
if not response.text:
raise ValueError("Model returned an empty text payload.")
return response.text
async def call_llm_cached(
contents: str,
system_prompt: str,
schema: Type[T],
model: str = "gemini-2.5-flash"
) -> T:
"""
Checks SQLite cache -> Calls LLM via thread pool with Tenacity retry -> Caches result.
"""
cache_key = make_cache_key(model, contents, system_prompt, schema.__name__)
# 1. Fast Cache Read (No Semaphore / No Token Usage)
cached_text = get_cached_response(cache_key)
if cached_text:
logger.debug(f"[CACHE HIT] Key: {cache_key[:8]}... Schema: {schema.__name__}")
return schema.model_validate_json(cached_text)
# 2. Acquire Concurrency Slot for Network Request
async with SEMAPHORE:
# Re-check cache inside semaphore to handle identical race conditions
rechecked_text = get_cached_response(cache_key)
if rechecked_text:
return schema.model_validate_json(rechecked_text)
logger.info(f"[API CALL] Querying {model} for schema: {schema.__name__}")
raw_text = await asyncio.to_thread(
_raw_api_call,
contents=contents,
system_prompt=system_prompt,
schema=schema,
model=model
)
# 3. Commit back to SQLite cache
set_cached_response(cache_key, model, contents, raw_text)
return schema.model_validate_json(raw_text)
3. Integrating with the Pipeline Steps
The fan-out functions simply swap their inner executor for call_llm_cached. Notice how the call signatures stay identical to normal Pydantic workflows:
from batch_schemas import (
RootDecomposition,
SubsystemBreakdown,
ComponentBreakdown,
FullComponent,
SubsystemStub,
ComponentStub
)
from llm_runner import call_llm_cached
async def get_root_subsystems(system_name: str) -> RootDecomposition:
prompt = (
f"Identify the 4-7 major subsystems of this machine: {system_name}. "
"Provide clear structural descriptions."
)
return await call_llm_cached(
contents=f"Root analysis: {system_name}",
system_prompt=prompt,
schema=RootDecomposition
)
async def get_components(system_name: str, sub: SubsystemStub) -> SubsystemBreakdown:
prompt = (
f"Deconstruct '{sub.name}' of a {system_name} ({sub.description}) "
"into 3-5 functional component assemblies."
)
return await call_llm_cached(
contents=f"Deconstruct subsystem: {sub.name}",
system_prompt=prompt,
schema=SubsystemBreakdown
)
async def get_parts_and_ideation(system_name: str, sub_name: str, comp: ComponentStub) -> FullComponent:
prompt = (
f"System: {system_name} | Subsystem: {sub_name} | Assembly: {comp.name}\n"
f"Description: {comp.description}\n"
"1. List 2-4 critical parts.\n"
"2. State Primary Useful Function.\n"
"3. Provide TRIZ Ideal Final Result (IFR).\n"
"4. Formulate technical SCAMPER variations."
)
breakdown = await call_llm_cached(
contents=f"Ideation for component: {comp.name}",
system_prompt=prompt,
schema=ComponentBreakdown
)
return FullComponent(name=comp.name, description=comp.description, parts=breakdown.parts)
How Interruption Recovery Works
Run 1 (Interrupted at 70%):
├── [API CALL] Root: Modern Electric Car ──▶ Cached
├── [API CALL] Subsystem: Battery Pack ──▶ Cached
├── [API CALL] Subsystem: Powertrain ──▶ Cached
└── [CRASH / SIGINT / Quota exhaustion]
Run 2 (Restarted immediately):
├── [CACHE HIT] Root: Modern Electric Car (0.01 ms)
├── [CACHE HIT] Subsystem: Battery Pack (0.01 ms)
├── [CACHE HIT] Subsystem: Powertrain (0.01 ms)
└── [API CALL] Subsystem: Chassis ──▶ Resumes fresh calls here
Zero Duplicate Spend: Re-running the script after a timeout, crash, or manual cancel will breeze through completed nodes instantly.
Cache Invalidation: Changing the system instructions or prompt text automatically computes a new SHA-256 hash, causing the pipeline to fetch fresh responses without needing to clear the database manually.
Bring it altogether. Provide the full Python application using Google Gemini API, and add JSON output for the results as well.
Here is the complete, production-ready, self-contained Python application. It uses the google-genai SDK, decomposes systems recursively via an async Map-Reduce tree, protects against API rate limits and failures using tenacity and an SQLite cache, and writes both a formatted Markdown document and a structured JSON output file.
Prerequisites
pip install google-genai pydantic tenacity
export GEMINI_API_KEY="your-api-key-here"
Complete Code (system_innovator.py)
import os
import sys
import json
import sqlite3
import hashlib
import asyncio
import logging
from typing import List, Optional, Type, TypeVar
from pydantic import BaseModel, Field
from google import genai
from google.genai import types
from google.genai.errors import APIError
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential,
retry_if_exception_type,
before_sleep_log,
)
# ---------------------------------------------------------------------------
# Logging & Database Initialization
# ---------------------------------------------------------------------------
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("SystemInnovator")
DB_FILE = "llm_cache.db"
MODEL_ID = "gemini-2.5-flash"
MAX_CONCURRENT_CALLS = 5
def init_cache_db():
"""Initializes SQLite cache with WAL mode for safe concurrent access."""
with sqlite3.connect(DB_FILE) as conn:
conn.execute("PRAGMA journal_mode=WAL;")
conn.execute(
"""
CREATE TABLE IF NOT EXISTS response_cache (
cache_key TEXT PRIMARY KEY,
model TEXT,
prompt TEXT,
response_text TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""
)
def make_cache_key(model: str, contents: str, system_prompt: str, schema_name: str) -> str:
payload = f"{model}::{system_prompt}::{contents}::{schema_name}"
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
def get_cached_response(cache_key: str) -> Optional[str]:
with sqlite3.connect(DB_FILE) as conn:
cur = conn.cursor()
cur.execute("SELECT response_text FROM response_cache WHERE cache_key = ?", (cache_key,))
row = cur.fetchone()
return row[0] if row else None
def set_cached_response(cache_key: str, model: str, contents: str, response_text: str):
with sqlite3.connect(DB_FILE) as conn:
conn.execute(
"""
INSERT OR REPLACE INTO response_cache (cache_key, model, prompt, response_text)
VALUES (?, ?, ?, ?)
""",
(cache_key, model, contents, response_text),
)
# ---------------------------------------------------------------------------
# Pydantic Schemas
# ---------------------------------------------------------------------------
class SubsystemStub(BaseModel):
name: str = Field(description="Name of the major subsystem.")
description: str = Field(description="Operational scope and primary purpose.")
class RootDecomposition(BaseModel):
system_name: str
summary: str
subsystems: List[SubsystemStub]
class ComponentStub(BaseModel):
name: str = Field(description="Name of the component assembly.")
description: str = Field(description="Role within the subsystem.")
class SubsystemBreakdown(BaseModel):
components: List[ComponentStub]
class ScamperTechnique(BaseModel):
substitute: str = Field(description="Alternative materials, energy sources, or mechanisms.")
combine: str = Field(description="Integration with adjacent parts or shared functions.")
adapt: str = Field(description="Borrowing mechanisms from other domains or nature.")
modify_magnify: str = Field(description="Altering shape, scale, motion, or frequency.")
put_to_another_use: str = Field(description="Secondary utilities or lifecycle repurposing.")
eliminate: str = Field(description="Removing the element and shifting its load to the system.")
reverse_rearrange: str = Field(description="Inverting the sequence, geometry, or operational order.")
class PartWithIdeation(BaseModel):
name: str
description: str
primary_function: str = Field(description="Primary Useful Function (PUF).")
triz_ifr: str = Field(description="TRIZ Ideal Final Result: Function delivers itself at zero cost/mass.")
scamper: ScamperTechnique
class ComponentBreakdown(BaseModel):
parts: List[PartWithIdeation]
class FullComponent(BaseModel):
name: str
description: str
parts: List[PartWithIdeation]
class FullSubsystem(BaseModel):
name: str
description: str
components: List[FullComponent]
class FullSystemReport(BaseModel):
system_name: str
summary: str
subsystems: List[FullSubsystem]
# ---------------------------------------------------------------------------
# Resilient LLM Execution Layer
# ---------------------------------------------------------------------------
T = TypeVar("T", bound=BaseModel)
semaphore = asyncio.Semaphore(MAX_CONCURRENT_CALLS)
client = genai.Client()
@retry(
retry=retry_if_exception_type((APIError, TimeoutError, ConnectionError)),
wait=wait_random_exponential(min=2, max=30),
stop=stop_after_attempt(5),
before_sleep=before_sleep_log(logger, logging.WARNING),
reraise=True,
)
def _raw_api_call(contents: str, system_prompt: str, schema: Type[T]) -> str:
"""Executes call via official google-genai SDK with structured output enforcement."""
response = client.models.generate_content(
model=MODEL_ID,
contents=contents,
config=types.GenerateContentConfig(
system_instruction=system_prompt,
response_mime_type="application/json",
response_schema=schema,
temperature=0.2,
),
)
if not response.text:
raise ValueError("Model returned an empty payload.")
return response.text
async def call_llm_cached(contents: str, system_prompt: str, schema: Type[T]) -> T:
cache_key = make_cache_key(MODEL_ID, contents, system_prompt, schema.__name__)
# Fast cache lookup
cached = get_cached_response(cache_key)
if cached:
logger.info(f"⚡ [CACHE HIT] {schema.__name__} for: {contents[:40]}...")
return schema.model_validate_json(cached)
# Throttled network call
async with semaphore:
# Check again in case another worker cached it while waiting
cached = get_cached_response(cache_key)
if cached:
return schema.model_validate_json(cached)
logger.info(f"🌐 [API CALL] {schema.__name__} for: {contents[:40]}...")
raw_text = await asyncio.to_thread(_raw_api_call, contents, system_prompt, schema)
set_cached_response(cache_key, MODEL_ID, contents, raw_text)
return schema.model_validate_json(raw_text)
# ---------------------------------------------------------------------------
# Fan-Out Tree Decomposition
# ---------------------------------------------------------------------------
async def step_root(system_name: str) -> RootDecomposition:
prompt = (
f"You are a systems engineer. Deconstruct '{system_name}' into 3-5 primary, "
"distinct functional/mechanical subsystems. Keep descriptions concise."
)
return await call_llm_cached(
contents=f"Root system decomposition: {system_name}",
system_prompt=prompt,
schema=RootDecomposition,
)
async def step_components(system_name: str, sub: SubsystemStub) -> SubsystemBreakdown:
prompt = (
f"System: {system_name} | Subsystem: {sub.name} ({sub.description}).\n"
"Identify 2-4 critical mechanical or operational components/assemblies inside this subsystem."
)
return await call_llm_cached(
contents=f"Components for: {sub.name}",
system_prompt=prompt,
schema=SubsystemBreakdown,
)
async def step_parts_and_ideation(system_name: str, sub_name: str, comp: ComponentStub) -> FullComponent:
prompt = (
f"Context: Machine = '{system_name}', Subsystem = '{sub_name}', Component = '{comp.name}'.\n"
f"Component Details: {comp.description}.\n"
"1. Identify 2-3 discrete parts.\n"
"2. State each part's Primary Useful Function.\n"
"3. Provide the TRIZ Ideal Final Result (IFR) formulation.\n"
"4. Provide technical, non-trivial SCAMPER suggestions."
)
breakdown = await call_llm_cached(
contents=f"Parts and ideation for: {comp.name}",
system_prompt=prompt,
schema=ComponentBreakdown,
)
return FullComponent(name=comp.name, description=comp.description, parts=breakdown.parts)
async def process_subsystem(system_name: str, sub: SubsystemStub) -> FullSubsystem:
comp_breakdown = await step_components(system_name, sub)
tasks = [step_parts_and_ideation(system_name, sub.name, comp) for comp in comp_breakdown.components]
components = await asyncio.gather(*tasks)
return FullSubsystem(name=sub.name, description=sub.description, components=components)
async def run_pipeline(system_name: str) -> FullSystemReport:
logger.info(f"Starting decomposition for '{system_name}'...")
root = await step_root(system_name)
sub_tasks = [process_subsystem(system_name, sub) for sub in root.subsystems]
subsystems = await asyncio.gather(*sub_tasks)
return FullSystemReport(
system_name=root.system_name,
summary=root.summary,
subsystems=subsystems,
)
# ---------------------------------------------------------------------------
# File Writers: Markdown & JSON
# ---------------------------------------------------------------------------
def write_markdown_report(report: FullSystemReport, filepath: str):
lines = [
f"# TRIZ & SCAMPER Systematic Innovation: {report.system_name.title()}",
f"> **System Overview:** {report.summary}\n",
"---\n",
]
for sub in report.subsystems:
lines.append(f"## Subsystem: {sub.name}")
lines.append(f"*{sub.description}*\n")
for comp in sub.components:
lines.append(f"### Component: {comp.name}")
lines.append(f"*{comp.description}*\n")
for part in comp.parts:
lines.append(f"#### Part: {part.name}")
lines.append(f"- **Description:** {part.description}")
lines.append(f"- **Primary Useful Function:** {part.primary_function}")
lines.append(f"- **TRIZ Ideal Final Result (IFR):** {part.triz_ifr}\n")
lines.append("| SCAMPER Lens | Technical Innovation Concept |")
lines.append("| :--- | :--- |")
lines.append(f"| **Substitute** | {part.scamper.substitute} |")
lines.append(f"| **Combine** | {part.scamper.combine} |")
lines.append(f"| **Adapt** | {part.scamper.adapt} |")
lines.append(f"| **Modify/Magnify** | {part.scamper.modify_magnify} |")
lines.append(f"| **Put to Another Use** | {part.scamper.put_to_another_use} |")
lines.append(f"| **Eliminate** | {part.scamper.eliminate} |")
lines.append(f"| **Reverse/Rearrange** | {part.scamper.reverse_rearrange} |\n")
with open(filepath, "w", encoding="utf-8") as f:
f.write("\n".join(lines))
logger.info(f"Markdown exported to: {filepath}")
def write_json_report(report: FullSystemReport, filepath: str):
with open(filepath, "w", encoding="utf-8") as f:
f.write(report.model_dump_json(indent=2))
logger.info(f"JSON data exported to: {filepath}")
# ---------------------------------------------------------------------------
# CLI Entrypoint
# ---------------------------------------------------------------------------
if __name__ == "__main__":
init_cache_db()
target_machine = (
sys.argv[1] if len(sys.argv) > 1
else input("Enter target system (e.g. Modern Electric Bicycle): ").strip()
) or "Modern Electric Bicycle"
base_slug = target_machine.lower().replace(" ", "_").replace("/", "_")
md_file = f"{base_slug}_report.md"
json_file = f"{base_slug}_data.json"
# Run pipeline
result = asyncio.run(run_pipeline(target_machine))
# Export both views
write_markdown_report(result, md_file)
write_json_report(result, json_file)
print(f"\n[DONE] Execution complete.")
print(f" - Markdown: {os.path.abspath(md_file)}")
print(f" - JSON: {os.path.abspath(json_file)}")
Example Usage & Artifacts
Run it directly from your terminal:
python system_innovator.py "Quadrotor Drone"
quadrotor_drone_report.md: A clean hierarchical report with tables ready for GitHub rendering, Obsidian, or documentation export.quadrotor_drone_data.json: A structured output matching the schema:
{
"system_name": "Quadrotor Drone",
"summary": "An unmanned multi-rotor aerial vehicle lifted and propelled by four rotors.",
"subsystems": [
{
"name": "Propulsion and Lift",
"description": "Generates vertical thrust and directional torque.",
"components": [
{
"name": "Brushless DC Motor Assembly",
"description": "Electromechanical actuator spinning the propellers.",
"parts": [
{
"name": "Neodymium Rotor Magnets",
"description": "Permanent magnets providing radial flux.",
"primary_function": "Generate magnetic field to drive rotation.",
"triz_ifr": "The propeller blades themselves are magnetically polarized, eliminating the motor rotor and central hub mass.",
"scamper": {
"substitute": "High-temperature ceramic magnets or Halbach array films to reduce stator iron weight.",
"combine": "Integrate magnets directly into the inner rim of duct shrouds.",
"adapt": "Mimic bacterial flagella rotary motor layouts for micro-scale scale-down.",
"modify_magnify": "Vary flux gradient along the axial length to optimize torque during sprint phases.",
"put_to_another_use": "Serve as inductive charging receiver nodes when grounded on a charging pad.",
"eliminate": "Replace with electrostatic ion-thruster blades without rotating magnets.",
"reverse_rearrange": "Invert inner/outer stator-rotor topology to maximize centrifugal air-cooling."
}
}
]
}
]
}
]
}
3. llm_cache.db: An SQLite file storing every node's JSON response, guaranteeing zero repeat billing if you tweak formatting or add post-processing later.
Comments
Post a Comment