How the LangGraph Workflow Processes Skills Through the Analysis Pipeline in SkillSpector
SkillSpector uses a compiled LangGraph StateGraph where deterministic nodes sequentially resolve input sources, build file context, execute static analyzers, optionally enrich findings with an LLM, and generate the final security report.
The NVIDIA/SkillSpector repository implements its core analysis engine as a modular LangGraph workflow. This architecture processes AI skills through a directed graph where each node receives a shared SkillspectorState dictionary, performs a specific transformation, and returns partial state updates. The pipeline is defined in skillspector/graph.py and invoked from the CLI, ensuring that every scan follows a consistent, extensible path from raw input to structured security findings.
LangGraph StateGraph Architecture
The workflow centers on a StateGraph that manages immutable state transitions across distinct processing phases. According to the SkillSpector source code, the graph is compiled once at module initialization and reused for every scan invocation.
The Shared State Schema
All nodes communicate through the SkillspectorState typed dictionary defined in skillspector/state.py. This shared container carries the skill's metadata, cached file contents, analysis findings, and configuration flags throughout the pipeline. Key fields include skill_path (the resolved local directory), file_cache (content mappings), findings (raw analyzer outputs), and filtered_findings (LLM-enriched results).
Graph Compilation and Reuse
The create_graph() function in skillspector/graph.py constructs the workflow by registering nodes and defining execution edges. The compiled graph object is stored as a module-level singleton (graph = create_graph()), allowing the CLI to invoke the same optimized graph instance repeatedly without reconstruction overhead.
Phase-by-Phase Pipeline Execution
The LangGraph workflow processes skills through five distinct phases, each implemented as a node that receives the current state and returns specific updates.
Input Resolution Phase
The resolve_input node normalizes the user's input_path (supporting Git URLs, zip archives, or local directories) into a local skill_path. This node, implemented in skillspector/nodes/resolve_input.py, also tracks temporary directories in temp_dir_for_cleanup for post-scan cleanup.
Context Building Phase
Next, the build_context node walks the resolved directory to populate the shared state. As implemented in skillspector/nodes/build_context.py, this step fills components (file listings), file_cache (content storage), manifest (parsed SKILL.md), and component_metadata (file types, line counts, executable flags).
Static Analysis Phase
The graph then fans out to multiple analyzer nodes registered under ANALYZER_NODE_IDS. Each static analyzer—such as static_yara—receives the populated state, reads from file_cache and components, and appends Finding objects to state["findings"]. The registry mapping IDs to callables lives in skillspector/nodes/analyzers/__init__.py.
LLM Enrichment Phase
The meta_analyzer node (in skillspector/nodes/meta_analyzer.py) optionally calls an LLM per file to confirm vulnerabilities and generate explanations. If state["use_llm"] is False or the call fails, a fallback copies raw findings to filtered_findings without modification.
Reporting Phase
Finally, the report node computes a risk score, builds a SARIF document, and formats output according to state["output_format"] (terminal, JSON, Markdown, or raw SARIF). This implementation in skillspector/nodes/report.py stores the final string in state["report_body"].
Code Implementation Details
The graph construction follows a declarative pattern where edges encode execution order. Here is the core implementation from src/skillspector/graph.py:
from langgraph.graph import END, START, StateGraph
from skillspector.state import SkillspectorState
from skillspector.nodes.resolve_input import resolve_input
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.analyzers import ANALYZER_NODE_IDS, ANALYZER_NODES
def create_graph():
workflow = StateGraph(SkillspectorState)
# Core pipeline 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)
# Register all static analyzer nodes
for analyzer_id in ANALYZER_NODE_IDS:
workflow.add_node(analyzer_id, ANALYZER_NODES[analyzer_id])
# Define execution order
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()
The ANALYZER_NODES dictionary in skillspector/nodes/analyzers/__init__.py maps string IDs to callable node functions, enabling easy extension:
from skillspector.nodes.analyzers.static_yara import node as static_yara_node
ANALYZER_NODES = {
"static_yara": static_yara_node,
# ... additional analyzers ...
}
Running the Pipeline Programmatically
While the CLI provides the primary interface, you can invoke the LangGraph workflow directly using the compiled graph and state helpers:
from skillspector.graph import graph
from skillspector.cli import _scan_state
# Build initial state with target and configuration
state = _scan_state(
input_path="https://github.com/example/my-skill",
format="json",
no_llm=False, # Enable LLM filtering
)
# Execute the compiled LangGraph workflow
result = graph.invoke(state)
# Access results from the final state
print(result["report_body"])
print("Risk score:", result["risk_score"])
print("Total findings:", len(result["filtered_findings"]))
The graph.invoke() call in skillspector/cli.py executes the entire pipeline synchronously, returning the populated SkillspectorState containing all analysis results.
Summary
- SkillSpector implements a deterministic LangGraph workflow where nodes transform shared state through distinct phases: input resolution, context building, static analysis, LLM enrichment, and reporting.
- The graph is compiled once in
skillspector/graph.pyviacreate_graph()and reused across scans for performance. - Static analyzers are dynamically registered through
ANALYZER_NODE_IDSandANALYZER_NODES, making the pipeline extensible without modifying core graph logic. - Each node receives the full
SkillspectorStatebut returns only partial updates, ensuring immutable state transitions and predictable data flow. - The final
reportnode aggregatesfiltered_findingsand produces structured output in SARIF, JSON, or Markdown formats.
Frequently Asked Questions
What is the role of SkillspectorState in the LangGraph workflow?
SkillspectorState is the typed dictionary that serves as the single source of truth throughout the pipeline. Every node receives the current state, reads cached data like file_cache or components, and appends results to fields like findings or filtered_findings. This shared container ensures type safety and consistent data flow between the input resolution, analysis, and reporting phases.
How does SkillSpector handle multiple static analyzers in the graph?
The workflow dynamically registers each analyzer defined in ANALYZER_NODE_IDS as a separate node using workflow.add_node(analyzer_id, ANALYZER_NODES[analyzer_id]). The graph edges ensure that after build_context completes, every analyzer runs (in parallel where possible) before converging at the meta_analyzer node. This pattern allows security checks to be added by simply extending the ANALYZER_NODES registry in skillspector/nodes/analyzers/__init__.py.
Can the LLM-based meta-analysis be disabled?
Yes. The meta_analyzer node checks state["use_llm"] before invoking the language model. If this flag is False or the LLM call errors, the node falls back to copying raw findings directly to filtered_findings without enrichment. You can disable LLM processing by passing no_llm=True to the state builder or setting the appropriate flag in the initial state dictionary.
Where is the LangGraph workflow compiled and stored?
The workflow is compiled in src/skillspector/graph.py by the create_graph() function, which returns a compiled graph object. This object is stored as a module-level singleton (graph = create_graph()) and imported by skillspector/cli.py. This compilation happens once at import time, allowing the CLI to call graph.invoke(state) repeatedly without reconstruction overhead.
Have a question about this repo?
These articles cover the highlights, but your codebase questions are specific. Give your agent direct access to the source. Share this with your agent to get started:
curl -s "https://instagit.com/install.md" Maintain an open-source project? Get it listed too →