BaseRagasEmbedding

BaseRagasEmbedding(cache: Optional[CacheInterface] = None)

基类： ABC

Ragas 嵌入实现的现代抽象基类。

该类为使用各种提供商嵌入文本提供一致接口。实现应提供嵌入单条文本的同步和异步方法，批量方法会自动提供。

参数： cache: 可选缓存后端用于缓存嵌入。使用 DiskCacheBackend() 做持久缓存。

源代码位于 src/ragas/embeddings/base.py

def __init__(self, cache: t.Optional[CacheInterface] = None):
    """Initialize embedding with optional caching.

    Args:
        cache: Optional cache backend for caching embeddings.
            Use DiskCacheBackend() for persistent caching.
    """
    self.cache = cache

    if self.cache is not None:
        self.embed_text = cacher(cache_backend=self.cache)(self.embed_text)
        self.aembed_text = cacher(cache_backend=self.cache)(self.aembed_text)

embed_text

embed_text(text: str, **kwargs: Any) -> List[float]

嵌入单条文本。

参数： text: 要嵌入的文本 **kwargs: 嵌入调用的额外参数

返回：表示嵌入的浮点数列表

源代码位于 src/ragas/embeddings/base.py

@abstractmethod
def embed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Embed a single text.

    Args:
        text: The text to embed
        **kwargs: Additional arguments for the embedding call

    Returns:
        List of floats representing the embedding
    """
    pass

aembed_text

aembed_text(text: str, **kwargs: Any) -> List[float]

异步嵌入单条文本。

参数： text: 要嵌入的文本 **kwargs: 嵌入调用的额外参数

返回：表示嵌入的浮点数列表

源代码位于 src/ragas/embeddings/base.py

@abstractmethod
async def aembed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Asynchronously embed a single text.

    Args:
        text: The text to embed
        **kwargs: Additional arguments for the embedding call

    Returns:
        List of floats representing the embedding
    """
    pass

embed_texts

embed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

嵌入多条文本。

默认实现逐条处理文本。可覆盖以实现批处理优化。

参数： texts: 要嵌入的文本列表 **kwargs: 嵌入调用的额外参数

返回：嵌入列表，每个输入文本一个

源代码位于 src/ragas/embeddings/base.py

def embed_texts(self, texts: t.List[str], **kwargs: t.Any) -> t.List[t.List[float]]:
    """Embed multiple texts.

    Default implementation processes texts individually. Override for
    batch optimization.

    Args:
        texts: List of texts to embed
        **kwargs: Additional arguments for the embedding calls

    Returns:
        List of embeddings, one for each input text
    """
    texts = validate_texts(texts)
    return [self.embed_text(text, **kwargs) for text in texts]

aembed_texts

aembed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

异步嵌入多条文本。

默认实现并发处理文本。可覆盖以实现批处理优化。

参数： texts: 要嵌入的文本列表 **kwargs: 嵌入调用的额外参数

返回：嵌入列表，每个输入文本一个

源代码位于 src/ragas/embeddings/base.py

async def aembed_texts(
    self, texts: t.List[str], **kwargs: t.Any
) -> t.List[t.List[float]]:
    """Asynchronously embed multiple texts.

    Default implementation processes texts concurrently. Override for
    batch optimization.

    Args:
        texts: List of texts to embed
        **kwargs: Additional arguments for the embedding calls

    Returns:
        List of embeddings, one for each input text
    """
    texts = validate_texts(texts)
    tasks = [self.aembed_text(text, **kwargs) for text in texts]
    return await asyncio.gather(*tasks)

BaseRagasEmbeddings

BaseRagasEmbeddings(cache: Optional[CacheInterface] = None)

基类： Embeddings, ABC

Ragas embeddings 的抽象基类。

该类扩展 Embeddings 类，提供嵌入文本和管理运行配置的方法。

属性： run_config (RunConfig): 嵌入操作的运行配置。

源代码位于 src/ragas/embeddings/base.py

