Source code for langchain_community.embeddings.spacy_embeddings
import importlib.util
from typing import Any, Dict, List, Optional
from langchain_core.embeddings import Embeddings
from langchain_core.pydantic_v1 import BaseModel, Extra, root_validator
[docs]class SpacyEmbeddings(BaseModel, Embeddings):
"""spaCy模型的嵌入。
属性:
model_name (str): spaCy模型的名称。
nlp (Any): 加载到内存中的spaCy模型。
方法:
embed_documents(texts: List[str]) -> List[List[float]]:
为文档列表生成嵌入。
embed_query(text: str) -> List[float]:
为单个文本生成嵌入。"""
model_name: str = "en_core_web_sm"
nlp: Optional[Any] = None
class Config:
"""此pydantic对象的配置。"""
extra = Extra.forbid # Forbid extra attributes during model initialization
@root_validator(pre=True)
def validate_environment(cls, values: Dict) -> Dict:
"""验证spaCy包和模型是否已安装。
参数:
values(字典):提供给类构造函数的值。
返回:
经过验证的值。
引发:
ValueError:如果spaCy包或模型未安装。
"""
if values.get("model_name") is None:
values["model_name"] = "en_core_web_sm"
model_name = values.get("model_name")
# Check if the spaCy package is installed
if importlib.util.find_spec("spacy") is None:
raise ValueError(
"SpaCy package not found. "
"Please install it with `pip install spacy`."
)
try:
# Try to load the spaCy model
import spacy
values["nlp"] = spacy.load(model_name)
except OSError:
# If the model is not found, raise a ValueError
raise ValueError(
f"SpaCy model '{model_name}' not found. "
f"Please install it with"
f" `python -m spacy download {model_name}`"
"or provide a valid spaCy model name."
)
return values # Return the validated values
[docs] def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""为文档列表生成嵌入。
参数:
texts(List[str]):要生成嵌入的文档。
返回:
一个嵌入列表,每个文档对应一个嵌入。
"""
return [self.nlp(text).vector.tolist() for text in texts] # type: ignore[misc]
[docs] def embed_query(self, text: str) -> List[float]:
"""为单个文本生成嵌入。
参数:
text (str): 要生成嵌入的文本。
返回:
文本的嵌入。
"""
return self.nlp(text).vector.tolist() # type: ignore[misc]
[docs] async def aembed_documents(self, texts: List[str]) -> List[List[float]]:
"""异步生成文档列表的嵌入。
该方法未实现并引发NotImplementedError。
参数:
texts(List[str]):要生成嵌入的文档列表。
引发:
NotImplementedError:该方法未实现。
"""
raise NotImplementedError("Asynchronous embedding generation is not supported.")
[docs] async def aembed_query(self, text: str) -> List[float]:
"""异步生成单个文本片段的嵌入。
此方法未实现,并引发NotImplementedError。
参数:
text(str):要生成嵌入的文本。
引发:
NotImplementedError:此方法未实现。
"""
raise NotImplementedError("Asynchronous embedding generation is not supported.")