Most teams treat embeddings as a black box POST request until they hit rate limits or dimension mismatches in production. This walkthrough covers embeddings api integration python end to end: installing the client, shaping inputs, batching, handling failures, and caching vectors without surprising your bill.
Setting up the client
The official openai SDK is the path of least resistance for embeddings api integration python. It speaks the OpenAI-compatible REST contract, which is also implemented by several gateways and self-hosted models.
pip install openaiInstantiate a client with your key. If you route through a gateway that exposes a single OpenAI-compatible endpoint, set base_url once and forget about provider-specific URLs.
from openai import OpenAI
client = OpenAI(
api_key="sk-...",
# base_url="https://api.n4n.ai/v1" # optional gateway
)Wrap this in a module-level singleton. Creating a client per request adds TLS handshake overhead and defeats connection pooling.
Choosing a model and making the first call
Embedding models differ in dimension, multilingual coverage, and price per token. For English-centric search, text-embedding-3-small (1536 dims) is a sane default. The call shape is identical across most compatible services.
resp = client.embeddings.create(
model="text-embedding-3-small",
input="What is the capital of France?",
)
vec = resp.data[0].embedding
assert len(vec) == 1536If you only need coarse similarity, request a truncated dimension to save storage:
resp = client.embeddings.create(
model="text-embedding-3-small",
input="Hello world",
dimensions=512,
)Tradeoff: lowering dimensions is a linear projection, not a magic compression. Recall on niche vocabularies drops. Benchmark on your own eval set before shipping.
Batching and input preparation
Sending one sentence per HTTP request will throttle your throughput and inflate latency. The API accepts a list of strings (or token arrays) in a single call.
texts = ["doc chunk 1", "doc chunk 2", "doc chunk 3"]
resp = client.embeddings.create(
model="text-embedding-3-small",
input=texts,
)
embeddings = [d.embedding for d in resp.data]The response data array preserves input order, so embeddings[i] maps to texts[i]. Do not rely on async unordered returns unless you pass user metadata and reconcile manually.
Provider limits still apply: OpenAI caps each input at 8191 tokens and recommends ≤2048 items per batch. In practice, keep batches between 100–500 items and parallelize with a worker pool.
from concurrent.futures import ThreadPoolExecutor
def embed_batch(client, batch):
return client.embeddings.create(model="text-embedding-3-small", input=batch)
def embed_all(client, texts, size=200):
with ThreadPoolExecutor(max_workers=4) as ex:
futures = [ex.submit(embed_batch, client, texts[i:i+size]) for i in range(0, len(texts), size)]
for f in futures:
yield from f.result().dataPre-chunk long documents before embedding. A 50-page PDF as a single string wastes tokens and degrades retrieval because the vector becomes a blurry average.
Chunking long documents
A 10k-token article shouldn’t be one embedding. Use a sliding window with overlap to preserve context across boundaries.
def chunk(text, max_tokens=500, overlap=50):
words = text.split()
step = max_tokens - overlap
for i in range(0, len(words), step):
yield " ".join(words[i:i+max_tokens])Overlap recovers context split across boundaries but multiplies embedding cost. Tune overlap to 10–15% of chunk size. Good embeddings api integration python code isolates chunking from the API call so you can swap strategies without touching retry or cache logic.
Handling errors and rate limits
Transient 429s are not exceptions; they are a scheduling signal. Wrap calls with exponential backoff. The SDK surfaces RateLimitError and APIError.
from openai import RateLimitError, APIError
import time
def embed_with_retry(client, texts, max_retries=4):
for attempt in range(max_retries):
try:
return client.embeddings.create(model="text-embedding-3-small", input=texts)
except RateLimitError:
time.sleep(2 ** attempt + 0.1)
except APIError as e:
if e.status_code >= 500:
time.sleep(2 ** attempt)
else:
raise
raise RuntimeError("embedding retry budget exhausted")If you sit behind a gateway such as n4n.ai, automatic fallback to a secondary provider on degradation means a plain retry often succeeds without you writing provider-specific logic. Still, cap retries to avoid queue buildup.
For batch jobs, persist progress (last processed offset) so a crash doesn’t force re-embedding millions of rows.
Caching and idempotency
Embeddings for a given model and text are deterministic. Cache them. The key must include the model identifier and any dimension override, because swapping models silently invalidates vectors.
import hashlib, json
def cache_key(model, text, dims=None):
raw = f"{model}|{dims}|{text}"
return hashlib.sha256(raw.encode()).hexdigest()
# Redis example
import redis
r = redis.Redis()
def cached_embed(client, model, text):
key = cache_key(model, text)
hit = r.get(key)
if hit:
return json.loads(hit)
resp = client.embeddings.create(model=model, input=text).data[0].embedding
r.set(key, json.dumps(resp), ex=86400*30)
return respLocal in-memory caches work for scripts; Redis or a Postgres column works for services. Watch token-count drift: if you change chunking strategy, old cache entries become lies. Version the cache key with a schema tag.
Production: dimensions, normalization, storage
Cosine similarity is the standard metric, but many vector stores accelerate dot product. Normalize vectors to unit length so dot product equals cosine.
import numpy as np
def normalize(v):
arr = np.array(v, dtype=np.float32)
return arr / np.linalg.norm(arr)Store the normalized vector. If you use pgvector:
CREATE TABLE docs (
id serial PRIMARY KEY,
content text,
embedding vector(1536)
);
CREATE INDEX ON docs USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);Query:
cur.execute(
"SELECT id, content FROM docs ORDER BY embedding <=> %s LIMIT 5",
(normalize(query_vec).tolist(),)
)IVFFlat needs a training step and isn’t exact; HNSW is better for dynamic data but heavier on RAM. Choose based on update frequency.
Dimension mismatches
Never mix embeddings from different models in one column. A text-embedding-3-small vector and a bge-large vector are numerically incompatible. Enforce at the application layer with a model registry.
Monitoring embedding drift
Models get updated; cosine distances shift. Log mean similarity of query to top doc weekly. If it drops >5% without a data change, investigate model version or prompt drift. Per-token usage metering (available on some gateways) helps attribute cost spikes to re-embedding jobs versus live traffic.
Common pitfalls
Token truncation silently. The API truncates inputs over the limit; it does not error. Your long contract clause gets cut, and retrieval misses it. Split text explicitly.
Embedding queries with a different model than documents. If docs used text-embedding-3-small and you embed the query with a local MiniLM, similarity scores are noise.
Ignoring per-token cost at scale. Embedding 10M support tickets once is a line item; re-embedding because of a bug is a second line item. Cache and version.
Assuming batch order is safe under manual concurrency. If you shard requests and merge responses by index, a dropped item shifts everything. Use the provided order or attach stable IDs.
Storing vectors without the source text. Debugging retrieval requires the original chunk. Keep content alongside embedding.
Not setting timeouts. The SDK defaults to a generous timeout. In a web request path, set max_retries=0 and a 2s timeout on the client, handle failure, and return degraded results.
A minimal end-to-end snippet
from openai import OpenAI
import numpy as np
client = OpenAI()
def embed(texts, model="text-embedding-3-small", dims=1536):
resp = client.embeddings.create(model=model, input=texts, dimensions=dims)
return [np.array(d.embedding, dtype=np.float32) for d in resp.data]
# embed a query and three docs
query_vec = embed(["refund policy"])[0]
doc_vecs = embed(["we refund in 30 days", "no returns on sale", "contact support"])
sims = [np.dot(query_vec, v) / (np.linalg.norm(query_vec)*np.linalg.norm(v)) for v in doc_vecs]That is the core of embeddings api integration python. The rest is plumbing: chunking, caching, storing, and monitoring drift.