langchain_graph_retriever.adapters

astra

Provides an adapter for AstraDB vector store integration.

AstraAdapter

AstraAdapter(vector_store: AstraDBVectorStore)

Bases: Adapter

Adapter for the AstraDB vector store.

This class integrates the LangChain AstraDB vector store with the graph retriever system, providing functionality for similarity search and document retrieval.

PARAMETER	DESCRIPTION
vector_store	The AstraDB vector store instance. TYPE: AstraDBVectorStore

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/astra.py

def __init__(self, vector_store: AstraDBVectorStore) -> None:
    self.vector_store = vector_store.copy(
        component_name="langchain_graph_retriever"
    )

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/astra.py

@override
async def aadjacent(
    self,
    edges: set[Edge],
    query_embedding: list[float],
    k: int,
    filter: dict[str, Any] | None,
    **kwargs: Any,
) -> Iterable[Content]:
    sort = self.vector_store.document_codec.encode_vector_sort(query_embedding)
    metadata, ids = _extract_queries(edges)
    filters = _queries(
        codec=self.vector_store.document_codec,
        user_filters=filter,
        metadata=metadata,
        ids=ids,
    )

    results = await self._aexecute_and_merge(
        filters=filters,
        sort=sort,
        limit=k,
    )
    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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/astra.py

@override
def adjacent(
    self,
    edges: set[Edge],
    query_embedding: list[float],
    k: int,
    filter: dict[str, Any] | None,
    **kwargs: Any,
) -> Iterable[Content]:
    sort = self.vector_store.document_codec.encode_vector_sort(query_embedding)
    metadata, ids = _extract_queries(edges)
    filters = _queries(
        codec=self.vector_store.document_codec,
        user_filters=filter,
        metadata=metadata,
        ids=ids,
    )

    results = self._execute_and_merge(
        filters=filters,
        sort=sort,
        limit=k,
    )
    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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/astra.py

@override
async def aget(
    self, ids: Sequence[str], filter: dict[str, Any] | None = None, **kwargs: Any
) -> list[Content]:
    return await self._aexecute_and_merge(
        _queries(
            codec=self.vector_store.document_codec,
            user_filters=filter,
            ids=set(ids),
        )
    )

asearch async

