diff --git a/deeptutor/services/llm/error_mapping.py b/deeptutor/services/llm/error_mapping.py index 0e8d2b5cc3..36111ee9b7 100644 --- a/deeptutor/services/llm/error_mapping.py +++ b/deeptutor/services/llm/error_mapping.py @@ -4,6 +4,7 @@ from __future__ import annotations +import asyncio from collections.abc import Callable from dataclasses import dataclass import logging @@ -14,6 +15,7 @@ LLMAuthenticationError, LLMError, LLMRateLimitError, + LLMTimeoutError, ProviderContextWindowError, ) @@ -52,6 +54,12 @@ def _classifier(exc: Exception) -> bool: _GLOBAL_RULES: list[MappingRule] = [ + MappingRule( + classifier=_instance_of(asyncio.TimeoutError, TimeoutError), + factory=lambda exc, provider: LLMTimeoutError( + str(exc) or "Request timed out", provider=provider + ), + ), MappingRule( classifier=_class_named("AuthenticationError", "AuthenticationStatusError"), factory=lambda exc, provider: LLMAuthenticationError(str(exc), provider=provider), diff --git a/deeptutor/services/llm/local_provider.py b/deeptutor/services/llm/local_provider.py index 75f098ff4f..484fbb9faf 100644 --- a/deeptutor/services/llm/local_provider.py +++ b/deeptutor/services/llm/local_provider.py @@ -121,6 +121,8 @@ async def complete( # Add optional parameters if kwargs.get("max_tokens"): data["max_tokens"] = kwargs["max_tokens"] + if isinstance(kwargs.get("response_format"), dict): + data["response_format"] = kwargs["response_format"] timeout_value = kwargs.get("timeout", DEFAULT_TIMEOUT) timeout_seconds = ( diff --git a/deeptutor/services/rag/pipelines/llamaindex/config.py b/deeptutor/services/rag/pipelines/llamaindex/config.py index 785916e977..9e093ff009 100644 --- a/deeptutor/services/rag/pipelines/llamaindex/config.py +++ b/deeptutor/services/rag/pipelines/llamaindex/config.py @@ -4,12 +4,26 @@ from dataclasses import dataclass import os +import sys VECTOR_PROFILE = "vector" HYBRID_PROFILE = "hybrid" SUPPORTED_RETRIEVAL_PROFILES = {VECTOR_PROFILE, HYBRID_PROFILE} +def should_show_progress() -> bool: + """Whether to emit LlamaIndex ``tqdm`` progress bars. + + tqdm writes carriage-return progress lines to ``sys.stdout``. When + DeepTutor runs as a server that stream is a pipe whose read end (the + launcher's relay thread) can close mid-indexing, and the next tqdm write + then raises :class:`BrokenPipeError`, killing document indexing. DeepTutor + reports indexing progress through its own ``ProgressTracker``, so the tqdm + output is only wanted in an interactive CLI/REPL session. + """ + return bool(getattr(sys.stdout, "isatty", lambda: False)()) + + @dataclass(frozen=True) class RetrievalConfig: """Runtime retrieval knobs for the LlamaIndex pipeline.""" diff --git a/deeptutor/services/rag/pipelines/llamaindex/ingestion.py b/deeptutor/services/rag/pipelines/llamaindex/ingestion.py index 56d6e0a8b1..69dd7da549 100644 --- a/deeptutor/services/rag/pipelines/llamaindex/ingestion.py +++ b/deeptutor/services/rag/pipelines/llamaindex/ingestion.py @@ -15,6 +15,7 @@ from llama_index.core.schema import BaseNode from . import vector_store +from .config import should_show_progress def build_ingestion_pipeline() -> IngestionPipeline: @@ -60,7 +61,9 @@ def _has_precomputed_embedding(document: Any) -> bool: return True -def documents_to_nodes(documents: list[Any], *, show_progress: bool = True) -> list[Any]: +def documents_to_nodes( + documents: list[Any], *, show_progress: bool = should_show_progress() +) -> list[Any]: """Convert LlamaIndex documents into embedded nodes. Pre-embedded nodes, such as ImageNode instances produced by the document @@ -80,7 +83,7 @@ def documents_to_nodes(documents: list[Any], *, show_progress: bool = True) -> l def create_index_from_documents( - documents: list[Any], storage_dir: Path, *, show_progress: bool = True + documents: list[Any], storage_dir: Path, *, show_progress: bool = should_show_progress() ) -> tuple[VectorStoreIndex, int]: """Create and persist a VectorStoreIndex from documents. @@ -97,7 +100,7 @@ def create_index_from_documents( def insert_documents_into_index( - index: Any, documents: list[Any], *, show_progress: bool = True + index: Any, documents: list[Any], *, show_progress: bool = should_show_progress() ) -> int: """Transform documents once, then insert nodes into an existing index.""" nodes = documents_to_nodes(documents, show_progress=show_progress) diff --git a/deeptutor/services/rag/pipelines/llamaindex/pipeline.py b/deeptutor/services/rag/pipelines/llamaindex/pipeline.py index 1932062260..a45a26d43f 100644 --- a/deeptutor/services/rag/pipelines/llamaindex/pipeline.py +++ b/deeptutor/services/rag/pipelines/llamaindex/pipeline.py @@ -21,7 +21,7 @@ from deeptutor.services.rag.kb_paths import resolve_kb_dir from . import storage -from .config import default_top_k +from .config import default_top_k, should_show_progress from .document_loader import LlamaIndexDocumentLoader from .embedding_adapter import ( configure_llamaindex_settings, @@ -102,7 +102,9 @@ async def initialize(self, kb_name: str, file_paths: List[str], **kwargs) -> boo loop = asyncio.get_running_loop() await loop.run_in_executor( None, - lambda: storage.create_index(documents, storage_dir, show_progress=True), + lambda: storage.create_index( + documents, storage_dir, show_progress=should_show_progress() + ), ) self.logger.info(f"Index persisted to {storage_dir}") @@ -260,7 +262,9 @@ async def add_documents(self, kb_name: str, file_paths: List[str], **kwargs) -> plan.storage_dir.mkdir(parents=True, exist_ok=True) num_added = await loop.run_in_executor( None, - lambda: storage.create_index(documents, plan.storage_dir, show_progress=True), + lambda: storage.create_index( + documents, plan.storage_dir, show_progress=should_show_progress() + ), ) self.logger.info(f"Created new index with {num_added} documents") if signature is not None: diff --git a/deeptutor/services/rag/pipelines/llamaindex/storage.py b/deeptutor/services/rag/pipelines/llamaindex/storage.py index 8be8b8648d..7bce4166e5 100644 --- a/deeptutor/services/rag/pipelines/llamaindex/storage.py +++ b/deeptutor/services/rag/pipelines/llamaindex/storage.py @@ -20,6 +20,7 @@ ) from . import ingestion, retrievers, vector_store +from .config import should_show_progress @dataclass(frozen=True) @@ -85,7 +86,9 @@ def resolve_add_storage_plan(kb_dir: Path, signature: EmbeddingSignature | None) return AddStoragePlan(existing_storage=existing_storage, storage_dir=storage_dir) -def create_index(documents: list[Any], storage_dir: Path, *, show_progress: bool = True) -> int: +def create_index( + documents: list[Any], storage_dir: Path, *, show_progress: bool = should_show_progress() +) -> int: index, count = ingestion.create_index_from_documents( documents, storage_dir, show_progress=show_progress ) @@ -97,7 +100,9 @@ def insert_documents(existing_storage: Path, storage_dir: Path, documents: list[ index = vector_store.load_index(existing_storage) _validate_persisted_embeddings(index, existing_storage) if hasattr(index, "insert_nodes"): - count = ingestion.insert_documents_into_index(index, documents, show_progress=True) + count = ingestion.insert_documents_into_index( + index, documents, show_progress=should_show_progress() + ) else: # Some tests use a tiny fake index that only implements insert(). for document in documents: diff --git a/tests/services/rag/test_llamaindex_ingestion.py b/tests/services/rag/test_llamaindex_ingestion.py index a24ad12806..1a2dd8483f 100644 --- a/tests/services/rag/test_llamaindex_ingestion.py +++ b/tests/services/rag/test_llamaindex_ingestion.py @@ -29,3 +29,50 @@ def run(self, *, documents, show_progress): assert captured["documents"] == [llama_document, plain_node] assert captured["show_progress"] is False assert nodes == ["chunked:Document", "chunked:TextNode", embedded_node] + + +def test_progress_disabled_when_stdout_is_not_a_tty(monkeypatch) -> None: + """tqdm progress bars must be suppressed in headless/server contexts. + + When DeepTutor runs as a server, stdout is a pipe whose read end can + close mid-indexing; a tqdm write then raises BrokenPipeError and kills + document indexing. ``should_show_progress`` must return False for a + non-interactive stream so no tqdm bar is ever created. + """ + from deeptutor.services.rag.pipelines.llamaindex.config import should_show_progress + + class _NonTtyStream: + def isatty(self) -> bool: + return False + + monkeypatch.setattr("sys.stdout", _NonTtyStream()) + assert should_show_progress() is False + + class _TtyStream: + def isatty(self) -> bool: + return True + + monkeypatch.setattr("sys.stdout", _TtyStream()) + assert should_show_progress() is True + + +def test_documents_to_nodes_defaults_to_no_progress_in_headless(monkeypatch) -> None: + """Default show_progress must be False so a broken stdout pipe can't crash indexing.""" + from deeptutor.services.rag.pipelines.llamaindex import ingestion + + captured: dict[str, object] = {} + + class FakePipeline: + def run(self, *, documents, show_progress): + captured["show_progress"] = show_progress + return list(documents) + + monkeypatch.setattr(ingestion, "build_ingestion_pipeline", lambda: FakePipeline()) + + class _NonTtyStream: + def isatty(self) -> bool: + return False + + monkeypatch.setattr("sys.stdout", _NonTtyStream()) + ingestion.documents_to_nodes([Document(text="x")]) + assert captured["show_progress"] is False