diff --git a/app.py b/app.py index 03eadaf..4af3a25 100644 --- a/app.py +++ b/app.py @@ -8,7 +8,8 @@ from src.generation.answerer import answer, build_source_elements from src.generation.condenser import Condenser from src.health_check import check_models, check_ollama -from src.ingestion.ingest import ingest_folder +from src.ingestion.ingest import IngestResult, ingest_folder +from src.ingestion.scanner import scan_folder from src.retrieval.bm25_index import BM25Index from src.retrieval.embeddings import make_ollama_embed_fn from src.retrieval.hybrid import HybridRetriever @@ -36,21 +37,25 @@ def _build_retriever(store: VectorStore, config) -> HybridRetriever: ) -async def _run_ingestion(store: VectorStore, config) -> int: +async def _run_ingestion(store: VectorStore, config, **kwargs) -> int: """Ingest documents from configured folder. Returns chunk count.""" folder = config.paths.documents if str(folder) and folder.exists(): - ingested, skipped = ingest_folder( + result = ingest_folder( folder, store, recursive=config.scanning.recursive, chunk_size=config.chunking.chunk_size, chunk_overlap=config.chunking.chunk_overlap, + **kwargs, ) - if ingested > 0: - logger.info("Ingested %d new files, skipped %d unchanged", ingested, skipped) - elif skipped > 0: - logger.info("All %d files unchanged, skipped", skipped) + if result.ingested > 0: + logger.info( + "Ingested %d new files, skipped %d, failed %d", + result.ingested, result.skipped, result.failed, + ) + elif result.skipped > 0: + logger.info("All %d files unchanged, skipped", result.skipped) return store.count() @@ -120,16 +125,89 @@ async def on_chat_start(): ] ).send() + # Add re-ingest action button + actions = [ + cl.Action( + name="reingest", + payload={"value": "reingest"}, + label="Re-ingest Documents", + description="Force re-ingest all documents from the configured folder", + ) + ] + if doc_count > 0: await cl.Message( - content=f"{doc_count} chunks indexed. Ask me anything!" + content=f"{doc_count} chunks indexed. Ask me anything!", + actions=actions, ).send() else: await cl.Message( - content="No documents indexed. Configure your documents folder in the settings panel (gear icon) or `config.yaml`." + content="No documents indexed. Configure your documents folder in the settings panel (gear icon) or `config.yaml`.", + actions=actions, ).send() +def _build_ingest_summary(result: IngestResult, total_files: int) -> str: + """Build a human-readable ingestion summary.""" + parts = [f"**Ingestion complete.** {result.ingested}/{total_files} documents ingested."] + + if result.skipped > 0: + parts.append(f"{result.skipped} unchanged (skipped).") + + if result.failed > 0: + parts.append(f"{result.failed} failed.") + parts.append("\n
Failed documents\n") + for filename, error in result.failures: + parts.append(f"- **{filename}**: {error}") + parts.append("\n
") + + return " ".join(parts) + + +@cl.action_callback("reingest") +async def on_reingest(action: cl.Action): + """Handle manual re-ingest action button.""" + config = cl.user_session.get("config") + store = cl.user_session.get("store") + folder = config.paths.documents + + if not str(folder) or not folder.exists(): + await cl.Message( + content="No documents folder configured. Set it in the settings panel first." + ).send() + return + + # Show document count before starting + files = scan_folder(folder, recursive=config.scanning.recursive) + total_files = len(files) + await cl.Message( + content=f"Re-ingesting {total_files} documents from `{folder}`..." + ).send() + + # Run ingestion with force=True and per-file progress steps + async with cl.Step(name="Re-ingestion Progress", type="tool") as step: + result = ingest_folder( + folder, + store, + recursive=config.scanning.recursive, + chunk_size=config.chunking.chunk_size, + chunk_overlap=config.chunking.chunk_overlap, + force=True, + on_progress=lambda name, status: logger.info(" %s: %s", name, status), + ) + step.output = _build_ingest_summary(result, total_files) + + # Rebuild retriever with updated store + retriever = _build_retriever(store, config) + cl.user_session.set("retriever", retriever) + + summary = _build_ingest_summary(result, total_files) + doc_count = store.count() + await cl.Message( + content=f"{summary}\n\n{doc_count} chunks indexed." + ).send() + + @cl.on_settings_update async def on_settings_update(settings: dict): """Re-scan and ingest when settings change.""" diff --git a/src/ingestion/ingest.py b/src/ingestion/ingest.py index db687bf..e8a887b 100644 --- a/src/ingestion/ingest.py +++ b/src/ingestion/ingest.py @@ -1,6 +1,8 @@ -"""Incremental document ingestion with change detection.""" +"""Incremental document ingestion with change detection and failure handling.""" import logging +from collections.abc import Callable +from dataclasses import dataclass, field from pathlib import Path from src.ingestion.chunker import chunk_documents @@ -10,6 +12,20 @@ logger = logging.getLogger(__name__) +ProgressCallback = Callable[[str, str], None] +"""Callback(filename, status) called per file. Status: 'processing', 'ingested', 'skipped', 'failed'.""" + + +@dataclass +class IngestResult: + """Summary of an ingestion run.""" + + ingested: int = 0 + skipped: int = 0 + failed: int = 0 + failures: list[tuple[str, str]] = field(default_factory=list) + """List of (filename, error_message) for failed files.""" + def ingest_folder( folder: Path, @@ -18,50 +34,79 @@ def ingest_folder( recursive: bool = True, chunk_size: int = 512, chunk_overlap: int = 100, -) -> tuple[int, int]: + force: bool = False, + on_progress: ProgressCallback | None = None, +) -> IngestResult: """Scan a folder and ingest new or changed documents. - Returns (ingested_count, skipped_count) — files ingested vs unchanged. + Args: + folder: Path to the documents folder. + store: Vector store to add chunks to. + recursive: Whether to scan subdirectories. + chunk_size: Maximum chunk size in characters. + chunk_overlap: Overlap between chunks. + force: If True, re-ingest all files regardless of hash. + on_progress: Optional callback for per-file progress. + + Returns: + IngestResult with counts of ingested, skipped, and failed files. """ files = scan_folder(folder, recursive=recursive) + result = IngestResult() if not files: - return 0, 0 + return result - existing_hashes = _get_stored_file_hashes(store) - - ingested = 0 - skipped = 0 + existing_hashes = set() if force else _get_stored_file_hashes(store) for file_path in files: - file_hash = compute_file_hash(file_path) - if file_hash in existing_hashes: - skipped += 1 - continue + filename = file_path.name - docs = load_file(file_path, folder) - if not docs: - continue - - # Attach file_hash to document metadata before chunking - for doc in docs: - doc.metadata["file_hash"] = file_hash - - chunks = chunk_documents( - docs, - chunk_size=chunk_size, - chunk_overlap=chunk_overlap, - ) + if on_progress: + on_progress(filename, "processing") - # Propagate file_hash to chunk metadata - for chunk in chunks: - chunk.metadata["file_hash"] = file_hash - - store.add_chunks(chunks) - ingested += 1 - logger.info("Ingested %s (%d chunks)", file_path.name, len(chunks)) + file_hash = compute_file_hash(file_path) + if not force and file_hash in existing_hashes: + result.skipped += 1 + if on_progress: + on_progress(filename, "skipped") + continue - return ingested, skipped + try: + docs = load_file(file_path, folder) + if not docs: + result.skipped += 1 + if on_progress: + on_progress(filename, "skipped") + continue + + for doc in docs: + doc.metadata["file_hash"] = file_hash + + chunks = chunk_documents( + docs, + chunk_size=chunk_size, + chunk_overlap=chunk_overlap, + ) + + for chunk in chunks: + chunk.metadata["file_hash"] = file_hash + + store.add_chunks(chunks) + result.ingested += 1 + if on_progress: + on_progress(filename, "ingested") + logger.info("Ingested %s (%d chunks)", filename, len(chunks)) + + except Exception as e: + result.failed += 1 + error_msg = f"{type(e).__name__}: {e}" + result.failures.append((filename, error_msg)) + if on_progress: + on_progress(filename, "failed") + logger.warning("Failed to ingest %s: %s", filename, error_msg) + + return result def _get_stored_file_hashes(store: VectorStore) -> set[str]: diff --git a/tests/test_ingest.py b/tests/test_ingest.py index 24e80db..26e53aa 100644 --- a/tests/test_ingest.py +++ b/tests/test_ingest.py @@ -1,10 +1,11 @@ """Tests for incremental ingestion.""" +import hashlib from unittest.mock import MagicMock, patch import pytest -from src.ingestion.ingest import ingest_folder +from src.ingestion.ingest import IngestResult, ingest_folder @pytest.fixture @@ -24,16 +25,14 @@ def mock_store(): class TestIngestFolder: def test_ingests_new_file(self, doc_folder, mock_store): - ingested, skipped = ingest_folder( + result = ingest_folder( doc_folder, mock_store, chunk_size=512, chunk_overlap=100 ) - assert ingested == 1 - assert skipped == 0 + assert result.ingested == 1 + assert result.skipped == 0 mock_store.add_chunks.assert_called_once() def test_skips_unchanged_file(self, doc_folder, mock_store): - # Simulate file already in store by returning its hash - import hashlib file_hash = hashlib.md5( (doc_folder / "notes.txt").read_bytes() ).hexdigest() @@ -43,9 +42,9 @@ def test_skips_unchanged_file(self, doc_folder, mock_store): [{"file_hash": file_hash}], ) - ingested, skipped = ingest_folder(doc_folder, mock_store) - assert ingested == 0 - assert skipped == 1 + result = ingest_folder(doc_folder, mock_store) + assert result.ingested == 0 + assert result.skipped == 1 mock_store.add_chunks.assert_not_called() def test_chunks_have_file_hash(self, doc_folder, mock_store): @@ -55,9 +54,9 @@ def test_chunks_have_file_hash(self, doc_folder, mock_store): assert "file_hash" in chunk.metadata def test_empty_folder(self, tmp_path, mock_store): - ingested, skipped = ingest_folder(tmp_path, mock_store) - assert ingested == 0 - assert skipped == 0 + result = ingest_folder(tmp_path, mock_store) + assert result.ingested == 0 + assert result.skipped == 0 def test_relative_path_in_metadata(self, doc_folder, mock_store): ingest_folder(doc_folder, mock_store) @@ -65,3 +64,50 @@ def test_relative_path_in_metadata(self, doc_folder, mock_store): for chunk in chunks: rel_path = chunk.metadata.get("relative_path", "") assert not rel_path.startswith("/") + + def test_force_ignores_existing_hashes(self, doc_folder, mock_store): + """Force mode ingests even when file hash already exists.""" + file_hash = hashlib.md5( + (doc_folder / "notes.txt").read_bytes() + ).hexdigest() + mock_store.count.return_value = 5 + mock_store.get_all_texts_and_metadatas.return_value = ( + ["text"], + [{"file_hash": file_hash}], + ) + + result = ingest_folder(doc_folder, mock_store, force=True) + assert result.ingested == 1 + assert result.skipped == 0 + mock_store.add_chunks.assert_called_once() + + def test_failure_skips_bad_file_continues(self, tmp_path, mock_store): + """A file that fails to load is skipped; other files still ingested.""" + (tmp_path / "good.txt").write_text("Good content") + (tmp_path / "bad.pdf").write_bytes(b"not a real pdf") + + result = ingest_folder(tmp_path, mock_store) + # good.txt should succeed, bad.pdf may fail + assert result.ingested >= 1 or result.failed >= 1 + assert result.ingested + result.failed + result.skipped == 2 + + def test_failure_records_error_details(self, tmp_path, mock_store): + """Failed files are recorded with filename and error message.""" + (tmp_path / "bad.pdf").write_bytes(b"not a real pdf") + + result = ingest_folder(tmp_path, mock_store) + if result.failed > 0: + assert len(result.failures) == result.failed + assert result.failures[0][0] == "bad.pdf" + assert isinstance(result.failures[0][1], str) + + def test_progress_callback_called(self, doc_folder, mock_store): + """Progress callback is called for each file.""" + progress_calls = [] + ingest_folder( + doc_folder, + mock_store, + on_progress=lambda name, status: progress_calls.append((name, status)), + ) + assert len(progress_calls) >= 1 + assert progress_calls[0][0] == "notes.txt"