Memory

NATS KV

KeyValueStore

NATS Key Value Store

Source code in germinate_ai/memory/kv.py

class KeyValueStore:
    """NATS Key Value Store"""

    connected: bool = False
    kv: _KeyValue = None

    def __init__(self, bucket_name: str, connection: NatsConnection = None):
        self.bucket_name = bucket_name
        if connection is None:
            connection = NatsConnection()
        self.connection = connection


    async def connect(self):
        """Connect to NATS cluster."""
        if not self.connection.is_connected:
            await self.connection.connect()
        self.kv = await self.connection.jetstream.create_key_value(bucket=self.bucket_name)
        self.connected = True

    async def get(self, key: str) -> str:
        """Get value corresponding to key."""
        entry =  await self.kv.get(key)
        return entry.value.decode()

    async def put(self, key: str, value: str):
        """Store key, value in store.""" 
        await self.kv.put(key, value.encode())

connect() async

Connect to NATS cluster.

Source code in germinate_ai/memory/kv.py

async def connect(self):
    """Connect to NATS cluster."""
    if not self.connection.is_connected:
        await self.connection.connect()
    self.kv = await self.connection.jetstream.create_key_value(bucket=self.bucket_name)
    self.connected = True

get(key) async

Get value corresponding to key.

Source code in germinate_ai/memory/kv.py

async def get(self, key: str) -> str:
    """Get value corresponding to key."""
    entry =  await self.kv.get(key)
    return entry.value.decode()

put(key, value) async

Store key, value in store.

Source code in germinate_ai/memory/kv.py

async def put(self, key: str, value: str):
    """Store key, value in store.""" 
    await self.kv.put(key, value.encode())

kv_story_factory(*, bucket_name, connection=None)

Creates a KV store.

Source code in germinate_ai/memory/kv.py

def kv_story_factory(*, bucket_name: str, connection: NatsConnection = None):
    """Creates a KV store."""
    kv = KeyValueStore(bucket_name, connection=connection)
    return kv

Weaviate

WeaviateCollection

Adapter for a Weaviate collection.

Source code in germinate_ai/memory/weaviate.py

class WeaviateCollection:
    """Adapter for a Weaviate collection."""

    def __init__(self, collection_name: str, *, host: str, port: str, grpc_port: int):
        self.collection_name = collection_name
        self.host = host
        self.port = port
        self.grpc_port = grpc_port
        self.client = None
        self.collection = None

    def connect(self):
        """Connect to Weaviate."""
        self.client = weaviate.connect_to_local(
            host=self.host,
            port=self.port,
            grpc_port=self.grpc_port,
        )

    def create_collection(self):
        """Create a Weaviate collection."""
        self.collection = self.client.collections.create(
            name=self.collection_name,
            vectorizer_config=wvc.config.Configure.Vectorizer.none(),
            vector_index_config=wvc.config.Configure.VectorIndex.hnsw(
                distance_metric=wvc.config.VectorDistances.COSINE  # select prefered distance metric
            ),
        )

    def query_near_vector(
        self, query_vector: typ.Sequence[float], *, filters=None, limit=10
    ):
        """Search by query vector."""
        resp = self.collection.query.near_vector(
            near_vector=query_vector,
            limit=limit,
            return_metadata=wvc.query.MetadataQuery(certainty=True),
            filters=filters,
        )
        return resp

connect()

Connect to Weaviate.

Source code in germinate_ai/memory/weaviate.py

def connect(self):
    """Connect to Weaviate."""
    self.client = weaviate.connect_to_local(
        host=self.host,
        port=self.port,
        grpc_port=self.grpc_port,
    )

create_collection()

Create a Weaviate collection.

Source code in germinate_ai/memory/weaviate.py

def create_collection(self):
    """Create a Weaviate collection."""
    self.collection = self.client.collections.create(
        name=self.collection_name,
        vectorizer_config=wvc.config.Configure.Vectorizer.none(),
        vector_index_config=wvc.config.Configure.VectorIndex.hnsw(
            distance_metric=wvc.config.VectorDistances.COSINE  # select prefered distance metric
        ),
    )

query_near_vector(query_vector, *, filters=None, limit=10)

Search by query vector.

Source code in germinate_ai/memory/weaviate.py

def query_near_vector(
    self, query_vector: typ.Sequence[float], *, filters=None, limit=10
):
    """Search by query vector."""
    resp = self.collection.query.near_vector(
        near_vector=query_vector,
        limit=limit,
        return_metadata=wvc.query.MetadataQuery(certainty=True),
        filters=filters,
    )
    return resp