pw.xpacks.llm.embedders

Pathway Live Data Framework wrapper for AWS Bedrock Embedding services (see Titan Embeddings docs).

Supports Amazon Titan embeddings and other embedding models available on Bedrock.

The capacity, retry_strategy, and cache_strategy need to be specified during object construction. AWS credentials can be provided explicitly or will be loaded from environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY) or IAM roles when running on AWS infrastructure.

• Parameters
  • capacity (int | None) – Maximum number of concurrent operations allowed. Defaults to None, indicating no specific limit.
  • retry_strategy (AsyncRetryStrategy | None) – Strategy for handling retries in case of failures. Defaults to ExponentialBackoffRetryStrategy.
  • cache_strategy (CacheStrategy | None) – Defines the caching mechanism. To enable caching, a valid CacheStrategy should be provided. Defaults to None.
  • model_id (str | None) – The Bedrock embedding model ID to use. Defaults to "amazon.titan-embed-text-v2:0". Other options include:
    • "amazon.titan-embed-text-v1"
    • "cohere.embed-english-v3"
    • "cohere.embed-multilingual-v3"
  • region_name (str | None) – AWS region where Bedrock is deployed (e.g., "us-east-1"). Can also be set via AWS_DEFAULT_REGION environment variable.
  • aws_access_key_id (str | None) – Optional AWS access key ID. If not provided, will use default credential chain.
  • aws_secret_access_key (str | None) – Optional AWS secret access key.
  • aws_session_token (str | None) – Optional AWS session token for temporary credentials.
  • dimensions – Output embedding dimensions (only supported by some models like Titan V2). If not specified, uses model default.
  • normalize – Whether to normalize the embedding vector (only supported by some models).

Example:

import pathway as pw
from pathway.xpacks.llm import embedders
embedder = embedders.BedrockEmbedder(
    model_id="amazon.titan-embed-text-v2:0",
    region_name="us-east-1"
)  
t = pw.debug.table_from_markdown('''
txt
Hello world
''')  
t.select(ret=embedder(pw.this.txt))

__call__(input, *args, **kwargs)

sourceEmbeds texts in a Column.

• Parameters
  input (ColumnExpression[str]) – Column with texts to embed

get_embedding_dimension(**kwargs)

sourceComputes number of embedder’s dimensions by asking the embedder to embed ".".

• Parameters
  **kwargs – parameters of the embedder, if unset defaults from the constructor will be taken.

Pathway Live Data Framework wrapper for Google Gemini Embedding services.

The capacity, retry_strategy and cache_strategy need to be specified during object construction. All other arguments can be overridden during application. Gemini API truncates the content in case the text length is larger than model’s context length.

• Parameters
  • capacity (int | None) – Maximum number of concurrent operations allowed. Defaults to None, indicating no specific limit.
  • retry_strategy (AsyncRetryStrategy | None) – Strategy for handling retries in case of failures. Defaults to the ExponentialRetryStrategy.
  • cache_strategy (CacheStrategy | None) – Defines the caching mechanism. To enable caching, a valid CacheStrategy should be provided. See Cache strategy for more information. Defaults to None.
  • model (str | None) – ID of the model to use. Check the Gemini documentation for list of available models. To specify the model in the UDF call, set it to None in the constructor.
  • api_key (str | None) – API key for Gemini API services. Can be provided in the constructor, in __call__ or by setting GOOGLE_API_KEY environment variable
  • gemini_kwargs – any other arguments accepted by gemini embedding service. Check the Gemini documentation for list of accepted arguments.

Example:

import pathway as pw
from pathway.xpacks.llm import embedders
embedder = embedders.GeminiEmbedder(model="models/text-embedding-004")
t = pw.debug.table_from_markdown('''
txt
Text
''')
t.select(ret=embedder(pw.this.txt))

import pathway as pw
from pathway.xpacks.llm import embedders
embedder = embedders.GeminiEmbedder()
t = pw.debug.table_from_markdown('''
txt  | model
Text | models/embedding-001
''')
t.select(ret=embedder(pw.this.txt, model=pw.this.model))

__call__(input, *args, **kwargs)

sourceEmbeds texts in a Column.

• Parameters
  input (ColumnExpression[str]) – Column with texts to embed

get_embedding_dimension(**kwargs)

sourceComputes number of embedder’s dimensions by asking the embedder to embed ".".

