"""用于Blob和Blob Loader的模式。
目标是促进内容加载与内容解析代码的解耦。
此外，内容加载代码应默认提供延迟加载接口。
"""
from __future__ import annotations
import contextlib
import mimetypes
from abc import ABC, abstractmethod
from io import BufferedReader, BytesIO
from pathlib import PurePath
from typing import Any, Dict, Generator, Iterable, Mapping, Optional, Union, cast
from langchain_core.pydantic_v1 import BaseModel, Field, root_validator
PathLike = Union[str, PurePath]
[docs]class Blob(BaseModel):
"""Blob代表通过引用或值表示的原始数据。
提供一个接口，以不同的表示形式实现Blob，并有助于将数据加载程序的开发与原始数据的下游解析分离。
灵感来自：https://developer.mozilla.org/en-US/docs/Web/API/Blob
"""
data: Union[bytes, str, None]
"""与blob相关的原始数据。"""
mimetype: Optional[str] = None
"""MimeType不要与文件扩展名混淆。"""
encoding: str = "utf-8"
"""如果将字节解码为字符串，则使用的编码。
如果解码为字符串，则使用utf-8作为默认编码。"""
path: Optional[PathLike] = None
"""原始内容的位置。"""
metadata: Dict[str, Any] = Field(default_factory=dict)
"""关于blob的元数据（例如，来源）"""
class Config:
arbitrary_types_allowed = True
frozen = True
@property
def source(self) -> Optional[str]:
"""如果已知blob的源位置，则返回字符串形式，否则返回None。
如果blob关联了路径，则默认为路径位置。
除非通过名为"source"的元数据字段显式设置，否则将使用该值。
"""
if self.metadata and "source" in self.metadata:
return cast(Optional[str], self.metadata["source"])
return str(self.path) if self.path else None
@root_validator(pre=True)
def check_blob_is_valid(cls, values: Mapping[str, Any]) -> Mapping[str, Any]:
"""验证是否提供了数据或路径。"""
if "data" not in values and "path" not in values:
raise ValueError("Either data or path must be provided")
return values
[docs] def as_string(self) -> str:
"""将数据读取为字符串。"""
if self.data is None and self.path:
with open(str(self.path), "r", encoding=self.encoding) as f:
return f.read()
elif isinstance(self.data, bytes):
return self.data.decode(self.encoding)
elif isinstance(self.data, str):
return self.data
else:
raise ValueError(f"Unable to get string for blob {self}")
[docs] def as_bytes(self) -> bytes:
"""以字节形式读取数据。"""
if isinstance(self.data, bytes):
return self.data
elif isinstance(self.data, str):
return self.data.encode(self.encoding)
elif self.data is None and self.path:
with open(str(self.path), "rb") as f:
return f.read()
else:
raise ValueError(f"Unable to get bytes for blob {self}")
[docs] @contextlib.contextmanager
def as_bytes_io(self) -> Generator[Union[BytesIO, BufferedReader], None, None]:
"""以字节流的形式读取数据。"""
if isinstance(self.data, bytes):
yield BytesIO(self.data)
elif self.data is None and self.path:
with open(str(self.path), "rb") as f:
yield f
else:
raise NotImplementedError(f"Unable to convert blob {self}")
[docs] @classmethod
def from_path(
cls,
path: PathLike,
*,
encoding: str = "utf-8",
mime_type: Optional[str] = None,
guess_type: bool = True,
metadata: Optional[dict] = None,
) -> Blob:
"""从类似路径对象中加载 blob。
参数:
path: 要读取的文件的路径对象
encoding: 如果将字节解码为字符串，则使用的编码
mime_type: 如果提供，将设置为数据的 mime 类型
guess_type: 如果为 True，则将从文件扩展名猜测 mime 类型，
如果未提供 mime 类型
metadata: 与 blob 关联的元数据
返回:
Blob 实例
"""
if mime_type is None and guess_type:
_mimetype = mimetypes.guess_type(path)[0] if guess_type else None
else:
_mimetype = mime_type
# We do not load the data immediately, instead we treat the blob as a
# reference to the underlying data.
return cls(
data=None,
mimetype=_mimetype,
encoding=encoding,
path=path,
metadata=metadata if metadata is not None else {},
)
[docs] @classmethod
def from_data(
cls,
data: Union[str, bytes],
*,
encoding: str = "utf-8",
mime_type: Optional[str] = None,
path: Optional[str] = None,
metadata: Optional[dict] = None,
) -> Blob:
"""从内存数据初始化blob。
参数：
data：与blob关联的内存数据
encoding：解码字节为字符串时要使用的编码
mime_type：如果提供，将设置为数据的mime类型
path：如果提供，将设置为数据来源的路径
metadata：要与blob关联的元数据
返回：
Blob实例
"""
return cls(
data=data,
mimetype=mime_type,
encoding=encoding,
path=path,
metadata=metadata if metadata is not None else {},
)
def __repr__(self) -> str:
"""定义blob表示形式。"""
str_repr = f"Blob {id(self)}"
if self.source:
str_repr += f" {self.source}"
return str_repr
[docs]class BlobLoader(ABC):
"""用于blob加载器实现的抽象接口。
实现者应该能够根据一些条件从存储系统加载原始内容，并将原始内容作为blob流进行惰性返回。
"""
[docs] @abstractmethod
def yield_blobs(
self,
) -> Iterable[Blob]:
"""一个用于表示LangChain的Blob对象的原始数据的惰性加载器。
返回：
一个blob的生成器
"""