Add DocEmbedder docs and examples

Signed-off-by: Chaitany patel <patelchaitany93@gmail.com>

Author: patelchaitany

docs/getting-started/genai.md

@@ -56,6 +56,53 @@ The transformation workflow typically involves:
 3. **Chunking**: Split documents into smaller, semantically meaningful chunks
 4. **Embedding Generation**: Convert text chunks into vector embeddings
 5. **Storage**: Store embeddings and metadata in Feast's feature store
+
+### DocEmbedder: End-to-End Document Ingestion Pipeline
+
+The `DocEmbedder` class provides an end-to-end pipeline for ingesting documents into Feast's online vector store. It handles chunking, embedding generation, and writing results -- all in a single step.
+
+#### Key Components
+
+* **`DocEmbedder`**: High-level orchestrator that runs the full pipeline: chunk → embed → schema transform → write to online store
+* **`BaseChunker` / `TextChunker`**: Pluggable chunking layer. `TextChunker` splits text by word count with configurable `chunk_size`, `chunk_overlap`, `min_chunk_size`, and `max_chunk_chars`
+* **`BaseEmbedder` / `MultiModalEmbedder`**: Pluggable embedding layer with modality routing. `MultiModalEmbedder` supports text (via sentence-transformers) and image (via CLIP) with lazy model loading
+* **`SchemaTransformFn`**: A user-defined function that transforms the chunked + embedded DataFrame into the format expected by the FeatureView schema
+
+#### Quick Example
+
+```python
+from feast import DocEmbedder
+import pandas as pd
+
+# Prepare your documents
+df = pd.DataFrame({
+    "id": ["doc1", "doc2"],
+    "text": ["First document content...", "Second document content..."],
+})
+
+# Create DocEmbedder -- automatically generates a FeatureView and applies the repo
+embedder = DocEmbedder(
+    repo_path="feature_repo/",
+    feature_view_name="text_feature_view",
+)
+
+# Embed and ingest documents in one step
+result = embedder.embed_documents(
+    documents=df,
+    id_column="id",
+    source_column="text",
+    column_mapping=("text", "text_embedding"),
+)
+```
+
+#### Features
+
+* **Auto-generates FeatureView**: Creates a Python file with Entity and FeatureView definitions compatible with `feast apply`
+* **Auto-applies repo**: Registers the generated FeatureView in the registry automatically
+* **Custom schema transform**: Provide your own `SchemaTransformFn` to control how chunked + embedded data maps to your FeatureView schema
+* **Extensible**: Subclass `BaseChunker` or `BaseEmbedder` to plug in your own chunking or embedding strategies
+
+For a complete walkthrough, see the [DocEmbedder tutorial notebook](../../examples/rag-retriever/rag_feast_docembedder.ipynb).
 ### Feature Transformation for LLMs
 
 Feast supports transformations that can be used to:
@@ -190,6 +237,7 @@ For more detailed information and examples:
 
 * [Vector Database Reference](../reference/alpha-vector-database.md)
 * [RAG Tutorial with Docling](../tutorials/rag-with-docling.md)
