Langchain + CTransformers embedding function

ctransformers_embeddings.py

from langchain_community.llms.ctransformers import CTransformers

from langchain_core.embeddings import Embeddings

from typing import List
import asyncio

class CTransformersEmbeddings(Embeddings):
    def __init__(self, llm):
        self.llm = llm

    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        res = []

        for t in texts:
            embedding = self.llm.client.embed(t)
            res.append(embedding)

        return res

    def embed_query(self, text: str) -> List[float]:
        return self.llm.client.embed(text)

    async def aembed_documents(self, texts: List[str]) -> List[List[float]]:
        """Asynchronous Embed search docs."""
        return await asyncio.get_running_loop().run_in_executor(
            None, self.embed_documents, texts
        )

    async def aembed_query(self, text: str) -> List[float]:
        """Asynchronous Embed query text."""
        return await asyncio.get_running_loop().run_in_executor(
            None, self.embed_query, text
        )