tessl/pypi-langchain-google-genai

An integration package connecting Google's genai package and LangChain

—

Pending

Overview

Eval results

Files

Embeddings

Name: tessl/pypi-langchain-google-genai
Author: tessl

High-quality text embeddings using Google's Generative AI models for semantic search, similarity analysis, clustering, and machine learning applications. Provides efficient batch processing and configurable task types for optimal performance.

Capabilities

GoogleGenerativeAIEmbeddings

Primary embeddings interface that extends LangChain's Embeddings class to provide access to Google's embedding models.

class GoogleGenerativeAIEmbeddings:
    def __init__(
        self,
        *,
        model: str,
        task_type: Optional[str] = None,
        google_api_key: Optional[SecretStr] = None,
        credentials: Any = None,
        client_options: Optional[Dict] = None,
        transport: Optional[str] = None,
        request_options: Optional[Dict] = None
    )

Parameters:

model (str): Embedding model name (e.g., "models/gemini-embedding-001")
task_type (Optional[str]): Task type for embeddings optimization
google_api_key (Optional[SecretStr]): Google API key (defaults to GOOGLE_API_KEY env var)
credentials (Any): Google authentication credentials object
client_options (Optional[Dict]): API client configuration options
transport (Optional[str]): Transport method ["rest", "grpc", "grpc_asyncio"]
request_options (Optional[Dict]): Request configuration options

Task Types:

"task_type_unspecified": Default, general-purpose embeddings
"retrieval_query": Optimized for search queries
"retrieval_document": Optimized for document content
"semantic_similarity": Optimized for similarity comparisons
"classification": Optimized for classification tasks
"clustering": Optimized for clustering applications

Core Methods

Single Query Embedding

def embed_query(
    self,
    text: str,
    *,
    task_type: Optional[str] = None,
    title: Optional[str] = None,
    output_dimensionality: Optional[int] = None
) -> List[float]

Generate embedding for a single query text.

Parameters:

text (str): Input text to embed
task_type (Optional[str]): Override default task type for this request
title (Optional[str]): Optional title for context
output_dimensionality (Optional[int]): Desired embedding dimensions

Returns: List of float values representing the embedding vector

Batch Document Embedding

def embed_documents(
    self,
    texts: List[str],
    *,
    batch_size: int = 100,
    task_type: Optional[str] = None,
    titles: Optional[List[str]] = None,
    output_dimensionality: Optional[int] = None
) -> List[List[float]]

Generate embeddings for multiple documents with automatic batching.

Parameters:

texts (List[str]): List of input texts to embed
batch_size (int): Batch size for processing (default: 100, max: based on token limits)
task_type (Optional[str]): Override default task type for this request
titles (Optional[List[str]]): Optional titles for each document
output_dimensionality (Optional[int]): Desired embedding dimensions

Returns: List of embedding vectors, one for each input text

Async Methods

async def aembed_query(
    self,
    text: str,
    *,
    task_type: Optional[str] = None,
    title: Optional[str] = None,
    output_dimensionality: Optional[int] = None
) -> List[float]

Async version of embed_query().

async def aembed_documents(
    self,
    texts: List[str],
    *,
    batch_size: int = 100,
    task_type: Optional[str] = None,
    titles: Optional[List[str]] = None,
    output_dimensionality: Optional[int] = None
) -> List[List[float]]

Async version of embed_documents().

Usage Examples

Basic Query Embedding

from langchain_google_genai import GoogleGenerativeAIEmbeddings

# Initialize embeddings model
embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001"
)

# Generate embedding for a query
query = "What is machine learning?"
query_embedding = embeddings.embed_query(query)

print(f"Query embedding dimension: {len(query_embedding)}")
print(f"First 5 values: {query_embedding[:5]}")

Document Embeddings

# Embed multiple documents
documents = [
    "Machine learning is a subset of artificial intelligence.",
    "Deep learning uses neural networks with multiple layers.",
    "Natural language processing focuses on text and speech.",
    "Computer vision enables machines to interpret visual information."
]

doc_embeddings = embeddings.embed_documents(documents)

print(f"Number of embeddings: {len(doc_embeddings)}")
print(f"Each embedding dimension: {len(doc_embeddings[0])}")

Task-Specific Embeddings

# Query embeddings optimized for retrieval
query_embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001",
    task_type="retrieval_query"
)

# Document embeddings optimized for retrieval
doc_embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001",
    task_type="retrieval_document"
)

# Generate embeddings with task-specific optimization
query_vector = query_embeddings.embed_query("Find information about AI")
doc_vectors = doc_embeddings.embed_documents([
    "AI article content here...",
    "Another AI document..."
])

Similarity Search

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

# Initialize embeddings
embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001",
    task_type="semantic_similarity"
)

# Documents to search
documents = [
    "Python is a programming language",
    "Machine learning algorithms learn from data",
    "Cats are domestic animals",
    "Data science involves analyzing large datasets"
]

# Query
query = "What is data analysis?"

# Generate embeddings
doc_embeddings = embeddings.embed_documents(documents)
query_embedding = embeddings.embed_query(query)

# Calculate similarities
similarities = cosine_similarity(
    [query_embedding], 
    doc_embeddings
)[0]

# Find most similar document
best_match_idx = np.argmax(similarities)
print(f"Most similar document: {documents[best_match_idx]}")
print(f"Similarity score: {similarities[best_match_idx]:.3f}")

