Embedding Service | Kamiwaza Docs

Overview

The Embedding Service (EmbeddingService) provides comprehensive text embedding functionality for the Kamiwaza AI Platform. Located in kamiwaza_sdk/services/embedding.py, this service handles text chunking, embedding generation, and provider management.

Key Features

• Provider-Based Architecture
• Text Chunking with Metadata
• Embedding Generation
• Batch Processing
• Multiple Provider Support
• Stateless Design

Getting Started

The embedding service requires initializing a provider before use:

from kamiwaza_sdk import KamiwazaClient

client = KamiwazaClient(api_key="your-key")

# Get an embedder instance (required first step)

embedder = client.embedding.get_embedder(

model="nomic-ai/nomic-embed-text-v1.5",  # optional, this is default

provider_type="sentencetransformers",     # optional, this is default

device="cuda"                             # optional, auto-detect if None

)

Text Processing

Available Methods

• chunk_text(text: str, max_length: int = 1024, overlap: int = 102, preamble_text: str = "", return_metadata: bool = False) -> Union[List[str], ChunkResponse]: Split text into chunks
• embed_chunks(text_chunks: List[str], batch_size: int = 64) -> List[List[float]]: Generate embeddings for chunks
• create_embedding(text: str, max_length: int = 1024, overlap: int = 102, preamble_text: str = "") -> EmbeddingOutput: Create embedding for text
• get_embedding(text: str, return_offset: bool = False) -> EmbeddingOutput: Generate embedding for text

# Initialize embedder first

embedder = client.embedding.get_embedder()

# Split text into chunks

chunks = embedder.chunk_text(

text="Long document text...",

max_length=1024,

overlap=102
)

# Or get chunks with metadata

chunk_response = embedder.chunk_text(

text="Long document text...",

max_length=1024,

overlap=102,

return_metadata=True
)

# Access: chunk_response.chunks, chunk_response.offsets, chunk_response.token_counts

# Generate embeddings for chunks

embeddings = embedder.embed_chunks(chunks, batch_size=64)

# Create single embedding

result = embedder.create_embedding("Sample text")

embedding_vector = result.embedding  # List[float]

# Generate embedding (alternative method)

result = embedder.get_embedding("Sample text")

embedding_vector = result.embedding

Provider Management

Getting an Embedder

The primary method for working with embeddings is through get_embedder():

# Get embedder with default settings

embedder = client.embedding.get_embedder()

# Get embedder with custom model

embedder = client.embedding.get_embedder(

model="sentence-transformers/all-mpnet-base-v2",

provider_type="sentencetransformers",

device="cuda"  # or "cpu", "mps", None for auto-detect

)

# List available providers

providers = client.embedding.get_providers()

Default Configuration

• Default Model: nomic-ai/nomic-embed-text-v1.5
• Default Provider: sentencetransformers
• Default Device: Auto-detected based on availability

Return Types

The service returns Pydantic models for structured data:

EmbeddingOutput

class EmbeddingOutput:

embedding: List[float]  # The embedding vector

offset: Optional[int]   # Offset in original text (if requested)

ChunkResponse

class ChunkResponse:

chunks: List[str]                    # Text chunks

offsets: Optional[List[int]]         # Start positions in original text

token_counts: Optional[List[int]]    # Token count per chunk

metadata: Optional[List[dict]]       # Additional metadata per chunk

Error Handling

The service uses a unified error handling approach:

from kamiwaza_sdk.exceptions import APIError

try:

embedder = client.embedding.get_embedder()

result = embedder.create_embedding("text")

except APIError as e:

print(f"Operation failed: {e}")

Deprecated Methods

The following methods are deprecated and should not be used:

HuggingFaceEmbedding()

• Status: Deprecated
• Replacement: Use get_embedder() instead
• Warning: Logs deprecation warning when called

reset_model()

• Status: Deprecated (no-op in stateless design)
• Returns: {"status": "no-op"}
• Note: Model state is now handled per request

call()

• Status: Deprecated
• Replacement: Use get_embedder() then call methods on the provider
• Raises: DeprecationWarning

Best Practices

1. Always initialize an embedder first using get_embedder()
2. Choose appropriate chunk sizes based on your model's context window
3. Use batch processing for multiple texts to improve performance
4. Handle overlaps properly to maintain context between chunks
5. Consider memory usage when processing large batches
6. Cache embeddings when possible to avoid recomputation
7. Use return_metadata=True when you need chunk offsets or token counts

Complete Example

from kamiwaza_sdk import KamiwazaClient

from kamiwaza_sdk.exceptions import APIError

# Initialize client

client = KamiwazaClient(api_key="your-key")

# Get embedder

embedder = client.embedding.get_embedder(

model="nomic-ai/nomic-embed-text-v1.5",

provider_type="sentencetransformers"
)

try:

# Process a document

document = "Your long document text here..."

# Chunk with metadata

chunk_response = embedder.chunk_text(

text=document,

max_length=512,

overlap=50,

return_metadata=True
    )

# Generate embeddings

embeddings = embedder.embed_chunks(

chunk_response.chunks,

batch_size=32
    )

# Process results

for i, (chunk, embedding) in enumerate(zip(chunk_response.chunks, embeddings)):

print(f"Chunk {i}: {len(embedding)} dimensions")

if chunk_response.offsets:

print(f"  Offset: {chunk_response.offsets[i]}")

if chunk_response.token_counts:

print(f"  Tokens: {chunk_response.token_counts[i]}")

except APIError as e:

print(f"Embedding operation failed: {e}")