Semantic Kernel memory tutorial: chunking documents

Semantic Kernel chunking documents memory tutorial content often skips the hard parts: choosing a strategy that survives real-world document variety, handling overlap without blowing up token budgets, and wiring the output into a vector store that doesn’t require constant babysitting. This guide walks through an ordered path from raw files to searchable chunks, with working code and the tradeoffs you’ll hit in production.

Understanding the chunking problem

Chunking is the step where raw text becomes retrieval units. Get it wrong and you either drown the model in irrelevant context or lose the answer entirely. Semantic Kernel Memory (SKM) provides abstractions, but the decisions — size, overlap, splitting logic — remain yours.

The core tension: larger chunks preserve context but waste tokens and dilute relevance scores. Smaller chunks increase precision but fragment concepts across boundaries. Most production systems settle between 300-800 tokens with 10-20% overlap, but your document type dictates the real answer.

Setting up the memory pipeline

Start with a minimal SKM configuration. You need a text embedding generator and a vector store. This example uses OpenAI embeddings and a local VolatileMemoryStore for development — swap the store for Qdrant, Pinecone, or Azure AI Search in production.

# requirements.txt
semantic-kernel[azure,openai,memory]
openai
tiktoken

import asyncio
from semantic_kernel import Kernel
from semantic_kernel.memory import SemanticTextMemory, VolatileMemoryStore
from semantic_kernel.connectors.ai.open_ai import OpenAITextEmbeddingGenerationService

async def build_memory() -> SemanticTextMemory:
    kernel = Kernel()
    
    # Embedding service — use text-embedding-3-small for cost/performance
    embedding_service = OpenAITextEmbeddingGenerationService(
        ai_model_id="text-embedding-3-small",
        api_key="your-key-here"
    )
    kernel.add_service(embedding_service)
    
    # Volatile store for dev; replace with persistent store for prod
    memory_store = VolatileMemoryStore()
    
    memory = SemanticTextMemory(
        storage=memory_store,
        embeddings_generator=embedding_service
    )
    
    return memory

Choosing a chunking strategy

SKM doesn’t enforce a chunker. You bring your own. Three strategies cover most cases:

Fixed-size token chunking

Predictable, fast, works well for homogeneous content like API docs or legal contracts. Use tiktoken to count tokens accurately.

import tiktoken
from typing import List

def chunk_by_tokens(
    text: str,
    max_tokens: int = 512,
    overlap_tokens: int = 64,
    encoding_name: str = "cl100k_base"
) -> List[str]:
    """Split text into overlapping token windows."""
    encoding = tiktoken.get_encoding(encoding_name)
    tokens = encoding.encode(text)
    
    if len(tokens) <= max_tokens:
        return [text]
    
    chunks = []
    start = 0
    while start < len(tokens):
        end = min(start + max_tokens, len(tokens))
        chunk_tokens = tokens[start:end]
        chunks.append(encoding.decode(chunk_tokens))
        
        if end == len(tokens):
            break
        start += max_tokens - overlap_tokens
    
    return chunks

Tradeoff: Splits mid-sentence, mid-table, mid-code-block. Overlap mitigates but doesn’t eliminate context loss at boundaries.

Recursive character splitting (LangChain-style)

Respects document structure by trying separators in order: \n\n, \n, . , , ``. Better for mixed-content PDFs, web pages, markdown.

from typing import List

def chunk_recursively(
    text: str,
    max_chars: int = 2000,
    overlap_chars: int = 200,
    separators: List[str] = None
) -> List[str]:
    """Split recursively by separators, then by character count."""
    if separators is None:
        separators = ["\n\n", "\n", ". ", " ", ""]
    
    def _split(text: str, seps: List[str]) -> List[str]:
        if not seps:
            # Fallback: hard character split
            return [text[i:i+max_chars] for i in range(0, len(text), max_chars - overlap_chars)]
        
        sep = seps[0]
        parts = text.split(sep)
        
        # If splitting worked and parts are small enough, recurse on each
        if len(parts) > 1 and all(len(p) <= max_chars for p in parts):
            result = []
            for part in parts:
                result.extend(_split(part, seps[1:]))
            return result
        
        # Otherwise try next separator
        return _split(text, seps[1:])
    
    initial_chunks = _split(text, separators)
    
    # Merge small chunks with overlap
    merged = []
    buffer = ""
    for chunk in initial_chunks:
        if len(buffer) + len(chunk) <= max_chars:
            buffer += chunk
        else:
            if buffer:
                merged.append(buffer)
            buffer = chunk
    if buffer:
        merged.append(buffer)
    
    return merged