def __init__(self, cache: t.Optional[CacheInterface] = None):
    super().__init__()
    self.cache = cache
    if self.cache is not None:
        self.embed_query = cacher(cache_backend=self.cache)(self.embed_query)
        self.embed_documents = cacher(cache_backend=self.cache)(
            self.embed_documents
        )
        self.aembed_query = cacher(cache_backend=self.cache)(self.aembed_query)
        self.aembed_documents = cacher(cache_backend=self.cache)(
            self.aembed_documents
        )

embed_text

embed_text(text: str, is_async=True) -> List[float]

嵌入单个文本字符串。

源代码位于 src/ragas/embeddings/base.py

async def embed_text(self, text: str, is_async=True) -> t.List[float]:
    """
    Embed a single text string.
    """
    embs = await self.embed_texts([text], is_async=is_async)
    return embs[0]

embed_texts

embed_texts(texts: List[str], is_async: bool = True) -> List[List[float]]

嵌入多条文本。

源代码位于 src/ragas/embeddings/base.py

async def embed_texts(
    self, texts: t.List[str], is_async: bool = True
) -> t.List[t.List[float]]:
    """
    Embed multiple texts.
    """
    if is_async:
        aembed_documents_with_retry = add_async_retry(
            self.aembed_documents, self.run_config
        )
        return await aembed_documents_with_retry(texts)
    else:
        loop = asyncio.get_event_loop()
        embed_documents_with_retry = add_retry(
            self.embed_documents, self.run_config
        )
        return await loop.run_in_executor(None, embed_documents_with_retry, texts)

set_run_config

set_run_config(run_config: RunConfig)

设置嵌入操作的运行配置。

源代码位于 src/ragas/embeddings/base.py

def set_run_config(self, run_config: RunConfig):
    """
    Set the run configuration for the embedding operations.
    """
    self.run_config = run_config

HuggingfaceEmbeddings

HuggingfaceEmbeddings(cache: Optional[CacheInterface] = None)

基类： BaseRagasEmbeddings

使用预训练模型生成嵌入的 Hugging Face embeddings 类。

该类提供加载并使用 Hugging Face 模型、为文本输入生成嵌入的功能。

参数：

名称	类型	说明	默认值
`model_name`	`str`	要使用的预训练模型名称，默认 DEFAULT_MODEL_NAME。	required
`cache_folder`	`str`	存放下载模型的路径。也可通过 SENTENCE_TRANSFORMERS_HOME 环境变量设置。	required
`model_kwargs`	`dict`	传递给模型的额外关键字参数。	required
`encode_kwargs`	`dict`	传递给编码方法的额外关键字参数。	required

属性：

名称	类型	说明
`model`	`Union[SentenceTransformer, CrossEncoder]`	已加载的 Hugging Face 模型。
`is_cross_encoder`	`bool`	指示模型是否为 cross-encoder 的标志。

方法：

Name	Description
`embed_query`	嵌入单条查询文本。
`embed_documents`	嵌入多份文档。
`predict`	使用 cross-encoder 模型做预测。

说明

该类需要安装 sentence_transformers 和 transformers 包。

示例：

>>> embeddings = HuggingfaceEmbeddings(model_name="bert-base-uncased")
>>> query_embedding = embeddings.embed_query("What is the capital of France?")
>>> doc_embeddings = embeddings.embed_documents(["Paris is the capital of France.", "London is the capital of the UK."])

源代码位于 src/ragas/embeddings/base.py

def __init__(self, cache: t.Optional[CacheInterface] = None):
    super().__init__()
    self.cache = cache
    if self.cache is not None:
        self.embed_query = cacher(cache_backend=self.cache)(self.embed_query)
        self.embed_documents = cacher(cache_backend=self.cache)(
            self.embed_documents
        )
        self.aembed_query = cacher(cache_backend=self.cache)(self.aembed_query)
        self.aembed_documents = cacher(cache_backend=self.cache)(
            self.aembed_documents
        )

