SkillSpector LangGraph Structure: How NVIDIA Built the Security Analysis Pipeline

SkillSpector implements its static analysis pipeline as a LangGraph state machine defined in src/skillspector/graph.py, using a linear flow from input resolution through parallel analyzers to final SARIF reporting.

The NVIDIA SkillSpector repository orchestrates skill security scanning through a sophisticated LangGraph structure that transforms raw input paths into comprehensive vulnerability reports. This workflow combines input normalization, context building, parallel static analysis, and meta-analysis into a single compiled state graph using LangGraph's StateGraph API.

LangGraph Workflow Architecture

SkillSpector composes its static-analysis pipeline as a LangGraph state machine. The core workflow definition lives in src/skillspector/graph.py and implements a five-phase execution model:

• Input resolution (resolve_input node): Normalizes paths, URLs, or ZIP archives to determine the target skill directory.
• Context building (build_context node): Walks the skill tree, extracts manifests, and builds file and AST caches.
• Static analysis (multiple analyzer nodes): Parallel inspection for vulnerabilities including static_patterns_prompt_injection, behavioral_ast, and semantic_security_discovery.
• Meta-analysis (meta_analyzer node): Aggregates findings, applies risk scoring, and filters duplicates.
• Reporting (report node): Formats results into SARIF or other output formats.

The workflow proceeds linearly from START → resolve_input → build_context, then fans out to each analyzer node, converges at meta_analyzer, and terminates at report → END.

Core Graph Definition

The graph is constructed using LangGraph’s StateGraph with the SkillspectorState schema. The create_graph() function in src/skillspector/graph.py wires the nodes and edges:

from langgraph.graph import END, START, StateGraph
from skillspector.nodes.analyzers import ANALYZER_NODE_IDS, ANALYZER_NODES
from skillspector.nodes.build_context import build_context
from skillspector.nodes.meta_analyzer import meta_analyzer
from skillspector.nodes.report import report
from skillspector.nodes.resolve_input import resolve_input
from skillspector.state import SkillspectorState

def create_graph():
    workflow = StateGraph(SkillspectorState)

    workflow.add_node("resolve_input", resolve_input)
    workflow.add_node("build_context", build_context)
    workflow.add_node("meta_analyzer", meta_analyzer)
    workflow.add_node("report", report)

    for analyzer_id in ANALYZER_NODE_IDS:
        workflow.add_node(analyzer_id, ANALYZER_NODES[analyzer_id])

    workflow.add_edge(START, "resolve_input")
    workflow.add_edge("resolve_input", "build_context")
    for analyzer_id in ANALYZER_NODE_IDS:
        workflow.add_edge("build_context", analyzer_id)
        workflow.add_edge(analyzer_id, "meta_analyzer")
    workflow.add_edge("meta_analyzer", "report")
    workflow.add_edge("report", END)
    return workflow.compile()

graph = create_graph()

Analyzer Registry

All analyzer nodes are registered in src/skillspector/nodes/analyzers/__init__.py. The list ANALYZER_NODE_IDS determines the order in which they are added to the graph, and ANALYZER_NODES maps each identifier to its concrete implementation:

# src/skillspector/nodes/analyzers/__init__.py

ANALYZER_NODE_IDS = [
    "static_patterns_prompt_injection",
    "static_patterns_data_exfiltration",
    # … (20 more IDs) …

    "semantic_quality_policy",
]

ANALYZER_NODES = {
    "static_patterns_prompt_injection": static_patterns_prompt_injection_node,
    # … (same keys as above) …

    "semantic_quality_policy": semantic_quality_policy_node,
}

State Schema Implementation

The graph state is a strongly-typed TypedDict (SkillspectorState) defined in src/skillspector/state.py. This schema carries input parameters, file caches, AST representations, component metadata, accumulated findings, risk scores, and the final report body between nodes. All nodes in the SkillSpector LangGraph read from and write to this shared state object.

Practical Usage Examples

Running the Full Pipeline

Invoke the compiled graph from Python to execute the complete workflow:

from skillspector.graph import graph
from skillspector.state import SkillspectorState

# Minimal input – path to a skill directory or archive

state: SkillspectorState = {
    "input_path": "/path/to/skill",
    "mode": "scan",
    "use_llm": True,
    "model_config": {"default": "gpt-4o-mini"},
}

# Execute the LangGraph workflow

result = graph.invoke(state)               # blocks until the graph finishes

print(result["report_body"])               # human‑readable SARIF output

This mirrors the internal behavior of the CLI wrapper found in src/skillspector/cli.py.

Debugging Individual Analyzers

For targeted debugging, invoke a single analyzer node directly without running the full graph:

from skillspector.nodes.analyzers import ANALYZER_NODES
from skillspector.state import SkillspectorState

state = {
    "components": ["my_component"],
    "file_cache": {...},
    "ast_cache": {...},
    "manifest": {...},
    "findings": [],
    "use_llm": False,
    # … other required fields …

}

# Run only the “static_patterns_prompt_injection” analyzer

node = ANALYZER_NODES["static_patterns_prompt_injection"]
partial_result = node(state)
print(partial_result["findings"])

CLI Integration

The public entry point bootstraps the graph via the command line:

skill-spector scan /path/to/skill --model default=gpt-4o-mini --output sarif

The CLI parses arguments, builds the initial SkillspectorState, and calls graph.invoke as implemented in src/skillspector/cli.py.

Summary

• SkillSpector LangGraph structure is defined in src/skillspector/graph.py using LangGraph's StateGraph class.
• The workflow follows a directed path: START → resolve_input → build_context → [parallel analyzers] → meta_analyzer → report → END.
• Analyzer nodes are dynamically registered via ANALYZER_NODE_IDS and ANALYZER_NODES in src/skillspector/nodes/analyzers/__init__.py.
• State is managed through the SkillspectorState TypedDict in src/skillspector/state.py, ensuring type safety across the pipeline.
• The compiled graph is exposed as graph and invoked via graph.invoke() in both programmatic and CLI contexts.

Frequently Asked Questions

How does SkillSpector execute multiple analyzers in parallel?

The LangGraph structure fans out from build_context to all registered analyzers simultaneously. Each analyzer node receives the complete SkillspectorState and writes findings back to the shared state before converging at meta_analyzer. This parallel execution pattern is defined in src/skillspector/graph.py where edges connect build_context to each analyzer ID, and each analyzer to meta_analyzer.

What data structure carries state between LangGraph nodes?

The graph uses a strongly-typed TypedDict called SkillspectorState defined in src/skillspector/state.py. This schema carries input parameters, file caches, AST representations, component metadata, accumulated findings, risk scores, and the final report body between nodes. All nodes in the SkillSpector LangGraph read from and write to this shared state object.

Can I customize which analyzers run in the SkillSpector pipeline?

Yes, the analyzer registry in src/skillspector/nodes/analyzers/__init__.py exposes ANALYZER_NODE_IDS, which controls which nodes are added to the graph. Modifying this list allows you to include or exclude specific security checks such as static_patterns_prompt_injection or semantic_quality_policy. The create_graph() function in src/skillspector/graph.py iterates over this list when wiring edges.

Where is the LangGraph workflow compiled and exposed?

The compiled graph is instantiated at module level in src/skillspector/graph.py through the create_graph() function, which returns workflow.compile(). This compiled graph object is imported by src/skillspector/cli.py and invoked via graph.invoke(state) to execute the full analysis pipeline. You can also import graph directly in Python scripts to programmatically trigger scans.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}