o +iu+@sdZddlmZddlmZmZddlmZddlm Z m Z ddl m Z ddl mZmZmZddlmZmZdd lmZdd lmZmZmZmZdd lmZe rXdd lmZmZeZ e!eZ"ee e"fZ#ee e"fZ$Gd ddeddZ%Gdddee e"feZ&dS)a,**Retriever** class returns Documents given a text **query**. It is more general than a vector store. A retriever does not need to be able to store documents, only to return (or retrieve) it. Vector stores can be used as the backbone of a retriever, but there are other types of retrievers as well. ) annotations)ABCabstractmethod) signature) TYPE_CHECKINGAny) ConfigDict)Self TypedDictoverride)AsyncCallbackManagerCallbackManager)Document)RunnableRunnableConfigRunnableSerializable ensure_config)run_in_executor)#AsyncCallbackManagerForRetrieverRunCallbackManagerForRetrieverRunc@s8eZdZUdZded< ded< ded< ded<dS) LangSmithRetrieverParamsz!LangSmith parameters for tracing.strls_retriever_namez str | Nonels_vector_store_providerls_embedding_providerls_embedding_modelN)__name__ __module__ __qualname____doc____annotations__r!r!V/var/www/html/psymed-ai/venv/lib/python3.10/site-packages/langchain_core/retrievers.pyr's rF)totalcseZdZUdZeddZdZded<dZded<dZ d ed < dZ d ed < e d(fdd Z d)ddZ e d*d+ddZe d*d+ddZed,d#d$Zd-d&d'ZZS). BaseRetrieveraAbstract base class for a document retrieval system. A retrieval system is defined as something that can take string queries and return the most 'relevant' documents from some source. Usage: A retriever follows the standard `Runnable` interface, and should be used via the standard `Runnable` methods of `invoke`, `ainvoke`, `batch`, `abatch`. Implementation: When implementing a custom retriever, the class should implement the `_get_relevant_documents` method to define the logic for retrieving documents. Optionally, an async native implementations can be provided by overriding the `_aget_relevant_documents` method. !!! example "Retriever that returns the first 5 documents from a list of documents" ```python from langchain_core.documents import Document from langchain_core.retrievers import BaseRetriever class SimpleRetriever(BaseRetriever): docs: list[Document] k: int = 5 def _get_relevant_documents(self, query: str) -> list[Document]: """Return the first k documents from the list of documents""" return self.docs[:self.k] async def _aget_relevant_documents(self, query: str) -> list[Document]: """(Optional) async native implementation.""" return self.docs[:self.k] ``` !!! example "Simple retriever based on a scikit-learn vectorizer" ```python from sklearn.metrics.pairwise import cosine_similarity class TFIDFRetriever(BaseRetriever, BaseModel): vectorizer: Any docs: list[Document] tfidf_array: Any k: int = 4 class Config: arbitrary_types_allowed = True def _get_relevant_documents(self, query: str) -> list[Document]: # Ip -- (n_docs,x), Op -- (n_docs,n_Feats) query_vec = self.vectorizer.transform([query]) # Op -- (n_docs,1) -- Cosine Sim with each doc results = cosine_similarity(self.tfidf_array, query_vec).reshape((-1,)) return [self.docs[i] for i in results.argsort()[-self.k :][::-1]] ``` T)arbitrary_types_allowedFbool_new_arg_supported_expects_other_argsNzlist[str] | Nonetagszdict[str, Any] | NonemetadatakwargsrreturnNonec srtjd i|t|jj}|ddu|_|js(|jtjkr(d dd }||_t t | hd d k|_ dS)N run_managerselfr queryrr,list[Document]cstd|j|IdHSN)r_get_relevant_documents)r/r0r!r!r"_aget_relevant_documentsszABaseRetriever.__init_subclass__.._aget_relevant_documents>r/r0r.rr!)r/r r0rr,r1) super__init_subclass__rr3 parametersgetr'r4r$lensetkeysr()clsr+r7r4 __class__r!r"r6s   zBaseRetriever.__init_subclass___kwargsrcKsH|}|dr|dd}n |dr|dd}|}t|dS)z Get standard params for tracing. Retriever Ni)r)get_name startswithendswithlowerr)r/r?default_retriever_namer!r!r"_get_ls_paramss    zBaseRetriever._get_ls_paramsinputrconfigRunnableConfig | Noner1c Kst|}i|dp i|jd i|}tj|dd|dd|d|j||jd}|jd||dp9|| d dd }z!|j rG|ni}|j rX|j |fd |i|}n |j |fi|}Wnt yt} z|| d} ~ ww|||S) aInvoke the retriever to get relevant documents. Main entry point for synchronous retriever invocations. Args: input: The query string. config: Configuration for the retriever. **kwargs: Additional arguments to pass to the retriever. Returns: List of relevant documents. Examples: ```python retriever.invoke("query") ``` r* callbacksNverboseFr)rLinheritable_tags local_tagsinheritable_metadatalocal_metadatarun_namerun_idnamerSr.r!)rr8rGr configurer)r*on_retriever_startrBpopr(r'r3 Exceptionon_retriever_erroron_retriever_end r/rHrIr+rPcallback_managerr.kwargs_resulter!r!r"invokesT     zBaseRetriever.invokec st|}i|dp i|jd i|}tj|dd|dd|d|j||jd}|jd||dp:|| d dd IdH}z'|j rK|ni}|j r_|j |fd |i|IdH}n |j |fi|IdH}Wnt y} z || IdHd} ~ ww||IdH|S) aAsynchronously invoke the retriever to get relevant documents. Main entry point for asynchronous retriever invocations. Args: input: The query string. config: Configuration for the retriever. **kwargs: Additional arguments to pass to the retriever. Returns: List of relevant documents. Examples: ```python await retriever.ainvoke("query") ``` r*rKNrLFr)rMrRrSrTr.r!)rr8rGr rVr)r*rWrBrXr(r'r4rYrZr[r\r!r!r"ainvokesV      zBaseRetriever.ainvoker0r.rcCsdS)zGet documents relevant to a query. Args: query: String to find relevant documents for. run_manager: The callback handler to use. Returns: List of relevant documents. Nr!r/r0r.r!r!r"r3#sz%BaseRetriever._get_relevant_documentsrcstd|j||dIdHS)zAsynchronously get documents relevant to a query. Args: query: String to find relevant documents for run_manager: The callback handler to use Returns: List of relevant documents N)r.)rr3get_syncrcr!r!r"r41s  z&BaseRetriever._aget_relevant_documents)r+rr,r-)r?rr,rr2)rHrrIrJr+rr,r1)r0rr.rr,r1)r0rr.rr,r1)rrrrr model_configr'r r(r)r*r r6rGrarbrr3r4 __classcell__r!r!r=r"r$4s. =       9<  r$N)'r __future__rabcrrinspectrtypingrrpydanticrtyping_extensionsr r r langchain_core.callbacks.managerr r langchain_core.documentsrlangchain_core.runnablesrrrrlangchain_core.runnables.configrrrrRetrieverInputlistRetrieverOutput RetrieverLikeRetrieverOutputLikerr$r!r!r!r"s&