Generating Embeddings With OpenAI API

Unit 10 showed the basic call that turns a string into a vector. This lesson treats embedding generation as a small piece of production infrastructure: batching requests efficiently, handling errors and rate limits, choosing vector dimensions deliberately, and structuring the code so it can be reused across the rest of this unit without rewriting the API call every time.

The Embeddings Endpoint, in Practical Terms

The embeddings endpoint accepts one or more strings and returns one vector per string. The important practical detail is that it accepts a list of inputs in a single request, and doing so is dramatically more efficient than calling the API once per string.

from openai import OpenAI

client = OpenAI()

response = client.embeddings.create(
    model="text-embedding-4",
    input=["Return policy for electronics", "How to reset your password"],
)

for item in response.data:
    print(item.index, len(item.embedding))

Note: Field names on the response object (data, index, embedding, usage) and the exact default vector length for text-embedding-4 can change between model versions. Confirm current field names and dimensions against the official OpenAI API reference before relying on them in production code.

Why batch instead of looping over individual calls? Each API call carries fixed overhead — network round-trip time, TLS handshake, request parsing — independent of how much text is inside it. Sending 100 strings in one request pays that overhead once; sending them one at a time pays it 100 times. For a pipeline that embeds thousands of document chunks (Lesson 8 covers chunking), the difference between batched and unbatched calls is often the difference between a job that finishes in seconds and one that takes minutes and burns through rate limits.

Batching in Practice: Respecting Size Limits

Real workloads exceed what fits in a single request. The endpoint has both a maximum number of inputs per request and a maximum total token count per request, so a robust pipeline chunks its input list before sending it.

def batch(items: list, batch_size: int = 100):
    """Yield successive slices of `items`, each up to `batch_size` long."""
    for start in range(0, len(items), batch_size):
        yield items[start:start + batch_size]

def embed_texts(client, texts: list[str], model: str = "text-embedding-4",
                 batch_size: int = 100) -> list[list[float]]:
    """Embed a large list of texts by sending it in fixed-size batches.

    Returns a flat list of embeddings in the same order as `texts`.
    """
    all_embeddings: list[list[float]] = []
    for text_batch in batch(texts, batch_size):
        response = client.embeddings.create(model=model, input=text_batch)
        all_embeddings.extend(item.embedding for item in response.data)
    return all_embeddings

batch() is a small generator that slices a list into fixed-size chunks without copying the whole list into memory at once — useful when items is large. embed_texts() uses it to keep each API call within safe limits while still batching far more efficiently than one-call-per-text. The order is preserved because each batch's results are appended in the same sequence the inputs were sliced, and within a batch, the API guarantees the response data list corresponds positionally to the input list (each item also carries an index field for a stricter check).

Handling Rate Limits and Transient Errors

API calls fail sometimes — a rate limit is hit, a network blip occurs, the service returns a transient 5xx error. Code that assumes every call succeeds will crash a batch job on its first hiccup and lose all the progress made before it. The standard fix is a retry with exponential backoff: wait briefly, retry; if it fails again, wait longer, retry again, up to a maximum number of attempts.

import time
import random

def call_with_retry(func, max_attempts: int = 5, base_delay: float = 1.0):
    """Call `func()` with exponential backoff on failure.

    `func` takes no arguments — wrap the real call in a lambda or
    a small closure at the call site.
    """
    for attempt in range(1, max_attempts + 1):
        try:
            return func()
        except Exception as exc:
            if attempt == max_attempts:
                raise
            delay = base_delay * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
            print(f"Attempt {attempt} failed ({exc!r}); retrying in {delay:.1f}s")
            time.sleep(delay)

def embed_batch_with_retry(client, texts: list[str], model: str = "text-embedding-4"):
    return call_with_retry(
        lambda: client.embeddings.create(model=model, input=texts)
    )

Why exponential backoff instead of a fixed delay? If a rate limit was hit because too many requests arrived in a short window, retrying immediately (or after the same fixed delay every time) keeps hitting the same limit. Doubling the delay after each failure gives the server's rate-limit window time to clear, and the small random jitter (random.uniform(0, 0.5)) prevents many parallel workers from retrying in lockstep and re-triggering the same limit together. Capping max_attempts matters too — without a cap, a request that fails for a non-transient reason (a malformed input, an authentication problem) would retry forever instead of surfacing the real error.

