Semantic Kernel separates memory from model invocation, but wiring up semantic kernel embedding generation n4n.ai requires pointing the embedding connector at an OpenAI-compatible endpoint. This tutorial builds a working in-memory vector store from scratch, using Python and the n4n.ai gateway as the embedding backend.
Prerequisites
- Python 3.10 or newer
pip install semantic-kernel==1.13.0(pinned to avoid connector churn)- An API key for n4n.ai exported as
N4N_API_KEY - Comfort with async Python and
asyncio.run
Project setup
Create a clean environment and install the framework.
python -m venv .venv
source .venv/bin/activate
pip install "semantic-kernel==1.13.0"
The memory connector API stabilized in the 1.x line, but minor versions renamed classes. The code here targets 1.13.0.
Configure the embedding connector
Semantic Kernel ships OpenAITextEmbedding, which speaks the OpenAI embeddings response shape. Point its endpoint at the gateway and pass your key.
import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAITextEmbedding
endpoint = "https://api.n4n.ai/v1"
api_key = os.environ["N4N_API_KEY"]
embedding_gen = OpenAITextEmbedding(
ai_model_id="text-embedding-3-small",
api_key=api_key,
endpoint=endpoint,
)
kernel = Kernel()
kernel.add_service(embedding_gen)
ai_model_id is the model name exposed by the gateway. text-embedding-3-small returns 1536-dimensional vectors. If you later switch to a model with a different dimension, you must recreate the collection—vector length is fixed per store.
Create the vector store and memory
SemanticTextMemory binds a storage backend to an embedding generator. For a local demo, VolatileMemoryStore holds everything in process memory.
from semantic_kernel.connectors.memory.volatile_memory_store import VolatileMemoryStore
from semantic_kernel.memory.semantic_text_memory import SemanticTextMemory
store = VolatileMemoryStore()
memory = SemanticTextMemory(storage=store, embeddings_generator=embedding_gen)
Volatile means non-persistent. Kill the process and the vectors evaporate. Fine for tests, unacceptable for production.
Ingest documents
Store two short facts. Real pipelines chunk larger text and attach metadata.
import asyncio
async def ingest():
collection = "sk_docs"
await memory.save_information(
collection,
id="doc1",
text="Semantic Kernel is a framework for building LLM agents with plugins.",
)
await memory.save_information(
collection,
id="doc2",
text="Vector databases store embeddings to support cosine similarity search.",
additional_metadata={"source": "notes", "topic": "retrieval"},
)
asyncio.run(ingest())
save_information computes the embedding inside the async call, then writes the record. For bulk ingestion, wrap calls in a bounded semaphore to avoid saturating the endpoint.
Query semantic memory
Search returns closest records by cosine distance. Set limit and an optional min_relevance_score.
async def query():
collection = "sk_docs"
results = await memory.search(
collection,
"How do I find similar text efficiently?",
limit=1,
min_relevance_score=0.0,
)
for r in results:
print(f"id={r.id} score={r.relevance:.3f} text={r.text}")
if r.additional_metadata:
print(" meta:", r.additional_metadata)
asyncio.run(query())
Expected output
id=doc2 score=0.842 text=Vector databases store embeddings to support cosine similarity search.
meta: {'source': 'notes', 'topic': 'retrieval'}
The score is normalized similarity (1.0 = identical). With tiny synthetic sentences, the vector-DB line beats the framework line. Tune min_relevance_score to drop noise; 0.0 keeps all hits.
Inspecting stored vectors
Debug by dumping the raw collection.
async def dump():
records = await store.get_all("sk_docs")
for rec in records:
print(rec.id, "dims=", len(rec.embedding))
asyncio.run(dump())
Output:
doc1 dims=1536
doc2 dims=1536
A dimension mismatch after a model swap is your signal to flush the store.
Production notes
The volatile store is a prototype toy. Back SemanticTextMemory with a durable vector DB—Postgres + pgvector, Qdrant, or Redis. The pattern for semantic kernel embedding generation n4n.ai stays identical; only the storage constructor changes.
Chunking strategy
Never embed a 200-page PDF as one vector. Split on semantic boundaries (headers, paragraphs) with 10–15% overlap. Store chunk offsets in metadata so you can return source spans to the user.
Model selection
text-embedding-3-small is a sane default. For code-heavy corpora, pick a code-specific embedding from the gateway catalog. Larger dimensions improve recall at the cost of RAM and latency.
Error handling
Network calls to the embedding endpoint fail. Wrap save_information in a retry with exponential backoff—Semantic Kernel does not retry for you.
Testing without the network
In unit tests, substitute a fake embedding generator that returns deterministic vectors. This keeps CI fast and avoids burning tokens.
class FakeEmbedding:
async def generate_embeddings(self, texts, **kwargs):
return [[0.1] * 1536 for _ in texts]
# memory = SemanticTextMemory(storage=store, embeddings_generator=FakeEmbedding())
Full script
import os
import asyncio
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAITextEmbedding
from semantic_kernel.connectors.memory.volatile_memory_store import VolatileMemoryStore
from semantic_kernel.memory.semantic_text_memory import SemanticTextMemory
endpoint = "https://api.n4n.ai/v1"
api_key = os.environ["N4N_API_KEY"]
embedding_gen = OpenAITextEmbedding(
ai_model_id="text-embedding-3-small",
api_key=api_key,
endpoint=endpoint,
)
kernel = Kernel()
kernel.add_service(embedding_gen)
store = VolatileMemoryStore()
memory = SemanticTextMemory(storage=store, embeddings_generator=embedding_gen)
async def main():
collection = "sk_docs"
await memory.save_information(collection, id="doc1", text="Semantic Kernel is a framework for building LLM agents with plugins.")
await memory.save_information(collection, id="doc2", text="Vector databases store embeddings to support cosine similarity search.", additional_metadata={"source": "notes"})
results = await memory.search(collection, "How do I find similar text efficiently?", limit=1)
for r in results:
print(f"id={r.id} score={r.relevance:.3f} text={r.text}")
if __name__ == "__main__":
asyncio.run(main())
Run it:
python memory_demo.py
You now have a minimal but real semantic memory pipeline. Swap the store, adjust chunking, add retries, and ship.