# How the LangGraph Workflow Orchestrates 20+ Analyzers in NVIDIA SkillSpector

> Discover how the LangGraph workflow orchestrates 20+ analyzers in NVIDIA SkillSpector. Learn about dynamic node registration, concurrent execution, and state aggregation for efficient risk scoring and reporting.

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

---

**The LangGraph workflow in SkillSpector dynamically registers over 20 analyzers as parallel nodes that execute concurrently after a shared context-building phase, aggregating findings through a typed state schema before final risk scoring and reporting.**

NVIDIA SkillSpector is an open-source security scanning framework that analyzes AI skills for vulnerabilities using static patterns, semantic analysis, and LLM-based detection. The project uses **LangGraph** as its execution engine to manage a growing suite of more than twenty analyzers without hardcoding each node into the pipeline logic.

## Dynamic Registry of Analyzer Nodes

Instead of manually wiring each analyzer into the graph, SkillSpector employs a registration pattern that decouples analyzer implementations from graph topology.

All analyzer implementations reside under `src/skillspector/nodes/analyzers/`. Each module exports a `node` object conforming to LangGraph's node signature. The package-wide registry in [`src/skillspector/nodes/analyzers/__init__.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/nodes/analyzers/__init__.py) aggregates these nodes into two data structures:

```python

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

ANALYZER_NODE_IDS = [
    "static_patterns_prompt_injection",
    "static_patterns_data_exfiltration",
    # … (20‑plus entries) …

    "semantic_quality_policy",
]

ANALYZER_NODES = {
    "static_patterns_prompt_injection": static_patterns_prompt_injection_node,
    # …

    "semantic_quality_policy": semantic_quality_policy_node,
}

```