embed_query

embed_query(text: str) -> List[float]

嵌入单条查询文本。

源代码位于 src/ragas/embeddings/base.py

def embed_query(self, text: str) -> t.List[float]:
    """
    Embed a single query text.
    """
    return self.embed_documents([text])[0]

embed_documents

embed_documents(texts: List[str]) -> List[List[float]]

嵌入多份文档。

源代码位于 src/ragas/embeddings/base.py

def embed_documents(self, texts: t.List[str]) -> t.List[t.List[float]]:
    """
    Embed multiple documents.
    """
    from sentence_transformers.SentenceTransformer import SentenceTransformer
    from torch import Tensor

    assert isinstance(self.model, SentenceTransformer), (
        "Model is not of the type Bi-encoder"
    )
    embeddings = self.model.encode(
        texts, normalize_embeddings=True, **self.encode_kwargs
    )

    assert isinstance(embeddings, Tensor)
    return embeddings.tolist()

predict

predict(texts: List[List[str]]) -> List[List[float]]

使用 cross-encoder 模型做预测。

源代码位于 src/ragas/embeddings/base.py

def predict(self, texts: t.List[t.List[str]]) -> t.List[t.List[float]]:
    """
    Make predictions using a cross-encoder model.
    """
    from sentence_transformers.cross_encoder import CrossEncoder
    from torch import Tensor

    assert isinstance(self.model, CrossEncoder), (
        "Model is not of the type CrossEncoder"
    )

    predictions = self.model.predict(texts, **self.encode_kwargs)

    assert isinstance(predictions, Tensor)
    return predictions.tolist()

GoogleEmbeddings

GoogleEmbeddings(client: Optional[Any] = None, model: str = 'gemini-embedding-001', use_vertex: bool = False, project_id: Optional[str] = None, location: Optional[str] = 'us-central1', cache: Optional[CacheInterface] = None, **kwargs: Any)

基类： BaseRagasEmbedding

使用 Vertex AI 或 Google AI（Gemini）的 Google embeddings。

同时支持 Vertex AI 和 Google AI（Gemini）嵌入模型。Vertex AI 需要 google-cloud-aiplatform 包。对 Google AI，二者都支持：

新 SDK（google-genai）：推荐，使用 genai.Client()
旧 SDK（google-generativeai）：已弃用（支持到 2025 年 8 月）

client 参数较灵活：

新 SDK：genai.Client(api_key="...") 实例
旧 SDK：None（自动导入）、genai 模块，或 GenerativeModel 实例
对于 Vertex：应为已配置的 vertex client

注意：与 LLM 生成不同，嵌入在两种 SDK 上都能正常工作。已知的 instructor safety settings 问题（github.com/567-labs/instructor/issues/1658）只影响 LLM 生成，不影响嵌入。

示例：

New SDK (google-genai) - recommended

from google import genai client = genai.Client(api_key="...") embeddings = GoogleEmbeddings(client=client, model="gemini-embedding-001")

# Old SDK (google-generativeai) - deprecated
import google.generativeai as genai
genai.configure(api_key="...")
embeddings = GoogleEmbeddings(client=genai, model="text-embedding-004")

# Auto-import (tries new SDK first, falls back to old)
embeddings = GoogleEmbeddings(model="text-embedding-004")

源代码位于 src/ragas/embeddings/google_provider.py

def __init__(
    self,
    client: t.Optional[t.Any] = None,
    model: str = "gemini-embedding-001",
    use_vertex: bool = False,
    project_id: t.Optional[str] = None,
    location: t.Optional[str] = "us-central1",
    cache: t.Optional[CacheInterface] = None,
    **kwargs: t.Any,
):
    super().__init__(cache=cache)
    self._original_client = client
    self.model = model
    self.use_vertex = use_vertex
    self.project_id = project_id
    self.location = location
    self.kwargs = kwargs

    # Track which SDK is being used (new google-genai vs old google-generativeai)
    self._use_new_sdk = False

    # Resolve the actual client to use
    self.client = self._resolve_client(client, use_vertex)

