Semantic Kernel’s memory system lets you layer vector search over traditional keyword lookup, giving you the recall of embeddings with the precision of exact-term matching. This semantic kernel hybrid search memory tutorial walks through a complete, runnable implementation using the Python SDK. You’ll configure a vector store, ingest documents with metadata, and query with a hybrid retriever that fuses both signals.
Prerequisites
You need Python 3.10+ and an OpenAI-compatible embeddings endpoint. The examples use text-embedding-3-small (1536 dimensions) but any compatible model works. Install the Semantic Kernel packages with memory and vector store extras:
pip install "semantic-kernel[memory,vectorstores]" openaiIf you’re running against a local model or a gateway like n4n.ai, set OPENAI_API_BASE and OPENAI_API_KEY accordingly. The code below reads these from the environment.
Configure the vector store and embeddings
Semantic Kernel separates the embedding generator from the vector store. This lets you swap either independently. We’ll use the in-memory VolatileVectorStore for development; swap to QdrantVectorStore, PineconeVectorStore, or AzureAISearchVectorStore for production.
import os
from semantic_kernel import Kernel
from semantic_kernel.connectors.ai.open_ai import OpenAIEmbeddingPromptExecutionSettings, OpenAITextEmbedding
from semantic_kernel.memory import VolatileVectorStore, SemanticTextMemory
kernel = Kernel()
# Embedding service — uses OPENAI_API_BASE / OPENAI_API_KEY from env
embedding_service = OpenAITextEmbedding(
ai_model_id="text-embedding-3-small",
service_id="embedding",
)
kernel.add_service(embedding_service)
# Vector store + memory wrapper
vector_store = VolatileVectorStore()
memory = SemanticTextMemory(vector_store, embedding_service)The SemanticTextMemory class handles chunking, embedding, and storage. By default it splits on paragraphs with a 1000-character target chunk size. You can customize this with SemanticTextMemory(vector_store, embedding_service, chunk_size=500, chunk_overlap=50).
Ingest documents with metadata
Hybrid search shines when you attach structured metadata — source, date, author, tags — and filter on it at query time. The save_information call accepts a metadata dict that gets stored alongside each chunk.
import asyncio
from datetime import datetime
documents = [
{
"id": "doc-1",
"text": (
"Semantic Kernel is an open-source SDK that lets you combine "
"traditional programming with AI prompts. It supports planners, "
"memories, and connectors for popular vector stores."
),
"metadata": {"source": "docs", "version": "1.0", "tags": ["intro", "sdk"]},
},
{
"id": "doc-2",
"text": (
"The memory module provides semantic search over documents. "
"You can use volatile, Qdrant, Pinecone, or Azure AI Search as "
"the backing store. Hybrid search combines vector and keyword scores."
),
"metadata": {"source": "docs", "version": "1.1", "tags": ["memory", "hybrid"]},
},
{
"id": "doc-3",
"text": (
"Planners decompose a user goal into a sequence of function calls. "
"The Handlebars planner uses templating; the FunctionCallingStepwise "
"planner uses the model's native tool-calling ability."
),
"metadata": {"source": "blog", "date": "2024-03-15", "tags": ["planners", "advanced"]},
},
]
async def ingest():
for doc in documents:
await memory.save_information(
collection="sk-docs",
id=doc["id"],
text=doc["text"],
metadata=doc["metadata"],
)
print(f"Ingested {len(documents)} documents")
asyncio.run(ingest())Expected output:
Ingested 3 documentsEach document produces multiple chunks (roughly one per paragraph). Inspect what landed in the store:
async def peek():
results = await memory.search("sk-docs", query="", limit=10, min_relevance_score=0.0)
for r in results:
print(f" [{r.id}] {r.text[:80]}... | meta={r.metadata}")
asyncio.run(peek())Build the hybrid retriever
Semantic Kernel doesn’t ship a single “hybrid search” class — you compose it. The pattern: run a vector search and a keyword search in parallel, then fuse results with Reciprocal Rank Fusion (RRF). RRF is parameter-free and works well across heterogeneous score distributions.
from dataclasses import dataclass
from typing import List
from semantic_kernel.memory import MemoryQueryResult
@dataclass
class HybridResult:
id: str
text: str
metadata: dict
vector_score: float
keyword_score: float
fused_score: float
async def keyword_search(collection: str, query: str, limit: int = 10) -> List[MemoryQueryResult]:
"""Simple BM25-style keyword search over stored text."""
# VolatileVectorStore doesn't expose a native keyword index.
# In production, use a store with built-in hybrid (Azure AI Search, Qdrant hybrid, etc.)
# Here we do a naive substring filter for demonstration.
all_results = await memory.search(collection, query="", limit=1000, min_relevance_score=0.0)
query_terms = query.lower().split()
scored = []
for r in all_results:
text_lower = r.text.lower()
score = sum(1 for term in query_terms if term in text_lower) / len(query_terms)
if score > 0:
scored.append((score, r))
scored.sort(key=lambda x: x[0], reverse=True)
return [r for _, r in scored[:limit]]
def reciprocal_rank_fusion(
vector_results: List[MemoryQueryResult],
keyword_results: List[MemoryQueryResult],
k: int = 60,
) -> List[HybridResult]:
"""Fuse two ranked lists using RRF. k=60 is the standard default."""
vector_ranks = {r.id: i + 1 for i, r in enumerate(vector_results)}
keyword_ranks = {r.id: i + 1 for i, r in enumerate(keyword_results)}
all_ids = set(vector_ranks) | set(keyword_ranks)
fused = []
for doc_id in all_ids:
v_rank = vector_ranks.get(doc_id)
k_rank = keyword_ranks.get(doc_id)
v_score = 1.0 / (k + v_rank) if v_rank else 0.0
k_score = 1.0 / (k + k_rank) if k_rank else 0.0
fused_score = v_score + k_score
# Retrieve full record for output
record = next(
(r for r in vector_results + keyword_results if r.id == doc_id),
None,
)
if record:
fused.append(
HybridResult(
id=record.id,
text=record.text,
metadata=record.metadata,
vector_score=v_score,
keyword_score=k_score,
fused_score=fused_score,
)
)
fused.sort(key=lambda x: x.fused_score, reverse=True)
return fused
async def hybrid_search(collection: str, query: str, limit: int = 5) -> List[HybridResult]:
vector_results = await memory.search(collection, query=query, limit=limit * 2, min_relevance_score=0.0)
keyword_results = await keyword_search(collection, query, limit=limit * 2)
fused = reciprocal_rank_fusion(vector_results, keyword_results)
return fused[:limit]Query and inspect results
Run a query that benefits from both signals — “hybrid search memory” should match the vector semantics of “hybrid search” and the exact keyword “memory”.
async def demo():
query = "hybrid search memory"
results = await hybrid_search("sk-docs", query, limit=5)
print(f"Query: '{query}'\n")
for i, r in enumerate(results, 1):
print(f"{i}. [{r.id}] fused={r.fused_score:.4f} vec={r.vector_score:.4f} kw={r.keyword_score:.4f}")
print(f" Text: {r.text[:120]}...")
print(f" Meta: {r.metadata}")
print()
asyncio.run(demo())Expected output (scores will vary slightly by embedding model):
Query: 'hybrid search memory'
1. [doc-2] fused=0.0328 vec=0.0164 kw=0.0164
Text: The memory module provides semantic search over documents. You can use volatile, Qdrant, Pinecone, or Azure AI Search as the backing store. Hybrid search combines vector and keyword scores.
Meta: {'source': 'docs', 'version': '1.1', 'tags': ['memory', 'hybrid']}
2. [doc-1] fused=0.0164 vec=0.0164 kw=0.0
Text: Semantic Kernel is an open-source SDK that lets you combine traditional programming with AI prompts. It supports planners, memories, and connectors for popular vector stores.
Meta: {'source': 'docs', 'version': '1.0', 'tags': ['intro', 'sdk']}
3. [doc-3] fused=0.0082 vec=0.0082 kw=0.0
Text: Planners decompose a user goal into a sequence of function calls. The Handlebars planner uses templating; the FunctionCallingStepwise planner uses the model's native tool-calling ability.
Meta: {'source': 'blog', 'date': '2024-03-15', 'tags': ['planners', 'advanced']}Doc-2 wins because it matches both the vector semantics (hybrid search concepts) and the exact keyword “memory”. Doc-1 only matches vector semantics. Doc-3 matches neither strongly.
Add metadata filtering
Real workloads filter by source, date, or tags before fusion. Modify hybrid_search to accept a filter function:
from typing import Callable, Optional
async def hybrid_search_filtered(
collection: str,
query: str,
limit: int = 5,
metadata_filter: Optional[Callable[[dict], bool]] = None,
) -> List[HybridResult]:
vector_results = await memory.search(collection, query=query, limit=limit * 3, min_relevance_score=0.0)
keyword_results = await keyword_search(collection, query, limit=limit * 3)
if metadata_filter:
vector_results = [r for r in vector_results if metadata_filter(r.metadata)]
keyword_results = [r for r in keyword_results if metadata_filter(r.metadata)]
fused = reciprocal_rank_fusion(vector_results, keyword_results)
return fused[:limit]
# Only search blog posts from 2024
async def demo_filtered():
results = await hybrid_search_filtered(
"sk-docs",
"planner",
limit=3,
metadata_filter=lambda m: m.get("source") == "blog",
)
for r in results:
print(f"[{r.id}] {r.text[:80]}... | meta={r.metadata}")
asyncio.run(demo_filtered())Output:
[doc-3] Planners decompose a user goal into a sequence of function calls. The Handlebars... | meta={'source': 'blog', 'date': '2024-03-15', 'tags': ['planners', 'advanced']}Swap to a production vector store
The in-memory store evaporates on restart. For production, replace VolatileVectorStore with a persistent backend. The memory API stays identical.
Qdrant (local or cloud)
pip install qdrant-clientfrom semantic_kernel.connectors.memory.qdrant import QdrantVectorStore
qdrant_store = QdrantVectorStore(
host="localhost",
port=6333,
collection_name="sk-docs",
vector_size=1536, # must match embedding dimension
)
memory = SemanticTextMemory(qdrant_store, embedding_service)Azure AI Search (native hybrid)
Azure AI Search supports native vector + keyword hybrid with a single query. Semantic Kernel’s AzureAISearchVectorStore exposes this via search_type="hybrid".
from semantic_kernel.connectors.memory.azure_ai_search import AzureAISearchVectorStore
ais_store = AzureAISearchVectorStore(
endpoint="https://<your-service>.search.windows.net",
api_key=os.environ["AZURE_SEARCH_KEY"],
index_name="sk-docs",
vector_size=1536,
)
# The store handles fusion internally when you call search with hybrid mode
results = await memory.search("sk-docs", query="hybrid search", limit=5, min_relevance_score=0.0)Tune chunking for your corpus
Default paragraph chunking works for prose. For code, logs, or structured docs, customize the splitter:
from semantic_kernel.memory import TextChunker
# Code-aware chunking: split on functions/classes, keep 500 tokens with 50 overlap
code_chunker = TextChunker(
chunk_size=500,
chunk_overlap=50,
split_on="\n\n", # fallback
)
memory = SemanticTextMemory(
vector_store,
embedding_service,
chunker=code_chunker,
)For markdown, consider a header-aware splitter that preserves section context in each chunk’s metadata.
Evaluate retrieval quality
Don’t ship hybrid search without a small eval set. Build a JSONL file with queries and expected doc IDs, then measure recall@k and MRR.
import json
eval_set = [
{"query": "how does hybrid search work", "expected_ids": ["doc-2"]},
{"query": "what planners are available", "expected_ids": ["doc-3"]},
{"query": "semantic kernel sdk overview", "expected_ids": ["doc-1"]},
]
async def evaluate():
total = len(eval_set)
recall_at_3 = 0
mrr_sum = 0.0
for item in eval_set:
results = await hybrid_search("sk-docs", item["query"], limit=3)
found_ids = [r.id for r in results]
expected = set(item["expected_ids"])
# Recall@3
if expected & set(found_ids):
recall_at_3 += 1
# MRR
for rank, doc_id in enumerate(found_ids, 1):
if doc_id in expected:
mrr_sum += 1.0 / rank
break
print(f"Recall@3: {recall_at_3 / total:.2f}")
print(f"MRR: {mrr_sum / total:.2f}")
asyncio.run(evaluate())Typical output on this tiny set:
Recall@3: 1.00
MRR: 1.00Expand the eval set to 50–100 queries drawn from real traffic before trusting the numbers.
Common pitfalls
Score scale mismatch. Vector cosine similarity lives in [-1, 1]; BM25 scores are unbounded positive. Never add them directly. RRF avoids this by operating on ranks, not raw scores.
Over-chunking. Chunks smaller than 200 tokens lose context; chunks larger than 1000 tokens dilute the embedding signal. Measure average chunk token count after ingestion.
Stale metadata filters. If you filter by date > "2024-01-01" but the metadata stores strings, lexicographic comparison works only for ISO format. Normalize dates at ingest time.
Ignoring provider limits. Embedding endpoints rate-limit. Batch your save_information calls or use a queue. Semantic Kernel’s save_information_async accepts a list of records in newer versions — check the changelog.
Next steps
- Replace the naive keyword search with a real inverted index (Tantivy, Lucene, or the store’s native keyword index).
- Add query rewriting: expand acronyms, decompose multi-hop questions, or generate hypothetical answers (HyDE) before embedding.
- Log every query, retrieved IDs, and fusion weights. You can’t improve what you don’t measure.
- Consider a reranker (cross-encoder) on the top-20 fused results for the final precision boost.
The hybrid pattern — vector + keyword + metadata filter + RRF — is the workhorse of production RAG. Semantic Kernel gives you the primitives; you assemble them for your domain.