Tradeoff: More complex, slower, but preserves semantic boundaries. Overlap implementation is trickier — the merge step above is naive.

Semantic chunking (embedding-based)

Cluster adjacent sentences by embedding similarity. Split where similarity drops. Highest quality, highest cost, non-deterministic.

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity

async def chunk_semantically(
    text: str,
    embedding_service,
    max_tokens: int = 512,
    similarity_threshold: float = 0.75
) -> List[str]:
    """Split where semantic similarity between adjacent sentences drops."""
    import nltk
    nltk.download('punkt_tab', quiet=True)
    from nltk.tokenize import sent_tokenize
    
    sentences = sent_tokenize(text)
    if len(sentences) <= 1:
        return [text]
    
    # Get embeddings for all sentences
    embeddings = []
    for sent in sentences:
        emb = await embedding_service.generate_embeddings([sent])
        embeddings.append(emb[0])
    
    embeddings = np.array(embeddings)
    
    # Find split points
    chunks = []
    current_chunk = [sentences[0]]
    current_tokens = len(tiktoken.get_encoding("cl100k_base").encode(sentences[0]))
    
    for i in range(1, len(sentences)):
        sim = cosine_similarity([embeddings[i-1]], [embeddings[i]])[0][0]
        sent_tokens = len(tiktoken.get_encoding("cl100k_base").encode(sentences[i]))
        
        # Split if similarity drops OR token limit reached
        if sim < similarity_threshold or current_tokens + sent_tokens > max_tokens:
            chunks.append(" ".join(current_chunk))
            current_chunk = [sentences[i]]
            current_tokens = sent_tokens
        else:
            current_chunk.append(sentences[i])
            current_tokens += sent_tokens
    
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    
    return chunks

Tradeoff: Requires an embedding call per sentence. Expensive for large corpora. Non-deterministic — same input can yield different chunks across runs. Cache embeddings if you retry.

Ingesting documents into Semantic Kernel Memory

Once you have chunks, write them to memory with metadata. The collection name acts as your namespace — use it for tenant isolation, document type, or versioning.

async def ingest_document(
    memory: SemanticTextMemory,
    collection: str,
    doc_id: str,
    text: str,
    chunker,
    metadata: dict = None
) -> List[str]:
    """Chunk a document and store each chunk with metadata."""
    chunks = chunker(text)
    chunk_ids = []
    
    for i, chunk in enumerate(chunks):
        chunk_id = f"{doc_id}#chunk-{i}"
        
        # Build metadata payload
        chunk_metadata = {
            "doc_id": doc_id,
            "chunk_index": i,
            "total_chunks": len(chunks),
            "char_count": len(chunk),
            "token_count": len(tiktoken.get_encoding("cl100k_base").encode(chunk)),
        }
        if metadata:
            chunk_metadata.update(metadata)
        
        await memory.save_information(
            collection=collection,
            id=chunk_id,
            text=chunk,
            description=json.dumps(chunk_metadata)  # SKM stores description as string
        )
        chunk_ids.append(chunk_id)
    
    return chunk_ids

Pitfall: SKM’s description field is a string, not structured metadata. If you need filterable fields (author, date, category), either serialize JSON in description or use a vector store that supports payload filtering (Qdrant, Pinecone, Azure AI Search) directly.

Searching with context reconstruction

Retrieval returns chunks. But the LLM often needs surrounding context. Two patterns:

Pattern 1: Return adjacent chunks at query time

Store chunk index in metadata. When a chunk hits, fetch neighbors.

