Azure OpenAI Embedder

AzureOpenAIEmbedderClient for OpenAI embeddings via Azure

Overview

The AzureOpenAIEmbedderClient provides embeddings using OpenAI models hosted on Azure, supporting both the native Azure OpenAI SDK and OpenAI’s v1 API compatibility endpoint.

Installation

pip install graphiti-core

The OpenAI SDK (which includes Azure support) is included by default.

Basic Usage

from graphiti_core.embedder import AzureOpenAIEmbedderClient
from openai import AsyncAzureOpenAI

# Create Azure OpenAI client
azure_client = AsyncAzureOpenAI(
    api_key="your-azure-key",
    api_version="2024-02-15-preview",
    azure_endpoint="https://your-resource.openai.azure.com"
)

# Initialize embedder
embedder = AzureOpenAIEmbedderClient(
    azure_client=azure_client,
    model="text-embedding-3-small"  # Your deployment name
)

# Single embedding
vector = await embedder.create("Hello, world!")
print(len(vector))  # Full dimension from Azure

# Batch embeddings  
texts = [
    "First document",
    "Second document",
    "Third document"
]
vectors = await embedder.create_batch(texts)
print(len(vectors))  # 3

Constructor

azure_client · AsyncAzureOpenAI | AsyncOpenAI · required

Pre-configured Azure OpenAI client. Must be either:

• AsyncAzureOpenAI for native Azure SDK
• AsyncOpenAI with Azure v1 API endpoint

model · str · default: "'text-embedding-3-small'"

Azure deployment name for the embedding model.

Unlike OpenAIEmbedder, this client does NOT support embedding dimension configuration. It returns embeddings in their native dimensions.

Azure SDK Setup

Option 1: AsyncAzureOpenAI (Recommended)

from openai import AsyncAzureOpenAI
from graphiti_core.embedder import AzureOpenAIEmbedderClient

azure_client = AsyncAzureOpenAI(
    api_key="your-azure-api-key",
    api_version="2024-02-15-preview",
    azure_endpoint="https://your-resource.openai.azure.com"
)

embedder = AzureOpenAIEmbedderClient(
    azure_client=azure_client,
    model="text-embedding-3-small-deployment"  # Your deployment name
)

Option 2: AsyncOpenAI with Azure v1 Endpoint

from openai import AsyncOpenAI
from graphiti_core.embedder import AzureOpenAIEmbedderClient

# Using Azure's OpenAI v1 compatibility endpoint
openai_client = AsyncOpenAI(
    api_key="your-azure-api-key",
    base_url="https://your-resource.openai.azure.com/openai/deployments/your-deployment"
)

embedder = AzureOpenAIEmbedderClient(
    azure_client=openai_client,
    model="text-embedding-3-small"
)

Supported Models

All OpenAI embedding models available on Azure:

• text-embedding-3-small (1536 dims)
• text-embedding-3-large (3072 dims)
• text-embedding-ada-002 (1536 dims, legacy)

Use your Azure deployment name as the model parameter, not the base model name.

Methods

create()

Generate a single embedding vector.

vector = await embedder.create("Your text here")
print(len(vector))  # Native dimensions (e.g., 1536 for text-embedding-3-small)

Parameters:

• input_data (str | list[str] | Any): Input to embed

Returns: list[float] - Embedding vector in native dimensions

Input handling:

• String: Direct embedding
• List of strings: Embeds all, returns first
• Other types: Converts to string

create_batch()

Generate embeddings for multiple texts.

texts = ["Text 1", "Text 2", "Text 3"]
vectors = await embedder.create_batch(texts)
print(len(vectors))  # 3
print(len(vectors[0]))  # Native dimensions

Parameters:

• input_data_list (list[str]): List of texts to embed

Returns: list[list[float]] - List of embedding vectors

Input Type Handling

The embedder handles different input types:

# String (standard)
vector = await embedder.create("Hello, world!")

# List of strings (embeds all, returns first)
input_list = ["Hello", "world"]
vector = await embedder.create(input_list)
# Creates embeddings for both, returns embedding of first

# Other types (converted to string)
vector = await embedder.create(12345)
# Converts to ["12345"] and embeds

Implementation:

if isinstance(input_data, str):
    text_input = [input_data]
elif isinstance(input_data, list) and all(isinstance(item, str) for item in input_data):
    text_input = input_data
else:
    text_input = [str(input_data)]

response = await self.azure_client.embeddings.create(
    model=self.model,
    input=text_input
)
return response.data[0].embedding

Error Handling

import logging

logger = logging.getLogger(__name__)

try:
    vector = await embedder.create("text")
except Exception as e:
    logger.error(f"Error in Azure OpenAI embedding: {e}")
    raise

Common errors:

• Authentication errors: Invalid API key or endpoint
• Deployment not found: Wrong deployment name
• Rate limit errors: Quota exceeded
• Input validation errors: Invalid input format

