How the LangGraph Workflow Orchestrates 20+ Analyzers in NVIDIA SkillSpector

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 aggregates these nodes into two data structures:

# 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 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. 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 assembles the workflow by combining fixed core nodes with the dynamic analyzer registry:

# 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:

# 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:

# 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 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:

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:

# 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 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.

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