Class LLMExtractor

explainshell/extraction/llm/extractor.py:188–568

LLM-based option extractor. Implements the base ``Extractor`` protocol via ``extract()``. Also satisfies ``BatchExtractor`` via ``prepare()``, ``finalize()``, and ``batch_provider``.

class LLMExtractor:
    """LLM-based option extractor.

    Implements the base ``Extractor`` protocol via ``extract()``.
    Also satisfies ``BatchExtractor`` via ``prepare()``, ``finalize()``,
    and ``batch_provider``.
    """

    def __init__(self, config: ExtractorConfig) -> None:
        self._model = config.model or ""
        self._run_dir = config.run_dir
        self._repo_root = config.repo_root
        self._debug = config.debug
        self.provider: LLMProvider = make_provider(self._model)
        try:
            self.batch_provider: BatchProvider = make_batch_provider(self._model)
        except ValueError:
            pass  # model doesn't support batch; accessed only via --batch flag
        self._cancelled = threading.Event()

    def cancel(self) -> None:
        """Signal all in-progress extract() calls to stop after their current
        LLM request completes.  Does not abort already in-flight HTTP calls,
        but prevents the next chunk from being submitted."""
        self._cancelled.set()

    def extract(self, gz_path: str) -> ExtractionResult:
        """Full extraction pipeline: prepare → LLM calls → finalize."""
        prepared = self.prepare(gz_path)
        basename = prepared.basename
        n_chunks = prepared.n_chunks

        logger.info(
            "%s: %d chars (%d numbered), %d chunk(s)",
            basename,
            prepared.plain_text_len,
            len(prepared.numbered_text),
            n_chunks,
        )

        stats = ExtractionStats(
            chunks=n_chunks,
            plain_text_len=prepared.plain_text_len,
        )

        all_chunk_data: list[ChunkResult] = []
        t0 = time.monotonic()

        for i, user_content in enumerate(prepared.requests):
            if self._cancelled.is_set():
                raise ExtractionError("cancelled", reason_class=FailureReason.CANCELLED)

            chunk_label = (
                f"chunk {i + 1}/{n_chunks}" if n_chunks > 1 else "single chunk"
            )
            logger.info(
                "%s: calling LLM (%s, %d chars)...",
                basename,