pip install langchain langchain-text-splitters tiktoken

from langchain_text_splitters import (
    RecursiveCharacterTextSplitter,
    CharacterTextSplitter,
    MarkdownHeaderTextSplitter,
    PythonCodeTextSplitter,
    TokenTextSplitter,
)

# ── 1. RecursiveCharacterTextSplitter — your default ──
# Tries to split on: \n\n, \n, " ", "" in that order
# Produces naturally bounded chunks (paragraphs, then sentences)
splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,       # max characters per chunk
    chunk_overlap=50,    # characters of overlap
    length_function=len,  # use len(str) — swap for token counter
    separators=["\n\n", "\n", " ", ""]   # priority order
)
chunks = splitter.split_text(long_text)
print(f"{len(chunks)} chunks, avg length: {sum(len(c) for c in chunks)//len(chunks)}")

# ── 2. Token-based splitting (recommended for LLM context) ──
# Characters are misleading — tokens are what the LLM actually counts
splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
    model_name="gpt-4",    # tokeniser to use
    chunk_size=400,        # max TOKENS per chunk
    chunk_overlap=40       # TOKENS of overlap
)
chunks = splitter.split_text(long_text)

# ── 3. Document splitting — preserves metadata ───────
from langchain_core.documents import Document

docs = [Document(page_content=text, metadata={"source": "dpdk_guide.pdf", "page": 1})]
split_docs = splitter.split_documents(docs)
# Each chunk keeps metadata from parent document
print(split_docs[0].metadata)   # {"source": "dpdk_guide.pdf", "page": 1}

pip install pymupdf python-docx beautifulsoup4 unstructured

# ── PDF — PyMuPDF (fastest, best quality) ────────────
import fitz   # PyMuPDF

def load_pdf(path: str) -> list[dict]:
    """Load PDF, return list of {text, page, source} dicts."""
    doc = fitz.open(path)
    pages = []
    for page_num, page in enumerate(doc):
        text = page.get_text("text")   # plain text extraction
        if text.strip():              # skip blank pages
            pages.append({
                "text":   text,
                "page":   page_num + 1,
                "source": path
            })
    doc.close()
    return pages

# ── DOCX ─────────────────────────────────────────────
from docx import Document as DocxDocument

def load_docx(path: str) -> str:
    doc = DocxDocument(path)
    paragraphs = [p.text for p in doc.paragraphs if p.text.strip()]
    return "\n\n".join(paragraphs)

# ── HTML ─────────────────────────────────────────────
from bs4 import BeautifulSoup
import requests

def load_html(url: str) -> str:
    response = requests.get(url, timeout=10)
    soup = BeautifulSoup(response.text, "html.parser")
    # Remove noise elements
    for tag in soup.find_all(["script", "style", "nav", "footer", "header"]):
        tag.decompose()
    return soup.get_text(separator="\n", strip=True)

# ── Markdown ──────────────────────────────────────────
from pathlib import Path

def load_markdown(path: str) -> str:
    return Path(path).read_text(encoding="utf-8")

# ── Directory loader — batch ingest ───────────────────
import os
from pathlib import Path

def load_directory(dir_path: str, extensions: list[str] = [".txt", ".md", ".pdf"]) -> list[dict]:
    docs = []
    for path in Path(dir_path).rglob("*"):
        if path.suffix in extensions and path.is_file():
            try:
                if path.suffix == ".pdf":
                    for page in load_pdf(str(path)):
                        docs.append(page)
                else:
                    text = path.read_text(encoding="utf-8", errors="ignore")
                    docs.append({"text": text, "source": str(path)})
            except Exception as e:
                print(f"Failed to load {path}: {e}")
    return docs

# Minimum metadata per chunk
chunk_metadata = {
    "source":      "dpdk-programmers-guide-v23.pdf",
    "source_type": "pdf",             # pdf | html | docx | md | code
    "chunk_idx":   42,                # position in document
    "char_count":  487,
}

# Good metadata per chunk (for serious RAG)
chunk_metadata = {
    "source":      "dpdk-programmers-guide-v23.pdf",
    "source_type": "pdf",
    "page":        47,
    "section":     "Memory Management",
    "subsection":  "Mempool Library",
    "chunk_idx":   42,
    "total_chunks": 380,
    "ingested_at": "2024-03-15T10:30:00Z",
    "doc_version": "23.11",
    "language":    "en",
    "token_count": 412,
}

