feat: add multi-format ingestion and index lifecycle

app/engine/chunking.py

@@ -0,0 +1,19 @@
+import uuid
+
+from langchain_core.documents import Document
+from langchain_text_splitters import RecursiveCharacterTextSplitter
+
+from app.core.config import settings
+
+
+def chunk_documents(documents: list[Document]) -> list[Document]:
+    splitter = RecursiveCharacterTextSplitter(
+        chunk_size=settings.CHUNK_SIZE,
+        chunk_overlap=settings.CHUNK_OVERLAP,
+    )
+    chunks = splitter.split_documents(documents)
+    for index, chunk in enumerate(chunks):
+        source = chunk.metadata.get("source", "unknown")
+        chunk.metadata["chunk_index"] = index
+        chunk.metadata["chunk_id"] = str(uuid.uuid5(uuid.NAMESPACE_URL, f"{source}:{index}:{chunk.page_content[:80]}"))
+    return chunks

app/engine/document_registry.py

@@ -0,0 +1,61 @@
+import hashlib
+import json
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+REGISTRY_PATH = Path("data/document_registry.json")
+
+
+@dataclass
+class DocumentRecord:
+    doc_id: str
+    source: str
+    source_path: str
+    content_hash: str
+    chunk_count: int
+    ingested_at: str
+
+
+def file_hash(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for block in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def load_registry(path: Path = REGISTRY_PATH) -> dict[str, dict]:
+    if not path.exists():
+        return {}
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def save_registry(registry: dict[str, dict], path: Path = REGISTRY_PATH) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(registry, indent=2, sort_keys=True), encoding="utf-8")
+
+
+def hash_exists(content_hash: str, registry: dict[str, dict]) -> bool:
+    return any(record["content_hash"] == content_hash for record in registry.values())
+
+
+def upsert_record(
+    doc_id: str,
+    source: str,
+    source_path: str,
+    content_hash: str,
+    chunk_count: int,
+    registry: dict[str, dict],
+) -> None:
+    registry[doc_id] = asdict(
+        DocumentRecord(
+            doc_id=doc_id,
+            source=source,
+            source_path=source_path,
+            content_hash=content_hash,
+            chunk_count=chunk_count,
+            ingested_at=datetime.now(timezone.utc).isoformat(),
+        )
+    )

app/engine/indexer.py

@@ -0,0 +1,87 @@
+from langchain_community.embeddings import FastEmbedEmbeddings
+from langchain_qdrant import FastEmbedSparse, QdrantVectorStore, RetrievalMode
+from qdrant_client import models
+
+from app.core.config import settings
+from app.core.dependencies import get_qdrant_client
+
+
+def qdrant_store_options() -> dict:
+    if settings.QDRANT_URL:
+        return {"url": settings.QDRANT_URL}
+    return {"path": settings.QDRANT_LOCATION}
+
+
+def retrieval_mode() -> RetrievalMode:
+    mode = settings.RETRIEVAL_MODE.lower()
+    if mode == "dense":
+        return RetrievalMode.DENSE
+    if mode == "sparse":
+        return RetrievalMode.SPARSE
+    return RetrievalMode.HYBRID
+
+
+def dense_embeddings() -> FastEmbedEmbeddings:
+    return FastEmbedEmbeddings(model_name=settings.DENSE_EMBEDDING_MODEL)
+
+
+def sparse_embeddings() -> FastEmbedSparse:
+    return FastEmbedSparse(model_name=settings.SPARSE_EMBEDDING_MODEL)
+
+
+def collection_exists() -> bool:
+    client = get_qdrant_client()
+    return any(collection.name == settings.COLLECTION_NAME for collection in client.get_collections().collections)
+
+
+def open_vector_store(validate_collection_config: bool = True) -> QdrantVectorStore:
+    return QdrantVectorStore(
+        client=get_qdrant_client(),
+        collection_name=settings.COLLECTION_NAME,
+        embedding=dense_embeddings(),
+        sparse_embedding=sparse_embeddings(),
+        retrieval_mode=retrieval_mode(),
+        validate_collection_config=validate_collection_config,
+    )
+
+
+def index_documents(documents, force_recreate: bool = False) -> None:
+    if force_recreate or not collection_exists():
+        QdrantVectorStore.from_documents(
+            documents,
+            embedding=dense_embeddings(),
+            sparse_embedding=sparse_embeddings(),
+            collection_name=settings.COLLECTION_NAME,
+            retrieval_mode=retrieval_mode(),
+            force_recreate=force_recreate,
+            **qdrant_store_options(),
+        )
+        return
+
+    store = open_vector_store()
+    store.add_documents(documents)
+
+
+def reset_collection() -> None:
+    client = get_qdrant_client()
+    if collection_exists():
+        client.delete_collection(settings.COLLECTION_NAME)
+
+
+def delete_document(doc_id: str) -> None:
+    client = get_qdrant_client()
+    if not collection_exists():
+        return
+    client.delete(
+        collection_name=settings.COLLECTION_NAME,
+        points_selector=models.FilterSelector(
+            filter=models.Filter(
+                must=[
+                    models.FieldCondition(
+                        key="metadata.doc_id",
+                        match=models.MatchValue(value=doc_id),
+                    )
+                ]
+            )
+        ),
+    )

