basic-usage

Installation

uv add llmemory
# or
pip install llmemory

Prerequisites:

• Python 3.10 or higher
• PostgreSQL 14+ (tested up to PostgreSQL 16)
• pgvector extension 0.5.0+
• OpenAI API key (or configure local embeddings)

Installing pgvector:

# Ubuntu/Debian
sudo apt-get install postgresql-16-pgvector

# macOS with Homebrew
brew install pgvector

# Or using CREATE EXTENSION in PostgreSQL:
psql -d your_database -c "CREATE EXTENSION IF NOT EXISTS vector;"

Verifying pgvector installation:

SELECT * FROM pg_extension WHERE extname = 'vector';
-- Should return one row if installed correctly

API Overview

This skill documents core llmemory operations:

• LLMemory - Main interface class
• DocumentType - Enum for document types
• SearchType - Enum for search modes
• ChunkingStrategy - Enum for chunking strategies
• add_document() - Add and process documents
• search() - Search for documents
• search_with_routing() - Search with automatic query routing (detects answerable queries)
• search_with_documents() - Search and return results with document metadata
• list_documents() - List documents with pagination
• get_document() - Retrieve a document (owner-scoped)
• get_document_chunks() - Get chunks with pagination (owner-scoped)
• get_chunk_count() - Get number of chunks for a document (owner-scoped)
• delete_document() / delete_documents() - Delete documents (owner-scoped)
• get_statistics() - Get owner statistics
• db_manager - Access underlying database manager
• initialize() / close() - Lifecycle management

Quick Start

import asyncio
from llmemory import LLMemory, DocumentType, SearchType

async def main():
    # Initialize
    memory = LLMemory(
        connection_string="postgresql://localhost/mydb",
        openai_api_key="sk-..."
    )
    await memory.initialize()

    # Add a document
    result = await memory.add_document(
        owner_id="workspace-1",
        id_at_origin="user-123",
        document_name="example.txt",
        document_type=DocumentType.TEXT,
        content="Your document content here...",
        metadata={"category": "example"}
    )
    print(f"Created document with {result.chunks_created} chunks")

    # Search
    results = await memory.search(
        owner_id="workspace-1",
        query_text="your search query",
        search_type=SearchType.HYBRID,
        limit=5
    )
    for result in results:
        print(f"[{result.score:.3f}] {result.content[:80]}...")

    # Clean up
    await memory.close()

asyncio.run(main())

Complete API Documentation

LLMemory

Main interface for document operations.

Constructor:

LLMemory(
    connection_string: Optional[str] = None,
    openai_api_key: Optional[str] = None,
    config: Optional[LLMemoryConfig] = None,
    db_manager: Optional[AsyncDatabaseManager] = None
)

Parameters:

