""" **Store** 实现了键-值存储和存储辅助功能。
该模块提供了符合简单键-值接口的各种键-值存储的实现。
这些存储的主要目标是支持缓存的实现。
"""
from abc import ABC, abstractmethod
from typing import (
Any,
AsyncIterator,
Dict,
Generic,
Iterator,
List,
Optional,
Sequence,
Tuple,
TypeVar,
Union,
)
from langchain_core.exceptions import LangChainException
from langchain_core.runnables import run_in_executor
K = TypeVar("K")
V = TypeVar("V")
[docs]class BaseStore(Generic[K, V], ABC):
"""用于键值存储的抽象接口。"""
[docs] @abstractmethod
def mget(self, keys: Sequence[K]) -> List[Optional[V]]:
"""获取与给定键相关联的值。
参数：
keys（Sequence[K]）：键的序列。
返回：
与键相关联的可选值序列。
如果未找到键，则相应的值将为None。
"""
[docs] async def amget(self, keys: Sequence[K]) -> List[Optional[V]]:
"""获取与给定键相关联的值。
参数：
keys（Sequence[K]）：键的序列。
返回：
与键相关联的可选值序列。
如果未找到键，则相应的值将为None。
"""
return await run_in_executor(None, self.mget, keys)
[docs] @abstractmethod
def mset(self, key_value_pairs: Sequence[Tuple[K, V]]) -> None:
"""设置给定键的值。
参数：
key_value_pairs（Sequence[Tuple[K，V]]）：键值对的序列。
"""
[docs] async def amset(self, key_value_pairs: Sequence[Tuple[K, V]]) -> None:
"""设置给定键的值。
参数：
key_value_pairs（Sequence[Tuple[K，V]]）：键值对的序列。
"""
return await run_in_executor(None, self.mset, key_value_pairs)
[docs] @abstractmethod
def mdelete(self, keys: Sequence[K]) -> None:
"""删除给定的键及其关联的值。
参数：
keys（Sequence[K]）：要删除的键的序列。
"""
[docs] async def amdelete(self, keys: Sequence[K]) -> None:
"""删除给定的键及其关联的值。
参数：
keys（Sequence[K]）：要删除的键的序列。
"""
return await run_in_executor(None, self.mdelete, keys)
[docs] @abstractmethod
def yield_keys(
self, *, prefix: Optional[str] = None
) -> Union[Iterator[K], Iterator[str]]:
"""获取与给定前缀匹配的键的迭代器。
参数：
prefix（str）：要匹配的前缀。
返回：
Iterator[K | str]：一个迭代器，用于匹配给定前缀的键。
该方法允许返回一个K或str类型的迭代器，具体取决于给定存储的情况。
"""
[docs] async def ayield_keys(
self, *, prefix: Optional[str] = None
) -> Union[AsyncIterator[K], AsyncIterator[str]]:
"""获取与给定前缀匹配的键的迭代器。
参数：
prefix（str）：要匹配的前缀。
返回：
Iterator[K | str]：一个迭代器，用于匹配给定前缀的键。
该方法允许返回一个K或str类型的迭代器，具体取决于给定存储的情况。
"""
iterator = await run_in_executor(None, self.yield_keys, prefix=prefix)
done = object()
while True:
item = await run_in_executor(None, lambda it: next(it, done), iterator)
if item is done:
break
yield item # type: ignore[misc]
ByteStore = BaseStore[str, bytes]
[docs]class InMemoryBaseStore(BaseStore[str, V], Generic[V]):
"""使用字典实现的BaseStore的内存实现。
属性：
store (Dict[str, Any]): 存储键值对的基础字典。
示例：
.. code-block:: python
from langchain.storage import InMemoryStore
store = InMemoryStore()
store.mset([('key1', 'value1'), ('key2', 'value2')])
store.mget(['key1', 'key2'])
# ['value1', 'value2']
store.mdelete(['key1'])
list(store.yield_keys())
# ['key2']
list(store.yield_keys(prefix='k'))
# ['key2']
"""
[docs] def __init__(self) -> None:
"""初始化一个空的存储。"""
self.store: Dict[str, V] = {}
[docs] def mget(self, keys: Sequence[str]) -> List[Optional[V]]:
"""获取与给定键相关联的值。
参数：
keys（Sequence[str]）：键的序列。
返回：
与键相关联的可选值序列。
如果未找到键，则相应的值将为None。
"""
return [self.store.get(key) for key in keys]
[docs] async def amget(self, keys: Sequence[str]) -> List[Optional[V]]:
"""获取与给定键相关联的值。
参数：
keys（Sequence[str]）：键的序列。
返回：
与键相关联的可选值序列。
如果未找到键，则相应的值将为None。
"""
return self.mget(keys)
[docs] def mset(self, key_value_pairs: Sequence[Tuple[str, V]]) -> None:
"""为给定的键设置值。
参数：
key_value_pairs（Sequence[Tuple[str, V]）：键值对的序列。
返回：
无
"""
for key, value in key_value_pairs:
self.store[key] = value
[docs] async def amset(self, key_value_pairs: Sequence[Tuple[str, V]]) -> None:
"""为给定的键设置值。
参数：
key_value_pairs（Sequence[Tuple[str, V]）：键值对的序列。
返回：
无
"""
return self.mset(key_value_pairs)
[docs] def mdelete(self, keys: Sequence[str]) -> None:
"""删除给定的键及其关联的值。
参数：
keys（Sequence[str]）：要删除的键的序列。
"""
for key in keys:
if key in self.store:
del self.store[key]
[docs] async def amdelete(self, keys: Sequence[str]) -> None:
"""删除给定的键及其关联的值。
参数：
keys（Sequence[str]）：要删除的键的序列。
"""
self.mdelete(keys)
[docs] def yield_keys(self, prefix: Optional[str] = None) -> Iterator[str]:
"""获取与给定前缀匹配的键的迭代器。
参数：
prefix（str，可选）：要匹配的前缀。默认为None。
返回：
Iterator[str]：一个迭代器，用于匹配给定前缀的键。
"""
if prefix is None:
yield from self.store.keys()
else:
for key in self.store.keys():
if key.startswith(prefix):
yield key
[docs] async def ayield_keys(self, prefix: Optional[str] = None) -> AsyncIterator[str]:
"""获取与给定前缀匹配的键的异步迭代器。
参数：
prefix（str，可选）：要匹配的前缀。默认为None。
返回：
AsyncIterator[str]：与给定前缀匹配的键的异步迭代器。
"""
if prefix is None:
for key in self.store.keys():
yield key
else:
for key in self.store.keys():
if key.startswith(prefix):
yield key
InMemoryStore = InMemoryBaseStore[Any]
InMemoryByteStore = InMemoryBaseStore[bytes]
[docs]class InvalidKeyException(LangChainException):
"""当键无效时引发；例如，使用了不正确的字符。"""