embed_text

embed_text(text: str, **kwargs: Any) -> List[float]

使用 Google 嵌入服务嵌入单条文本。

源代码位于 src/ragas/embeddings/google_provider.py

def embed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Embed a single text using Google's embedding service."""
    if self.use_vertex:
        return self._embed_text_vertex(text, **kwargs)
    else:
        return self._embed_text_genai(text, **kwargs)

aembed_text

aembed_text(text: str, **kwargs: Any) -> List[float]

使用 Google 嵌入服务异步嵌入单条文本。

Google 的 SDK 不提供原生异步支持，因此我们使用 ThreadPoolExecutor。

源代码位于 src/ragas/embeddings/google_provider.py

async def aembed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Asynchronously embed a single text using Google's embedding service.

    Google's SDK doesn't provide native async support, so we use ThreadPoolExecutor.
    """
    return await run_sync_in_async(self.embed_text, text, **kwargs)

embed_texts

embed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

使用 Google 嵌入服务嵌入多条文本。

源代码位于 src/ragas/embeddings/google_provider.py

def embed_texts(self, texts: t.List[str], **kwargs: t.Any) -> t.List[t.List[float]]:
    """Embed multiple texts using Google's embedding service."""
    texts = validate_texts(texts)
    if not texts:
        return []

    if self.use_vertex:
        return self._embed_texts_vertex(texts, **kwargs)
    else:
        return self._embed_texts_genai(texts, **kwargs)

aembed_texts

aembed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

使用 Google 嵌入服务异步嵌入多条文本。

源代码位于 src/ragas/embeddings/google_provider.py

async def aembed_texts(
    self, texts: t.List[str], **kwargs: t.Any
) -> t.List[t.List[float]]:
    """Asynchronously embed multiple texts using Google's embedding service."""
    texts = validate_texts(texts)
    if not texts:
        return []

    return await run_sync_in_async(self.embed_texts, texts, **kwargs)

HaystackEmbeddingsWrapper

HaystackEmbeddingsWrapper(embedder: Union[AzureOpenAITextEmbedder, HuggingFaceAPITextEmbedder, OpenAITextEmbedder, SentenceTransformersTextEmbedder], run_config: Optional[RunConfig] = None, cache: Optional[CacheInterface] = None)

基类： BaseRagasEmbeddings

在 Ragas 框架中使用 Haystack embedder 的包装器。

该类允许你通过 Haystack embedder，使用同步和异步方法（embed_query/embed_documents 和 aembed_query/aembed_documents）生成嵌入。

参数：

