# SkillSpector LangGraph Structure: How NVIDIA Built the Security Analysis Pipeline

> Explore the LangGraph structure NVIDIA built for the SkillSpector security analysis pipeline. See the linear flow from input resolution to SARIF reporting in src/skillspector/graph.py.

- Repository: [NVIDIA Corporation/SkillSpector](https://github.com/NVIDIA/SkillSpector)
- Tags: architecture
- Published: 2026-06-23

---

**SkillSpector implements its static analysis pipeline as a LangGraph state machine defined in [`src/skillspector/graph.py`](https://github.com/NVIDIA/SkillSpector/blob/main/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`](https://github.com/NVIDIA/SkillSpector/blob/main/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`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/graph.py) wires the nodes and edges:

```python
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`](https://github.com/NVIDIA/SkillSpector/blob/main/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:

```python

# 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`](https://github.com/NVIDIA/SkillSpector/blob/main/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:

```python
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`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/cli.py).

### Debugging Individual Analyzers

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

```python
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:

```bash
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`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/cli.py).

## Summary

- **SkillSpector LangGraph structure** is defined in [`src/skillspector/graph.py`](https://github.com/NVIDIA/SkillSpector/blob/main/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`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/nodes/analyzers/__init__.py).
- State is managed through the `SkillspectorState` TypedDict in [`src/skillspector/state.py`](https://github.com/NVIDIA/SkillSpector/blob/main/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`](https://github.com/NVIDIA/SkillSpector/blob/main/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`](https://github.com/NVIDIA/SkillSpector/blob/main/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`](https://github.com/NVIDIA/SkillSpector/blob/main/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`](https://github.com/NVIDIA/SkillSpector/blob/main/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`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/graph.py) through the `create_graph()` function, which returns `workflow.compile()`. This compiled graph object is imported by [`src/skillspector/cli.py`](https://github.com/NVIDIA/SkillSpector/blob/main/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.