fix(rag): improve local LlamaIndex pipeline compatibility

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),

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 = (

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."""

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)

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:

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:

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