async def search_with_context(
    memory: SemanticTextMemory,
    collection: str,
    query: str,
    limit: int = 5,
    context_window: int = 1
) -> List[dict]:
    """Retrieve chunks plus neighbors for context."""
    results = await memory.search(
        collection=collection,
        query=query,
        limit=limit,
        min_relevance_score=0.7
    )
    
    enriched = []
    for result in results:
        # Parse stored metadata
        meta = json.loads(result.description) if result.description else {}
        doc_id = meta.get("doc_id")
        chunk_idx = meta.get("chunk_index", 0)
        total = meta.get("total_chunks", 1)
        
        # Fetch neighbors
        context_chunks = [result.text]
        for offset in range(-context_window, context_window + 1):
            if offset == 0:
                continue
            neighbor_idx = chunk_idx + offset
            if 0 <= neighbor_idx < total:
                neighbor_id = f"{doc_id}#chunk-{neighbor_idx}"
                neighbor = await memory.get(collection, neighbor_id)
                if neighbor:
                    context_chunks.insert(0 if offset < 0 else len(context_chunks), neighbor.text)
        
        enriched.append({
            "text": "\n\n".join(context_chunks),
            "score": result.relevance,
            "metadata": meta
        })
    
    return enriched

Pattern 2: Parent-document retrieval (recommended for production)

Store two collections: one with small chunks for retrieval, one with large sections (or full documents) for generation. Link them via doc_id and section_id.