asearch(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/astra.py

@override
@backoff.on_exception(backoff.expo, _EXCEPTIONS_TO_RETRY, max_tries=_MAX_RETRIES)
async def asearch(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    if k == 0:
        return []

    docs_with_embeddings = (
        await self.vector_store.asimilarity_search_with_embedding_by_vector(
            embedding=embedding,
            k=k,
            filter=filter,
            **kwargs,
        )
    )
    return self._build_contents(docs_with_embeddings)

asearch_with_embedding async

asearch_with_embedding(
    query: str,
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/astra.py

@override
@backoff.on_exception(backoff.expo, _EXCEPTIONS_TO_RETRY, max_tries=_MAX_RETRIES)
async def asearch_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    # Work around the fact that `k == 0` is rejected by Astra.
    # AstraDBVectorStore has a similar work around for non-vectorize path, but
    # we want it to apply in both cases.
    query_k = k
    if query_k == 0:
        query_k = 1
    (
        query_embedding,
        docs_with_embeddings,
    ) = await self.vector_store.asimilarity_search_with_embedding(
        query=query,
        k=query_k,
        filter=filter,
        **kwargs,
    )
    if k == 0:
        return query_embedding, []
    return query_embedding, self._build_contents(docs_with_embeddings)

get

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/astra.py

@override
def get(
    self, ids: Sequence[str], filter: dict[str, Any] | None = None, **kwargs: Any
) -> list[Content]:
    return self._execute_and_merge(
        _queries(
            codec=self.vector_store.document_codec,
            user_filters=filter,
            ids=set(ids),
        )
    )

search

search(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/astra.py

@override
@backoff.on_exception(backoff.expo, _EXCEPTIONS_TO_RETRY, max_tries=_MAX_RETRIES)
def search(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    if k == 0:
        return []

    docs_with_embeddings = (
        self.vector_store.similarity_search_with_embedding_by_vector(
            embedding=embedding,
            k=k,
            filter=filter,
            **kwargs,
        )
    )
    return self._build_contents(docs_with_embeddings)

search_with_embedding

search_with_embedding(
    query: str,
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/astra.py

@override
@backoff.on_exception(backoff.expo, _EXCEPTIONS_TO_RETRY, max_tries=_MAX_RETRIES)
def search_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    # Work around the fact that `k == 0` is rejected by Astra.
    # AstraDBVectorStore has a similar work around for non-vectorize path, but
    # we want it to apply in both cases.
    query_k = k
    if query_k == 0:
        query_k = 1
    query_embedding, docs_with_embeddings = (
        self.vector_store.similarity_search_with_embedding(
            query=query,
            k=query_k,
            filter=filter,
            **kwargs,
        )
    )
    if k == 0:
        return query_embedding, []

    return query_embedding, self._build_contents(docs_with_embeddings)

cassandra

Provides an adapter for Cassandra vector store integration.

CassandraAdapter

CassandraAdapter(
    vector_store: StoreT,
    shredder: ShreddingTransformer | None = None,
    nested_metadata_fields: set[str] = set(),
)

Bases: ShreddedLangchainAdapter[Cassandra]

Adapter for the Apache Cassandra vector store.

This class integrates the LangChain Cassandra vector store with the graph retriever system, providing functionality for similarity search and document retrieval.

PARAMETER	DESCRIPTION
vector_store	The Cassandra vector store instance. TYPE: StoreT
shredder	An instance of the ShreddingTransformer used for doc insertion. If not passed then a default instance of ShreddingTransformer is used. TYPE: ShreddingTransformer | None DEFAULT: None

Initialize the base adapter.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def __init__(
    self,
    vector_store: StoreT,
    shredder: ShreddingTransformer | None = None,
    nested_metadata_fields: set[str] = set(),
):
    """Initialize the base adapter."""
    super().__init__(vector_store=vector_store)
    self.shredder = ShreddingTransformer() if shredder is None else shredder
    self.nested_metadata_fields = nested_metadata_fields

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

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

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,
    )

aembed_query async

aembed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

async def aembed_query(self, query: str):
    """Return the embedding of the query."""
    return await self._safe_embedding.aembed_query(query)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def aget(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = await self._aget(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

asearch async

asearch(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    if k == 0:
        return []

    docs = await self._asearch(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = await self.aembed_query(query)
    docs = await self.asearch(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

embed_query

embed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def embed_query(self, query: str):
    """Return the embedding of the query."""
    return self._safe_embedding.embed_query(query)

format_documents_hook

format_documents_hook(
    docs: list[Document],
) -> list[Content]

Format the documents as content after executing the query.

PARAMETER	DESCRIPTION
docs	The documents returned from the vector store TYPE: list[Document]

RETURNS	DESCRIPTION
list[Content]	The formatted content.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def format_documents_hook(self, docs: list[Document]) -> list[Content]:
    restored = list(self.shredder.restore_documents(documents=docs))
    return super().format_documents_hook(restored)

get

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def get(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = self._get(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

search

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

Return contents most similar to the query vector.

PARAMETER	DESCRIPTION
embedding	Embedding to look up documents similar to. TYPE: list[float]
k	Number of Documents to return. TYPE: int DEFAULT: 4
filter	Filter on the metadata to apply. TYPE: dict[str, str] | None DEFAULT: None
kwargs	Additional keyword arguments. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
list[Content]	List of Contents most similar to the query vector.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Return contents most similar to the query vector.

    Parameters
    ----------
    embedding :
        Embedding to look up documents similar to.
    k :
        Number of Documents to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    :
        List of Contents most similar to the query vector.
    """
    if k == 0:
        return []

    docs = self._search(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

search_with_embedding

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = self.embed_query(query)
    docs = self.search(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

update_filter_hook

update_filter_hook(
    filter: dict[str, str] | None,
) -> dict[str, str] | None

Update the metadata filter before executing the query.

PARAMETER	DESCRIPTION
filter	Filter on the metadata to update. TYPE: dict[str, Any] | None

RETURNS	DESCRIPTION
dict[str, Any] | None	The updated filter on the metadata to apply.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def update_filter_hook(
    self, filter: dict[str, str] | None
) -> dict[str, str] | None:
    if filter is None:
        return None

    shredded_filter = {}
    for key, value in filter.items():
        if key in self.nested_metadata_fields:
            shredded_filter[self.shredder.shredded_key(key, value)] = (
                self.shredder.shredded_value()
            )
        else:
            shredded_filter[key] = value
    return shredded_filter

chroma

Provides an adapter for Chroma vector store integration.

ChromaAdapter

ChromaAdapter(
    vector_store: StoreT,
    shredder: ShreddingTransformer | None = None,
    nested_metadata_fields: set[str] = set(),
)

Bases: ShreddedLangchainAdapter[Chroma]

Adapter for Chroma vector store.

This adapter integrates the LangChain Chroma vector store with the graph retriever system, allowing for similarity search and document retrieval.

PARAMETER	DESCRIPTION
vector_store	The Chroma vector store instance. TYPE: StoreT
shredder	An instance of the ShreddingTransformer used for doc insertion. If not passed then a default instance of ShreddingTransformer is used. TYPE: ShreddingTransformer | None DEFAULT: None

Initialize the base adapter.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def __init__(
    self,
    vector_store: StoreT,
    shredder: ShreddingTransformer | None = None,
    nested_metadata_fields: set[str] = set(),
):
    """Initialize the base adapter."""
    super().__init__(vector_store=vector_store)
    self.shredder = ShreddingTransformer() if shredder is None else shredder
    self.nested_metadata_fields = nested_metadata_fields

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

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

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,
    )

aembed_query async

aembed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

async def aembed_query(self, query: str):
    """Return the embedding of the query."""
    return await self._safe_embedding.aembed_query(query)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def aget(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = await self._aget(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

asearch async

asearch(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    if k == 0:
        return []

    docs = await self._asearch(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = await self.aembed_query(query)
    docs = await self.asearch(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

embed_query

embed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def embed_query(self, query: str):
    """Return the embedding of the query."""
    return self._safe_embedding.embed_query(query)

format_documents_hook

format_documents_hook(
    docs: list[Document],
) -> list[Content]

Format the documents as content after executing the query.

PARAMETER	DESCRIPTION
docs	The documents returned from the vector store TYPE: list[Document]

RETURNS	DESCRIPTION
list[Content]	The formatted content.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def format_documents_hook(self, docs: list[Document]) -> list[Content]:
    restored = list(self.shredder.restore_documents(documents=docs))
    return super().format_documents_hook(restored)

get

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def get(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = self._get(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

search

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

Return contents most similar to the query vector.

PARAMETER	DESCRIPTION
embedding	Embedding to look up documents similar to. TYPE: list[float]
k	Number of Documents to return. TYPE: int DEFAULT: 4
filter	Filter on the metadata to apply. TYPE: dict[str, str] | None DEFAULT: None
kwargs	Additional keyword arguments. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
list[Content]	List of Contents most similar to the query vector.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Return contents most similar to the query vector.

    Parameters
    ----------
    embedding :
        Embedding to look up documents similar to.
    k :
        Number of Documents to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    :
        List of Contents most similar to the query vector.
    """
    if k == 0:
        return []

    docs = self._search(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

search_with_embedding

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = self.embed_query(query)
    docs = self.search(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

update_filter_hook

update_filter_hook(
    filter: dict[str, Any] | None,
) -> dict[str, Any] | None

Update the metadata filter before executing the query.

PARAMETER	DESCRIPTION
filter	Filter on the metadata to update. TYPE: dict[str, Any] | None

RETURNS	DESCRIPTION
dict[str, Any] | None	The updated filter on the metadata to apply.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/chroma.py

@override
def update_filter_hook(
    self, filter: dict[str, Any] | None
) -> dict[str, Any] | None:
    filter = super().update_filter_hook(filter)
    if not filter or len(filter) <= 1:
        return filter

    conjoined = [{k: v} for k, v in filter.items()]
    return {"$and": conjoined}

in_memory

Provides an adapter for the InMemoryVectorStore integration.

InMemoryAdapter

InMemoryAdapter(vector_store: StoreT)

Bases: LangchainAdapter[InMemoryVectorStore]

Adapter for InMemoryVectorStore vector store.

This adapter integrates the LangChain In-Memory vector store with the graph retriever system, enabling similarity search and document retrieval.

PARAMETER	DESCRIPTION
vector_store	The in-memory vector store instance. TYPE: StoreT

Initialize the base adapter.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def __init__(
    self,
    vector_store: StoreT,
):
    """Initialize the base adapter."""
    self.vector_store = vector_store

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

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

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,
    )

aembed_query async

aembed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

async def aembed_query(self, query: str):
    """Return the embedding of the query."""
    return await self._safe_embedding.aembed_query(query)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def aget(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = await self._aget(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

asearch async

asearch(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    if k == 0:
        return []

    docs = await self._asearch(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = await self.aembed_query(query)
    docs = await self.asearch(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

embed_query

embed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def embed_query(self, query: str):
    """Return the embedding of the query."""
    return self._safe_embedding.embed_query(query)

format_documents_hook

format_documents_hook(
    docs: list[Document],
) -> list[Content]

Format the documents as content after executing the query.

PARAMETER	DESCRIPTION
docs	The documents returned from the vector store TYPE: list[Document]

RETURNS	DESCRIPTION
list[Content]	The formatted content.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def format_documents_hook(self, docs: list[Document]) -> list[Content]:
    """
    Format the documents as content after executing the query.

    Parameters
    ----------
    docs :
        The documents returned from the vector store

    Returns
    -------
    :
        The formatted content.
    """
    return [doc_to_content(doc) for doc in docs]

get

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def get(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = self._get(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

search

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

Return contents most similar to the query vector.

PARAMETER	DESCRIPTION
embedding	Embedding to look up documents similar to. TYPE: list[float]
k	Number of Documents to return. TYPE: int DEFAULT: 4
filter	Filter on the metadata to apply. TYPE: dict[str, str] | None DEFAULT: None
kwargs	Additional keyword arguments. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
list[Content]	List of Contents most similar to the query vector.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Return contents most similar to the query vector.

    Parameters
    ----------
    embedding :
        Embedding to look up documents similar to.
    k :
        Number of Documents to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    :
        List of Contents most similar to the query vector.
    """
    if k == 0:
        return []

    docs = self._search(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

search_with_embedding

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = self.embed_query(query)
    docs = self.search(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

update_filter_hook

update_filter_hook(
    filter: dict[str, Any] | None,
) -> dict[str, Any] | None

Update the metadata filter before executing the query.

PARAMETER	DESCRIPTION
filter	Filter on the metadata to update. TYPE: dict[str, Any] | None

RETURNS	DESCRIPTION
dict[str, Any] | None	The updated filter on the metadata to apply.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def update_filter_hook(
    self, filter: dict[str, Any] | None
) -> dict[str, Any] | None:
    """
    Update the metadata filter before executing the query.

    Parameters
    ----------
    filter :
        Filter on the metadata to update.

    Returns
    -------
    :
        The updated filter on the metadata to apply.
    """
    return filter

inference

Infers the appropriate adapter for a given vector store.

infer_adapter

infer_adapter(store: Adapter | VectorStore) -> Adapter

Dynamically infer the adapter for a given vector store.

This function identifies the correct adapter based on the vector store type and instantiates it with the provided arguments.

PARAMETER	DESCRIPTION
store	The vector store instance. TYPE: Adapter | VectorStore

RETURNS	DESCRIPTION
Adapter	The initialized adapter for the given vector store.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/inference.py

def infer_adapter(store: Adapter | VectorStore) -> Adapter:
    """
    Dynamically infer the adapter for a given vector store.

    This function identifies the correct adapter based on the vector store type
    and instantiates it with the provided arguments.

    Parameters
    ----------
    store :
        The vector store instance.

    Returns
    -------
    :
        The initialized adapter for the given vector store.
    """
    if isinstance(store, Adapter):
        return store

    module_name, class_name = _infer_adapter_name(store.__class__)
    adapter_module = importlib.import_module(module_name)
    adapter_class = getattr(adapter_module, class_name)
    return adapter_class(store)

langchain

Defines the base class for vector store adapters.

LangchainAdapter

LangchainAdapter(vector_store: StoreT)

Bases: Generic[StoreT], Adapter

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.

PARAMETER	DESCRIPTION
vector_store	The vector store instance. TYPE: StoreT

Initialize the base adapter.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def __init__(
    self,
    vector_store: StoreT,
):
    """Initialize the base adapter."""
    self.vector_store = vector_store

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

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

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,
    )

aembed_query async

aembed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

async def aembed_query(self, query: str):
    """Return the embedding of the query."""
    return await self._safe_embedding.aembed_query(query)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def aget(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = await self._aget(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

asearch async

asearch(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    if k == 0:
        return []

    docs = await self._asearch(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = await self.aembed_query(query)
    docs = await self.asearch(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

embed_query

embed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def embed_query(self, query: str):
    """Return the embedding of the query."""
    return self._safe_embedding.embed_query(query)

format_documents_hook

format_documents_hook(
    docs: list[Document],
) -> list[Content]

Format the documents as content after executing the query.

PARAMETER	DESCRIPTION
docs	The documents returned from the vector store TYPE: list[Document]

RETURNS	DESCRIPTION
list[Content]	The formatted content.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def format_documents_hook(self, docs: list[Document]) -> list[Content]:
    """
    Format the documents as content after executing the query.

    Parameters
    ----------
    docs :
        The documents returned from the vector store

    Returns
    -------
    :
        The formatted content.
    """
    return [doc_to_content(doc) for doc in docs]

get

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def get(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = self._get(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

search

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

Return contents most similar to the query vector.

PARAMETER	DESCRIPTION
embedding	Embedding to look up documents similar to. TYPE: list[float]
k	Number of Documents to return. TYPE: int DEFAULT: 4
filter	Filter on the metadata to apply. TYPE: dict[str, str] | None DEFAULT: None
kwargs	Additional keyword arguments. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
list[Content]	List of Contents most similar to the query vector.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Return contents most similar to the query vector.

    Parameters
    ----------
    embedding :
        Embedding to look up documents similar to.
    k :
        Number of Documents to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    :
        List of Contents most similar to the query vector.
    """
    if k == 0:
        return []

    docs = self._search(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

search_with_embedding

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = self.embed_query(query)
    docs = self.search(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

update_filter_hook

update_filter_hook(
    filter: dict[str, Any] | None,
) -> dict[str, Any] | None

Update the metadata filter before executing the query.

PARAMETER	DESCRIPTION
filter	Filter on the metadata to update. TYPE: dict[str, Any] | None

RETURNS	DESCRIPTION
dict[str, Any] | None	The updated filter on the metadata to apply.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def update_filter_hook(
    self, filter: dict[str, Any] | None
) -> dict[str, Any] | None:
    """
    Update the metadata filter before executing the query.

    Parameters
    ----------
    filter :
        Filter on the metadata to update.

    Returns
    -------
    :
        The updated filter on the metadata to apply.
    """
    return filter

ShreddedLangchainAdapter

ShreddedLangchainAdapter(
    vector_store: StoreT,
    shredder: ShreddingTransformer | None = None,
    nested_metadata_fields: set[str] = set(),
)

Bases: LangchainAdapter[StoreT]

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 that do not support searching on list-based metadata values.

PARAMETER	DESCRIPTION
vector_store	The vector store instance. TYPE: StoreT
shredder	An instance of the ShreddingTransformer used for doc insertion. If not passed then a default instance of ShreddingTransformer is used. TYPE: ShreddingTransformer | None DEFAULT: None
nested_metadata_fields	The set of metadata fields that contain nested values. TYPE: set[str] DEFAULT: set()

Initialize the base adapter.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def __init__(
    self,
    vector_store: StoreT,
    shredder: ShreddingTransformer | None = None,
    nested_metadata_fields: set[str] = set(),
):
    """Initialize the base adapter."""
    super().__init__(vector_store=vector_store)
    self.shredder = ShreddingTransformer() if shredder is None else shredder
    self.nested_metadata_fields = nested_metadata_fields

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

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

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,
    )

aembed_query async

aembed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

async def aembed_query(self, query: str):
    """Return the embedding of the query."""
    return await self._safe_embedding.aembed_query(query)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def aget(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = await self._aget(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

asearch async

asearch(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    if k == 0:
        return []

    docs = await self._asearch(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = await self.aembed_query(query)
    docs = await self.asearch(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

embed_query

embed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def embed_query(self, query: str):
    """Return the embedding of the query."""
    return self._safe_embedding.embed_query(query)

format_documents_hook

format_documents_hook(
    docs: list[Document],
) -> list[Content]

Format the documents as content after executing the query.

PARAMETER	DESCRIPTION
docs	The documents returned from the vector store TYPE: list[Document]

RETURNS	DESCRIPTION
list[Content]	The formatted content.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def format_documents_hook(self, docs: list[Document]) -> list[Content]:
    restored = list(self.shredder.restore_documents(documents=docs))
    return super().format_documents_hook(restored)

get

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def get(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = self._get(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

search

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

Return contents most similar to the query vector.

PARAMETER	DESCRIPTION
embedding	Embedding to look up documents similar to. TYPE: list[float]
k	Number of Documents to return. TYPE: int DEFAULT: 4
filter	Filter on the metadata to apply. TYPE: dict[str, str] | None DEFAULT: None
kwargs	Additional keyword arguments. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
list[Content]	List of Contents most similar to the query vector.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Return contents most similar to the query vector.

    Parameters
    ----------
    embedding :
        Embedding to look up documents similar to.
    k :
        Number of Documents to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    :
        List of Contents most similar to the query vector.
    """
    if k == 0:
        return []

    docs = self._search(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

search_with_embedding

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = self.embed_query(query)
    docs = self.search(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

update_filter_hook

update_filter_hook(
    filter: dict[str, str] | None,
) -> dict[str, str] | None

Update the metadata filter before executing the query.

PARAMETER	DESCRIPTION
filter	Filter on the metadata to update. TYPE: dict[str, Any] | None

RETURNS	DESCRIPTION
dict[str, Any] | None	The updated filter on the metadata to apply.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def update_filter_hook(
    self, filter: dict[str, str] | None
) -> dict[str, str] | None:
    if filter is None:
        return None

    shredded_filter = {}
    for key, value in filter.items():
        if key in self.nested_metadata_fields:
            shredded_filter[self.shredder.shredded_key(key, value)] = (
                self.shredder.shredded_value()
            )
        else:
            shredded_filter[key] = value
    return shredded_filter

open_search

Provides an adapter for OpenSearch vector store integration.

OpenSearchAdapter

OpenSearchAdapter(vector_store: OpenSearchVectorSearch)

Bases: LangchainAdapter[OpenSearchVectorSearch]

Adapter to traverse OpenSearch vector stores.

This adapter enables similarity search and document retrieval using an OpenSearch vector store.

PARAMETER	DESCRIPTION
vector_store	The OpenSearch vector store instance. TYPE: OpenSearchVectorSearch

Notes

Graph Traversal is only supported when using either the "lucene" or "faiss" engine.

For more info, see the OpenSearch Documentation

Initialize the base adapter.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/open_search.py

def __init__(self, vector_store: OpenSearchVectorSearch):
    if vector_store.engine not in ["lucene", "faiss"]:
        msg = (
            f"Invalid engine for Traversal: '{self.vector_store.engine}'"
            " please instantiate the Open Search Vector Store with"
            " either the 'lucene' or 'faiss' engine"
        )
        raise ValueError(msg)
    super().__init__(vector_store)

    if vector_store.is_aoss:
        self._id_filed = "id"
    else:
        self._id_field = "_id"

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

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

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,
    )

aembed_query async

aembed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

async def aembed_query(self, query: str):
    """Return the embedding of the query."""
    return await self._safe_embedding.aembed_query(query)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def aget(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = await self._aget(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

asearch async

asearch(
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | 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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    if k == 0:
        return []

    docs = await self._asearch(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
async def asearch_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = await self.aembed_query(query)
    docs = await self.asearch(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

embed_query

embed_query(query: str)

Return the embedding of the query.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def embed_query(self, query: str):
    """Return the embedding of the query."""
    return self._safe_embedding.embed_query(query)

format_documents_hook

format_documents_hook(
    docs: list[Document],
) -> list[Content]

Format the documents as content after executing the query.

PARAMETER	DESCRIPTION
docs	The documents returned from the vector store TYPE: list[Document]

RETURNS	DESCRIPTION
list[Content]	The formatted content.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def format_documents_hook(self, docs: list[Document]) -> list[Content]:
    """
    Format the documents as content after executing the query.

    Parameters
    ----------
    docs :
        The documents returned from the vector store

    Returns
    -------
    :
        The formatted content.
    """
    return [doc_to_content(doc) for doc in docs]

get

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def get(
    self,
    ids: Sequence[str],
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> list[Content]:
    docs = self._get(self._remove_duplicates(ids), filter, **kwargs)
    return self.format_documents_hook(docs)

search

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

Return contents most similar to the query vector.

PARAMETER	DESCRIPTION
embedding	Embedding to look up documents similar to. TYPE: list[float]
k	Number of Documents to return. TYPE: int DEFAULT: 4
filter	Filter on the metadata to apply. TYPE: dict[str, str] | None DEFAULT: None
kwargs	Additional keyword arguments. TYPE: Any DEFAULT: {}

RETURNS	DESCRIPTION
list[Content]	List of Contents most similar to the query vector.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search(
    self,
    embedding: list[float],
    k: int = 4,
    filter: dict[str, str] | None = None,
    **kwargs: Any,
) -> list[Content]:
    """
    Return contents most similar to the query vector.

    Parameters
    ----------
    embedding :
        Embedding to look up documents similar to.
    k :
        Number of Documents to return.
    filter :
        Filter on the metadata to apply.
    kwargs :
        Additional keyword arguments.

    Returns
    -------
    :
        List of Contents most similar to the query vector.
    """
    if k == 0:
        return []

    docs = self._search(
        embedding=embedding,
        k=k,
        filter=self.update_filter_hook(filter),
        **kwargs,
    )
    return self.format_documents_hook(docs)

search_with_embedding

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/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

@override
def search_with_embedding(
    self,
    query: str,
    k: int = 4,
    filter: dict[str, Any] | None = None,
    **kwargs: Any,
) -> tuple[list[float], list[Content]]:
    query_embedding = self.embed_query(query)
    docs = self.search(
        embedding=query_embedding,
        k=k,
        filter=filter,
        **kwargs,
    )
    return query_embedding, docs

update_filter_hook

update_filter_hook(
    filter: dict[str, Any] | None,
) -> dict[str, Any] | None

Update the metadata filter before executing the query.

PARAMETER	DESCRIPTION
filter	Filter on the metadata to update. TYPE: dict[str, Any] | None

RETURNS	DESCRIPTION
dict[str, Any] | None	The updated filter on the metadata to apply.

Source code in packages/langchain-graph-retriever/src/langchain_graph_retriever/adapters/langchain.py

def update_filter_hook(
    self, filter: dict[str, Any] | None
) -> dict[str, Any] | None:
    """
    Update the metadata filter before executing the query.

    Parameters
    ----------
    filter :
        Filter on the metadata to update.

    Returns
    -------
    :
        The updated filter on the metadata to apply.
    """
    return filter