Overview
The Embedding module provides abstract and concrete embedding model implementations for converting text into vector representations, including OpenAI and Manifest backends.
Description
This module defines the embedding layer used by Guardrails for vector similarity operations:
EmbeddingBase ABC: Abstract base class providing the interface and shared utilities for embedding models. Key features include:
embed(texts) and embed_query(query) abstract methods._len_safe_get_embedding: Splits long text into token chunks, embeds each chunk, and averages the resulting vectors (weighted by chunk length) to produce a single normalized embedding._chunked_tokens: Uses tiktoken to tokenize text and yield decoded token chunks._batched: A generic utility for batching iterables into fixed-size groups.
OpenAIEmbedding: Concrete implementation using the OpenAI embeddings API. Defaults to the text-embedding-ada-002 model with cl100k_base encoding and 8191 max tokens. Provides output dimension lookup for common OpenAI embedding models.
ManifestEmbedding: Concrete implementation using the manifest-ml library, which supports multiple embedding backends and optional caching. Output dimension is determined dynamically via a test embedding.
Usage
Use EmbeddingBase subclasses to generate text embeddings for vector similarity search in document stores. OpenAIEmbedding is the default choice and is used by Text2Sql and DocumentStore. Use ManifestEmbedding when you need alternative backends or local caching.
Code Reference
Source Location
- Repository: Guardrails
- File:
guardrails/embedding.py
- Lines: 1-218
Signature
class EmbeddingBase(ABC):
def __init__(self, model=None, encoding_name=None, max_tokens=None):
def embed(self, texts: List[str]) -> List[List[float]]: ...
def embed_query(self, query: str) -> List[float]: ...
def _len_safe_get_embedding(self, text, embedder, average=True) -> List[float]:
def _chunked_tokens(text, encoding_name, chunk_length):
def _batched(iterable, n):
def output_dim(self) -> int:
class OpenAIEmbedding(EmbeddingBase):
def __init__(self, model="text-embedding-ada-002", encoding_name="cl100k_base",
max_tokens=8191, api_key=None, api_base=None):
def embed(self, texts: List[str]) -> List[List[float]]:
def embed_query(self, query: str) -> List[float]:
def output_dim(self) -> int:
class ManifestEmbedding(EmbeddingBase):
def __init__(self, client_name="openai", client_connection=None,
cache_name=None, cache_connection=None, engine="text-embedding-ada-002",
encoding_name="cl100k_base", max_tokens=8191):
def embed(self, texts: List[str]) -> List[List[float]]:
def embed_query(self, query: str) -> List[float]:
def output_dim(self) -> int:
Import
from guardrails.embedding import EmbeddingBase, OpenAIEmbedding, ManifestEmbedding
I/O Contract
EmbeddingBase.__init__
| Parameter |
Type |
Default |
Description
|
model |
Optional[str] |
None |
Name of the embedding model.
|
encoding_name |
Optional[str] |
None |
Tiktoken encoding name for tokenization.
|
max_tokens |
Optional[int] |
None |
Maximum tokens per chunk for long text splitting.
|
embed
| Parameter |
Type |
Description
|
texts |
List[str] |
List of text strings to embed.
|
| Return Type |
Description
|
List[List[float]] |
List of embedding vectors, one per input text.
|
embed_query
| Parameter |
Type |
Description
|
query |
str |
A single text string to embed.
|
| Return Type |
Description
|
List[float] |
The embedding vector for the query.
|
output_dim (OpenAIEmbedding)
| Model |
Output Dimension
|
text-embedding-ada-002 |
1536
|
Other ada models |
1024
|
babbage models |
2048
|
curie models |
4096
|
davinci models |
12288
|
OpenAIEmbedding.__init__
| Parameter |
Type |
Default |
Description
|
model |
str |
"text-embedding-ada-002" |
OpenAI embedding model name.
|
encoding_name |
str |
"cl100k_base" |
Tiktoken encoding for tokenization.
|
max_tokens |
int |
8191 |
Maximum tokens per embedding chunk.
|
api_key |
Optional[str] |
None |
OpenAI API key override.
|
api_base |
Optional[str] |
None |
OpenAI API base URL override.
|
ManifestEmbedding.__init__
| Parameter |
Type |
Default |
Description
|
client_name |
str |
"openai" |
Manifest client backend name.
|
client_connection |
Optional[str] |
None |
Connection string for the client.
|
cache_name |
Optional[str] |
None |
Name of the caching backend.
|
cache_connection |
Optional[str] |
None |
Connection string for the cache.
|
engine |
Optional[str] |
"text-embedding-ada-002" |
Embedding engine/model name.
|
encoding_name |
Optional[str] |
"cl100k_base" |
Tiktoken encoding for tokenization.
|
max_tokens |
Optional[int] |
8191 |
Maximum tokens per embedding chunk.
|
Usage Examples
from guardrails.embedding import OpenAIEmbedding, ManifestEmbedding
# Using OpenAI embeddings
embedder = OpenAIEmbedding(
model="text-embedding-ada-002",
api_key="sk-..."
)
# Embed a single query
vector = embedder.embed_query("What is the total revenue?")
print(len(vector)) # 1536
# Embed multiple texts
vectors = embedder.embed(["Hello world", "Goodbye world"])
print(len(vectors)) # 2
# Get output dimension
print(embedder.output_dim) # 1536
# Using Manifest embeddings with caching
manifest_embedder = ManifestEmbedding(
client_name="openai",
cache_name="sqlite",
cache_connection="/tmp/embeddings_cache.db",
)
vector = manifest_embedder.embed_query("Find all active users")
Related Pages