async def ingest_parent_child(
    memory: SemanticTextMemory,
    doc_id: str,
    full_text: str,
    child_chunker,
    parent_chunker
):
    """Store both granular chunks and parent sections."""
    # Children: small chunks for retrieval
    child_chunks = child_chunker(full_text)
    for i, chunk in enumerate(child_chunks):
        await memory.save_information(
            collection="chunks",
            id=f"{doc_id}#child-{i}",
            text=chunk,
            description=json.dumps({"doc_id": doc_id, "parent_index": i // 3})  # 3 children per parent
        )
    
    # Parents: larger sections for generation context
    parent_chunks = parent_chunker(full_text)
    for i, chunk in enumerate(parent_chunks):
        await memory.save_information(
            collection="parents",
            id=f"{doc_id}#parent-{i}",
            text=chunk,
            description=json.dumps({"doc_id": doc_id, "parent_index": i})
        )

async def retrieve_parent_context(
    memory: SemanticTextMemory,
    query: str,
    child_limit: int = 10
) -> List[str]:
    """Retrieve child chunks, map to unique parents, return parent text."""
    child_results = await memory.search("chunks", query, limit=child_limit, min_relevance_score=0.65)
    
    parent_indices = set()
    for result in child_results:
        meta = json.loads(result.description) if result.description else {}
        parent_indices.add(meta.get("parent_index", 0))
    
    parents = []
    for idx in sorted(parent_indices):
        parent = await memory.get("parents", f"{meta['doc_id']}#parent-{idx}")
        if parent:
            parents.append(parent.text)
    
    return parents

Tradeoff: Double storage, double ingestion cost. But generation context is cleaner — no stitching fragments, no duplicate overlap text.

Handling document formats

Real systems ingest PDFs, Word, HTML, markdown, code. Don’t chunk raw binary. Extract text first, preserve structure where it helps.

import pypdf
from docx import Document
from bs4 import BeautifulSoup
import markdown

def extract_text(file_path: str, mime_type: str) -> str:
    """Extract plain text from common formats."""
    if mime_type == "application/pdf":
        with open(file_path, "rb") as f:
            reader = pypdf.PdfReader(f)
            return "\n\n".join(page.extract_text() or "" for page in reader.pages)
    
    elif mime_type == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
        doc = Document(file_path)
        return "\n\n".join(p.text for p in doc.paragraphs if p.text.strip())
    
    elif mime_type == "text/html":
        with open(file_path, "r", encoding="utf-8") as f:
            soup = BeautifulSoup(f.read(), "html.parser")
            # Remove scripts, styles, nav
            for tag in soup(["script", "style", "nav", "footer", "header"]):
                tag.decompose()
            return soup.get_text(separator="\n\n", strip=True)
    
    elif mime_type == "text/markdown":
        with open(file_path, "r", encoding="utf-8") as f:
            html = markdown.markdown(f.read())
            soup = BeautifulSoup(html, "html.parser")
            return soup.get_text(separator="\n\n", strip=True)
    
    else:
        # Plain text, code, etc.
        with open(file_path, "r", encoding="utf-8") as f:
            return f.read()

Pitfall: PDF extraction is lossy. Tables, columns, and multi-page figures break. For high-value PDFs, consider a layout-aware extractor (Azure Document Intelligence, Unstructured.io) and chunk at the element level, not page level.

Token budgeting and cost control

Embedding costs scale with token count. Chunking directly controls this. A 10,000-page corpus at 512 tokens/chunk with 20% overlap = ~23M tokens. At text-embedding-3-small pricing ($0.02/1M tokens), that’s ~$0.46 per embedding run. Re-embedding on schema changes adds up.

def estimate_embedding_cost(
    texts: List[str],
    model: str = "text-embedding-3-small",
    pricing_per_million: float = 0.02
) -> dict:
    """Estimate tokens and cost for a batch of texts."""
    encoding = tiktoken.get_encoding("cl100k_base")
    total_tokens = sum(len(encoding.encode(t)) for t in texts)
    cost = (total_tokens / 1_000_000) * pricing_per_million
    return {
        "total_chunks": len(texts),
        "total_tokens": total_tokens,
        "estimated_cost_usd": round(cost, 4)
    }

Optimization: Deduplicate before embedding. Hash chunks (xxhash or SHA256) and skip identical content. Common in versioned docs, boilerplate legal text, generated API references.

Common pitfalls

Evaluating chunking quality

Don’t guess. Measure.

async def evaluate_chunking(
    memory: SemanticTextMemory,
    collection: str,
    eval_queries: List[dict],  # [{"query": "...", "relevant_doc_ids": [...], "relevant_chunk_ids": [...]}]
    k: int = 5
) -> dict:
    """Compute recall@k for chunk retrieval."""
    hits = 0
    total = 0
    
    for item in eval_queries:
        results = await memory.search(collection, item["query"], limit=k, min_relevance_score=0.0)
        retrieved_ids = [r.id for r in results]
        
        # Check if any relevant chunk was retrieved
        relevant = set(item.get("relevant_chunk_ids", []))
        if relevant & set(retrieved_ids):
            hits += 1
        total += 1
    
    return {
        "recall_at_k": hits / total if total else 0,
        "total_queries": total,
        "hits": hits
    }

Run this after every chunking strategy change. A 5% recall drop isn’t worth a 20% token savings.

Production checklist

Before shipping:

1. Persistent vector store — VolatileMemoryStore loses data on restart. Use Qdrant, Pinecone, Weaviate, or Azure AI Search.
2. Idempotent ingestion — Re-running ingestion must not create duplicates. Upsert by deterministic chunk ID.
3. Monitoring — Log chunk counts, token totals, embedding latency, search latency, relevance scores per query.
4. Fallback embedding provider — If your primary embedding endpoint fails or rate-limits, have a secondary. This is where a gateway like n4n.ai helps — it routes embedding requests across providers with automatic fallback and per-token metering.
5. Versioning — Tag collections or chunk metadata with schema_version. When you change chunking logic, increment and re-ingest incrementally.
6. Deletion — Implement delete_document(doc_id) that removes all chunks across collections. GDPR/CCPA requires it.

Summary

Chunking is the fulcrum of RAG quality. Start with recursive character splitting at 512 tokens / 64 overlap for general text. Move to semantic chunking for heterogeneous corpora where boundaries matter. Use parent-child retrieval for generation context. Measure recall@k on a fixed eval set. Store metadata in your vector store’s payload, not in SKM’s description field. And never re-embed the whole corpus without a hash-based diff.

The code above runs. Adapt the chunkers, plug in your vector store, and you have a production ingestion path.