Skip to content

neurotrace.core.vector_memory

neurotrace.core.vector_memory

BaseVectorMemoryAdapter

Bases: ABC

Abstract base class for vector memory storage adapters.

This class defines the interface for storing and retrieving messages using vector embeddings. Implementations should provide concrete methods for adding, searching, and optionally deleting messages from the vector store.

Source code in neurotrace/core/vector_memory.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
class BaseVectorMemoryAdapter(ABC):
    """Abstract base class for vector memory storage adapters.

    This class defines the interface for storing and retrieving messages using
    vector embeddings. Implementations should provide concrete methods for
    adding, searching, and optionally deleting messages from the vector store.
    """

    @abstractmethod
    def add_messages(self, messages: List[Message]) -> None:
        """Add a list of messages to the vector memory store.

        Args:
            messages (List[Message]): List of Message objects to be added to
                the vector store. Each message will be embedded and stored.
        """
        pass

    @abstractmethod
    def search(self, query: str, k: int = 5) -> List[Message]:
        """Search the vector memory for the most relevant messages.

        Performs a similarity search in the vector store using the provided
        query string. The query will be embedded and compared against stored
        message embeddings.

        Args:
            query (str): The search query string.
            k (int, optional): Maximum number of results to return. Defaults to 5.

        Returns:
            List[Message]: List of messages ranked by similarity to the query,
                limited to k results.
        """
        pass

    @abstractmethod
    def delete(self, ids: List[str]) -> None:
        """Delete messages from the vector store by their IDs.

        Args:
            ids (List[str]): List of message IDs to be deleted from the store.

        Note:
            This is an optional operation that might not be supported by all
            vector store implementations.
        """
        pass

add_messages(messages) abstractmethod

Add a list of messages to the vector memory store.

Parameters:

Name Type Description Default
messages List[Message]

List of Message objects to be added to the vector store. Each message will be embedded and stored.

 required
Source code in neurotrace/core/vector_memory.py 
18
19
20
21
22
23
24
25
26
@abstractmethod
def add_messages(self, messages: List[Message]) -> None:
    """Add a list of messages to the vector memory store.

    Args:
        messages (List[Message]): List of Message objects to be added to
            the vector store. Each message will be embedded and stored.
    """
    pass

delete(ids) abstractmethod

Delete messages from the vector store by their IDs.

Parameters:

Name Type Description Default
ids List[str]

List of message IDs to be deleted from the store.

 required
Note

This is an optional operation that might not be supported by all vector store implementations.

Source code in neurotrace/core/vector_memory.py 
46
47
48
49
50
51
52
53
54
55
56
57
@abstractmethod
def delete(self, ids: List[str]) -> None:
    """Delete messages from the vector store by their IDs.

    Args:
        ids (List[str]): List of message IDs to be deleted from the store.

    Note:
        This is an optional operation that might not be supported by all
        vector store implementations.
    """
    pass

search(query, k=5) abstractmethod

Search the vector memory for the most relevant messages.

Performs a similarity search in the vector store using the provided query string. The query will be embedded and compared against stored message embeddings.

Parameters:

Name Type Description Default
query str

The search query string.

 required
k int

Maximum number of results to return. Defaults to 5.

 5

Returns:

Type Description
List[Message]

List[Message]: List of messages ranked by similarity to the query, limited to k results.
Source code in neurotrace/core/vector_memory.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
@abstractmethod
def search(self, query: str, k: int = 5) -> List[Message]:
    """Search the vector memory for the most relevant messages.

    Performs a similarity search in the vector store using the provided
    query string. The query will be embedded and compared against stored
    message embeddings.

    Args:
        query (str): The search query string.
        k (int, optional): Maximum number of results to return. Defaults to 5.

    Returns:
        List[Message]: List of messages ranked by similarity to the query,
            limited to k results.
    """
    pass

VectorMemoryAdapter

Bases: BaseVectorMemoryAdapter

Concrete implementation of vector memory storage using LangChain components.

This adapter wraps a LangChain-compatible vector store and embedding model to provide vector-based message storage and retrieval.

Parameters:

Name Type Description Default
vector_store VectorStore

LangChain vector store implementation for storing embeddings.

 required
embedding_model Embeddings

LangChain embeddings model for converting text to vectors.

 required
