Source code for langchain.embeddings.self_hosted
from typing import Any, Callable, List
from langchain.llms import SelfHostedPipeline
from langchain.pydantic_v1 import Extra
from langchain.schema.embeddings import Embeddings
def _embed_documents(pipeline: Any, *args: Any, **kwargs: Any) -> List[List[float]]:
    """Inference function to send to the remote hardware.
    Accepts a sentence_transformer model_id and
    returns a list of embeddings for each document in the batch.
    """
    return pipeline(*args, **kwargs)
[docs]class SelfHostedEmbeddings(SelfHostedPipeline, Embeddings):
    """Custom embedding models on self-hosted remote hardware.
    Supported hardware includes auto-launched instances on AWS, GCP, Azure,
    and Lambda, as well as servers specified
    by IP address and SSH credentials (such as on-prem, or another
    cloud like Paperspace, Coreweave, etc.).
    To use, you should have the ``runhouse`` python package installed.
    Example using a model load function:
        .. code-block:: python
            from langchain.embeddings import SelfHostedEmbeddings
            from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline
            import runhouse as rh
            gpu = rh.cluster(name="rh-a10x", instance_type="A100:1")
            def get_pipeline():
                model_id = "facebook/bart-large"
                tokenizer = AutoTokenizer.from_pretrained(model_id)
                model = AutoModelForCausalLM.from_pretrained(model_id)
                return pipeline("feature-extraction", model=model, tokenizer=tokenizer)
            embeddings = SelfHostedEmbeddings(
                model_load_fn=get_pipeline,
                hardware=gpu