Source code for langchain_community.embeddings.dashscope
from __future__ import annotations
import logging
from typing import (
Any,
Callable,
Dict,
List,
Optional,
)
from langchain_core.embeddings import Embeddings
from langchain_core.pydantic_v1 import BaseModel, Extra, root_validator
from langchain_core.utils import get_from_dict_or_env
from requests.exceptions import HTTPError
from tenacity import (
before_sleep_log,
retry,
retry_if_exception_type,
stop_after_attempt,
wait_exponential,
)
logger = logging.getLogger(__name__)
def _create_retry_decorator(embeddings: DashScopeEmbeddings) -> Callable[[Any], Any]:
multiplier = 1
min_seconds = 1
max_seconds = 4
# Wait 2^x * 1 second between each retry starting with
# 1 seconds, then up to 4 seconds, then 4 seconds afterwards
return retry(
reraise=True,
stop=stop_after_attempt(embeddings.max_retries),
wait=wait_exponential(multiplier, min=min_seconds, max=max_seconds),
retry=(retry_if_exception_type(HTTPError)),
before_sleep=before_sleep_log(logger, logging.WARNING),
)
[docs]def embed_with_retry(embeddings: DashScopeEmbeddings, **kwargs: Any) -> Any:
"""使用tenacity来重试嵌入调用。"""
retry_decorator = _create_retry_decorator(embeddings)
@retry_decorator
def _embed_with_retry(**kwargs: Any) -> Any:
result = []
i = 0
input_data = kwargs["input"]
while i < len(input_data):
kwargs["input"] = input_data[i : i + 25]
resp = embeddings.client.call(**kwargs)
if resp.status_code == 200:
result += resp.output["embeddings"]
elif resp.status_code in [400, 401]:
raise ValueError(
f"status_code: {resp.status_code} \n "
f"code: {resp.code} \n message: {resp.message}"
)
else:
raise HTTPError(
f"HTTP error occurred: status_code: {resp.status_code} \n "
f"code: {resp.code} \n message: {resp.message}",
response=resp,
)
i += 25
return result
return _embed_with_retry(**kwargs)
[docs]class DashScopeEmbeddings(BaseModel, Embeddings):
"""DashScope嵌入模型。
要使用,您应该已安装``dashscope`` python包,并且
环境变量``DASHSCOPE_API_KEY``设置为您的API密钥,或将其传递
作为构造函数的命名参数。
示例:
.. code-block:: python
from langchain_community.embeddings import DashScopeEmbeddings
embeddings = DashScopeEmbeddings(dashscope_api_key="my-api-key")
示例:
.. code-block:: python
import os
os.environ["DASHSCOPE_API_KEY"] = "your DashScope API KEY"
from langchain_community.embeddings.dashscope import DashScopeEmbeddings
embeddings = DashScopeEmbeddings(
model="text-embedding-v1",
)
text = "This is a test query."
query_result = embeddings.embed_query(text)"""
client: Any #: :meta private:
"""DashScope客户端。"""
model: str = "text-embedding-v1"
dashscope_api_key: Optional[str] = None
max_retries: int = 5
"""生成时最大的重试次数。"""
class Config:
"""此pydantic对象的配置。"""
extra = Extra.forbid
@root_validator()
def validate_environment(cls, values: Dict) -> Dict:
import dashscope
"""Validate that api key and python package exists in environment."""
values["dashscope_api_key"] = get_from_dict_or_env(
values, "dashscope_api_key", "DASHSCOPE_API_KEY"
)
dashscope.api_key = values["dashscope_api_key"]
try:
import dashscope
values["client"] = dashscope.TextEmbedding
except ImportError:
raise ImportError(
"Could not import dashscope python package. "
"Please install it with `pip install dashscope`."
)
return values
[docs] def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""调用DashScope的嵌入端点以嵌入搜索文档。
参数:
texts: 要嵌入的文本列表。
chunk_size: 嵌入的块大小。如果为None,则使用类指定的块大小。
返回:
每个文本的嵌入列表。
"""
embeddings = embed_with_retry(
self, input=texts, text_type="document", model=self.model
)
embedding_list = [item["embedding"] for item in embeddings]
return embedding_list
[docs] def embed_query(self, text: str) -> List[float]:
"""调用DashScope的嵌入端点以嵌入查询文本。
参数:
text:要嵌入的文本。
返回:
文本的嵌入。
"""
embedding = embed_with_retry(
self, input=text, text_type="query", model=self.model
)[0]["embedding"]
return embedding