Overview
The DocumentStore module provides abstract and concrete implementations for storing documents with metadata and performing vector similarity searches, backed by SQLAlchemy and a pluggable vector database.
Description
This module defines the full document storage stack for Guardrails:
Document dataclass: Represents a document with an id, a dictionary of pages (page number to text), and optional metadata.
PageCoordinates namedtuple: Locates a page within a document via doc_id and page_num.
Page dataclass: Represents a single page with its coordinates, text content, and metadata.
DocumentStoreBase ABC: Abstract base class defining the interface for document stores: add_document, search, add_text, add_texts, and flush.
EphemeralDocumentStore: Concrete implementation using a VectorDBBase for similarity search and a SQLAlchemy-backed SQLMetadataStore for metadata persistence. Documents are MD5-hashed for deduplication.
SQLMetadataStore: SQLAlchemy-backed metadata storage with a documents table mapping document pages to vector indices.
The module includes a fallback mechanism: if SQLAlchemy is not installed, the exported classes raise ImportError at instantiation time rather than at import time.
Usage
Use DocumentStore classes when you need to store text documents and retrieve them by semantic similarity. This is used by the Text2Sql application to store and retrieve example query pairs, and can be used in any Guardrails application that needs retrieval-augmented generation.
Code Reference
Source Location
- Repository: Guardrails
- File:
guardrails/document_store.py
- Lines: 1-265
Signature
@dataclass
class Document:
id: str
pages: Dict[int, str]
metadata: Dict[Any, Any] = Field(default_factory=dict)
PageCoordinates = namedtuple("PageCoordinates", ["doc_id", "page_num"])
@dataclass
class Page:
cordinates: PageCoordinates
text: str
metadata: Dict[Any, Any]
class DocumentStoreBase(ABC):
def __init__(self, vector_db: VectorDBBase, path: Optional[str] = None): ...
def add_document(self, document: Document) -> None: ...
def search(self, query: str, k: int = 4) -> List[Page]: ...
def add_text(self, text: str, meta: Dict[Any, Any]) -> str: ...
def add_texts(self, texts: Dict[str, Dict[Any, Any]]) -> List[str]: ...
def flush(): ...
class EphemeralDocumentStore(DocumentStoreBase):
def __init__(self, vector_db: VectorDBBase, path: Optional[str] = None):
def add_document(self, document: Document):
def add_text(self, text: str, meta: Dict[Any, Any]) -> str:
def add_texts(self, texts: Dict[str, Dict[Any, Any]]) -> List[str]:
def search(self, query: str, k: int = 4) -> List[Page]:
def search_with_threshold(self, query: str, threshold: float, k: int = 4) -> List[Page]:
def flush(self, path: Optional[str] = None):
Import
from guardrails.document_store import (
Document,
Page,
PageCoordinates,
DocumentStoreBase,
EphemeralDocumentStore,
)
I/O Contract
Document Fields
| Field |
Type |
Description
|
id |
str |
Unique identifier for the document.
|
pages |
Dict[int, str] |
Mapping of page numbers to page text content.
|
metadata |
Dict[Any, Any] |
Arbitrary metadata associated with the document.
|
Page Fields
| Field |
Type |
Description
|
cordinates |
PageCoordinates |
Location of this page within its document (doc_id, page_num).
|
text |
str |
The text content of the page.
|
metadata |
Dict[Any, Any] |
Metadata associated with this page.
|
search
| Parameter |
Type |
Default |
Description
|
query |
str |
required |
Text query for similarity search.
|
k |
int |
4 |
Number of similar pages to return.
|
| Return Type |
Description
|
List[Page] |
List of pages with text similar to the query.
|
add_text
| Parameter |
Type |
Description
|
text |
str |
Text content to add.
|
meta |
Dict[Any, Any] |
Metadata to associate with the text.
|
| Return Type |
Description
|
str |
The MD5-based ID of the added text.
|
SQL Document Schema
| Column |
Type |
Description
|
id |
INTEGER (PK) |
Document identifier.
|
page_num |
INTEGER (PK) |
Page number within the document.
|
text |
String |
Page text content.
|
meta |
PickleType |
Serialized metadata dictionary.
|
vector_index |
Integer |
Index into the vector database.
|
Usage Examples
from guardrails.document_store import (
Document,
EphemeralDocumentStore,
)
from guardrails.vectordb import Faiss
from guardrails.embedding import OpenAIEmbedding
# Set up the embedding and vector database
embedding = OpenAIEmbedding()
vector_db = Faiss.new_flat_l2_index(embedding.output_dim, embedder=embedding)
# Create the document store
store = EphemeralDocumentStore(vector_db)
# Add individual texts with metadata
doc_id = store.add_text(
"How to count users",
{"ctx": "SELECT COUNT(*) FROM users"}
)
# Add multiple texts at once
store.add_texts({
"Get active users": {"ctx": "SELECT * FROM users WHERE active = 1"},
"Total revenue": {"ctx": "SELECT SUM(amount) FROM orders"},
})
# Search for similar documents
results = store.search("Find all users", k=2)
for page in results:
print(page.text, page.metadata)
# Persist vector database to disk
store.flush("/path/to/save")
Related Pages