Source code in neurotrace/core/vector_memory.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class VectorMemoryAdapter(BaseVectorMemoryAdapter):
    """Concrete implementation of vector memory storage using LangChain components.

    This adapter wraps a LangChain-compatible vector store and embedding model
    to provide vector-based message storage and retrieval.

    Args:
        vector_store (VectorStore): LangChain vector store implementation for
            storing embeddings.
        embedding_model (Embeddings): LangChain embeddings model for converting
            text to vectors.
    """

    def __init__(self, vector_store: VectorStore):
        """
        Vector memory adapter that wraps a LangChain-compatible vector store.

        Args:
            vector_store (VectorStore): Any LangChain-compatible vector store.
            embedding_model (Embeddings): Embedding model to generate embeddings.
        """
        self.vector_store = vector_store
        self.embedding_model = vector_store.embeddings

    def add_messages(self, messages: List[Message]) -> None:
        """Add multiple messages to the vector store.

        Converts the messages to LangChain Document format and adds them to
        the underlying vector store. The documents will be automatically
        embedded using the configured embedding model.

        Args:
            messages (List[Message]): List of messages to be added to the
                vector store.
        """
        documents = [msg.to_document() for msg in messages]
        self.vector_store.add_documents(documents)

    def search(self, query: str, k: int = 5) -> List[Message]:
        """Search for similar messages in the vector store.

        Performs a similarity search using the query string. The query is
        embedded and compared against stored message embeddings to find
        the most similar messages.

        Args:
            query (str): The search query string.
            k (int, optional): Maximum number of results to return. Defaults to 5.

        Returns:
            List[Message]: List of messages ranked by similarity to the query,
                limited to k results.

        Note:
            TODO: Add support for enhancing the prompt for vector search using LLM.
        """
        # todo: add support for enhancing the prompt for vector search using llm
        results = self.vector_store.similarity_search(query=query, k=k)
        return [Message.from_document(doc) for doc in results]

    def delete(self, ids: List[str]) -> None:
        """Delete messages from the vector store by their IDs.

        Args:
            ids (List[str]): List of message IDs to be deleted.

        Raises:
            NotImplementedError: If the underlying vector store doesn't
                support deletion operations.
        """
        if hasattr(self.vector_store, "delete"):
            self.vector_store.delete(ids)
        else:
            raise NotImplementedError(f"Delete not supported by {type(self.vector_store)}.")

__init__(vector_store)

Vector memory adapter that wraps a LangChain-compatible vector store.

Parameters:

Name Type Description Default
vector_store VectorStore

Any LangChain-compatible vector store.

 required
embedding_model Embeddings

Embedding model to generate embeddings.

 required
Source code in neurotrace/core/vector_memory.py 
73
74
75
76
77
78
79
80
81
82
def __init__(self, vector_store: VectorStore):
    """
    Vector memory adapter that wraps a LangChain-compatible vector store.

    Args:
        vector_store (VectorStore): Any LangChain-compatible vector store.
        embedding_model (Embeddings): Embedding model to generate embeddings.
    """
    self.vector_store = vector_store
    self.embedding_model = vector_store.embeddings

add_messages(messages)

Add multiple messages to the vector store.

Converts the messages to LangChain Document format and adds them to the underlying vector store. The documents will be automatically embedded using the configured embedding model.

Parameters:

Name Type Description Default
messages List[Message]

List of messages to be added to the vector store.

 required
Source code in neurotrace/core/vector_memory.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
def add_messages(self, messages: List[Message]) -> None:
    """Add multiple messages to the vector store.

    Converts the messages to LangChain Document format and adds them to
    the underlying vector store. The documents will be automatically
    embedded using the configured embedding model.

    Args:
        messages (List[Message]): List of messages to be added to the
            vector store.
    """
    documents = [msg.to_document() for msg in messages]
    self.vector_store.add_documents(documents)

delete(ids)

Delete messages from the vector store by their IDs.

Parameters:

Name Type Description Default
ids List[str]

List of message IDs to be deleted.

 required

Raises:

Type Description
NotImplementedError

If the underlying vector store doesn't support deletion operations.
Source code in neurotrace/core/vector_memory.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def delete(self, ids: List[str]) -> None:
    """Delete messages from the vector store by their IDs.

    Args:
        ids (List[str]): List of message IDs to be deleted.

    Raises:
        NotImplementedError: If the underlying vector store doesn't
            support deletion operations.
    """
    if hasattr(self.vector_store, "delete"):
        self.vector_store.delete(ids)
    else:
        raise NotImplementedError(f"Delete not supported by {type(self.vector_store)}.")

search(query, k=5)

Search for similar messages in the vector store.

Performs a similarity search using the query string. The query is embedded and compared against stored message embeddings to find the most similar messages.

Parameters:

Name Type Description Default
query str

The search query string.

 required
k int

Maximum number of results to return. Defaults to 5.

 5

Returns:

Type Description
List[Message]

List[Message]: List of messages ranked by similarity to the query, limited to k results.
Note

TODO: Add support for enhancing the prompt for vector search using LLM.

Source code in neurotrace/core/vector_memory.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def search(self, query: str, k: int = 5) -> List[Message]:
    """Search for similar messages in the vector store.

    Performs a similarity search using the query string. The query is
    embedded and compared against stored message embeddings to find
    the most similar messages.

    Args:
        query (str): The search query string.
        k (int, optional): Maximum number of results to return. Defaults to 5.

    Returns:
        List[Message]: List of messages ranked by similarity to the query,
            limited to k results.

    Note:
        TODO: Add support for enhancing the prompt for vector search using LLM.
    """
    # todo: add support for enhancing the prompt for vector search using llm
    results = self.vector_store.similarity_search(query=query, k=k)
    return [Message.from_document(doc) for doc in results]