• Parameters
  **kwargs – parameters of the embedder, if unset defaults from the constructor will be taken.

Pathway Live Data Framework wrapper for litellm.embedding.

Model has to be specified either in constructor call or in each application, no default is provided. The capacity, retry_strategy and cache_strategy need to be specified during object construction. All other arguments can be overridden during application.

• Parameters
  • capacity (int | None) – Maximum number of concurrent operations allowed. Defaults to None, indicating no specific limit.
  • retry_strategy (AsyncRetryStrategy | None) – Strategy for handling retries in case of failures. Defaults to the ExponentialRetryStrategy.
  • cache_strategy (CacheStrategy | None) – Defines the caching mechanism. To enable caching, a valid CacheStrategy should be provided. See Cache strategy for more information. Defaults to None.
  • model (str | None) – The embedding model to use.
  • timeout – The timeout value for the API call, default 10 mins
  • litellm_call_id – The call ID for litellm logging.
  • litellm_logging_obj – The litellm logging object.
  • logger_fn – The logger function.
  • api_base – Optional. The base URL for the API.
  • api_version – Optional. The version of the API.
  • api_key – Optional. The API key to use.
  • api_type – Optional. The type of the API.
  • custom_llm_provider – The custom llm provider.

Any arguments can be provided either to the constructor or in the UDF call. To specify the model in the UDF call, set it to None.

Example:

import pathway as pw
from pathway.xpacks.llm import embedders
embedder = embedders.LiteLLMEmbedder(model="text-embedding-3-small")
t = pw.debug.table_from_markdown('''
txt
Text
''')
t.select(ret=embedder(pw.this.txt))

import pathway as pw
from pathway.xpacks.llm import embedders
embedder = embedders.LiteLLMEmbedder()
t = pw.debug.table_from_markdown('''
txt  | model
Text | text-embedding-3-small
''')
t.select(ret=embedder(pw.this.txt, model=pw.this.model))

__call__(input, *args, **kwargs)

sourceEmbeds texts in a Column.

• Parameters
  input (ColumnExpression[str]) – Column with texts to embed

get_embedding_dimension(**kwargs)

sourceComputes number of embedder’s dimensions by asking the embedder to embed ".".

• Parameters
  **kwargs – parameters of the embedder, if unset defaults from the constructor will be taken.

Embed text using the TwelveLabs Marengo multimodal embedding model.

Marengo returns 512-dimensional embeddings in a shared multimodal space, so the text it produces is directly comparable with image, audio and video embeddings from the same model. This makes it a natural retriever embedder for pipelines that index video with TwelveLabsVideoParser.

• Parameters
  • model (str) – Marengo model name. Defaults to "marengo3.0".
  • api_key (str | None) – TwelveLabs API key. If None, the SDK reads it from the TWELVELABS_API_KEY environment variable.
  • capacity (int | None) – Maximum number of concurrent requests to the TwelveLabs API. Defaults to 16, which stays clear of the API rate limits; raise it if your account allows more.
  • retry_strategy (AsyncRetryStrategy | None) – Strategy for handling retries. Defaults to ExponentialBackoffRetryStrategy.
  • cache_strategy (CacheStrategy | None) – Pathway caching strategy. Defaults to None. In production consider pw.udfs.DiskCache() so restarts do not re-embed all documents.
  • embedding_dimension (int | None) – Dimension of the embeddings reported to index factories without calling the API. Defaults to 512 (all current Marengo models). Pass None to probe the API with a live request instead.

Example:

import pathway as pw  
from pathway.xpacks.llm.embedders import MarengoEmbedder  
embedder = MarengoEmbedder(cache_strategy=pw.udfs.DiskCache())

__call__(input, *args, **kwargs)

sourceEmbeds texts in a Column.

• Parameters
  input (ColumnExpression[str]) – Column with texts to embed

get_embedding_dimension(**kwargs)

sourceReturn the embedding dimension (512 for Marengo).

By default this returns the known dimension from embedding_dimension without any network call, so building the pipeline does not depend on the TwelveLabs API being reachable. When the embedder is constructed with embedding_dimension=None, the dimension is probed with a single request instead — a one-time, setup-time call, not on the per-document hot path (the base implementation cannot be reused for the probe because this embedder’s __wrapped__ takes a batch of strings, not a single string).

