How SkillSpector Implements Async Batch Processing with arun_batches and max_concurrency

SkillSpector uses an asyncio semaphore inside LLMAnalyzerBase.arun_batches to limit concurrent LLM calls to a configurable max_concurrency (default 10), enabling efficient parallel processing of code analysis batches while respecting provider rate limits.

The NVIDIA SkillSpector framework analyzes source code using LLM-driven analyzers that must process hundreds of fragments efficiently. To prevent sequential bottlenecks while avoiding provider rate limits, the framework implements sophisticated async batch processing with arun_batches and max_concurrency controls in the base analyzer class.

How arun_batches Controls Concurrency

The core implementation resides in src/skillspector/llm_analyzer_base.py, where the arun_batches method orchestrates concurrent execution through three coordinated steps.

Step 1: Semaphore Initialization

The method creates an asyncio.Semaphore using the max_concurrency parameter to cap simultaneous LLM requests. According to the source code at lines 87-88, the default value is 10, though callers can adjust this based on their provider limits or throughput requirements.

sem = asyncio.Semaphore(max_concurrency)

Step 2: The Worker Coroutine

For each batch, an inner _process coroutine acquires the semaphore context, builds the prompt, and invokes the language model. As implemented in lines 89-103, the worker automatically releases the semaphore when exiting the context manager, ensuring max_concurrency calls never exceed the limit simultaneously.

The coroutine selects between structured and plain LLM outputs:

# Inside _process coroutine

if self._structured_llm:
    response = await self._structured_llm.ainvoke(prompt)
else:
    response = await self._llm.ainvoke(prompt)

After parsing through parse_response(), it returns a tuple of (batch, parsed_results).

Step 3: Parallel Execution with Gather

The method fires all worker coroutines simultaneously using asyncio.gather, as shown at lines 104-106. The semaphore guarantees that while hundreds of batches may queue, only max_concurrency workers execute at any moment:

return list(await asyncio.gather(*[_process(b) for b in batches]))

This returns a list of (Batch, parsed_result) tuples, maintaining API parity with the synchronous run_batches method documented in lines 73-86.

Configuring max_concurrency for Your Environment

The max_concurrency parameter provides fine-grained control over resource utilization. Lower values prevent rate-limiting errors on constrained tiers, while higher values maximize throughput on enterprise plans.

Default behavior: When omitted, max_concurrency falls back to 10 concurrent requests.

Typical invocation patterns in concrete analyzers like SemanticQualityPolicy or MetaAnalyzer follow this structure:

batches = analyzer.get_batches(files, file_cache)
results = await analyzer.arun_batches(batches, max_concurrency=20)
findings = analyzer.collect_findings(results)

Practical Implementation Examples

Basic Custom Analyzer

This minimal example demonstrates creating a custom analyzer that processes batches with controlled concurrency:

from skillspector.llm_analyzer_base import LLMAnalyzerBase, Batch

class EchoAnalyzer(LLMAnalyzerBase):
    base_prompt = "Return the text you receive unchanged."

    def parse_response(self, response, batch):
        # The LLM returns plain text; wrap it as a finding

        return [self.Finding(message=response, line=batch.start_line)]

# Prepare a few tiny batches

batches = [
    Batch(file_path="a.py", content="print('hello')", start_line=1, end_line=1),
    Batch(file_path="b.py", content="x = 42", start_line=1, end_line=1),
]

# Run up to 5 concurrent LLM calls

results = await EchoAnalyzer().arun_batches(batches, max_concurrency=5)
print(results)   # → [(Batch(...), [Finding(...)]), ...]

Real-World MetaAnalyzer Usage

The MetaAnalyzer class demonstrates production usage of the async batch processing system:

from skillspector.nodes.meta_analyzer import MetaAnalyzer

# Assume `files` is a dict mapping file paths → source text

meta = MetaAnalyzer()
batches = meta.get_batches(files, file_cache={})

# Process up to 15 batches simultaneously

batch_results = await meta.arun_batches(batches, max_concurrency=15)

# Turn per-batch findings into a flat list

all_findings = meta.collect_findings(batch_results)
print(f"Discovered {len(all_findings)} issues")

Key Source Files

Understanding the async batch processing architecture requires examining these specific files:

• src/skillspector/llm_analyzer_base.py: Implements arun_batches with semaphore-controlled concurrency (lines 87-106)
• src/skillspector/nodes/meta_analyzer.py: Demonstrates high-level consumption of the async batch executor
• src/skillspector/nodes/analyzers/semantic_quality_policy.py: Production analyzer utilizing max_concurrency controls
• tests/nodes/test_llm_analyzer_base.py: Unit tests verifying concurrent execution limits and response parsing

Summary

• SkillSpector implements async batch processing through LLMAnalyzerBase.arun_batches in src/skillspector/llm_analyzer_base.py
• Concurrency control uses asyncio.Semaphore(max_concurrency) with a default limit of 10 simultaneous LLM calls
• Execution model queues all batches immediately but limits active workers via _process coroutines managed by asyncio.gather
• API consistency ensures arun_batches returns the same (Batch, result) tuple format as the synchronous run_batches method
• Configuration allows runtime adjustment of max_concurrency to match provider rate limits or throughput requirements

Frequently Asked Questions

What is the default max_concurrency value in SkillSpector?

The default value is 10. This is hardcoded in the arun_batches method signature at line 87 of src/skillspector/llm_analyzer_base.py. Users can override this when calling the method to accommodate different LLM provider tiers or rate limiting requirements.

How does the semaphore prevent rate limiting?

The asyncio.Semaphore acts as a gatekeeper that allows only max_concurrency number of coroutines to enter the critical section (the LLM invocation) simultaneously. When the limit is reached, additional batches wait until an active task releases the semaphore, preventing your application from exceeding provider rate limits or connection pools.

Can I use arun_batches with custom analyzer classes?

Yes. Any class inheriting from LLMAnalyzerBase can utilize arun_batches immediately. You must implement the abstract methods get_batches and parse_response, then call await self.arun_batches(batches, max_concurrency=N) to process your data with controlled concurrency.

What is the difference between arun_batches and run_batches?

run_batches provides synchronous batch processing, while arun_batches offers asynchronous execution with concurrency limits. Both methods return identical data structures—lists of (Batch, parsed_result) tuples—but arun_batches uses asyncio.gather with semaphore-controlled workers to execute multiple LLM calls in parallel rather than sequentially.

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