Skip to content

graph_retriever.adapters

Adapter 

Adapter()

Bases: ABC

Base adapter for integrating vector stores with the graph retriever system.

This class provides a foundation for custom adapters, enabling consistent interaction with various vector store implementations.

Source code in packages/graph-retriever/src/graph_retriever/adapters/base.py 
24
25
def __init__(self) -> None:
    pass

aadjacent async 

aadjacent(
    edges: set[Edge],
    query_embedding: list[float],
    k: int,
    filter: dict[str, Any] | None,
    **kwargs: Any,
) -> Iterable[Content]

Asynchronously return the content items with at least one matching edge.

PARAMETER DESCRIPTION
edges

The edges to look for.

TYPE: set[Edge]

query_embedding

The query embedding used for selecting the most relevant content.

TYPE: list[float]

k

The number of relevant content items to select for the edges.

TYPE: int

filter

Optional metadata to filter the results.

TYPE: dict[str, Any] | None

kwargs

Keyword arguments to pass to the similarity search.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
Iterable[Content]

Iterable of adjacent content items.
RAISES DESCRIPTION
ValueError

If unsupported edge types are encountered.
Source code in packages/graph-retriever/src/graph_retriever/adapters/base.py 
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
async def aadjacent(
    self,
    edges: set[Edge],
    query_embedding: list[float],
    k: int,
    filter: dict[str, Any] | None,
    **kwargs: Any,
) -> Iterable[Content]:
    """
    Asynchronously return the content items with at least one matching edge.

    Parameters
    ----------
    edges :
        The edges to look for.
    query_embedding :
        The query embedding used for selecting the most relevant content.
    k :
        The number of relevant content items to select for the edges.
    filter :
        Optional metadata to filter the results.
    kwargs :
        Keyword arguments to pass to the similarity search.

    Returns
    -------
    :
        Iterable of adjacent content items.

    Raises
    ------
    ValueError
        If unsupported edge types are encountered.
    """
    tasks = []
    ids = []
    for edge in edges:
        if isinstance(edge, MetadataEdge):
            tasks.append(
                self.asearch(
                    embedding=query_embedding,
                    k=k,
                    filter=self._metadata_filter(base_filter=filter, edge=edge),
                    **kwargs,
                )
            )
        elif isinstance(edge, IdEdge):
            ids.append(edge.id)
        else:
            raise ValueError(f"Unsupported edge: {edge}")

    if ids:
        tasks.append(self.aget(ids, filter))

    results: list[Content] = [
        c
        for completed_task in asyncio.as_completed(tasks)
        for c in await completed_task
    ]

    return top_k(
        results,
        embedding=query_embedding,
        k=k,
    )

adjacent 

adjacent(
    edges: set[Edge],
    query_embedding: list[float],
    k: int,
    filter: dict[str, Any] | None,
    **kwargs: Any,
) -> Iterable[Content]

Return the content items with at least one matching incoming edge.

PARAMETER DESCRIPTION
edges

The edges to look for.

TYPE: set[Edge]

query_embedding

The query embedding used for selecting the most relevant content.

TYPE: list[float]

k

The number of relevant content items to select.

TYPE: int

filter

Optional metadata to filter the results.

TYPE: dict[str, Any] | None

kwargs

Keyword arguments to pass to the similarity search.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
Iterable[Content]

Iterable of adjacent content items.
RAISES DESCRIPTION
ValueError

If unsupported edge types are encountered.
Source code in packages/graph-retriever/src/graph_retriever/adapters/base.py 
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def adjacent(
    self,
    edges: set[Edge],
    query_embedding: list[float],
    k: int,
    filter: dict[str, Any] | None,
    **kwargs: Any,
) -> Iterable[Content]:
    """
    Return the content items with at least one matching incoming edge.

    Parameters
    ----------
    edges :
        The edges to look for.
    query_embedding :
        The query embedding used for selecting the most relevant content.
    k :
        The number of relevant content items to select.
    filter :
        Optional metadata to filter the results.
    kwargs :
        Keyword arguments to pass to the similarity search.

    Returns
    -------
    :
        Iterable of adjacent content items.

    Raises
    ------
    ValueError
        If unsupported edge types are encountered.
    """
    results: list[Content] = []

    ids = []
    for edge in edges:
        if isinstance(edge, MetadataEdge):
            docs = self.search(
                embedding=query_embedding,
                k=k,
                filter=self._metadata_filter(base_filter=filter, edge=edge),
                **kwargs,
            )
            results.extend(docs)
        elif isinstance(edge, IdEdge):
            ids.append(edge.id)
        else:
            raise ValueError(f"Unsupported edge: {edge}")

    if ids:
        results.extend(self.get(ids, filter=filter))

    return top_k(
        results,
        embedding=query_embedding,
        k=k,
    )

aget async 

aget(
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]

Asynchronously get content items by ID.

Fewer content items may be returned than requested if some IDs are not found or if there are duplicated IDs. This method should NOT raise exceptions if no content items are found for some IDs.

Users should not assume that the order of the returned content items matches the order of the input IDs. Instead, users should rely on the ID field of the returned content items.

PARAMETER DESCRIPTION
ids

List of IDs to get.

TYPE: Sequence[str]

filter

Filter on the metadata to apply.

TYPE: dict[str, Any] | None DEFAULT: None

kwargs

Additional keyword arguments. These are up to the implementation.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
list[Content]

