diff --git a/src/skillspector/llm_analyzer_base.py b/src/skillspector/llm_analyzer_base.py index 8f42a997..6af01bb4 100644 --- a/src/skillspector/llm_analyzer_base.py +++ b/src/skillspector/llm_analyzer_base.py @@ -28,6 +28,7 @@ from __future__ import annotations import asyncio +import os from collections import defaultdict from dataclasses import dataclass, field from typing import Any, Literal, cast @@ -50,6 +51,37 @@ logger = get_logger(__name__) +DEFAULT_MAX_LLM_CONCURRENCY = 10 + + +def resolve_max_concurrency() -> int: + """Resolve the LLM fan-out concurrency from ``SKILLSPECTOR_MAX_LLM_CONCURRENCY``. + + Defaults to :data:`DEFAULT_MAX_LLM_CONCURRENCY`. Users on rate-limited + providers (free tiers with a low RPM) can set it to ``1`` to serialize + requests instead of bursting up to 10 in parallel — a burst that otherwise + guarantees 429s, and 429'd batches are dropped from the result (see the + analyzer fan-out below). Invalid values fall back to the default; values + below 1 are clamped to 1. + """ + raw = os.environ.get("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "").strip() + if not raw: + return DEFAULT_MAX_LLM_CONCURRENCY + try: + value = int(raw) + except ValueError: + logger.warning( + "Invalid SKILLSPECTOR_MAX_LLM_CONCURRENCY=%r (not an int); using %d", + raw, + DEFAULT_MAX_LLM_CONCURRENCY, + ) + return DEFAULT_MAX_LLM_CONCURRENCY + if value < 1: + logger.warning("SKILLSPECTOR_MAX_LLM_CONCURRENCY=%d < 1; clamping to 1", value) + return 1 + return value + + # OpenAI suggests ~4 chars per token for English text with BPE tokenizers. CHARS_PER_TOKEN = 4 CHUNK_OVERLAP_LINES = 50 @@ -546,7 +578,7 @@ async def arun_batches( self, batches: list[Batch], *, - max_concurrency: int = 10, + max_concurrency: int | None = None, **kwargs: object, ) -> list[tuple[Batch, list]]: """Execute LLM calls for all *batches* concurrently. @@ -555,6 +587,11 @@ async def arun_batches( *max_concurrency* LLM requests in parallel. Both cross-file and cross-chunk batches are parallelized in a single gather call. + When *max_concurrency* is ``None`` (the default) it is resolved from + ``SKILLSPECTOR_MAX_LLM_CONCURRENCY`` via :func:`resolve_max_concurrency`, + so users on rate-limited providers can serialize the fan-out; an + explicit argument still wins. + Failures are isolated per batch: a transient error (timeout, 429, oversized-chunk 400, ...) costs only its own batch, which is logged and omitted from the result, so one bad call cannot cancel the rest @@ -565,6 +602,8 @@ async def arun_batches( The return type mirrors :meth:`run_batches`. """ + if max_concurrency is None: + max_concurrency = resolve_max_concurrency() outcome = await self.arun_batches_detailed( batches, max_concurrency=max_concurrency, **kwargs ) @@ -575,7 +614,7 @@ async def arun_batches_detailed( self, batches: list[Batch], *, - max_concurrency: int = 10, + max_concurrency: int = DEFAULT_MAX_LLM_CONCURRENCY, **kwargs: object, ) -> BatchExecutionResult: """Execute batches concurrently and retain sanitized per-batch failures.""" diff --git a/tests/nodes/test_llm_analyzer_base.py b/tests/nodes/test_llm_analyzer_base.py index d552079c..1d9b6a23 100644 --- a/tests/nodes/test_llm_analyzer_base.py +++ b/tests/nodes/test_llm_analyzer_base.py @@ -25,6 +25,7 @@ from skillspector.inspection_ledger import LedgerReason, finalize_ledger from skillspector.llm_analyzer_base import ( + DEFAULT_MAX_LLM_CONCURRENCY, Batch, BatchExecutionResult, BatchFailure, @@ -36,6 +37,7 @@ findings_in_range, ledger_events_for_batches, number_lines, + resolve_max_concurrency, ) from skillspector.models import Finding from skillspector.nodes.meta_analyzer import ( @@ -50,6 +52,28 @@ # --------------------------------------------------------------------------- +class TestResolveMaxConcurrency: + def test_unset_uses_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", raising=False) + assert resolve_max_concurrency() == DEFAULT_MAX_LLM_CONCURRENCY + + def test_valid_value(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "1") + assert resolve_max_concurrency() == 1 + + def test_blank_uses_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", " ") + assert resolve_max_concurrency() == DEFAULT_MAX_LLM_CONCURRENCY + + def test_invalid_falls_back_to_default(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "abc") + assert resolve_max_concurrency() == DEFAULT_MAX_LLM_CONCURRENCY + + def test_below_one_clamps_to_one(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SKILLSPECTOR_MAX_LLM_CONCURRENCY", "0") + assert resolve_max_concurrency() == 1 + + class TestEstimateTokens: def test_empty_string(self) -> None: assert estimate_tokens("") == 0