This registry drives the dynamic construction of the LangGraph workflow. When the graph is instantiated, it iterates over `ANALYZER_NODE_IDS` to create nodes, eliminating the need to modify [`graph.py`](https://github.com/NVIDIA/SkillSpector/blob/main/graph.py) when adding new analyzers.

## Typed State Schema for Cross-Node Communication

The workflow shares a strongly-typed dictionary called `SkillspectorState` defined in [`src/skillspector/state.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/state.py). This state object carries input parameters, intermediate artifacts, and a mutable `findings` list that each analyzer appends to during execution.

The schema enforces contracts between nodes, preventing accidental mutation of unrelated fields while allowing safe concurrent access to shared caches. Analyzers read from pre-computed fields like `components`, `file_cache`, and `manifest` without interfering with each other's operations.

## Graph Construction and Wiring Logic

The `create_graph()` function in [`src/skillspector/graph.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/graph.py) assembles the workflow by combining fixed core nodes with the dynamic analyzer registry:

```python

# src/skillspector/graph.py

workflow = StateGraph(SkillspectorState)

# Core nodes

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)

# Dynamically add every analyzer from the registry

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

```

Edge definitions establish the execution order and parallelism strategy:

```python

# Sequential initialization

workflow.add_edge(START, "resolve_input")
workflow.add_edge("resolve_input", "build_context")

# Parallel analyzer branches

for analyzer_id in ANALYZER_NODE_IDS:
    workflow.add_edge("build_context", analyzer_id)      # Fan-out

    workflow.add_edge(analyzer_id, "meta_analyzer")      # Fan-in

# Final aggregation and output

workflow.add_edge("meta_analyzer", "report")
workflow.add_edge("report", END)

```

Because every analyzer connects to the same predecessor (`build_context`) and shares the same successor (`meta_analyzer`), LangGraph recognizes them as independent branches eligible for concurrent execution.

## Execution Flow and Parallel Processing

The LangGraph workflow executes in five distinct phases, with the analyzer phase running in parallel:

1. **Resolve Input**: The `resolve_input` node normalizes user input (file, zip, URL, or Git repository) and populates `skill_path` and `temp_dir_for_cleanup` in the state.

2. **Build Context**: The `build_context` node walks the skill directory, parses manifests, builds AST/file caches, and extracts component metadata into the shared state.

3. **Analyzer Execution**: All 20+ analyzers run **concurrently** (subject to the runtime's thread pool). Each analyzer reads from the shared caches, performs pattern matching, YARA scans, or LLM calls, and **appends** findings to `state["findings"]` using list extension operations that are thread-safe for appends.

4. **Meta Analysis**: The `meta_analyzer` node performs deduplication, risk scoring, and filtering based on user flags, producing `filtered_findings` and an overall `risk_score`.

5. **Report Generation**: The `report` node formats the final SARIF/JSON output and writes it to `state["report_body"]` before reaching `END`.

## Adding New Analyzers Without Graph Modification

The registration pattern enables zero-touch extension of the pipeline. To add a new analyzer:

```python

# 1. Implement node in src/skillspector/nodes/analyzers/my_new_analyzer.py

# 2. Export in __init__.py

from .my_new_analyzer import node as my_new_analyzer_node

ANALYZER_NODE_IDS.append("my_new_analyzer")
ANALYZER_NODES["my_new_analyzer"] = my_new_analyzer_node

```

No changes to [`src/skillspector/graph.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/graph.py) are required. The `ANALYZER_NODE_IDS` list drives automatic inclusion in the next graph compilation.

## Programmatic Usage Examples

Running the pipeline from Python provides direct access to the LangGraph workflow:

```python
from skillspector.graph import graph
from skillspector.state import SkillspectorState

initial_state: SkillspectorState = {
    "input_path": "./my_skill",
    "mode": "scan",
    "use_llm": True,
}

# Execute the compiled LangGraph

result = graph.invoke(initial_state)

# Access the SARIF report

sarif = result["sarif_report"]
print(sarif["runs"][0]["results"][0]["message"]["text"])

```

Command-line invocation triggers the same graph construction:

```bash

# Scan local directory with all analyzers

skillspector scan ./my_skill

# Skip LLM-based analyzers

skillspector scan ./my_skill --no-llm

```

## Summary

- **Dynamic Registration**: The `ANALYZER_NODE_IDS` list in [`src/skillspector/nodes/analyzers/__init__.py`](https://github.com/NVIDIA/SkillSpector/blob/main/src/skillspector/nodes/analyzers/__init__.py) enables the LangGraph workflow to discover and wire over 20 analyzers automatically.
- **Parallel Execution**: Analyzers fan out from `build_context` and fan in to `meta_analyzer`, allowing LangGraph to schedule them concurrently without interference.
- **State-Driven Architecture**: The `SkillspectorState` schema provides thread-safe append operations for findings while isolating read-only caches.
- **Zero-Touch Extension**: New analyzers require only implementation and registry entry—no graph topology changes needed.
- **Deterministic Ordering**: Input resolution and context building always execute sequentially before parallel analysis begins.

## Frequently Asked Questions

### How does LangGraph handle concurrency between 20+ analyzers?

LangGraph recognizes that all analyzer nodes share the same predecessor (`build_context`) and have no dependencies on each other. The runtime schedules them concurrently within its thread pool, subject to available resources. Each analyzer only performs read operations on shared caches and append operations on the `findings` list, preventing race conditions.

### What happens if an analyzer requires an external API that is unavailable?

Individual analyzers implement availability checks (e.g., `requires_api_key` or `is_available()` methods) and silently skip execution or emit warnings when external services like LLM endpoints are unreachable. The graph continues execution with available analyzers, ensuring partial results rather than pipeline failures.

### Can I run the SkillSpector LangGraph workflow without CLI arguments?

Yes. The `graph` object exported from `skillspector.graph` is a compiled LangGraph instance that can be invoked directly with a `SkillspectorState` dictionary. This programmatic approach allows custom orchestration, such as running within Jupyter notebooks or integrating into larger data processing pipelines.

### Why use LangGraph instead of a simple loop for analyzer execution?

LangGraph provides **deterministic state management** through the typed `SkillspectorState` schema, **automatic parallelization** of independent branches, and **visualization** of the execution graph. The graph structure also enables future extensibility for conditional branching, retry logic, or human-in-the-loop breakpoints that sequential loops cannot easily accommodate.