Implementation:Unstructured IO Unstructured BaseEmbeddingEncoder

Knowledge Sources	Unstructured
Domains	NLP, RAG, Embeddings
Last Updated	2026-02-12 00:00 GMT

Overview

Concrete tool defining the abstract embedding encoder interface provided by the Unstructured library.

Description

The BaseEmbeddingEncoder is an abstract dataclass that defines the contract all embedding providers must implement. It is paired with EmbeddingConfig, a Pydantic BaseModel that holds provider-specific configuration (API keys, model names, region). Concrete implementations include OpenAIEmbeddingEncoder, BedrockEmbeddingEncoder, HuggingFaceEmbeddingEncoder, VertexAIEmbeddingEncoder, VoyageAIEmbeddingEncoder, and MixedBreadAIEmbeddingEncoder.

Usage

Import these base classes when implementing a new embedding provider or when writing code that operates generically over any embedding provider. For using a specific provider, import the concrete class directly (e.g., OpenAIEmbeddingEncoder).

Code Reference

Source Location

• Repository: unstructured
• File: unstructured/embed/interfaces.py
• Lines: 10-39

Signature

class EmbeddingConfig(BaseModel):
    """Base configuration for embedding providers. Subclass for provider-specific fields."""
    pass

@dataclass
class BaseEmbeddingEncoder(ABC):
    config: EmbeddingConfig

    @abstractmethod
    def initialize(self):
        """Initialize the embedding client connection."""

    @property
    @abstractmethod
    def num_of_dimensions(self) -> Tuple[int]:
        """Return the embedding vector dimensionality."""

    @property
    @abstractmethod
    def is_unit_vector(self) -> bool:
        """Return whether embeddings are L2-normalized."""

    @abstractmethod
    def embed_documents(self, elements: List[Element]) -> List[Element]:
        """Embed document elements, storing vectors in element.embeddings."""

    @abstractmethod
    def embed_query(self, query: str) -> List[float]:
        """Embed a single query string, returning the vector."""

Import

from unstructured.embed.interfaces import BaseEmbeddingEncoder, EmbeddingConfig

I/O Contract

Inputs (embed_documents)

Name	Type	Required	Description
elements	List[Element]	Yes	Document elements to embed (text converted via str())

Inputs (embed_query)

Name	Type	Required	Description
query	str	Yes	Query text to embed

Outputs

Name	Type	Description
embed_documents return	List[Element]	Same elements with element.embeddings populated (list[float])
embed_query return	List[float]	Embedding vector for the query string
num_of_dimensions	Tuple[int]	Vector dimensionality
is_unit_vector	bool	Whether vectors are L2-normalized

Usage Examples

Using a Concrete Provider (OpenAI)

from unstructured.embed.openai import OpenAIEmbeddingConfig, OpenAIEmbeddingEncoder
from unstructured.partition.auto import partition

# 1. Configure the provider
config = OpenAIEmbeddingConfig(api_key="sk-...")

# 2. Create encoder
encoder = OpenAIEmbeddingEncoder(config=config)
encoder.initialize()

# 3. Partition and embed
elements = partition(filename="report.pdf")
embedded_elements = encoder.embed_documents(elements)

# 4. Access embeddings
for el in embedded_elements:
    if el.embeddings:
        print(f"Dimensions: {len(el.embeddings)}")

Implementing a Custom Provider

from dataclasses import dataclass
from typing import List, Tuple
from unstructured.embed.interfaces import BaseEmbeddingEncoder, EmbeddingConfig
from unstructured.documents.elements import Element

class CustomConfig(EmbeddingConfig):
    api_endpoint: str
    api_key: str

@dataclass
class CustomEmbeddingEncoder(BaseEmbeddingEncoder):
    config: CustomConfig

    def initialize(self):
        self.client = create_client(self.config.api_endpoint, self.config.api_key)

    @property
    def num_of_dimensions(self) -> Tuple[int]:
        return (768,)

    @property
    def is_unit_vector(self) -> bool:
        return True

    def embed_documents(self, elements: List[Element]) -> List[Element]:
        texts = [str(e) for e in elements]
        vectors = self.client.embed_batch(texts)
        for el, vec in zip(elements, vectors):
            el.embeddings = vec
        return elements

    def embed_query(self, query: str) -> List[float]:
        return self.client.embed(query)

Related Pages

Implements Principle

• Principle:Unstructured_IO_Unstructured_Embedding_Provider_Interface