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.")