diff --git a/src/skillspector/llm_analyzer_base.py b/src/skillspector/llm_analyzer_base.py index c5ab9dce..ad16b06b 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 Literal @@ -42,6 +43,36 @@ 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 @@ -398,7 +429,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. @@ -407,6 +438,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 @@ -417,6 +453,8 @@ async def arun_batches( The return type mirrors :meth:`run_batches`. """ + if max_concurrency is None: + max_concurrency = resolve_max_concurrency() sem = asyncio.Semaphore(max_concurrency) async def _process(batch: Batch) -> tuple[Batch, list]: diff --git a/tests/nodes/test_llm_analyzer_base.py b/tests/nodes/test_llm_analyzer_base.py index e344e654..8d7962ea 100644 --- a/tests/nodes/test_llm_analyzer_base.py +++ b/tests/nodes/test_llm_analyzer_base.py @@ -24,6 +24,7 @@ from langchain_core.messages import AIMessage from skillspector.llm_analyzer_base import ( + DEFAULT_MAX_LLM_CONCURRENCY, Batch, LLMAnalysisResult, LLMAnalyzerBase, @@ -32,6 +33,7 @@ estimate_tokens, findings_in_range, number_lines, + resolve_max_concurrency, ) from skillspector.models import Finding from skillspector.nodes.meta_analyzer import ( @@ -46,6 +48,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