🛠️ API Reference

🛠 API Reference

Store Classes

AgentVectorDBStore

Core class for managing vector database operations.

Constructor

AgentVectorDBStore(
    db_path: str = "./vector_db"
) -> AgentVectorDBStore

Parameters:

  • db_path (str): Path to database directory

Returns:

  • AgentVectorDBStore: Store instance

Raises:

  • InitializationError: If database initialization fails
  • PermissionError: If directory access is denied

Methods

get_or_create_collection

Creates a new collection or returns existing one.

def get_or_create_collection(
    name: str,
    embedding_function: Optional[BaseEmbeddingFunction] = None,
    recreate: bool = False,
    schema: Optional[MemoryEntrySchema] = None
) -> AgentMemoryCollection

Parameters:

  • name (str): Unique collection identifier
  • embedding_function (BaseEmbeddingFunction, optional): Function for generating embeddings
  • recreate (bool): If True, deletes existing collection
  • schema (MemoryEntrySchema, optional): Custom schema definition

Returns:

  • AgentMemoryCollection: Collection instance

Raises:

  • SchemaError: If schema validation fails
  • OperationError: If collection creation fails

Example:

ef = DefaultTextEmbeddingFunction(dimension=384)
collection = store.get_or_create_collection(
    name="agent_memories",
    embedding_function=ef
)
list_collections

Returns list of available collections.

def list_collections() -> List[str]

Returns:

  • List[str]: Names of available collections

Example:

collections = store.list_collections()
print(f"Available collections: {collections}")

AsyncAgentVectorDBStore

Asynchronous version of AgentVectorDBStore.

Constructor

AsyncAgentVectorDBStore(
    db_path: str = "./vector_db"
) -> AsyncAgentVectorDBStore

Parameters: Same as AgentVectorDBStore

Methods

All methods are async versions of AgentVectorDBStore methods.

Collection Classes

AgentMemoryCollection

Manages memory entries within a collection.

Methods

add

Adds a single memory entry.

def add(
    content: str,
    type: str = "memory",
    importance_score: float = 0.5,
    metadata: Optional[Dict[str, Any]] = None,
    vector: Optional[np.ndarray] = None
) -> Dict[str, Any]

Parameters:

  • content (str): Text content of memory
  • type (str): Category/type of memory
  • importance_score (float): Value between 0.0 and 1.0
  • metadata (Dict[str, Any], optional): Additional metadata
  • vector (np.ndarray, optional): Pre-computed embedding vector

Returns:

  • Dict[str, Any]: Added memory entry

Raises:

  • SchemaError: If data validation fails
  • EmbeddingError: If vector generation fails

Example:

entry = collection.add(
    content="User preference noted",
    type="user_pref",
    importance_score=0.8,
    metadata={"category": "ui", "user_id": "123"}
)
add_batch

Adds multiple memories efficiently.

def add_batch(
    entries: List[Dict[str, Any]],
    batch_size: int = 100
) -> List[Dict[str, Any]]

Parameters:

  • entries (List[Dict]): List of memory entries
  • batch_size (int): Size of processing batches

Returns:

  • List[Dict]: Added memory entries

Note: Requires minimum 8 entries for KMeans index creation

Example:

memories = [
    {
        "content": "First observation",
        "type": "observation",
        "importance_score": 0.7
    },
    # ... at least 7 more entries
]
results = collection.add_batch(memories)
query

Performs semantic search.

def query(
    query_text: Optional[str] = None,
    query_vector: Optional[np.ndarray] = None,
    k: int = 5,
    filter_sql: Optional[str] = None,
    include_vectors: bool = False,
    select_columns: Optional[List[str]] = None
) -> List[Dict[str, Any]]

Parameters:

  • query_text (str, optional): Text to search for
  • query_vector (np.ndarray, optional): Pre-computed query vector
  • k (int): Number of results to return
  • filter_sql (str, optional): SQL filter expression
  • include_vectors (bool): Include embedding vectors in results
  • select_columns (List[str], optional): Specific columns to return

Returns:

  • List[Dict]: Matching memory entries

Example:

# Basic query
results = collection.query(
    query_text="system performance",
    k=3
)
 
# Advanced query
results = collection.query(
    query_text="errors",
    k=5,
    filter_sql="type = 'error' AND importance_score > 0.8",
    include_vectors=True
)

Embedding Functions

BaseEmbeddingFunction

Abstract base class for embedding functions.

class BaseEmbeddingFunction:
    def __init__(self, dimension: int):
        """
        Args:
            dimension (int): Embedding vector dimension
        """
        self._dimension = dimension
    
    @property
    def dimension(self) -> int:
        """Get embedding dimension"""
        return self._dimension
    
    def embed(self, texts: List[str]) -> np.ndarray:
        """
        Generate embeddings for texts.
        
        Args:
            texts (List[str]): Texts to embed
            
        Returns:
            np.ndarray: Embedding vectors
            
        Raises:
            EmbeddingError: If embedding generation fails
        """
        raise NotImplementedError

DefaultTextEmbeddingFunction

Default embedding implementation.

class DefaultTextEmbeddingFunction(BaseEmbeddingFunction):
    def __init__(
        self,
        dimension: int = 384,
        normalize: bool = True
    ):
        """
        Args:
            dimension (int): Embedding dimension
            normalize (bool): Normalize vectors
        """
        super().__init__(dimension)
        self.normalize = normalize

Schema Management

MemoryEntrySchema

Base schema for memory entries.

class MemoryEntrySchema:
    def __init__(
        self,
        vector_dimension: int,
        additional_fields: Optional[Dict[str, Any]] = None
    ):
        """
        Args:
            vector_dimension (int): Embedding dimension
            additional_fields (Dict): Extra schema fields
        """

Exception Classes

AgentVectorDBException

Base exception class.

class AgentVectorDBException(Exception):
    """Base exception for AgentVectorDB"""
    pass

Specific Exceptions

  • InitializationError: Database initialization failed
  • SchemaError: Schema validation failed
  • QueryError: Invalid query
  • OperationError: Operation failed
  • EmbeddingError: Embedding generation failed

Type Definitions

MemoryEntry = TypedDict('MemoryEntry', {
    'content': str,
    'type': str,
    'importance_score': float,
    'metadata': Optional[Dict[str, Any]],
    'vector': Optional[List[float]]
})
 
QueryResult = TypedDict('QueryResult', {
    'content': str,
    'type': str,
    '_distance': float,
    'metadata': Optional[Dict[str, Any]]
})

Constants

DEFAULT_DIMENSION = 384
MIN_KMEANS_ENTRIES = 8
DEFAULT_BATCH_SIZE = 100

Configuration

Logging Setup

def configure_logging(
    level: str = "INFO",
    log_file: Optional[str] = None,
    format: str = "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
) -> None:
    """
    Configure logging settings.
    
    Args:
        level (str): Log level
        log_file (str, optional): Log file path
        format (str): Log message format
    """

For more examples and detailed usage, check the Examples section.

📖 Guides🧪 Examples