app/engine/ingestion.py

@@ -1,46 +1,133 @@
-import os
-from langchain_community.document_loaders import DirectoryLoader, TextLoader
-from langchain_text_splitters import RecursiveCharacterTextSplitter
-from langchain_community.embeddings import FastEmbedEmbeddings
-from langchain_qdrant import FastEmbedSparse, QdrantVectorStore, RetrievalMode
+import argparse
+import uuid
+from collections import Counter
+from pathlib import Path
 
-from app.core.config import settings
+from langchain_core.documents import Document
 
-def ingest_documents(data_dir: str = "data/docs"):
+from app.engine.chunking import chunk_documents
+from app.engine.document_registry import (
+    file_hash,
+    hash_exists,
+    load_registry,
+    save_registry,
+    upsert_record,
+)
+from app.engine.indexer import delete_document, index_documents, reset_collection
+from app.engine.loaders import SUPPORTED_EXTENSIONS, load_document
+
+
+def ingest_documents(data_dir: str = "data/docs", force: bool = False) -> None:
     print(f"Loading documents from {data_dir}...")
-    if not os.path.exists(data_dir):
-        os.makedirs(data_dir)
-        
-    loader = DirectoryLoader(data_dir, glob="**/*.txt", loader_cls=TextLoader)
-    documents = loader.load()
-    
-    if not documents:
-        print("No documents found. Please place some text documents into data/docs first.")
+    root = Path(data_dir)
+    root.mkdir(parents=True, exist_ok=True)
+
+    registry = {} if force else load_registry()
+    source_documents: list[Document] = []
+    doc_id_by_source: dict[str, str] = {}
+
+    if force:
+        reset_collection()
+
+    for path in sorted(root.rglob("*")):
+        if not path.is_file() or path.suffix.lower() not in SUPPORTED_EXTENSIONS:
+            continue
+
+        content_hash = file_hash(path)
+        if not force and hash_exists(content_hash, registry):
+            print(f"Skipping unchanged document: {path.name}")
+            continue
+
+        doc_id = str(uuid.uuid5(uuid.NAMESPACE_URL, f"{path.name}:{content_hash}"))
+        loaded = load_document(path)
+        for document in loaded:
+            document.metadata["doc_id"] = doc_id
+            document.metadata["content_hash"] = content_hash
+        source_documents.extend(loaded)
+        doc_id_by_source[str(path)] = doc_id
+
+    if not source_documents:
+        print("No new documents found.")
         return
 
-    text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
-    chunks = text_splitter.split_documents(documents)
+    chunks = chunk_documents(source_documents)
     print(f"Split documents into {len(chunks)} chunks.")
 
-    dense_embeddings = FastEmbedEmbeddings(model_name=settings.DENSE_EMBEDDING_MODEL)
-    sparse_embeddings = FastEmbedSparse(model_name=settings.SPARSE_EMBEDDING_MODEL)
-
-    store_options = {
-        "url": settings.QDRANT_URL,
-    } if settings.QDRANT_URL else {
-        "path": settings.QDRANT_LOCATION,
-    }
-
-    QdrantVectorStore.from_documents(
-        chunks,
-        embedding=dense_embeddings,
-        sparse_embedding=sparse_embeddings,
-        collection_name=settings.COLLECTION_NAME,
-        retrieval_mode=RetrievalMode.HYBRID,
-        force_recreate=True,
-        **store_options,
-    )
-    print("Ingestion complete! Hybrid index is built.")
+    index_documents(chunks, force_recreate=force)
+
+    chunk_counts = Counter(chunk.metadata["doc_id"] for chunk in chunks)
+    for source_path, doc_id in doc_id_by_source.items():
+        path = Path(source_path)
+        upsert_record(
+            doc_id=doc_id,
+            source=path.name,
+            source_path=str(path),
+            content_hash=file_hash(path),
+            chunk_count=chunk_counts[doc_id],
+            registry=registry,
+        )
+    save_registry(registry)
+    print("Ingestion complete. Hybrid index is built.")
+
+
+def list_documents() -> None:
+    registry = load_registry()
+    if not registry:
+        print("No indexed documents found.")
+        return
+    for record in registry.values():
+        print(f"{record['doc_id']} | {record['source']} | chunks={record['chunk_count']}")
+
+
+def delete_indexed_document(doc_id: str) -> None:
+    registry = load_registry()
+    if doc_id not in registry:
+        print(f"Document not found: {doc_id}")
+        return
+    delete_document(doc_id)
+    del registry[doc_id]
+    save_registry(registry)
+    print(f"Deleted document: {doc_id}")
+
+
+def reset_index() -> None:
+    reset_collection()
+    save_registry({})
+    print("Vector collection and document registry reset.")
+
+
+def build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(description="Manage support document ingestion.")
+    subparsers = parser.add_subparsers(dest="command")
+
+    ingest_parser = subparsers.add_parser("ingest", help="Ingest documents into Qdrant.")
+    ingest_parser.add_argument("--data-dir", default="data/docs")
+    ingest_parser.add_argument("--force", action="store_true", help="Recreate the collection before ingesting.")
+
+    subparsers.add_parser("list", help="List indexed documents.")
+
+    delete_parser = subparsers.add_parser("delete", help="Delete one indexed document.")
+    delete_parser.add_argument("--doc-id", required=True)
+
+    subparsers.add_parser("reset", help="Delete the vector collection and registry.")
+    return parser
+
+
+def main() -> None:
+    parser = build_parser()
+    args = parser.parse_args()
+
+    if args.command in {None, "ingest"}:
+        ingest_documents(data_dir=getattr(args, "data_dir", "data/docs"), force=getattr(args, "force", False))
+    elif args.command == "list":
+        list_documents()
+    elif args.command == "delete":
+        delete_indexed_document(args.doc_id)
+    elif args.command == "reset":
+        reset_index()
+    else:
+        parser.print_help()
+
 
 if __name__ == "__main__":
