Source code for langchain_core._api.beta_decorator
"""Helper functions for marking parts of the LangChain API as beta.This module was loosely adapted from matplotlibs _api/deprecation.py module:https://github.com/matplotlib/matplotlib/blob/main/lib/matplotlib/_api/deprecation.py.. warning:: This module is for internal use only. Do not use it in your own code. We may change the API at any time with no warning."""importcontextlibimportfunctoolsimportinspectimportwarningsfromcollections.abcimportGeneratorfromtypingimportAny,Callable,TypeVar,Union,castfromlangchain_core._api.internalimportis_caller_internalclassLangChainBetaWarning(DeprecationWarning):"""A class for issuing beta warnings for LangChain users."""# PUBLIC APIT=TypeVar("T",bound=Union[Callable[...,Any],type])defbeta(*,message:str="",name:str="",obj_type:str="",addendum:str="",)->Callable[[T],T]:"""Decorator to mark a function, a class, or a property as beta. When marking a classmethod, a staticmethod, or a property, the ``@beta`` decorator should go *under* ``@classmethod`` and ``@staticmethod`` (i.e., `beta` should directly decorate the underlying callable), but *over* ``@property``. When marking a class ``C`` intended to be used as a base class in a multiple inheritance hierarchy, ``C`` *must* define an ``__init__`` method (if ``C`` instead inherited its ``__init__`` from its own base class, then ``@beta`` would mess up ``__init__`` inheritance when installing its own (annotation-emitting) ``C.__init__``). Args: message : str, optional Override the default beta message. The %(since)s, %(name)s, %(alternative)s, %(obj_type)s, %(addendum)s, and %(removal)s format specifiers will be replaced by the values of the respective arguments passed to this function. name : str, optional The name of the beta object. obj_type : str, optional The object type being beta. addendum : str, optional Additional text appended directly to the final message. Examples: .. code-block:: python @beta def the_function_to_annotate(): pass """defbeta(obj:T,*,_obj_type:str=obj_type,_name:str=name,_message:str=message,_addendum:str=addendum,)->T:"""Implementation of the decorator returned by `beta`."""defemit_warning()->None:"""Emit the warning."""warn_beta(message=_message,name=_name,obj_type=_obj_type,addendum=_addendum,)warned=Falsedefwarning_emitting_wrapper(*args:Any,**kwargs:Any)->Any:"""Wrapper for the original wrapped callable that emits a warning. Args: *args: The positional arguments to the function. **kwargs: The keyword arguments to the function. Returns: The return value of the function being wrapped. """nonlocalwarnedifnotwarnedandnotis_caller_internal():warned=Trueemit_warning()returnwrapped(*args,**kwargs)asyncdefawarning_emitting_wrapper(*args:Any,**kwargs:Any)->Any:"""Same as warning_emitting_wrapper, but for async functions."""nonlocalwarnedifnotwarnedandnotis_caller_internal():warned=Trueemit_warning()returnawaitwrapped(*args,**kwargs)ifisinstance(obj,type):ifnot_obj_type:_obj_type="class"wrapped=obj.__init__# type: ignore[misc]_name=_nameorobj.__qualname__old_doc=obj.__doc__deffinalize(wrapper:Callable[...,Any],new_doc:str)->T:# noqa: ARG001"""Finalize the annotation of a class."""# Can't set new_doc on some extension objects.withcontextlib.suppress(AttributeError):obj.__doc__=new_docdefwarn_if_direct_instance(self:Any,*args:Any,**kwargs:Any)->Any:"""Warn that the class is in beta."""nonlocalwarnedifnotwarnedandtype(self)isobjandnotis_caller_internal():warned=Trueemit_warning()returnwrapped(self,*args,**kwargs)obj.__init__=functools.wraps(obj.__init__)(# type: ignore[misc]warn_if_direct_instance)returncast("T",obj)elifisinstance(obj,property):# note(erick): this block doesn't seem to be used?ifnot_obj_type:_obj_type="attribute"wrapped=None_name=_nameorobj.fget.__qualname__old_doc=obj.__doc__class_BetaProperty(property):"""A beta property."""def__init__(self,fget:Union[Callable[[Any],Any],None]=None,fset:Union[Callable[[Any,Any],None],None]=None,fdel:Union[Callable[[Any],None],None]=None,doc:Union[str,None]=None,)->None:super().__init__(fget,fset,fdel,doc)self.__orig_fget=fgetself.__orig_fset=fsetself.__orig_fdel=fdeldef__get__(self,instance:Any,owner:Union[type,None]=None)->Any:ifinstanceisnotNoneorownerisnotNone:emit_warning()returnself.fget(instance)def__set__(self,instance:Any,value:Any)->None:ifinstanceisnotNone:emit_warning()returnself.fset(instance,value)def__delete__(self,instance:Any)->None:ifinstanceisnotNone:emit_warning()returnself.fdel(instance)def__set_name__(self,owner:Union[type,None],set_name:str)->None:nonlocal_nameif_name=="<lambda>":_name=set_namedeffinalize(wrapper:Callable[...,Any],new_doc:str)->Any:# noqa: ARG001"""Finalize the property."""return_BetaProperty(fget=obj.fget,fset=obj.fset,fdel=obj.fdel,doc=new_doc)else:_name=_nameorobj.__qualname__ifnot_obj_type:# edge case: when a function is within another function# within a test, this will call it a "method" not a "function"_obj_type="function"if"."notin_nameelse"method"wrapped=objold_doc=wrapped.__doc__deffinalize(wrapper:Callable[...,Any],new_doc:str)->T:"""Wrap the wrapped function using the wrapper and update the docstring. Args: wrapper: The wrapper function. new_doc: The new docstring. Returns: The wrapped function. """wrapper=functools.wraps(wrapped)(wrapper)wrapper.__doc__=new_docreturncast("T",wrapper)old_doc=inspect.cleandoc(old_docor"").strip("\n")or""components=[message,addendum]details=" ".join([component.strip()forcomponentincomponentsifcomponent])new_doc=f".. beta::\n{details}\n\n{old_doc}\n"ifinspect.iscoroutinefunction(obj):returnfinalize(awarning_emitting_wrapper,new_doc)returnfinalize(warning_emitting_wrapper,new_doc)returnbeta@contextlib.contextmanagerdefsuppress_langchain_beta_warning()->Generator[None,None,None]:"""Context manager to suppress LangChainDeprecationWarning."""withwarnings.catch_warnings():warnings.simplefilter("ignore",LangChainBetaWarning)yielddefwarn_beta(*,message:str="",name:str="",obj_type:str="",addendum:str="",)->None:"""Display a standardized beta annotation. Arguments: message : str, optional Override the default beta message. The %(name)s, %(obj_type)s, %(addendum)s format specifiers will be replaced by the values of the respective arguments passed to this function. name : str, optional The name of the annotated object. obj_type : str, optional The object type being annotated. addendum : str, optional Additional text appended directly to the final message. """ifnotmessage:message=""ifobj_type:message+=f"The {obj_type} `{name}`"else:message+=f"`{name}`"message+=" is in beta. It is actively being worked on, so the API may change."ifaddendum:message+=f" {addendum}"warning=LangChainBetaWarning(message)warnings.warn(warning,category=LangChainBetaWarning,stacklevel=4)defsurface_langchain_beta_warnings()->None:"""Unmute LangChain beta warnings."""warnings.filterwarnings("default",category=LangChainBetaWarning,)