"""用于废弃LangChain API 的辅助函数。
此模块是从matplotlib的_api/deprecation.py模块调整而来：
https://github.com/matplotlib/matplotlib/blob/main/lib/matplotlib/_api/deprecation.py
.. 警告::
该模块仅供内部使用。请勿在您自己的代码中使用。
我们可能随时更改API，而不发出警告。
"""
import contextlib
import functools
import inspect
import warnings
from typing import Any, Callable, Generator, Type, TypeVar, Union, cast
from langchain_core._api.internal import is_caller_internal
class LangChainDeprecationWarning(DeprecationWarning):
"""一个用于向LangChain用户发出弃用警告的类。"""
class LangChainPendingDeprecationWarning(PendingDeprecationWarning):
"""一个用于向LangChain用户发出弃用警告的类。"""
# PUBLIC API
T = TypeVar("T", bound=Union[Type, Callable[..., Any]])
def deprecated(
since: str,
*,
message: str = "",
name: str = "",
alternative: str = "",
alternative_import: str = "",
pending: bool = False,
obj_type: str = "",
addendum: str = "",
removal: str = "",
package: str = "",
) -> Callable[[T], T]:
"""装饰器用于标记函数、类或属性为已弃用。
当将classmethod、staticmethod或属性标记为已弃用时，应该将``@deprecated``装饰器放在``@classmethod``和``@staticmethod``之*下*（即，`deprecated`应该直接装饰基础可调用对象），但应该放在``@property``之*上*。
当将一个类``C``标记为已弃用，该类预期用作多重继承层次结构中的基类时，``C`` *必须*定义一个``__init__``方法（如果``C``从自己的基类继承了``__init__``，那么当安装自己的（发出弃用警告的）``C.__init__``时，``@deprecated``会破坏``__init__``的继承）。
参数与`warn_deprecated`相同，除了如果装饰一个类，则*obj_type*默认为'class'，如果装饰一个属性，则为'attribute'，否则为'function'。
参数:
since : str
此API变为已弃用的版本。
message : str, optional
覆盖默认的弃用消息。格式说明符%(since)s、%(name)s、%(alternative)s、%(obj_type)s、%(addendum)s和%(removal)s将被传递给此函数的相应参数的值替换。
name : str, optional
被弃用对象的名称。
alternative : str, optional
用户可以在弃用API的位置使用的替代API。如果提供了替代API，弃用警告将告知用户有关此替代API。
pending : bool, optional
如果为True，则使用PendingDeprecationWarning而不是DeprecationWarning。不能与removal一起使用。
obj_type : str, optional
被弃用对象的类型。
addendum : str, optional
直接附加到最终消息的附加文本。
removal : str, optional
预期的移除版本。默认情况下（空字符串），移除版本会根据since自动计算。设置为其他假值以不安排移除日期。不能与pending一起使用。
示例
--------
.. code-block:: python
@deprecated('1.4.0')
def the_function_to_deprecate():
pass
"""
def deprecate(
obj: T,
*,
_obj_type: str = obj_type,
_name: str = name,
_message: str = message,
_alternative: str = alternative,
_alternative_import: str = alternative_import,
_pending: bool = pending,
_addendum: str = addendum,
_package: str = package,
) -> T:
"""实现了`deprecated`返回的装饰器。"""
def emit_warning() -> None:
"""发出警告。"""
warn_deprecated(
since,
message=_message,
name=_name,
alternative=_alternative,
alternative_import=_alternative_import,
pending=_pending,
obj_type=_obj_type,
addendum=_addendum,
removal=removal,
package=_package,
)
warned = False
def warning_emitting_wrapper(*args: Any, **kwargs: Any) -> Any:
"""对原始包装的可调用对象进行包装，发出警告。
参数：
*args: 函数的位置参数。
**kwargs: 函数的关键字参数。
返回值：
被包装函数的返回值。
"""
nonlocal warned
if not warned and not is_caller_internal():
warned = True
emit_warning()
return wrapped(*args, **kwargs)
async def awarning_emitting_wrapper(*args: Any, **kwargs: Any) -> Any:
"""与warning_emitting_wrapper相同，但适用于异步函数。"""
nonlocal warned
if not warned and not is_caller_internal():
warned = True
emit_warning()
return await wrapped(*args, **kwargs)
_package = _package or obj.__module__.split(".")[0].replace("_", "-")
if isinstance(obj, type):
if not _obj_type:
_obj_type = "class"
wrapped = obj.__init__ # type: ignore
_name = _name or obj.__qualname__
old_doc = obj.__doc__
def finalize(wrapper: Callable[..., Any], new_doc: str) -> T:
"""完成对一个类的弃用。"""
try:
obj.__doc__ = new_doc
except AttributeError: # Can't set on some extension objects.
pass
def warn_if_direct_instance(
self: Any, *args: Any, **kwargs: Any
) -> Any:
"""警告该类处于测试阶段。"""
nonlocal warned
if not warned and type(self) is obj and not is_caller_internal():
warned = True
emit_warning()
return wrapped(self, *args, **kwargs)
obj.__init__ = functools.wraps(obj.__init__)( # type: ignore[misc]
warn_if_direct_instance
)
return cast(T, obj)
elif isinstance(obj, property):
if not _obj_type:
_obj_type = "attribute"
wrapped = None
_name = _name or obj.fget.__qualname__
old_doc = obj.__doc__
class _deprecated_property(property):
"""一个不推荐使用的属性。"""
def __init__(self, fget=None, fset=None, fdel=None, doc=None):
super().__init__(fget, fset, fdel, doc)
self.__orig_fget = fget
self.__orig_fset = fset
self.__orig_fdel = fdel
def __get__(self, instance, owner=None):
if instance is not None or owner is not None:
emit_warning()
return self.fget(instance)
def __set__(self, instance, value):
if instance is not None:
emit_warning()
return self.fset(instance, value)
def __delete__(self, instance):
if instance is not None:
emit_warning()
return self.fdel(instance)
def __set_name__(self, owner, set_name):
nonlocal _name
if _name == "<lambda>":
_name = set_name
def finalize(wrapper: Callable[..., Any], new_doc: str) -> Any:
"""完成属性的设置。"""
return _deprecated_property(
fget=obj.fget, fset=obj.fset, fdel=obj.fdel, doc=new_doc
)
else:
_name = _name or obj.__qualname__
if not _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 "." not in _name else "method"
wrapped = obj
old_doc = wrapped.__doc__
def finalize(wrapper: Callable[..., Any], new_doc: str) -> T:
"""使用包装器(wrapper)包装被包装的函数，并更新文档字符串。
参数:
wrapper: 包装器函数。
new_doc: 新的文档字符串。
返回:
包装后的函数。
"""
wrapper = functools.wraps(wrapped)(wrapper)
wrapper.__doc__ = new_doc
return cast(T, wrapper)
old_doc = inspect.cleandoc(old_doc or "").strip("\n")
# old_doc can be None
if not old_doc:
old_doc = ""
# Modify the docstring to include a deprecation notice.
notes_header = "\nNotes\n-----"
components = [
message,
f"Use {alternative} instead." if alternative else "",
addendum,
]
details = " ".join([component.strip() for component in components if component])
package = (
_package or _name.split(".")[0].replace("_", "-") if "." in _name else None
)
since_str = f"{package}=={since}" if package else since
new_doc = (
f"[*Deprecated*] {old_doc}\n"
f"{notes_header if notes_header not in old_doc else ''}\n"
f".. deprecated:: {since_str}\n"
f" {details}"
)
if inspect.iscoroutinefunction(obj):
finalized = finalize(awarning_emitting_wrapper, new_doc)
else:
finalized = finalize(warning_emitting_wrapper, new_doc)
return cast(T, finalized)
return deprecate
@contextlib.contextmanager
def suppress_langchain_deprecation_warning() -> Generator[None, None, None]:
"""上下文管理器，用于抑制LangChainDeprecationWarning。"""
with warnings.catch_warnings():
warnings.simplefilter("ignore", LangChainDeprecationWarning)
warnings.simplefilter("ignore", LangChainPendingDeprecationWarning)
yield
def warn_deprecated(
since: str,
*,
message: str = "",
name: str = "",
alternative: str = "",
alternative_import: str = "",
pending: bool = False,
obj_type: str = "",
addendum: str = "",
removal: str = "",
package: str = "",
) -> None:
"""显示一个标准化的弃用。
参数：
since : str
此API被弃用的发布版本。
message : str, optional
覆盖默认的弃用消息。%(since)s, %(name)s, %(alternative)s, %(obj_type)s, %(addendum)s,
和 %(removal)s 格式说明符将被传递给此函数的相应参数的值替换。
name : str, optional
被弃用对象的名称。
alternative : str, optional
用户可以在被弃用API的位置使用的替代API。如果提供，弃用警告将告知用户有关此替代项。
pending : bool, optional
如果为True，则使用PendingDeprecationWarning而不是DeprecationWarning。不能与removal一起使用。
obj_type : str, optional
被弃用的对象类型。
addendum : str, optional
直接附加到最终消息的附加文本。
removal : str, optional
预期的移除版本。默认情况下（空字符串），将从since自动计算移除版本。设置为其他假值以不安排移除日期。不能与pending一起使用。
"""
if pending and removal:
raise ValueError("A pending deprecation cannot have a scheduled removal")
if alternative and alternative_import:
raise ValueError("Cannot specify both alternative and alternative_import")
if alternative_import and "." not in alternative_import:
raise ValueError("alternative_import must be a fully qualified module path")
if not pending:
if not removal:
removal = f"in {removal}" if removal else "within ?? minor releases"
raise NotImplementedError(
f"Need to determine which default deprecation schedule to use. "
f"{removal}"
)
else:
removal = f"in {removal}"
if not message:
message = ""
_package = (
package or name.split(".")[0].replace("_", "-")
if "." in name
else "LangChain"
)
if obj_type:
message += f"The {obj_type} `{name}`"
else:
message += f"`{name}`"
if pending:
message += " will be deprecated in a future version"
else:
message += f" was deprecated in {_package} {since}"
if removal:
message += f" and will be removed {removal}"
if alternative_import:
alt_package = alternative_import.split(".")[0].replace("_", "-")
if alt_package == _package:
message += f". Use {alternative_import} instead."
else:
alt_module, alt_name = alternative_import.rsplit(".", 1)
message += (
f". An updated version of the {obj_type} exists in the "
f"{alt_package} package and should be used instead. To use it run "
f"`pip install -U {alt_package}` and import as "
f"`from {alt_module} import {alt_name}`."
)
elif alternative:
message += f". Use {alternative} instead."
if addendum:
message += f" {addendum}"
warning_cls = (
LangChainPendingDeprecationWarning if pending else LangChainDeprecationWarning
)
warning = warning_cls(message)
warnings.warn(warning, category=LangChainDeprecationWarning, stacklevel=2)
def surface_langchain_deprecation_warnings() -> None:
"""取消静音LangChain弃用警告。"""
warnings.filterwarnings(
"default",
category=LangChainPendingDeprecationWarning,
)
warnings.filterwarnings(
"default",
category=LangChainDeprecationWarning,
)