名称	类型	说明	默认值
`embedder`	`AzureOpenAITextEmbedder	HuggingFaceAPITextEmbedder	OpenAITextEmbedder
`run_config`	`RunConfig`	用于管理嵌入执行设置的配置对象，默认 None。	`None`
`cache`	`CacheInterface`	用于存储和检索嵌入结果的缓存实例，默认 None。	`None`

源代码位于 src/ragas/embeddings/haystack_wrapper.py

def __init__(
    self,
    embedder: t.Union[
        "AzureOpenAITextEmbedder",
        "HuggingFaceAPITextEmbedder",
        "OpenAITextEmbedder",
        "SentenceTransformersTextEmbedder",
    ],
    run_config: t.Optional[RunConfig] = None,
    cache: t.Optional[CacheInterface] = None,
):
    super().__init__(cache=cache)

    # Lazy Import of required Haystack components
    try:
        from haystack import AsyncPipeline
        from haystack.components.embedders.azure_text_embedder import (
            AzureOpenAITextEmbedder,
        )
        from haystack.components.embedders.hugging_face_api_text_embedder import (
            HuggingFaceAPITextEmbedder,
        )
        from haystack.components.embedders.openai_text_embedder import (
            OpenAITextEmbedder,
        )
        from haystack.components.embedders.sentence_transformers_text_embedder import (
            SentenceTransformersTextEmbedder,
        )
    except ImportError as exc:
        raise ImportError(
            "Haystack is not installed. Please install it with `pip install haystack-ai`."
        ) from exc

    # Validate embedder type
    if not isinstance(
        embedder,
        (
            AzureOpenAITextEmbedder,
            HuggingFaceAPITextEmbedder,
            OpenAITextEmbedder,
            SentenceTransformersTextEmbedder,
        ),
    ):
        raise TypeError(
            "Expected 'embedder' to be one of: AzureOpenAITextEmbedder, "
            "HuggingFaceAPITextEmbedder, OpenAITextEmbedder, or "
            f"SentenceTransformersTextEmbedder, but got {type(embedder).__name__}."
        )

    self.embedder = embedder

    # Initialize an asynchronous pipeline and add the embedder component
    self.async_pipeline = AsyncPipeline()
    self.async_pipeline.add_component("embedder", self.embedder)  # type: ignore[reportArgumentType]

    # Set or create the run configuration
    if run_config is None:
        run_config = RunConfig()
    self.set_run_config(run_config)

HuggingFaceEmbeddings

HuggingFaceEmbeddings(model: str, use_api: bool = False, api_key: Optional[str] = None, device: Optional[str] = None, normalize_embeddings: bool = True, batch_size: int = 32, cache: Optional[CacheInterface] = None, **model_kwargs: Any)

基类： BaseRagasEmbedding

同时支持本地和基于 API 的模型的 HuggingFace embeddings。

本地模型支持 sentence-transformers，托管模型支持 HuggingFace API。提供高效批处理和缓存。

源代码位于 src/ragas/embeddings/huggingface_provider.py

def __init__(
    self,
    model: str,
    use_api: bool = False,
    api_key: t.Optional[str] = None,
    device: t.Optional[str] = None,
    normalize_embeddings: bool = True,
    batch_size: int = 32,
    cache: t.Optional[CacheInterface] = None,
    **model_kwargs: t.Any,
):
    super().__init__(cache=cache)
    self.model = model
    self.use_api = use_api
    self.api_key = api_key
    self.device = device
    self.normalize_embeddings = normalize_embeddings
    self.batch_size = batch_size
    self.model_kwargs = model_kwargs

    if use_api:
        self._setup_api_client()
    else:
        self._setup_local_model()

embed_text

embed_text(text: str, **kwargs: Any) -> List[float]

使用 HuggingFace 嵌入单条文本。

源代码位于 src/ragas/embeddings/huggingface_provider.py

def embed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Embed a single text using HuggingFace."""
    if self.use_api:
        return self._embed_text_api(text, **kwargs)
    else:
        return self._embed_text_local(text, **kwargs)

aembed_text

aembed_text(text: str, **kwargs: Any) -> List[float]

使用 HuggingFace 异步嵌入单条文本。

源代码位于 src/ragas/embeddings/huggingface_provider.py

async def aembed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Asynchronously embed a single text using HuggingFace."""
    if self.use_api:
        return await self._aembed_text_api(text, **kwargs)
    else:
        return await run_sync_in_async(self._embed_text_local, text, **kwargs)

embed_texts

embed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

使用 HuggingFace 分批嵌入多条文本。

源代码位于 src/ragas/embeddings/huggingface_provider.py

def embed_texts(self, texts: t.List[str], **kwargs: t.Any) -> t.List[t.List[float]]:
    """Embed multiple texts using HuggingFace with batching."""
    texts = validate_texts(texts)
    if not texts:
        return []

    if self.use_api:
        return self._embed_texts_api(texts, **kwargs)
    else:
        return self._embed_texts_local(texts, **kwargs)

aembed_texts

aembed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

使用 HuggingFace 异步嵌入多条文本。

源代码位于 src/ragas/embeddings/huggingface_provider.py

async def aembed_texts(
    self, texts: t.List[str], **kwargs: t.Any
) -> t.List[t.List[float]]:
    """Asynchronously embed multiple texts using HuggingFace."""
    texts = validate_texts(texts)
    if not texts:
        return []

    if self.use_api:
        return await run_sync_in_async(self._embed_texts_api, texts, **kwargs)
    else:
        return await run_sync_in_async(self._embed_texts_local, texts, **kwargs)

LiteLLMEmbeddings

LiteLLMEmbeddings(model: str, api_key: Optional[str] = None, api_base: Optional[str] = None, api_version: Optional[str] = None, timeout: int = 600, max_retries: int = 3, batch_size: Optional[int] = None, cache: Optional[CacheInterface] = None, **litellm_params: Any)

基类： BaseRagasEmbedding

使用 LiteLLM 的通用嵌入接口。

支持 OpenAI、Azure、Google、Cohere、Anthropic 等 100+ 模型。提供智能分批和提供商特定优化。

源代码位于 src/ragas/embeddings/litellm_provider.py

def __init__(
    self,
    model: str,
    api_key: t.Optional[str] = None,
    api_base: t.Optional[str] = None,
    api_version: t.Optional[str] = None,
    timeout: int = 600,
    max_retries: int = 3,
    batch_size: t.Optional[int] = None,
    cache: t.Optional[CacheInterface] = None,
    **litellm_params: t.Any,
):
    super().__init__(cache=cache)
    self.litellm = safe_import("litellm", "litellm")
    self.model = model
    self.api_key = api_key
    self.api_base = api_base
    self.api_version = api_version
    self.timeout = timeout
    self.max_retries = max_retries
    self.batch_size = batch_size or get_optimal_batch_size("litellm", model)
    self.litellm_params = litellm_params

embed_text

embed_text(text: str, **kwargs: Any) -> List[float]

使用 LiteLLM 嵌入单条文本。

源代码位于 src/ragas/embeddings/litellm_provider.py

def embed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Embed a single text using LiteLLM."""
    call_kwargs = self._prepare_kwargs(**kwargs)
    response = self.litellm.embedding(input=[text], **call_kwargs)
    return response.data[0]["embedding"]

aembed_text

aembed_text(text: str, **kwargs: Any) -> List[float]

使用 LiteLLM 异步嵌入单条文本。

源代码位于 src/ragas/embeddings/litellm_provider.py

async def aembed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Asynchronously embed a single text using LiteLLM."""
    call_kwargs = self._prepare_kwargs(**kwargs)
    response = await self.litellm.aembedding(input=[text], **call_kwargs)
    return response.data[0]["embedding"]

embed_texts

embed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

使用 LiteLLM 智能分批嵌入多条文本。

源代码位于 src/ragas/embeddings/litellm_provider.py

def embed_texts(self, texts: t.List[str], **kwargs: t.Any) -> t.List[t.List[float]]:
    """Embed multiple texts using LiteLLM with intelligent batching."""
    texts = validate_texts(texts)
    if not texts:
        return []

    embeddings = []
    batches = batch_texts(texts, self.batch_size)

    for batch in batches:
        call_kwargs = self._prepare_kwargs(**kwargs)
        response = self.litellm.embedding(input=batch, **call_kwargs)
        embeddings.extend([item["embedding"] for item in response.data])

    return embeddings

aembed_texts

aembed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

使用 LiteLLM 智能分批异步嵌入多条文本。

源代码位于 src/ragas/embeddings/litellm_provider.py

async def aembed_texts(
    self, texts: t.List[str], **kwargs: t.Any
) -> t.List[t.List[float]]:
    """Asynchronously embed multiple texts using LiteLLM with intelligent batching."""
    texts = validate_texts(texts)
    if not texts:
        return []

    embeddings = []
    batches = batch_texts(texts, self.batch_size)

    for batch in batches:
        call_kwargs = self._prepare_kwargs(**kwargs)
        response = await self.litellm.aembedding(input=batch, **call_kwargs)
        embeddings.extend([item["embedding"] for item in response.data])

    return embeddings

OpenAIEmbeddings

OpenAIEmbeddings(client: Any, model: str = 'text-embedding-3-small', cache: Optional[CacheInterface] = None)

基类： BaseRagasEmbedding

带批处理优化的 OpenAI embeddings 实现。

自动检测并同时支持同步和异步 OpenAI 客户端。提供优化批处理以提升性能。

源代码位于 src/ragas/embeddings/openai_provider.py

def __init__(
    self,
    client: t.Any,
    model: str = "text-embedding-3-small",
    cache: t.Optional[CacheInterface] = None,
):
    super().__init__(cache=cache)
    self.client = client
    self.model = model
    self.is_async = self._check_client_async(client)

embed_text

embed_text(text: str, **kwargs: Any) -> List[float]

使用 OpenAI 嵌入单条文本。

对于异步客户端，这将在适当的事件循环中运行异步方法。

源代码位于 src/ragas/embeddings/openai_provider.py

def embed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Embed a single text using OpenAI.

    For async clients, this will run the async method in the appropriate event loop.
    """
    if self.is_async:
        result = self._run_async_in_current_loop(self.aembed_text(text, **kwargs))
    else:
        response = self.client.embeddings.create(
            input=text, model=self.model, **kwargs
        )
        result = response.data[0].embedding

    # Track usage
    track(
        EmbeddingUsageEvent(
            provider="openai",
            model=self.model,
            embedding_type="modern",
            num_requests=1,
            is_async=self.is_async,
        )
    )
    return result

aembed_text

aembed_text(text: str, **kwargs: Any) -> List[float]

使用 OpenAI 异步嵌入单条文本。

源代码位于 src/ragas/embeddings/openai_provider.py

async def aembed_text(self, text: str, **kwargs: t.Any) -> t.List[float]:
    """Asynchronously embed a single text using OpenAI."""
    if not self.is_async:
        raise TypeError(
            "Cannot use aembed_text() with a synchronous client. Use embed_text() instead."
        )

    response = await self.client.embeddings.create(
        input=text, model=self.model, **kwargs
    )
    result = response.data[0].embedding

    # Track usage
    track(
        EmbeddingUsageEvent(
            provider="openai",
            model=self.model,
            embedding_type="modern",
            num_requests=1,
            is_async=True,
        )
    )
    return result

embed_texts

embed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

使用 OpenAI batch API 嵌入多条文本以优化。

源代码位于 src/ragas/embeddings/openai_provider.py

def embed_texts(self, texts: t.List[str], **kwargs: t.Any) -> t.List[t.List[float]]:
    """Embed multiple texts using OpenAI's batch API for optimization."""
    texts = validate_texts(texts)
    if not texts:
        return []

    if self.is_async:
        result = self._run_async_in_current_loop(self.aembed_texts(texts, **kwargs))
    else:
        # OpenAI supports batch embedding natively
        response = self.client.embeddings.create(
            input=texts, model=self.model, **kwargs
        )
        result = [item.embedding for item in response.data]

    # Track usage
    track(
        EmbeddingUsageEvent(
            provider="openai",
            model=self.model,
            embedding_type="modern",
            num_requests=len(texts),
            is_async=self.is_async,
        )
    )
    return result

aembed_texts

aembed_texts(texts: List[str], **kwargs: Any) -> List[List[float]]

使用 OpenAI batch API 异步嵌入多条文本。

源代码位于 src/ragas/embeddings/openai_provider.py

async def aembed_texts(
    self, texts: t.List[str], **kwargs: t.Any
) -> t.List[t.List[float]]:
    """Asynchronously embed multiple texts using OpenAI's batch API."""
    texts = validate_texts(texts)
    if not texts:
        return []

    if not self.is_async:
        raise TypeError(
            "Cannot use aembed_texts() with a synchronous client. Use embed_texts() instead."
        )

    response = await self.client.embeddings.create(
        input=texts, model=self.model, **kwargs
    )
    result = [item.embedding for item in response.data]

    # Track usage
    track(
        EmbeddingUsageEvent(
            provider="openai",
            model=self.model,
            embedding_type="modern",
            num_requests=len(texts),
            is_async=True,
        )
    )
    return result

batch_texts

batch_texts(texts: List[str], batch_size: int) -> List[List[str]]

将文本列表分批成更小的 chunk。

参数： texts: 要分批的文本列表 batch_size: 每批大小

返回：批次列表，每批是文本列表

源代码位于 src/ragas/embeddings/utils.py

def batch_texts(texts: t.List[str], batch_size: int) -> t.List[t.List[str]]:
    """Batch a list of texts into smaller chunks.

    Args:
        texts: List of texts to batch
        batch_size: Size of each batch

    Returns:
        List of batches, where each batch is a list of texts
    """
    if batch_size <= 0:
        raise ValueError("Batch size must be positive")

    batches = []
    for i in range(0, len(texts), batch_size):
        batches.append(texts[i : i + batch_size])
    return batches

get_optimal_batch_size

get_optimal_batch_size(provider: str, model: str) -> int

获取 provider/model 组合的最优批次大小。

参数： provider: 嵌入提供商 model: 模型名称

返回：该 provider/model 的最优批次大小

源代码位于 src/ragas/embeddings/utils.py

def get_optimal_batch_size(provider: str, model: str) -> int:
    """Get optimal batch size for a provider/model combination.

    Args:
        provider: The embedding provider
        model: The model name

    Returns:
        Optimal batch size for the provider/model
    """
    provider_lower = provider.lower()

    # Provider-specific batch sizes
    if "openai" in provider_lower:
        return 100  # OpenAI supports large batches
    elif "cohere" in provider_lower:
        return 96  # Cohere's documented limit
    elif "google" in provider_lower or "vertex" in provider_lower:
        return 5  # Google/Vertex AI is more conservative
    elif "huggingface" in provider_lower:
        return 32  # HuggingFace default
    else:
        return 10  # Conservative default for unknown providers

validate_texts

validate_texts(texts: Union[str, List[str]]) -> List[str]

校验并规范化文本输入。

参数： texts: 单条文本或文本列表

返回：校验后的文本列表

抛出： ValueError: 若文本无效

源代码位于 src/ragas/embeddings/utils.py

def validate_texts(texts: t.Union[str, t.List[str]]) -> t.List[str]:
    """Validate and normalize text inputs.

    Args:
        texts: Single text or list of texts

    Returns:
        List of validated texts

    Raises:
        ValueError: If texts are invalid
    """
    if isinstance(texts, str):
        texts = [texts]

    if not isinstance(texts, list):
        raise ValueError("Texts must be a string or list of strings")

    if not texts:
        raise ValueError("Texts list cannot be empty")

    for i, text in enumerate(texts):
        if not isinstance(text, str):
            raise ValueError(f"Text at index {i} must be a string, got {type(text)}")
        if not text.strip():
            raise ValueError(f"Text at index {i} cannot be empty or whitespace only")

    return texts

embedding_factory

embedding_factory(*args, **kwargs)

已弃用：请直接使用 base 模块中的 embedding_factory。

源代码位于 src/ragas/embeddings/__init__.py

def embedding_factory(*args, **kwargs):
    """Deprecated: Use embedding_factory from base module directly."""
    import warnings

    warnings.warn(
        "Importing embedding_factory from ragas.embeddings is deprecated. "
        "Import directly from ragas.embeddings.base or use modern providers: "
        "from ragas.embeddings import OpenAIEmbeddings, GoogleEmbeddings, HuggingFaceEmbeddings",
        DeprecationWarning,
        stacklevel=2,
    )
    return _embedding_factory(*args, **kwargs)