Dimension Handling

Unlike OpenAIEmbedder, this client does NOT truncate dimensions:

# Returns full 1536 dimensions for text-embedding-3-small
vector = await embedder.create("text")
print(len(vector))  # 1536 (not truncated)

# If you need truncation, do it manually:
embedding_dim = 1024
vector = (await embedder.create("text"))[:embedding_dim]
print(len(vector))  # 1024

Or wrap in a custom class:

class TruncatedAzureEmbedder:
    def __init__(self, azure_client, model, embedding_dim):
        self.embedder = AzureOpenAIEmbedderClient(azure_client, model)
        self.embedding_dim = embedding_dim
    
    async def create(self, input_data):
        vector = await self.embedder.create(input_data)
        return vector[:self.embedding_dim]
    
    async def create_batch(self, input_data_list):
        vectors = await self.embedder.create_batch(input_data_list)
        return [v[:self.embedding_dim] for v in vectors]

Batch Processing

Efficiently process multiple texts:

# Single API call for 100 texts
texts = [f"Document {i}" for i in range(100)]
vectors = await embedder.create_batch(texts)
print(len(vectors))  # 100

Implementation:

response = await self.azure_client.embeddings.create(
    model=self.model,
    input=input_data_list
)
return [embedding.embedding for embedding in response.data]

Example: Document Embedding

from graphiti_core.embedder import AzureOpenAIEmbedderClient
from openai import AsyncAzureOpenAI
import os

# Setup Azure client
azure_client = AsyncAzureOpenAI(
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    api_version="2024-02-15-preview",
    azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT")
)

# Create embedder
embedder = AzureOpenAIEmbedderClient(
    azure_client=azure_client,
    model="text-embedding-3-small"  # Your deployment
)

# Prepare documents
documents = [
    "Artificial intelligence is transforming technology.",
    "Machine learning enables computers to learn from data.",
    "Natural language processing helps machines understand text."
]

# Generate embeddings
vectors = await embedder.create_batch(documents)

# Compute similarity
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

vectors_np = np.array(vectors)
similarity_matrix = cosine_similarity(vectors_np)

print("Document similarities:")
for i in range(len(documents)):
    for j in range(i + 1, len(documents)):
        print(f"Doc {i} <-> Doc {j}: {similarity_matrix[i][j]:.4f}")

Use with Graphiti

from graphiti_core import Graphiti
from graphiti_core.embedder import AzureOpenAIEmbedderClient
from openai import AsyncAzureOpenAI
import os

# Setup Azure client
azure_client = AsyncAzureOpenAI(
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    api_version="2024-02-15-preview",
    azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT")
)

# Create embedder
embedder = AzureOpenAIEmbedderClient(
    azure_client=azure_client,
    model="text-embedding-3-small-deployment"
)

# Initialize Graphiti with Azure embeddings
graphiti = Graphiti(
    uri="neo4j://localhost:7687",
    user="neo4j",
    password="password",
    embedder=embedder
)

# Azure embeddings used for all operations
await graphiti.add_episode(
    name="episode1",
    episode_body="Your text here...",
    source_description="source1"
)

Comparison with OpenAIEmbedder

Feature	OpenAIEmbedder	AzureOpenAIEmbedderClient
Client type	AsyncOpenAI	AsyncAzureOpenAI or AsyncOpenAI
Model parameter	Base model name	Azure deployment name
Dimension control	Yes (via config)	No (returns native dims)
API endpoint	api.openai.com	Azure resource endpoint
Configuration class	OpenAIEmbedderConfig	None (direct params)
Azure support	Via client param	Native

Performance Tips

1. Use batch processing: Always prefer create_batch() for multiple inputs
2. Monitor Azure quotas: Check deployment TPM/RPM limits
3. Choose appropriate model:
   • text-embedding-3-small for cost-effectiveness
   • text-embedding-3-large for better quality
4. Deploy in same region: Reduce latency by deploying Azure resources nearby
5. Use multiple deployments: Distribute load across deployments

Troubleshooting

Authentication Errors

# Verify API key and endpoint
azure_client = AsyncAzureOpenAI(
    api_key="your-key",  # From Azure portal
    api_version="2024-02-15-preview",
    azure_endpoint="https://your-resource.openai.azure.com"  # Full URL
)

Deployment Not Found

# Use your deployment name, not base model
embedder = AzureOpenAIEmbedderClient(
    azure_client=azure_client,
    model="my-embedding-deployment"  # Your deployment name
)

Rate Limiting

# Check Azure portal for quotas:
# - Tokens per minute (TPM)
# - Requests per minute (RPM)
# Implement backoff or use multiple deployments
import asyncio
from tenacity import retry, wait_exponential, stop_after_attempt

@retry(wait=wait_exponential(min=1, max=60), stop=stop_after_attempt(5))
async def embed_with_retry(text):
    return await embedder.create(text)