List of content items that were found.
Source code in packages/graph-retriever/src/graph_retriever/adapters/base.py 
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
async def aget(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Asynchronously get content items by ID.

    Fewer content items may be returned than requested if some IDs are
    not found or if there are duplicated IDs. This method should **NOT**
    raise exceptions if no content items are found for some IDs.

    Users should not assume that the order of the returned content items
    matches  the order of the input IDs. Instead, users should rely on
    the ID field of the returned content items.

    Parameters
    ----------
    ids :
        List of IDs to get.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments. These are up to the implementation.

    Returns
    -------
    :
        List of content items that were found.
    """
    return await run_in_executor(
        None,
        self.get,
        ids,
        filter,
        **kwargs,
    )

asearch async 

asearch(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]

Asynchronously return content items most similar to the query vector.

PARAMETER DESCRIPTION
embedding

The query embedding used for selecting the most relevant content.

TYPE: list[float]

k

Number of content items to return.

TYPE: int DEFAULT: 4

filter

Filter on the metadata to apply.

TYPE: dict[str, Any] | None DEFAULT: None

kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
list[Content]

List of content items most similar to the query vector.
Source code in packages/graph-retriever/src/graph_retriever/adapters/base.py 
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
async def asearch(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Asynchronously return content items most similar to the query vector.

    Parameters
    ----------
    embedding :
        The query embedding used for selecting the most relevant content.
    k :
        Number of content items to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    :
        List of content items most similar to the query vector.
    """
    return await run_in_executor(
        None,
        self.search,
        embedding,
        k,
        filter,
        **kwargs,
    )

asearch_with_embedding async 

asearch_with_embedding(
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]

Asynchronously return content items most similar to the query.

Also returns the embedded query vector.

PARAMETER DESCRIPTION
query

Input text.

TYPE: str

k

Number of content items to return.

TYPE: int DEFAULT: 4

filter

Filter on the metadata to apply.

TYPE: dict[str, Any] | None DEFAULT: None

kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
query_embedding

The query embedding used for selecting the most relevant content.

TYPE: list[float]

contents

List of up to k content items most similar to the query vector.

TYPE: list[Content]

Source code in packages/graph-retriever/src/graph_retriever/adapters/base.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
async def asearch_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    """
    Asynchronously return content items most similar to the query.

    Also returns the embedded query vector.

    Parameters
    ----------
    query :
        Input text.
    k :
        Number of content items to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    query_embedding :
        The query embedding used for selecting the most relevant content.
    contents :
        List of up to `k` content items most similar to the query
        vector.
    """
    return await run_in_executor(
        None, self.search_with_embedding, query, k, filter, **kwargs
    )

get abstractmethod 

get(
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]

Get content items by ID.

Fewer content items may be returned than requested if some IDs are not found or if there are duplicated IDs. This method should NOT raise exceptions if no content items are found for some IDs.

Users should not assume that the order of the returned content items matches the order of the input IDs. Instead, users should rely on the ID field of the returned content items.

PARAMETER DESCRIPTION
ids

List of IDs to get.

TYPE: Sequence[str]

filter

Filter on the metadata to apply.

TYPE: dict[str, Any] | None DEFAULT: None

kwargs

Additional keyword arguments. These are up to the implementation.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
list[Content]

List of content items that were found.
Source code in packages/graph-retriever/src/graph_retriever/adapters/base.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
@abc.abstractmethod
def get(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Get content items by ID.

    Fewer content items may be returned than requested if some IDs are
    not found or if there are duplicated IDs. This method should **NOT**
    raise exceptions if no content items are found for some IDs.

    Users should not assume that the order of the returned content items
    matches  the order of the input IDs. Instead, users should rely on
    the ID field of the returned content items.

    Parameters
    ----------
    ids :
        List of IDs to get.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments. These are up to the implementation.

    Returns
    -------
    :
        List of content items that were found.
    """
    ...

search abstractmethod 

search(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]

Return content items most similar to the query vector.

PARAMETER DESCRIPTION
embedding

The query embedding used for selecting the most relevant content.

TYPE: list[float]

k

Number of content items to return.

TYPE: int DEFAULT: 4

filter

Filter on the metadata to apply.

TYPE: dict[str, Any] | None DEFAULT: None

kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
list[Content]

List of content items most similar to the query vector.
Source code in packages/graph-retriever/src/graph_retriever/adapters/base.py 
 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
@abc.abstractmethod
def search(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Return content items most similar to the query vector.

    Parameters
    ----------
    embedding :
        The query embedding used for selecting the most relevant content.
    k :
        Number of content items to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    :
        List of content items most similar to the query vector.
    """
    ...

search_with_embedding abstractmethod 

search_with_embedding(
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]

Return content items most similar to the query.

Also returns the embedded query vector.

PARAMETER DESCRIPTION
query

Input text.

TYPE: str

k

Number of content items to return.

TYPE: int DEFAULT: 4

filter

Filter on the metadata to apply.

TYPE: dict[str, Any] | None DEFAULT: None

kwargs

Additional keyword arguments.

TYPE: Any DEFAULT: {}

RETURNS DESCRIPTION
query_embedding

The query embedding used for selecting the most relevant content.

TYPE: list[float]

contents

List of up to k content items most similar to the query vector.

TYPE: list[Content]

Source code in packages/graph-retriever/src/graph_retriever/adapters/base.py 
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
58
@abc.abstractmethod
def search_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    """
    Return content items most similar to the query.

    Also returns the embedded query vector.

    Parameters
    ----------
    query :
        Input text.
    k :
        Number of content items to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    query_embedding :
        The query embedding used for selecting the most relevant content.
    contents :
        List of up to `k` content items most similar to the query vector.
    """
    ...