langchain_core.vectorstores
.VectorStoreRetriever¶
Note
VectorStoreRetriever implements the standard Runnable Interface
. 🏃
The Runnable Interface
has additional methods that are available on runnables, such as with_types
, with_retry
, assign
, bind
, get_graph
, and more.
- class langchain_core.vectorstores.VectorStoreRetriever[source]¶
Bases:
BaseRetriever
Base Retriever class for VectorStore.
Create a new model by parsing and validating input data from keyword arguments.
Raises ValidationError if the input data cannot be parsed to form a valid model.
- param metadata: Optional[Dict[str, Any]] = None¶
Optional metadata associated with the retriever. Defaults to None. This metadata will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks. You can use these to eg identify a specific instance of a retriever with its use case.
- param search_kwargs: dict [Optional]¶
Keyword arguments to pass to the search function.
- param search_type: str = 'similarity'¶
Type of search to perform. Defaults to “similarity”.
- param tags: Optional[List[str]] = None¶
Optional list of tags associated with the retriever. Defaults to None. These tags will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks. You can use these to eg identify a specific instance of a retriever with its use case.
- param vectorstore: VectorStore [Required]¶
VectorStore to use for retrieval.
- async aadd_documents(documents: List[Document], **kwargs: Any) List[str] [source]¶
Add documents to the vectorstore.
- Parameters
documents (List[Document]) – Documents to add to the vectorstore.
kwargs (Any) –
- Returns
List of IDs of the added texts.
- Return type
List[str]
- async abatch(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) List[Output] ¶
Default implementation runs ainvoke in parallel using asyncio.gather.
The default implementation of batch works well for IO bound runnables.
Subclasses should override this method if they can batch more efficiently; e.g., if the underlying runnable uses an API which supports a batch mode.
- Parameters
inputs (List[Input]) –
config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) –
return_exceptions (bool) –
kwargs (Optional[Any]) –
- Return type
List[Output]
- async abatch_as_completed(inputs: Sequence[Input], config: Optional[Union[RunnableConfig, Sequence[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) AsyncIterator[Tuple[int, Union[Output, Exception]]] ¶
Run ainvoke in parallel on a list of inputs, yielding results as they complete.
- Parameters
inputs (Sequence[Input]) –
config (Optional[Union[RunnableConfig, Sequence[RunnableConfig]]]) –
return_exceptions (bool) –
kwargs (Optional[Any]) –
- Return type
AsyncIterator[Tuple[int, Union[Output, Exception]]]
- add_documents(documents: List[Document], **kwargs: Any) List[str] [source]¶
Add documents to the vectorstore.
- Parameters
documents (List[Document]) – Documents to add to the vectorstore.
kwargs (Any) –
- Returns
List of IDs of the added texts.
- Return type
List[str]
- async aget_relevant_documents(query: str, *, callbacks: Callbacks = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, **kwargs: Any) List[Document] ¶
[Deprecated] Asynchronously get documents relevant to a query.
Users should favor using .ainvoke or .abatch rather than aget_relevant_documents directly.
- Parameters
query (str) – string to find relevant documents for
callbacks (Callbacks) – Callback manager or list of callbacks
tags (Optional[List[str]]) – Optional list of tags associated with the retriever. Defaults to None These tags will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks.
metadata (Optional[Dict[str, Any]]) – Optional metadata associated with the retriever. Defaults to None This metadata will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks.
run_name (Optional[str]) – Optional name for the run.
kwargs (Any) –
- Returns
List of relevant documents
- Return type
List[Document]
Notes
Deprecated since version langchain-core==0.1.46: Use ainvoke instead.
- async ainvoke(input: str, config: Optional[RunnableConfig] = None, **kwargs: Any) List[Document] ¶
Asynchronously invoke the retriever to get relevant documents.
Main entry point for asynchronous retriever invocations.
- Parameters
input (str) – The query string
config (Optional[RunnableConfig]) – Configuration for the retriever
**kwargs (Any) – Additional arguments to pass to the retriever
- Returns
List of relevant documents
- Return type
List[Document]
Examples:
await retriever.ainvoke("query")
- async astream(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Optional[Any]) AsyncIterator[Output] ¶
Default implementation of astream, which calls ainvoke. Subclasses should override this method if they support streaming output.
- Parameters
input (Input) –
config (Optional[RunnableConfig]) –
kwargs (Optional[Any]) –
- Return type
AsyncIterator[Output]
- astream_events(input: Any, config: Optional[RunnableConfig] = None, *, version: Literal['v1', 'v2'], include_names: Optional[Sequence[str]] = None, include_types: Optional[Sequence[str]] = None, include_tags: Optional[Sequence[str]] = None, exclude_names: Optional[Sequence[str]] = None, exclude_types: Optional[Sequence[str]] = None, exclude_tags: Optional[Sequence[str]] = None, **kwargs: Any) AsyncIterator[StreamEvent] ¶
[Beta] Generate a stream of events.
Use to create an iterator over StreamEvents that provide real-time information about the progress of the runnable, including StreamEvents from intermediate results.
A StreamEvent is a dictionary with the following schema:
event
: str - Event names are of theformat: on_[runnable_type]_(start|stream|end).
name
: str - The name of the runnable that generated the event.run_id
: str - randomly generated ID associated with the given execution ofthe runnable that emitted the event. A child runnable that gets invoked as part of the execution of a parent runnable is assigned its own unique ID.
parent_ids
: List[str] - The IDs of the parent runnables thatgenerated the event. The root runnable will have an empty list. The order of the parent IDs is from the root to the immediate parent. Only available for v2 version of the API. The v1 version of the API will return an empty list.
tags
: Optional[List[str]] - The tags of the runnable that generatedthe event.
metadata
: Optional[Dict[str, Any]] - The metadata of the runnablethat generated the event.
data
: Dict[str, Any]
Below is a table that illustrates some evens that might be emitted by various chains. Metadata fields have been omitted from the table for brevity. Chain definitions have been included after the table.
ATTENTION This reference table is for the V2 version of the schema.
event
name
chunk
input
output
on_chat_model_start
[model name]
{“messages”: [[SystemMessage, HumanMessage]]}
on_chat_model_stream
[model name]
AIMessageChunk(content=”hello”)
on_chat_model_end
[model name]
{“messages”: [[SystemMessage, HumanMessage]]}
AIMessageChunk(content=”hello world”)
on_llm_start
[model name]
{‘input’: ‘hello’}
on_llm_stream
[model name]
‘Hello’
on_llm_end
[model name]
‘Hello human!’
on_chain_start
format_docs
on_chain_stream
format_docs
“hello world!, goodbye world!”
on_chain_end
format_docs
[Document(…)]
“hello world!, goodbye world!”
on_tool_start
some_tool
{“x”: 1, “y”: “2”}
on_tool_end
some_tool
{“x”: 1, “y”: “2”}
on_retriever_start
[retriever name]
{“query”: “hello”}
on_retriever_end
[retriever name]
{“query”: “hello”}
[Document(…), ..]
on_prompt_start
[template_name]
{“question”: “hello”}
on_prompt_end
[template_name]
{“question”: “hello”}
ChatPromptValue(messages: [SystemMessage, …])
Here are declarations associated with the events shown above:
format_docs:
def format_docs(docs: List[Document]) -> str: '''Format the docs.''' return ", ".join([doc.page_content for doc in docs]) format_docs = RunnableLambda(format_docs)
some_tool:
@tool def some_tool(x: int, y: str) -> dict: '''Some_tool.''' return {"x": x, "y": y}
prompt:
template = ChatPromptTemplate.from_messages( [("system", "You are Cat Agent 007"), ("human", "{question}")] ).with_config({"run_name": "my_template", "tags": ["my_template"]})
Example:
from langchain_core.runnables import RunnableLambda async def reverse(s: str) -> str: return s[::-1] chain = RunnableLambda(func=reverse) events = [ event async for event in chain.astream_events("hello", version="v2") ] # will produce the following events (run_id, and parent_ids # has been omitted for brevity): [ { "data": {"input": "hello"}, "event": "on_chain_start", "metadata": {}, "name": "reverse", "tags": [], }, { "data": {"chunk": "olleh"}, "event": "on_chain_stream", "metadata": {}, "name": "reverse", "tags": [], }, { "data": {"output": "olleh"}, "event": "on_chain_end", "metadata": {}, "name": "reverse", "tags": [], }, ]
- Parameters
input (Any) – The input to the runnable.
config (Optional[RunnableConfig]) – The config to use for the runnable.
version (Literal['v1', 'v2']) – The version of the schema to use either v2 or v1. Users should use v2. v1 is for backwards compatibility and will be deprecated in 0.4.0. No default will be assigned until the API is stabilized.
include_names (Optional[Sequence[str]]) – Only include events from runnables with matching names.
include_types (Optional[Sequence[str]]) – Only include events from runnables with matching types.
include_tags (Optional[Sequence[str]]) – Only include events from runnables with matching tags.
exclude_names (Optional[Sequence[str]]) – Exclude events from runnables with matching names.
exclude_types (Optional[Sequence[str]]) – Exclude events from runnables with matching types.
exclude_tags (Optional[Sequence[str]]) – Exclude events from runnables with matching tags.
kwargs (Any) – Additional keyword arguments to pass to the runnable. These will be passed to astream_log as this implementation of astream_events is built on top of astream_log.
- Returns
An async stream of StreamEvents.
- Return type
AsyncIterator[StreamEvent]
Notes
- batch(inputs: List[Input], config: Optional[Union[RunnableConfig, List[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) List[Output] ¶
Default implementation runs invoke in parallel using a thread pool executor.
The default implementation of batch works well for IO bound runnables.
Subclasses should override this method if they can batch more efficiently; e.g., if the underlying runnable uses an API which supports a batch mode.
- Parameters
inputs (List[Input]) –
config (Optional[Union[RunnableConfig, List[RunnableConfig]]]) –
return_exceptions (bool) –
kwargs (Optional[Any]) –
- Return type
List[Output]
- batch_as_completed(inputs: Sequence[Input], config: Optional[Union[RunnableConfig, Sequence[RunnableConfig]]] = None, *, return_exceptions: bool = False, **kwargs: Optional[Any]) Iterator[Tuple[int, Union[Output, Exception]]] ¶
Run invoke in parallel on a list of inputs, yielding results as they complete.
- Parameters
inputs (Sequence[Input]) –
config (Optional[Union[RunnableConfig, Sequence[RunnableConfig]]]) –
return_exceptions (bool) –
kwargs (Optional[Any]) –
- Return type
Iterator[Tuple[int, Union[Output, Exception]]]
- configurable_alternatives(which: ConfigurableField, *, default_key: str = 'default', prefix_keys: bool = False, **kwargs: Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]) RunnableSerializable[Input, Output] ¶
Configure alternatives for runnables that can be set at runtime.
from langchain_anthropic import ChatAnthropic from langchain_core.runnables.utils import ConfigurableField from langchain_openai import ChatOpenAI model = ChatAnthropic( model_name="claude-3-sonnet-20240229" ).configurable_alternatives( ConfigurableField(id="llm"), default_key="anthropic", openai=ChatOpenAI() ) # uses the default model ChatAnthropic print(model.invoke("which organization created you?").content) # uses ChatOpenAI print( model.with_config( configurable={"llm": "openai"} ).invoke("which organization created you?").content )
- Parameters
which (ConfigurableField) –
default_key (str) –
prefix_keys (bool) –
kwargs (Union[Runnable[Input, Output], Callable[[], Runnable[Input, Output]]]) –
- Return type
RunnableSerializable[Input, Output]
- configurable_fields(**kwargs: Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) RunnableSerializable[Input, Output] ¶
Configure particular runnable fields at runtime.
from langchain_core.runnables import ConfigurableField from langchain_openai import ChatOpenAI model = ChatOpenAI(max_tokens=20).configurable_fields( max_tokens=ConfigurableField( id="output_token_number", name="Max tokens in the output", description="The maximum number of tokens in the output", ) ) # max_tokens = 20 print( "max_tokens_20: ", model.invoke("tell me something about chess").content ) # max_tokens = 200 print("max_tokens_200: ", model.with_config( configurable={"output_token_number": 200} ).invoke("tell me something about chess").content )
- Parameters
kwargs (Union[ConfigurableField, ConfigurableFieldSingleOption, ConfigurableFieldMultiOption]) –
- Return type
RunnableSerializable[Input, Output]
- get_relevant_documents(query: str, *, callbacks: Callbacks = None, tags: Optional[List[str]] = None, metadata: Optional[Dict[str, Any]] = None, run_name: Optional[str] = None, **kwargs: Any) List[Document] ¶
[Deprecated] Retrieve documents relevant to a query.
Users should favor using .invoke or .batch rather than get_relevant_documents directly.
- Parameters
query (str) – string to find relevant documents for
callbacks (Callbacks) – Callback manager or list of callbacks
tags (Optional[List[str]]) – Optional list of tags associated with the retriever. Defaults to None These tags will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks.
metadata (Optional[Dict[str, Any]]) – Optional metadata associated with the retriever. Defaults to None This metadata will be associated with each call to this retriever, and passed as arguments to the handlers defined in callbacks.
run_name (Optional[str]) – Optional name for the run.
kwargs (Any) –
- Returns
List of relevant documents
- Return type
List[Document]
Notes
Deprecated since version langchain-core==0.1.46: Use invoke instead.
- invoke(input: str, config: Optional[RunnableConfig] = None, **kwargs: Any) List[Document] ¶
Invoke the retriever to get relevant documents.
Main entry point for synchronous retriever invocations.
- Parameters
input (str) – The query string
config (Optional[RunnableConfig]) – Configuration for the retriever
**kwargs (Any) – Additional arguments to pass to the retriever
- Returns
List of relevant documents
- Return type
List[Document]
Examples:
retriever.invoke("query")
- stream(input: Input, config: Optional[RunnableConfig] = None, **kwargs: Optional[Any]) Iterator[Output] ¶
Default implementation of stream, which calls invoke. Subclasses should override this method if they support streaming output.
- Parameters
input (Input) –
config (Optional[RunnableConfig]) –
kwargs (Optional[Any]) –
- Return type
Iterator[Output]
- to_json() Union[SerializedConstructor, SerializedNotImplemented] ¶
Serialize the runnable to JSON.
- Return type
- allowed_search_types: ClassVar[Collection[str]] = ('similarity', 'similarity_score_threshold', 'mmr')¶