-    ingest_documents()
+    main()

app/engine/loaders.py

@@ -0,0 +1,81 @@
+from pathlib import Path
+
+from langchain_core.documents import Document
+
+
+SUPPORTED_EXTENSIONS = {".txt", ".md", ".pdf", ".docx", ".html", ".htm"}
+
+
+def load_documents(data_dir: str) -> list[Document]:
+    root = Path(data_dir)
+    root.mkdir(parents=True, exist_ok=True)
+
+    documents: list[Document] = []
+    for path in sorted(root.rglob("*")):
+        if not path.is_file() or path.suffix.lower() not in SUPPORTED_EXTENSIONS:
+            continue
+        documents.extend(load_document(path))
+    return documents
+
+
+def load_document(path: Path) -> list[Document]:
+    suffix = path.suffix.lower()
+    if suffix in {".txt", ".md"}:
+        return [_text_document(path)]
+    if suffix == ".pdf":
+        return _pdf_documents(path)
+    if suffix == ".docx":
+        return [_docx_document(path)]
+    if suffix in {".html", ".htm"}:
+        return [_html_document(path)]
+    return []
+
+
+def _base_metadata(path: Path) -> dict:
+    return {
+        "source": path.name,
+        "source_path": str(path),
+        "file_type": path.suffix.lower().lstrip("."),
+    }
+
+
+def _text_document(path: Path) -> Document:
+    return Document(page_content=path.read_text(encoding="utf-8", errors="ignore"), metadata=_base_metadata(path))
+
+
+def _pdf_documents(path: Path) -> list[Document]:
+    try:
+        from pypdf import PdfReader
+    except ImportError as exc:
+        raise RuntimeError("Install pypdf to ingest PDF files.") from exc
+
+    reader = PdfReader(str(path))
+    documents: list[Document] = []
+    for index, page in enumerate(reader.pages, start=1):
+        metadata = _base_metadata(path)
+        metadata["page"] = index
+        documents.append(Document(page_content=page.extract_text() or "", metadata=metadata))
+    return documents
+
+
+def _docx_document(path: Path) -> Document:
+    try:
+        from docx import Document as DocxDocument
+    except ImportError as exc:
+        raise RuntimeError("Install python-docx to ingest DOCX files.") from exc
+
+    doc = DocxDocument(str(path))
+    text = "\n".join(paragraph.text for paragraph in doc.paragraphs if paragraph.text.strip())
+    return Document(page_content=text, metadata=_base_metadata(path))
+
+
+def _html_document(path: Path) -> Document:
+    try:
+        from bs4 import BeautifulSoup
+    except ImportError as exc:
+        raise RuntimeError("Install beautifulsoup4 to ingest HTML files.") from exc
+
+    soup = BeautifulSoup(path.read_text(encoding="utf-8", errors="ignore"), "html.parser")
+    for tag in soup(["script", "style"]):
+        tag.decompose()
+    return Document(page_content=soup.get_text("\n", strip=True), metadata=_base_metadata(path))

requirements.txt

@@ -19,3 +19,7 @@ jinja2==3.1.3
 tabulate==0.9.0
 streamlit==1.32.2
 requests==2.34.2
+pypdf==4.3.1
+python-docx==1.1.2
+beautifulsoup4==4.12.3
+pytest==8.2.2