• connection_string (str, optional): PostgreSQL connection URL (format: postgresql://user:pass@host:port/database). Ignored if db_manager provided.
• openai_api_key (str, optional): OpenAI API key for embeddings. Can also be set via OPENAI_API_KEY environment variable.
• config (LLMemoryConfig, optional): Configuration object. Defaults to config from environment if not provided.
• db_manager (AsyncDatabaseManager, optional): Existing database manager from shared pool (for production apps with multiple services).

Raises:

• ConfigurationError: If neither connection_string nor db_manager provided, or if configuration is invalid.

Example:

from llmemory import LLMemory

# Simple initialization
memory = LLMemory(
    connection_string="postgresql://localhost/mydb",
    openai_api_key="sk-..."
)
await memory.initialize()

LLMemory.from_db_manager()

Create instance from existing AsyncDatabaseManager (shared pool pattern).

Signature:

@classmethod
def from_db_manager(
    cls,
    db_manager: AsyncDatabaseManager,
    openai_api_key: Optional[str] = None,
    config: Optional[LLMemoryConfig] = None
) -> LLMemory

Parameters:

• db_manager (AsyncDatabaseManager, required): Existing database manager with schema already set
• openai_api_key (str, optional): OpenAI API key
• config (LLMemoryConfig, optional): Configuration object

Returns:

• LLMemory: Configured instance

Example:

from pgdbm import AsyncDatabaseManager, DatabaseConfig
from llmemory import LLMemory

# Create shared pool
config = DatabaseConfig(connection_string="postgresql://localhost/mydb")
shared_pool = await AsyncDatabaseManager.create_shared_pool(config)

# Create llmemory with shared pool
db_manager = AsyncDatabaseManager(pool=shared_pool, schema="llmemory")
memory = LLMemory.from_db_manager(
    db_manager,
    openai_api_key="sk-..."
)
await memory.initialize()

db_manager

Get the underlying database manager for health checks and monitoring.

Property:

@property
def db_manager(self) -> Optional[AsyncDatabaseManager]

Returns:

• Optional[AsyncDatabaseManager]: Database manager instance if initialized, None otherwise

Example:

from llmemory import LLMemory

memory = LLMemory(connection_string="postgresql://localhost/mydb")
await memory.initialize()

# Access underlying database manager
db_mgr = memory.db_manager
if db_mgr:
    # Check connection pool status
    pool_status = await db_mgr.get_pool_status()
    print(f"Active connections: {pool_status['active']}")
    print(f"Idle connections: {pool_status['idle']}")

    # Run health check
    is_healthy = await db_mgr.health_check()
    print(f"Database healthy: {is_healthy}")

When to use:

• Health monitoring and observability
• Accessing connection pool metrics
• Database diagnostics
• Integration with monitoring systems

initialize()

Initialize the library and database schema.

Signature:

async def initialize() -> None

Raises:

• DatabaseError: If database initialization fails
• ConfigurationError: If configuration is invalid

Example:

memory = LLMemory(connection_string="postgresql://localhost/mydb")
await memory.initialize()  # Sets up tables, migrations, indexes

close()

Close all connections and cleanup resources.

Signature:

async def close() -> None

Example:

await memory.close()

Context Manager Pattern (Recommended):

async with LLMemory(connection_string="...") as memory:
    # Use memory here
    results = await memory.search(...)
# Automatically closed

Document Types

class DocumentType(str, Enum):
    PDF = "pdf"
    MARKDOWN = "markdown"
    CODE = "code"
    TEXT = "text"
    HTML = "html"
    DOCX = "docx"
    EMAIL = "email"
    REPORT = "report"
    CHAT = "chat"
    PRESENTATION = "presentation"
    LEGAL_DOCUMENT = "legal_document"
    TECHNICAL_DOC = "technical_doc"
    BUSINESS_REPORT = "business_report"
    UNKNOWN = "unknown"

Search Types

class SearchType(str, Enum):
    VECTOR = "vector"     # Vector similarity search only
    TEXT = "text"         # Full-text search only
    HYBRID = "hybrid"     # Combines vector + text (recommended)

Chunking Strategies

class ChunkingStrategy(str, Enum):
    HIERARCHICAL = "hierarchical"      # Default - Creates parent and child chunks for better context
    FIXED_SIZE = "fixed_size"          # Fixed-size chunks with overlap
    SEMANTIC = "semantic"              # Chunks based on semantic boundaries (slower, higher quality)
    SLIDING_WINDOW = "sliding_window"  # Sliding window with configurable overlap

Strategy descriptions:

• HIERARCHICAL (default): Creates hierarchical parent and child chunks. Parent chunks provide broader context while child chunks are used for precise retrieval. Best for most use cases.
• FIXED_SIZE: Creates fixed-size chunks with configurable overlap. Simple and fast, good for uniform documents.
• SEMANTIC: Chunks based on semantic boundaries (paragraphs, sections). Slower but produces higher quality chunks that respect document structure.
• SLIDING_WINDOW: Creates overlapping chunks using a sliding window approach. Good for ensuring no information is lost at chunk boundaries.

Usage:

from llmemory import ChunkingStrategy

# Use enum value
result = await memory.add_document(
    owner_id="workspace-1",
    id_at_origin="user-123",
    document_name="example.txt",
    document_type=DocumentType.TEXT,
    content="Your document content...",
    chunking_strategy=ChunkingStrategy.SEMANTIC  # Use enum
)

# Or use string value (also valid)
result = await memory.add_document(
    owner_id="workspace-1",
    id_at_origin="user-123",
    document_name="example.txt",
    document_type=DocumentType.TEXT,
    content="Your document content...",
    chunking_strategy="hierarchical"  # String also works
)

Model Classes

SearchResult

Search result from any search operation.

Fields:

• chunk_id (UUID): Chunk identifier
• document_id (UUID): Document identifier
• content (str): Chunk content
• metadata (Dict[str, Any]): Chunk metadata
• score (float): Overall relevance score
• similarity (float, optional): Vector similarity score (0-1)
• text_rank (float, optional): Full-text search rank
• rrf_score (float, optional): Reciprocal Rank Fusion score
• rerank_score (float, optional): Reranker score (when reranking enabled)
• summary (str, optional): Chunk summary if generated
• parent_chunks (List[DocumentChunk]): Surrounding chunks if requested

EnrichedSearchResult

Extended search result with document metadata (inherits from SearchResult).

Additional Fields:

• document_name (str): Name of the source document
• document_type (str): Type of document
• document_metadata (Dict[str, Any]): Document-level metadata

When used: Returned by search_with_documents()

SearchResultWithDocuments

Container for enriched search results.

Fields:

• results (List[EnrichedSearchResult]): Enriched search results
• total (int): Total number of results

DocumentAddResult

Result of adding a document.

Fields:

• document (Document): Created document object with all fields
• chunks_created (int): Number of chunks created
• embeddings_created (int): Number of embeddings generated
• processing_time_ms (float): Processing time in milliseconds

DocumentListResult

Result of listing documents with pagination.

Fields:

• documents (List[Document]): Document objects
• total (int): Total matching documents (before pagination)
• limit (int): Applied limit
• offset (int): Applied offset

DocumentWithChunks

Document with optional chunks.

Fields:

• document (Document): Document object
• chunks (Optional[List[DocumentChunk]]): Chunks if requested
• chunk_count (int): Total number of chunks

OwnerStatistics

Statistics for an owner's documents.

Fields:

• document_count (int): Total documents
• chunk_count (int): Total chunks
• total_size_bytes (int): Estimated total size
• document_type_breakdown (Optional[Dict[DocumentType, int]]): Count by document type
• created_date_range (Optional[Tuple[datetime, datetime]]): (min_date, max_date) of document creation

DeleteResult

Result of batch delete operation.

Fields:

• deleted_count (int): Number of documents deleted
• deleted_document_ids (List[UUID]): IDs of deleted documents

EmbeddingStatus

Enum for embedding generation status.

class EmbeddingStatus(str, Enum):
    PENDING = "pending"        # Job queued but not started
    PROCESSING = "processing"  # Currently generating embeddings
    COMPLETED = "completed"    # Successfully completed
    FAILED = "failed"          # Failed with error

EmbeddingJob

Represents a background embedding generation job.

Fields:

• chunk_id (UUID): Chunk being processed
• provider_id (str): Embedding provider ID
• status (EmbeddingStatus): Current status
• retry_count (int): Number of retries attempted
• error_message (Optional[str]): Error details if failed
• created_at (datetime): When job was created
• processed_at (Optional[datetime]): When processing finished

SearchQuery

Internal search query model (rarely used directly).

Fields:

• owner_id (str): Owner identifier
• query_text (str): Search query text
• search_type (SearchType): Type of search
• limit (int): Maximum results
• alpha (float): Hybrid search weight
• metadata_filter (Optional[Dict[str, Any]]): Metadata filter
• id_at_origin (Optional[str]): Single origin filter
• id_at_origins (Optional[List[str]]): Multiple origins filter
• date_from (Optional[datetime]): Start date
• date_to (Optional[datetime]): End date
• include_parent_context (bool): Include parent chunks
• context_window (int): Number of parent chunks
• rerank (bool): Enable reranking
• enable_query_expansion (bool): Enable query expansion
• max_query_variants (int): Max query variants

add_document()

Add a document and process it into searchable chunks.

Signature:

async def add_document(
    owner_id: str,
    id_at_origin: str,
    document_name: str,
    document_type: Union[DocumentType, str],
    content: str,
    document_date: Optional[datetime] = None,
    metadata: Optional[Dict[str, Any]] = None,
    chunking_strategy: str = "hierarchical",
    chunking_config: Optional[ChunkingConfig] = None,
    generate_embeddings: bool = True
) -> DocumentAddResult

Parameters:

• owner_id (str, required): Owner identifier for multi-tenancy (e.g., "workspace-123", "tenant-abc")
• id_at_origin (str, required): Origin identifier within owner (e.g., "user-456", "thread-789")
• document_name (str, required): Name of the document
• document_type (DocumentType or str, required): Type of document
• content (str, required): Full document content
• document_date (datetime, optional): Document date for temporal filtering
• metadata (Dict[str, Any], optional): Custom metadata (searchable via metadata_filter)
• chunking_strategy (str, default: "hierarchical"): Chunking strategy to use
• chunking_config (ChunkingConfig, optional): Custom chunking configuration
• generate_embeddings (bool, default: True): Generate embeddings immediately

Returns:

• DocumentAddResult with:
  • document (Document): Created document object
  • chunks_created (int): Number of chunks created
  • embeddings_created (int): Number of embeddings generated
  • processing_time_ms (float): Processing time in milliseconds

Raises:

• ValidationError: If input validation fails (invalid owner_id, empty content, etc.)
• DatabaseError: If database operation fails
• EmbeddingError: If embedding generation fails

Example:

from llmemory import DocumentType
from datetime import datetime

result = await memory.add_document(
    owner_id="workspace-1",
    id_at_origin="user-123",
    document_name="Q4 Report.pdf",
    document_type=DocumentType.PDF,
    content="Full document text here...",
    document_date=datetime(2024, 10, 1),
    metadata={
        "category": "financial",
        "department": "finance",
        "confidential": False
    }
)

print(f"Document ID: {result.document.document_id}")
print(f"Chunks: {result.chunks_created}")
print(f"Embeddings: {result.embeddings_created}")
print(f"Time: {result.processing_time_ms:.2f}ms")

search()

Search for documents.

Signature:

async def search(
    owner_id: str,
    query_text: str,
    search_type: Union[SearchType, str] = SearchType.HYBRID,
    limit: int = 10,
    id_at_origin: Optional[str] = None,
    id_at_origins: Optional[List[str]] = None,
    metadata_filter: Optional[Dict[str, Any]] = None,
    date_from: Optional[datetime] = None,
    date_to: Optional[datetime] = None,
    include_parent_context: bool = False,
    context_window: int = 2,
    alpha: float = 0.5,
    query_expansion: Optional[bool] = None,
    max_query_variants: Optional[int] = None,
    rerank: Optional[bool] = None,
    rerank_top_k: Optional[int] = None,
    rerank_return_k: Optional[int] = None
) -> List[SearchResult]

Parameters:

• owner_id (str, required): Owner identifier for filtering
• query_text (str, required): Search query text
• search_type (SearchType or str, default: HYBRID): Type of search to perform
• limit (int, default: 10): Maximum number of results
• id_at_origin (str, optional): Filter by single origin ID
• id_at_origins (List[str], optional): Filter by multiple origin IDs
• metadata_filter (Dict[str, Any], optional): Filter by metadata (e.g., {"category": "financial"})
• date_from (datetime, optional): Start date filter
• date_to (datetime, optional): End date filter
• include_parent_context (bool, default: False): Include surrounding chunks
• context_window (int, default: 2): Number of surrounding chunks to include
• alpha (float, default: 0.5): Hybrid search weight (0=text only, 1=vector only)
• query_expansion (bool, optional): Enable query expansion (None = follow config)
• max_query_variants (int, optional): Max query variants for expansion
• rerank (bool, optional): Enable reranking (None = follow config)
• rerank_top_k (int, optional): Candidates for reranker
• rerank_return_k (int, optional): Results after reranking

Returns:

• List[SearchResult] where each result has:
  • chunk_id (UUID): Chunk identifier
  • document_id (UUID): Document identifier
  • content (str): Chunk content
  • metadata (Dict[str, Any]): Chunk metadata
  • score (float): Overall relevance score
  • similarity (float, optional): Vector similarity score
  • text_rank (float, optional): Text search rank
  • rrf_score (float, optional): Reciprocal Rank Fusion score
  • rerank_score (float, optional): Reranker score (when reranking enabled)
  • summary (str, optional): Chunk summary if available
  • parent_chunks (List[DocumentChunk]): Surrounding chunks if requested

Raises:

• ValidationError: If input validation fails
• SearchError: If search operation fails

Example:

from llmemory import SearchType

# Basic search
results = await memory.search(
    owner_id="workspace-1",
    query_text="quarterly revenue trends",
    search_type=SearchType.HYBRID,
    limit=5
)

for result in results:
    print(f"Score: {result.score:.3f}")
    print(f"Content: {result.content[:100]}...")
    print(f"Metadata: {result.metadata}")
    print("---")

# Advanced search with filters
results = await memory.search(
    owner_id="workspace-1",
    query_text="product launch strategy",
    search_type=SearchType.HYBRID,
    limit=10,
    metadata_filter={"category": "strategy", "department": "product"},
    date_from=datetime(2024, 1, 1),
    date_to=datetime(2024, 12, 31),
    alpha=0.7  # Favor vector search slightly
)

search_with_documents()

Search and return results enriched with document metadata.

Signature:

async def search_with_documents(
    owner_id: str,
    query_text: str,
    search_type: Union[SearchType, str] = SearchType.HYBRID,
    limit: int = 10,
    metadata_filter: Optional[Dict[str, Any]] = None,
    include_document_metadata: bool = True
) -> SearchResultWithDocuments

Parameters:

• owner_id (str, required): Owner identifier
• query_text (str, required): Search query text
• search_type (SearchType or str, default: HYBRID): Type of search
• limit (int, default: 10): Maximum results
• metadata_filter (Dict[str, Any], optional): Filter by metadata
• include_document_metadata (bool, default: True): Include document-level metadata

Returns:

• SearchResultWithDocuments with:
  • results (List[EnrichedSearchResult]): Enriched search results
  • total (int): Total number of results

EnrichedSearchResult fields:

• All fields from SearchResult (chunk_id, content, score, etc.)
• document_name (str): Name of the source document
• document_type (str): Type of document
• document_metadata (Dict[str, Any]): Document-level metadata

Raises:

• ValidationError: If input validation fails
• SearchError: If search operation fails

Example:

# Search with document context
results_with_docs = await memory.search_with_documents(
    owner_id="workspace-1",
    query_text="quarterly financial performance",
    search_type=SearchType.HYBRID,
    limit=10
)

print(f"Found {results_with_docs.total} results")

for result in results_with_docs.results:
    print(f"Document: {result.document_name}")
    print(f"Type: {result.document_type}")
    print(f"Score: {result.score:.3f}")
    print(f"Content: {result.content[:100]}...")
    print(f"Metadata: {result.document_metadata}")
    print("---")

When to use:

• When you need document context along with search results
• Building UI that shows source documents
• Grouping results by document
• When document metadata is needed for filtering or display

list_documents()

List documents with pagination and filtering.

Signature:

async def list_documents(
    owner_id: str,
    limit: int = 20,
    offset: int = 0,
    document_type: Optional[DocumentType] = None,
    order_by: Literal["created_at", "updated_at", "document_name"] = "created_at",
    order_desc: bool = True,
    metadata_filter: Optional[Dict[str, Any]] = None
) -> DocumentListResult

Parameters:

• owner_id (str, required): Owner identifier
• limit (int, default: 20): Maximum documents to return
• offset (int, default: 0): Number of documents to skip (for pagination)
• document_type (DocumentType, optional): Filter by document type
• order_by (str, default: "created_at"): Field to sort by
• order_desc (bool, default: True): Sort descending
• metadata_filter (Dict[str, Any], optional): Filter by metadata

Returns:

• DocumentListResult with:
  • documents (List[Document]): Document objects
  • total (int): Total matching documents
  • limit (int): Applied limit
  • offset (int): Applied offset

Raises:

• ValidationError: If parameters are invalid

Example:

# List recent documents
result = await memory.list_documents(
    owner_id="workspace-1",
    limit=20,
    offset=0,
    order_by="created_at",
    order_desc=True
)

print(f"Total documents: {result.total}")
for doc in result.documents:
    print(f"{doc.document_name} - {doc.document_type.value}")

# Filter by type and metadata
result = await memory.list_documents(
    owner_id="workspace-1",
    document_type=DocumentType.PDF,
    metadata_filter={"category": "financial"},
    limit=50
)

get_document()

Retrieve a specific document with optional chunks.

Signature:

async def get_document(
    owner_id: str,
    document_id: Union[str, UUID],
    include_chunks: bool = False,
    include_embeddings: bool = False
) -> DocumentWithChunks

Parameters:

• owner_id (str, required): Owner/workspace identifier (required for access control)
• document_id (str or UUID, required): Document identifier
• include_chunks (bool, default: False): Include all chunks for this document
• include_embeddings (bool, default: False): Include embeddings with chunks (requires include_chunks=True)

Returns:

• DocumentWithChunks with:
  • document (Document): Document object
  • chunks (List[DocumentChunk], optional): Chunks if requested
  • chunk_count (int): Total number of chunks

**Rai

<!-- Content truncated for initial SEO render. Open the source file tab for the full file. -->