+* [DocEmbedder Tutorial Notebook](../../examples/rag-retriever/rag_feast_docembedder.ipynb)
 * [RAG Fine Tuning with Feast and Milvus](../../examples/rag-retriever/README.md)
 * [Milvus Quickstart Example](https://github.com/feast-dev/feast/tree/master/examples/rag/milvus-quickstart.ipynb)
 * [Feast + Ray: Distributed Processing for RAG Applications](https://feast.dev/blog/feast-ray-distributed-processing/)

docs/tutorials/rag-with-docling.md

@@ -409,6 +409,234 @@ response = client.chat.completions.create(
 print('\n'.join([c.message.content for c in response.choices]))
 ```
 
+## Alternative: Using DocEmbedder for Simplified Ingestion
+
+Instead of manually chunking, embedding, and writing documents as shown above, you can use Feast's `DocEmbedder` class to handle the entire pipeline in a single step. `DocEmbedder` automates chunking, embedding generation, FeatureView creation, and writing to the online store.
+
+### Install Dependencies
+
+```bash
+pip install feast[milvus,rag]
+```
+
+### Set Up and Ingest with DocEmbedder
+
+```python
+from feast import DocEmbedder
+import pandas as pd
+
+# Prepare your documents as a DataFrame
+df = pd.DataFrame({
+    "id": ["doc1", "doc2", "doc3"],
+    "text": [
+        "Aaron is a prophet, high priest, and the brother of Moses...",
+        "God at Sinai granted Aaron the priesthood for himself...",
+        "His rod turned into a snake. Then he stretched out...",
+    ],
+})
+
+# DocEmbedder handles everything: generates FeatureView, applies repo,
+# chunks text, generates embeddings, and writes to the online store
+embedder = DocEmbedder(
+    repo_path="feature_repo/",
+    feature_view_name="text_feature_view",
+)
+
+result = embedder.embed_documents(
+    documents=df,
+    id_column="id",
+    source_column="text",
+    column_mapping=("text", "text_embedding"),
+)
+```
+
+### Retrieve and Query
+
+Once documents are ingested, you can retrieve them the same way as shown in Step 5 above:
+
+```python
+from feast import FeatureStore
+
+store = FeatureStore("feature_repo/")
+
+query_embedding = embed_text("Who are the authors of the paper?")
+context_data = store.retrieve_online_documents_v2(
+    features=[
+        "text_feature_view:embedding",
+        "text_feature_view:text",
+        "text_feature_view:source_id",
+    ],
+    query=query_embedding,
+    top_k=3,
+    distance_metric="COSINE",
+).to_df()
+```
+
+### Customizing the Pipeline
+
+`DocEmbedder` is extensible at every stage. Below are examples of how to create custom components and wire them together.
+
+#### Custom Chunker
+
+Subclass `BaseChunker` to implement your own chunking strategy. The `load_parse_and_chunk` method receives each document and must return a list of chunk dictionaries.
+
+```python
+from feast.chunker import BaseChunker, ChunkingConfig
+from typing import Any, Optional
+
+class SentenceChunker(BaseChunker):
+    """Chunks text by sentences instead of word count."""
+
+    def load_parse_and_chunk(
+        self,
+        source: Any,
+        source_id: str,
+        source_column: str,
+        source_type: Optional[str] = None,
+    ) -> list[dict]:
+        import re
+
+        text = str(source)
+        # Split on sentence boundaries
+        sentences = re.split(r'(?<=[.!?])\s+', text)
+
+        chunks = []
+        current_chunk = []
+        chunk_index = 0
+
+        for sentence in sentences:
+            current_chunk.append(sentence)
+            combined = " ".join(current_chunk)
+
+            if len(combined.split()) >= self.config.chunk_size:
+                chunks.append({
+                    "chunk_id": f"{source_id}_{chunk_index}",
+                    "original_id": source_id,
+                    source_column: combined,
+                    "chunk_index": chunk_index,
+                })
+                # Keep overlap by retaining the last sentence
+                current_chunk = [sentence]
+                chunk_index += 1
+
+        # Don't forget the last chunk
+        if current_chunk and len(" ".join(current_chunk).split()) >= self.config.min_chunk_size:
+            chunks.append({
+                "chunk_id": f"{source_id}_{chunk_index}",
+                "original_id": source_id,
+                source_column: " ".join(current_chunk),
+                "chunk_index": chunk_index,
+            })
+
+        return chunks
+```
+
+Or simply configure the built-in `TextChunker`:
+
+```python
+from feast import TextChunker, ChunkingConfig
+
+chunker = TextChunker(config=ChunkingConfig(
+    chunk_size=200,
+    chunk_overlap=50,
+    min_chunk_size=30,
+    max_chunk_chars=1000,
+))
+```
+
+#### Custom Embedder
+
+Subclass `BaseEmbedder` to use a different embedding model. Register modality handlers in `_register_default_modalities` and implement the `embed` method.
+
+```python
+from feast.embedder import BaseEmbedder, EmbeddingConfig
+from typing import Any, List, Optional
+import numpy as np
+
+class OpenAIEmbedder(BaseEmbedder):
+    """Embedder that uses the OpenAI API for text embeddings."""
+
+    def __init__(self, model: str = "text-embedding-3-small", config: Optional[EmbeddingConfig] = None):
+        self.model = model
+        self._client = None
+        super().__init__(config)
+
+    def _register_default_modalities(self) -> None:
+        self.register_modality("text", self._embed_text)
+
+    @property
+    def client(self):
+        if self._client is None:
+            from openai import OpenAI
+            self._client = OpenAI()
+        return self._client
+
+    def get_embedding_dim(self, modality: str) -> Optional[int]:
+        # text-embedding-3-small produces 1536-dim vectors
+        if modality == "text":
+            return 1536
+        return None
+
+    def embed(self, inputs: List[Any], modality: str) -> np.ndarray:
+        if modality not in self._modality_handlers:
+            raise ValueError(f"Unsupported modality: '{modality}'")
+        return self._modality_handlers[modality](inputs)
+
+    def _embed_text(self, inputs: List[str]) -> np.ndarray:
+        response = self.client.embeddings.create(input=inputs, model=self.model)
+        return np.array([item.embedding for item in response.data])
+```
+
+#### Custom Logical Layer Function
+
+The schema transform function transforms the chunked + embedded DataFrame into the exact schema your FeatureView expects. It must accept a `pd.DataFrame` and return a `pd.DataFrame`.
+
+```python
+import pandas as pd
+from datetime import datetime, timezone
+
+def my_schema_transform_fn(df: pd.DataFrame) -> pd.DataFrame:
+    """Map chunked + embedded columns to the FeatureView schema."""
+    return pd.DataFrame({
+        "passage_id": df["chunk_id"],
+        "text": df["text"],
+        "embedding": df["text_embedding"],
+        "event_timestamp": [datetime.now(timezone.utc)] * len(df),
+        "source_id": df["original_id"],
+        # Add any extra columns your FeatureView expects
+        "chunk_index": df["chunk_index"],
+    })
+```
+
+#### Putting It All Together
+
+Pass your custom components to `DocEmbedder`:
+
+```python
+from feast import DocEmbedder
+
+embedder = DocEmbedder(
+    repo_path="feature_repo/",
+    feature_view_name="text_feature_view",
+    chunker=SentenceChunker(config=ChunkingConfig(chunk_size=150, min_chunk_size=20)),
+    embedder=OpenAIEmbedder(model="text-embedding-3-small"),
+    schema_transform_fn=my_schema_transform_fn,
+    vector_length=1536,  # Match the OpenAI embedding dimension
+)
+
+# Embed and ingest
+result = embedder.embed_documents(
+    documents=df,
+    id_column="id",
+    source_column="text",
+    column_mapping=("text", "text_embedding"),
+)
+```
+
+> **Note:** When using a custom `schema_transform_fn`, ensure the returned DataFrame columns match your FeatureView schema. When using a custom embedder with a different output dimension, set `vector_length` accordingly (or let it auto-detect via `get_embedding_dim`).
+
+For a complete end-to-end example, see the [DocEmbedder notebook](https://github.com/feast-dev/feast/tree/master/examples/rag-retriever/rag_feast_docembedder.ipynb).
+
 ## Why Feast for RAG?
 
 Feast makes it remarkably easy to set up and manage a RAG system by:

examples/rag-retriever/README.md

@@ -62,6 +62,59 @@ Navigate to the examples/rag-retriever directory. Here you will find the followi
 
 Open `rag_feast.ipynb` and follow the steps in the notebook to run the example.
 
+## Using DocEmbedder for Simplified Ingestion
+
+As an alternative to the manual data preparation steps in the notebook above, Feast provides the `DocEmbedder` class that automates the entire document-to-embeddings pipeline: chunking, embedding generation, FeatureView creation, and writing to the online store.
+
+### Install Dependencies
+
+```bash
+pip install feast[milvus,rag]
+```
+
+### Quick Start
+
+```python
+from feast import DocEmbedder
+from datasets import load_dataset
+
+# Load your dataset
+dataset = load_dataset("facebook/wiki_dpr", "psgs_w100.nq.exact", split="train[:1%]",
+                       with_index=False, trust_remote_code=True)
+df = dataset.select(range(100)).to_pandas()
+
+# DocEmbedder handles everything in one step
+embedder = DocEmbedder(
+    repo_path="feature_repo_docembedder/",
+    feature_view_name="text_feature_view",
+)
+
+result = embedder.embed_documents(
+    documents=df,
+    id_column="id",
+    source_column="text",
+    column_mapping=("text", "text_embedding"),
+)
+```
+
+### What DocEmbedder Does
+
+1. **Generates a FeatureView**: Automatically creates a Python file with Entity and FeatureView definitions compatible with `feast apply`
+2. **Applies the repo**: Registers the FeatureView in the Feast registry and deploys infrastructure (e.g., Milvus collection)
+3. **Chunks documents**: Splits text into smaller passages using `TextChunker` (configurable chunk size, overlap, etc.)
+4. **Generates embeddings**: Produces vector embeddings using `MultiModalEmbedder` (defaults to `all-MiniLM-L6-v2`)
+5. **Writes to online store**: Stores the processed data in your configured online store (e.g., Milvus)
+
+### Customization
+
+* **Custom Chunker**: Subclass `BaseChunker` for your own chunking strategy
+* **Custom Embedder**: Subclass `BaseEmbedder` to use a different embedding model
+* **Logical Layer Function**: Provide a `SchemaTransformFn` to control how the output maps to your FeatureView schema
+
+### Example Notebook
+
+See **`rag_feast_docembedder.ipynb`** for a complete end-to-end example that uses DocEmbedder with the Wiki DPR dataset and then queries the results using `FeastRAGRetriever`.
+
 ## FeastRagRetriver Low Level Design
 
 <img src="images/FeastRagRetriever.png" width="800" height="450" alt="Low level design for feast rag retriever">

sdk/python/feast/__init__.py

@@ -17,7 +17,7 @@
 from .chunker import BaseChunker, ChunkingConfig, TextChunker
 from .data_source import KafkaSource, KinesisSource, PushSource, RequestSource
 from .dataframe import DataFrameEngine, FeastDataFrame
-from .doc_embedder import DocEmbedder, LogicalLayerFn
+from .doc_embedder import DocEmbedder, SchemaTransformFn
 from .embedder import BaseEmbedder, EmbeddingConfig, MultiModalEmbedder
 from .entity import Entity
 from .feature import Feature
@@ -66,7 +66,7 @@
     "Project",
     "FeastVectorStore",
     "DocEmbedder",
-    "LogicalLayerFn",
+    "SchemaTransformFn",
     "BaseChunker",
     "TextChunker",
     "ChunkingConfig",