What happens if this is removed? Without retry logic, embedding 50,000 document chunks becomes fragile: any single transient network error anywhere in the run stops the whole job, and the pipeline has to be restarted from scratch (or from wherever manual bookkeeping left off). At production scale, transient failures are not an edge case — they are a certainty over a large enough batch, so retry logic is not optional polish; it is a correctness requirement.

Choosing a Vector Dimension

Some embedding models, text-embedding-4 included, support requesting a shorter vector than the model's default by passing a dimensions parameter.

response = client.embeddings.create(
    model="text-embedding-4",
    input="Shorter vectors trade a little accuracy for speed and storage.",
    dimensions=512,
)

Note: Not every embedding model supports the dimensions parameter, and the default/maximum dimension size for text-embedding-4 should be confirmed against current OpenAI documentation before being hard-coded into a schema.

Why would you ask for fewer dimensions than the default? Vector size directly affects three costs: storage (each dimension is a float, and databases charge for it), memory (an index of a million 1536-dimension vectors is twice the size of the same index at 768 dimensions), and search speed (comparing shorter vectors is faster). If a shorter vector loses only a small amount of ranking accuracy for a given use case, it is often a good trade — Lesson 9 shows how to measure that accuracy loss quantitatively instead of guessing.

When should you keep the default (larger) dimension? When retrieval quality is the priority and the corpus is small enough that storage and search-speed costs are negligible, or when the application is still in the evaluation stage and you have not yet measured whether a smaller size hurts accuracy. Decide the dimension after running the kind of evaluation shown in Lesson 9, not before.

Building a Reusable Embedding Client

Tying batching, retries, and model/dimension choice together into one small wrapper keeps the rest of this unit's lessons from repeating the same boilerplate.

from dataclasses import dataclass

@dataclass
class EmbeddingClient:
    client: object
    model: str = "text-embedding-4"
    dimensions: int | None = None
    batch_size: int = 100

    def embed(self, texts: list[str]) -> list[list[float]]:
        results: list[list[float]] = []
        for text_batch in batch(texts, self.batch_size):
            kwargs = {"model": self.model, "input": text_batch}
            if self.dimensions is not None:
                kwargs["dimensions"] = self.dimensions
            response = call_with_retry(lambda: self.client.embeddings.create(**kwargs))
            results.extend(item.embedding for item in response.data)
        return results

Because EmbeddingClient takes client as a constructor argument rather than importing OpenAI internally, it can be tested with a fake object that mimics the real client's shape, without making network calls.

class FakeEmbeddingsAPI:
    def create(self, model, input, **kwargs):
        class Item:
            def __init__(self, embedding):
                self.embedding = embedding
        class Response:
            def __init__(self, items):
                self.data = items
        return Response([Item([0.1, 0.2, 0.3]) for _ in input])

class FakeClient:
    def __init__(self):
        self.embeddings = FakeEmbeddingsAPI()

def test_embedding_client_batches_and_preserves_count():
    fake = FakeClient()
    ec = EmbeddingClient(client=fake, batch_size=2)
    result = ec.embed(["a", "b", "c", "d", "e"])
    assert len(result) == 5
    assert all(len(vec) == 3 for vec in result)
    print("PASS: EmbeddingClient batches correctly against a fake API")

test_embedding_client_batches_and_preserves_count()

FakeEmbeddingsAPI and FakeClient stand in for the real openai.OpenAI client, returning a fixed fake vector for every input instead of calling the network. This dependency-injection pattern — passing the client in rather than constructing it inside the class — is what makes this test fast, deterministic, and free to run as often as needed, unlike a test that made real API calls.

Common Mistakes

• Calling the API once per string in a loop. This multiplies fixed per-request overhead and is far slower than batching the same inputs into fewer requests.
• No retry logic around network calls. Treating every embedding call as guaranteed to succeed turns a large batch job into something that fails unpredictably and has to be restarted by hand.
• Changing the embedding model or dimension after vectors are already stored. Vectors from different models (or different dimension settings) are not comparable to each other. Mixing them in the same index silently corrupts search results — Lesson 6 covers how to version stored embeddings to avoid this.

Best Practices

• Centralize embedding calls behind one function or class. A single EmbeddingClient-style wrapper makes it possible to change model, batch size, or retry policy in one place instead of hunting through every call site.
• Log or track token usage per batch. The response's usage information (subject to the field-name caveat above) is the basis for the cost tracking introduced in Unit 1 — apply the same habit here since embedding a large corpus is a real, measurable cost.
• Test embedding logic against a fake client, not the real API. Batching, retry, and error-handling logic can all be verified without spending money or depending on network availability, as shown above.