Pathway Live Data Framework wrapper for OpenAI Embedding services.

The capacity, retry_strategy and cache_strategy need to be specified during object construction, and API key must be provided to the constructor with the api_key argument or set in the OPENAI_API_KEY environment variable. All other arguments can be overridden during application.

• Parameters
  • capacity (int | None) – Maximum number of concurrent operations allowed. Defaults to None, indicating no specific limit.
  • retry_strategy (AsyncRetryStrategy | None) – Strategy for handling retries in case of failures. Defaults to the ExponentialRetryStrategy.
  • cache_strategy (CacheStrategy | None) – Defines the caching mechanism. To enable caching, a valid CacheStrategy should be provided. See Cache strategy for more information. Defaults to None.
  • model (str | None) – ID of the model to use. You can use the List models API to see all of your available models, or see Model overview for descriptions of them.
  • api_key – API key to be used for API calls to OpenAI. It must be either provided in the constructor or set in the OPENAI_API_KEY environment variable.
  • truncation_keep_strategy (Optional[Literal['start', 'end']]) – Strategy to keep the part of the text if truncation is necessary. If set, only documents that are longer than model’s supported context will be truncated. Can be "start", "end" or None. "start" will keep the first part of the text and remove the rest. "end" will keep the last part of the text. If None, no truncation will be applied to any of the documents, this may cause API exceptions.
  • batch_size (int) – maximum size of a single batch to be sent to the embedder. Bigger batches may reduce the time needed for embedding.
  • encoding_format – The format to return the embeddings in. Can be either float or base64.
  • user – A unique identifier representing your end-user, which can help OpenAI to monitor and detect abuse. Learn more.
  • extra_headers – Send extra headers
  • extra_query – Add additional query parameters to the request
  • extra_body – Add additional JSON properties to the request
  • timeout – Timeout for requests, in seconds

Any arguments can be provided either to the constructor or in the UDF call. To specify the model in the UDF call, set it to None.

Example:

import pathway as pw
from pathway.xpacks.llm import embedders
embedder = embedders.OpenAIEmbedder(model="text-embedding-3-small")
t = pw.debug.table_from_markdown('''
txt
Text
''')
t.select(ret=embedder(pw.this.txt))

import pathway as pw
from pathway.xpacks.llm import embedders
embedder = embedders.OpenAIEmbedder()
t = pw.debug.table_from_markdown('''
txt  | model
Text | text-embedding-3-small
''')
t.select(ret=embedder(pw.this.txt, model=pw.this.model))

__call__(input, *args, **kwargs)

sourceEmbeds texts in a Column.

• Parameters
  input (ColumnExpression[str]) – Column with texts to embed

get_embedding_dimension(**kwargs)

sourceComputes number of embedder’s dimensions by asking the embedder to embed ".".

• Parameters
  **kwargs – parameters of the embedder, if unset defaults from the constructor will be taken.

static truncate_context(model, text, strategy)

sourceMaybe truncate the given text from the end, or from the start. "strategy" determines which part of the text will be kept.

The Pathway Live Data Framework wrapper for Sentence-Transformers embedder.

• Parameters
  • model (str) – model name or path
  • call_kwargs (dict) – kwargs that will be passed to each call of encode. These can be overridden during each application. For possible arguments check the Sentence-Transformers documentation.
  • device (str) – defines which device will be used to run the Pipeline
  • batch_size (int) – maximum size of a single batch to be sent to the embedder. Bigger batches may reduce the time needed for embedding, especially on GPU.
  • sentencetransformer_kwargs – kwargs accepted during initialization of SentenceTransformers. For possible arguments check the Sentence-Transformers documentation

Example:

import pathway as pw  
from pathway.xpacks.llm import embedders  
embedder = embedders.SentenceTransformerEmbedder(model="intfloat/e5-large-v2")  
t = pw.debug.table_from_markdown('''
txt
Text
''')  
t.select(ret=embedder(pw.this.txt))

__call__(input, *args, **kwargs)

sourceEmbeds texts in a Column.

• Parameters
  input (ColumnExpression[str]) – Column with texts to embed

get_embedding_dimension(**kwargs)

sourceComputes number of embedder’s dimensions by asking the embedder to embed ".".

• Parameters
  **kwargs – parameters of the embedder, if unset defaults from the constructor will be taken.