# For web content
web_chunk_metadata = {
    "url":         "https://doc.dpdk.org/guides/prog_guide/mempool_lib.html",
    "title":       "Mempool Library — DPDK documentation",
    "scraped_at":  "2024-03-15",
    "domain":      "doc.dpdk.org",
    "section":     "Programmer's Guide",
}

import asyncio, hashlib, json
from pathlib import Path
from dataclasses import dataclass, field
from typing import Optional
import chromadb
from chromadb.utils import embedding_functions
from langchain_text_splitters import RecursiveCharacterTextSplitter

@dataclass
class IngestionConfig:
    chunk_size:    int   = 500
    chunk_overlap: int   = 50
    min_chunk_len: int   = 100     # discard shorter chunks
    embedding_model: str = "text-embedding-3-small"
    collection_name: str = "documents"
    chroma_path:    str  = "./chroma_db"
    add_context:    bool = False   # enable LLM context enrichment

class DocumentIngestionPipeline:
    def __init__(self, config: IngestionConfig = IngestionConfig()):
        self.config   = config
        self.splitter = RecursiveCharacterTextSplitter.from_tiktoken_encoder(
            model_name="gpt-4",
            chunk_size=config.chunk_size,
            chunk_overlap=config.chunk_overlap,
        )
        self.chroma   = chromadb.PersistentClient(path=config.chroma_path)
        self.ef       = embedding_functions.OpenAIEmbeddingFunction(
            api_key=os.environ["OPENAI_API_KEY"],
            model_name=config.embedding_model
        )
        self.collection = self.chroma.get_or_create_collection(
            name=config.collection_name,
            embedding_function=self.ef,
            metadata={"hnsw:space": "cosine"}
        )

    def _doc_id(self, text: str, meta: dict) -> str:
        """Stable ID based on content hash — prevents duplicate ingestion."""
        key = json.dumps({"text": text, "source": meta.get("source", "")})
        return hashlib.md5(key.encode()).hexdigest()

    def ingest_text(self, text: str, metadata: dict = {}) -> int:
        """Ingest a single text string. Returns number of chunks added."""
        # Clean
        text = clean_text(text)
        if not text.strip():
            return 0

        # Chunk
        chunks = self.splitter.split_text(text)
        chunks = [c for c in chunks if len(c) >= self.config.min_chunk_len]

        if not chunks:
            return 0

        # Build IDs, documents, metadatas
        ids, docs, metas = [], [], []
        for i, chunk in enumerate(chunks):
            chunk_meta = {
                **metadata,
                "chunk_idx":    i,
                "total_chunks": len(chunks),
                "char_count":   len(chunk),
                "ingested_at":  datetime.utcnow().isoformat(),
            }
            ids.append(self._doc_id(chunk, chunk_meta))
            docs.append(chunk)
            metas.append(chunk_meta)

        # Add to ChromaDB (skips existing IDs automatically)
        self.collection.upsert(ids=ids, documents=docs, metadatas=metas)
        return len(chunks)

    def ingest_file(self, path: str) -> int:
        """Auto-detect file type and ingest."""
        path = Path(path)
        meta = {"source": str(path), "filename": path.name}

        if path.suffix == ".pdf":
            total = 0
            for page in load_pdf(str(path)):
                total += self.ingest_text(page["text"], {**meta, "page": page["page"]})
            return total
        elif path.suffix == ".docx":
            text = load_docx(str(path))
        elif path.suffix in (".md", ".txt"):
            text = path.read_text(encoding="utf-8")
        else:
            raise ValueError(f"Unsupported file type: {path.suffix}")

        return self.ingest_text(text, meta)

    def ingest_directory(self, dir_path: str) -> dict:
        """Ingest all supported files in a directory."""
        results = {"files": 0, "chunks": 0, "errors": []}
        for path in Path(dir_path).rglob("*"):
            if path.suffix in (".pdf", ".docx", ".md", ".txt") and path.is_file():
                try:
                    n = self.ingest_file(str(path))
                    results["chunks"] += n
                    results["files"]  += 1
                except Exception as e:
                    results["errors"].append({"file": str(path), "error": str(e)})
        return results

    def query(self, text: str, n_results: int = 5, where: dict = None) -> list[dict]:
        """Semantic search — returns list of {text, score, metadata}."""
        kwargs = {"query_texts": [text], "n_results": n_results,
                  "include": ["documents", "distances", "metadatas"]}
        if where:
            kwargs["where"] = where
        results = self.collection.query(**kwargs)
        return [
            {"text": doc, "score": 1 - dist, "meta": meta}
            for doc, dist, meta in zip(
                results["documents"][0],
                results["distances"][0],
                results["metadatas"][0]
            )
        ]