Batch Processing with Custom Sizes

# Large document collection
large_doc_collection = [f"Document {i} content..." for i in range(1000)]

# Process in smaller batches to manage memory and API limits
embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001"
)

# Embed with custom batch size
all_embeddings = embeddings.embed_documents(
    large_doc_collection,
    batch_size=50  # Smaller batches for large collections
)

print(f"Generated {len(all_embeddings)} embeddings")

Clustering Application

from sklearn.cluster import KMeans
import numpy as np

# Prepare embeddings for clustering
embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001",
    task_type="clustering"
)

texts = [
    "Machine learning and AI research",
    "Deep learning neural networks",
    "Cooking recipes and food preparation",
    "Baking techniques and desserts",
    "Space exploration and astronomy",
    "Planetary science and cosmology"
]

# Generate embeddings optimized for clustering
vectors = embeddings.embed_documents(texts)

# Perform clustering
kmeans = KMeans(n_clusters=3, random_state=42)
clusters = kmeans.fit_predict(vectors)

# Display results
for i, (text, cluster) in enumerate(zip(texts, clusters)):
    print(f"Cluster {cluster}: {text}")

Async Processing

import asyncio

async def process_embeddings_async():
    embeddings = GoogleGenerativeAIEmbeddings(
        model="models/gemini-embedding-001"
    )
    
    # Process query and documents concurrently
    query_task = embeddings.aembed_query("What is AI?")
    docs_task = embeddings.aembed_documents([
        "AI is artificial intelligence",
        "Machine learning is part of AI",
        "Deep learning uses neural networks"
    ])
    
    # Wait for both to complete
    query_vector, doc_vectors = await asyncio.gather(query_task, docs_task)
    
    print(f"Query embedding: {len(query_vector)} dimensions")
    print(f"Document embeddings: {len(doc_vectors)} vectors")

# Run async example
asyncio.run(process_embeddings_async())

Custom Configuration

# Configure with custom client options
embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001",
    client_options={
        "api_endpoint": "https://generativelanguage.googleapis.com"
    },
    transport="rest",
    request_options={
        "timeout": 30
    }
)

# Use with custom configuration
vectors = embeddings.embed_documents(["Custom configured embedding"])

Integration with Vector Stores

from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document

# Initialize embeddings
embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001",
    task_type="retrieval_document"
)

# Create documents
docs = [
    Document(page_content="AI is transforming technology", metadata={"source": "doc1"}),
    Document(page_content="Machine learning enables predictions", metadata={"source": "doc2"}),
    Document(page_content="Deep learning uses neural networks", metadata={"source": "doc3"})
]

# Create vector store
vector_store = FAISS.from_documents(docs, embeddings)

# Perform similarity search
results = vector_store.similarity_search("What is artificial intelligence?", k=2)

for result in results:
    print(f"Content: {result.page_content}")
    print(f"Source: {result.metadata['source']}")

Dimensionality Control

# Control output dimensions (if supported by model)
embeddings = GoogleGenerativeAIEmbeddings(
    model="models/gemini-embedding-001"
)

# Generate embeddings with specific dimensions
vector = embeddings.embed_query(
    "Sample text",
    output_dimensionality=512  # Request 512-dimensional embeddings
)

print(f"Requested dimensions: 512, Got: {len(vector)}")

Constants and Limits

_MAX_TOKENS_PER_BATCH = 20000  # Maximum tokens per batch request
_DEFAULT_BATCH_SIZE = 100      # Default batch size for document processing

The embedding service automatically manages token limits and batch sizes to ensure efficient processing while staying within API constraints.

Error Handling

from langchain_google_genai import GoogleGenerativeAIEmbeddings

try:
    embeddings = GoogleGenerativeAIEmbeddings(
        model="models/gemini-embedding-001"
    )
    
    vectors = embeddings.embed_documents(["text1", "text2"])
    
except Exception as e:
    if "quota" in str(e).lower() or "rate" in str(e).lower():
        print(f"API limit error: {e}")
    elif "auth" in str(e).lower():
        print(f"Authentication error: {e}")
    else:
        print(f"Embedding error: {e}")

Best Practices

Choose appropriate task types for your use case to get optimized embeddings
Use batch processing for multiple documents to improve efficiency
Consider async methods for high-throughput applications
Monitor token usage when processing large document collections
Cache embeddings when possible to avoid redundant API calls
Use appropriate batch sizes based on your document lengths and API limits

Install with Tessl CLI

npx tessl i tessl/pypi-langchain-google-genai

docs

tessl/pypi-langchain-google-genai

embeddings.md.css-3qkkll{font-size:var(--chakra-font-sizes-sm);font-weight:var(--chakra-font-weights-normal);color:var(--chakra-colors-gray-300);}docs/

Embeddings

Capabilities

GoogleGenerativeAIEmbeddings

Core Methods

Single Query Embedding

Batch Document Embedding

Async Methods

Usage Examples

Basic Query Embedding

Document Embeddings

Task-Specific Embeddings

Similarity Search

Batch Processing with Custom Sizes

Clustering Application

Async Processing

Custom Configuration

Integration with Vector Stores

Dimensionality Control

Constants and Limits

Error Handling

Best Practices

embeddings.mddocs/