Source code for langchain_community.embeddings.anyscale

"""Anyscale嵌入包装器。"""
from __future__ import annotations

from typing import Dict

from langchain_core.pydantic_v1 import Field, SecretStr, root_validator
from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env

from langchain_community.embeddings.openai import OpenAIEmbeddings
from langchain_community.utils.openai import is_openai_v1

DEFAULT_API_BASE = "https://api.endpoints.anyscale.com/v1"
DEFAULT_MODEL = "thenlper/gte-large"


[docs]class AnyscaleEmbeddings(OpenAIEmbeddings): """`Anyscale` 嵌入式 API。""" anyscale_api_key: SecretStr = Field(default=None) """任何规模的端点 API 密钥。""" model: str = Field(default=DEFAULT_MODEL) """要使用的模型名称。""" anyscale_api_base: str = Field(default=DEFAULT_API_BASE) """API请求的基础URL路径。""" tiktoken_enabled: bool = False """将此设置为False,用于嵌入API的非OpenAI实现。""" embedding_ctx_length: int = 500 """一次嵌入的最大令牌数。""" @property def lc_secrets(self) -> Dict[str, str]: return { "anyscale_api_key": "ANYSCALE_API_KEY", } @root_validator() def validate_environment(cls, values: dict) -> dict: """验证环境中是否存在API密钥和Python包。""" values["anyscale_api_key"] = convert_to_secret_str( get_from_dict_or_env( values, "anyscale_api_key", "ANYSCALE_API_KEY", ) ) values["anyscale_api_base"] = get_from_dict_or_env( values, "anyscale_api_base", "ANYSCALE_API_BASE", default=DEFAULT_API_BASE, ) try: import openai except ImportError: raise ImportError( "Could not import openai python package. " "Please install it with `pip install openai`." ) if is_openai_v1(): # For backwards compatibility. client_params = { "api_key": values["anyscale_api_key"].get_secret_value(), "base_url": values["anyscale_api_base"], } values["client"] = openai.OpenAI(**client_params).embeddings else: values["openai_api_base"] = values["anyscale_api_base"] values["openai_api_key"] = values["anyscale_api_key"].get_secret_value() values["client"] = openai.Embedding return values @property def _llm_type(self) -> str: return "anyscale-embedding"