o Zh_A@sdZddlmZddlZddlmZmZddlmZddl m Z m Z m Z ddl mZddlmZmZmZdd lmZdd lmZdd lmZdd lmZmZmZmZdd lmZe rbddl m!Z!m"Z"e#Z$e%eZ&ee$e&fZ'ee e&fZ(GdddeddZ)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. **Class hierarchy:** .. code-block:: BaseRetriever --> Retriever # Examples: ArxivRetriever, MergerRetriever **Main helpers:** .. code-block:: RetrieverInput, RetrieverOutput, RetrieverLike, RetrieverOutputLike, Document, Serializable, Callbacks, CallbackManagerForRetrieverRun, AsyncCallbackManagerForRetrieverRun ) annotationsN)ABCabstractmethod) signature) TYPE_CHECKINGAnyOptional) ConfigDict)Self TypedDictoverride) deprecated) Callbacks)Document)RunnableRunnableConfigRunnableSerializable ensure_config)run_in_executor)#AsyncCallbackManagerForRetrieverRunCallbackManagerForRetrieverRunc@s8eZdZUdZded< ded< ded< ded<dS) LangSmithRetrieverParamsz!LangSmith parameters for tracing.strls_retriever_name Optional[str]Zls_vector_store_providerZls_embedding_providerZls_embedding_modelN)__name__ __module__ __qualname____doc____annotations__r r P/var/www/html/lang_env/lib/python3.10/site-packages/langchain_core/retrievers.pyr7s rF)totalcseZdZUdZeddZdZded<dZded<dZ d ed < dZ d ed < e d6fdd Z d7ddZ e d8d9ddZe d8d9ddZed:d#d$Zd;d&d'Zed(d)d*d+ddddd,d 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: A simple retriever based on a scikit-learn vectorizer .. code-block:: 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)Zarbitrary_types_allowedFbool_new_arg_supported_expects_other_argsNOptional[list[str]]tagsOptional[dict[str, Any]]metadatakwargsrreturnNonec stjdi||jtjkr!tjdtdd|j}tj|_||_t|dr>|j tj kr>tjdtdd|j }tj |_ ||_ t |jj }| ddu|_|js]|j tj kr]dd d}||_ tt|hddk|_dS)NzgRetrievers must implement abstract `_get_relevant_documents` method instead of `get_relevant_documents`) stacklevelaget_relevant_documentsziRetrievers must implement abstract `_aget_relevant_documents` method instead of `aget_relevant_documents` run_managerselfr queryrr,list[Document]cstd|j|IdHSN)r_get_relevant_documents)r2r3r r r!_aget_relevant_documentsszABaseRetriever.__init_subclass__.._aget_relevant_documents>r2r1r3rr )r2r r3rr,r4)super__init_subclass__get_relevant_documentsr#warningswarnDeprecationWarningr6hasattrr0r7r parametersgetr%lensetkeysr&)clsr+ZswapZaswapr?r7 __class__r r!r9s@     zBaseRetriever.__init_subclass___kwargsrcKsH|}|dr|dd}n |dr|dd}|}t|dS)z Get standard params for tracing.Z Retriever Ni)r)get_name startswithendswithlowerr)r2rGZdefault_retriever_namer r r!_get_ls_paramss    zBaseRetriever._get_ls_paramsinputrconfigOptional[RunnableConfig]r4c Ksddlm}t|}i|dpi|jdi|}|j|dd|dd|d|j||jd }|jd||d p?| | d dd }z!|j rM|ni}|j r^|j |fd |i|} n |j |fi|} Wntyz} 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. Defaults to None. kwargs: Additional arguments to pass to the retriever. Returns: List of relevant documents. Examples: .. code-block:: python retriever.invoke("query") r)CallbackManagerr* callbacksNverboseFr(rSZinheritable_tagsZ local_tagsinheritable_metadataZlocal_metadatarun_namerun_idnamerWr1r ) langchain_core.callbacks.managerrQrr@rM configurer(r*on_retriever_startrIpopr&r%r6 Exceptionon_retriever_erroron_retriever_end) r2rNrOr+rQrUcallback_managerr1rGresulter r r!invokesV      zBaseRetriever.invokec s$ddlm}t|}i|dpi|jdi|}|j|dd|dd|d|j||jd }|jd||d p@| | d dd IdH}z'|j rQ|ni}|j re|j |fd |i|IdH} n |j |fi|IdH} Wnty} 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. Defaults to None. kwargs: Additional arguments to pass to the retriever. Returns: List of relevant documents. Examples: .. code-block:: python await retriever.ainvoke("query") r)AsyncCallbackManagerr*rRNrSFr(rTrVrWrXr1r )rZrerr@rMr[r(r*r\rIr]r&r%r7r^r_r`) r2rNrOr+rerUrar1rGrbrcr r r!ainvokesX       zBaseRetriever.ainvoker3r1rcCsdS)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 r2r3r1r r r!r6Qsz%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)r1)rr6Zget_syncrgr r r!r7_s  z&BaseRetriever._aget_relevant_documentsz0.1.46rdz1.0)Zsince alternativeZremoval)rRr(r*rVrRrrVrcKsHi}|r||d<|r||d<|r||d<|r||d<|j||fi|S)a Retrieve documents relevant to a query. Users should favor using `.invoke` or `.batch` rather than `get_relevant_documents directly`. Args: query: string to find relevant documents for. callbacks: Callback manager or list of callbacks. Defaults to None. tags: Optional list of tags associated with the retriever. These tags will be associated with each call to this retriever, and passed as arguments to the handlers defined in `callbacks`. Defaults to None. metadata: Optional metadata associated with the retriever. This metadata will be associated with each call to this retriever, and passed as arguments to the handlers defined in `callbacks`. Defaults to None. run_name: Optional name for the run. Defaults to None. kwargs: Additional arguments to pass to the retriever. Returns: List of relevant documents. rRr(r*rV)rdr2r3rRr(r*rVr+rOr r r!r:qs!z$BaseRetriever.get_relevant_documentsrfcsPi}|r ||d<|r||d<|r||d<|r||d<|j||fi|IdHS)aAsynchronously get documents relevant to a query. Users should favor using `.ainvoke` or `.abatch` rather than `aget_relevant_documents directly`. Args: query: string to find relevant documents for. callbacks: Callback manager or list of callbacks. tags: Optional list of tags associated with the retriever. These tags will be associated with each call to this retriever, and passed as arguments to the handlers defined in `callbacks`. Defaults to None. metadata: Optional metadata associated with the retriever. This metadata will be associated with each call to this retriever, and passed as arguments to the handlers defined in `callbacks`. Defaults to None. run_name: Optional name for the run. Defaults to None. kwargs: Additional arguments to pass to the retriever. Returns: List of relevant documents. rRr(r*rVN)rfrir r r!r0s!z%BaseRetriever.aget_relevant_documents)r+rr,r-)rGrr,rr5)rNrrOrPr+rr,r4)r3rr1rr,r4)r3rr1rr,r4)r3rrRrr(r'r*r)rVrr+rr,r4)rrrrr Z model_configr%rr&r(r*r r9rMrdrfrr6r7r r:r0 __classcell__r r rEr!r#DsF <     2 <?   +r#)+r __future__rr;abcrrinspectrtypingrrrZpydanticr Ztyping_extensionsr r r Zlangchain_core._apir Zlangchain_core.callbacksrZlangchain_core.documentsrZlangchain_core.runnablesrrrrZlangchain_core.runnables.configrrZrrrZRetrieverInputlistZRetrieverOutputZ RetrieverLikeZRetrieverOutputLikerr#r r r r!s*