Cookbook

Chunking Large Documents with Contextualized Retrieval

Ingest documents larger than 10,000 characters using semantic chunking and LLM-powered contextualization

The graph.add endpoint has a 10,000-character limit per request. Split larger documents before ingestion.

This cookbook uses contextualized retrieval. A large language model adds document context to each chunk before Zep ingests it.

This approach produces richer knowledge graphs with better entity and relationship extraction compared to naive chunking.

Chunk documents with zep-ingest

Use ingest_documents for paragraph-aware chunking, optional LLM context, preview, and submission.

View the complete source code on GitHub: Python | TypeScript | Go

Overview

The ingestion pipeline follows these steps:

  1. Read the document from a text file
  2. Chunk the document into smaller pieces using paragraph-aware splitting
  3. Contextualize each chunk using an LLM to add situational context
  4. Add each chunk to Zep via graph.add

Setup

Install the required dependencies:

pip install zep-cloud openai python-dotenv

Initialize the clients:

import os
from openai import OpenAI
from zep_cloud.client import Zep
from dotenv import load_dotenv
load_dotenv()
openai_client = OpenAI(api_key=os.environ.get("OPENAI_API_KEY"))
zep_client = Zep(api_key=os.environ.get("ZEP_API_KEY"))

Chunking the Document

Alternative chunking libraries: If you prefer using an established library over the custom implementation below, consider LangChain, LlamaIndex, Unstructured, or Chonkie.

The chunking algorithm splits text at paragraph boundaries first, then falls back to sentence boundaries for long paragraphs. This preserves semantic coherence better than fixed-size splitting.

import re
from typing import Generator
def chunk_document(
    text: str,
    chunk_size: int = 500,
    chunk_overlap: int = 50
) -> Generator[tuple[int, str], None, None]:
    """
    Split a document into chunks with configurable size and overlap.
    Args:
        text: The full document text
        chunk_size: Maximum characters per chunk (default 500)
        chunk_overlap: Characters to overlap between chunks for continuity
    Yields:
        Tuple of (chunk_index, chunk_text)
    """
    if not text:
        return
    text = text.strip()
    paragraphs = text.split('\n\n')
    current_chunk = ""
    chunk_index = 0
    for paragraph in paragraphs:
        paragraph = paragraph.strip()
        if not paragraph:
            continue
        # If adding this paragraph exceeds chunk_size, yield current chunk
        if len(current_chunk) + len(paragraph) + 2 > chunk_size:
            if current_chunk:
                yield (chunk_index, current_chunk.strip())
                chunk_index += 1
                # Start new chunk with overlap from previous
                if chunk_overlap > 0 and len(current_chunk) > chunk_overlap:
                    overlap_text = current_chunk[-chunk_overlap:]
                    first_space = overlap_text.find(' ')
                    if first_space > 0:
                        overlap_text = overlap_text[first_space + 1:]
                    current_chunk = overlap_text + "\n\n"
                else:
                    current_chunk = ""
            # Handle single paragraphs longer than chunk_size
            if len(paragraph) > chunk_size:
                for sub_chunk in split_long_paragraph(paragraph, chunk_size, chunk_overlap):
                    yield (chunk_index, sub_chunk)
                    chunk_index += 1
                current_chunk = ""
            else:
                current_chunk += paragraph
        else:
            if current_chunk:
                current_chunk += "\n\n" + paragraph
            else:
                current_chunk = paragraph
    # Yield final chunk
    if current_chunk.strip():
        yield (chunk_index, current_chunk.strip())
def split_long_paragraph(
    paragraph: str,
    chunk_size: int,
    chunk_overlap: int
) -> Generator[str, None, None]:
    """Split a long paragraph by sentences."""
    sentences = re.split(r'(?<=[.!?])\s+', paragraph)
    current_chunk = ""
    for sentence in sentences:
        if len(current_chunk) + len(sentence) + 1 > chunk_size:
            if current_chunk:
                yield current_chunk.strip()
                if chunk_overlap > 0:
                    overlap = current_chunk[-chunk_overlap:]
                    first_space = overlap.find(' ')
                    if first_space > 0:
                        current_chunk = overlap[first_space + 1:] + " "
                    else:
                        current_chunk = ""
                else:
                    current_chunk = ""
        current_chunk += sentence + " "
    if current_chunk.strip():
        yield current_chunk.strip()

Contextualizing Chunks

This is the key step that improves retrieval quality. For each chunk, we ask the LLM to generate a short context that situates it within the full document. This context is prepended to the chunk before adding to Zep.

Cost optimization: When contextualizing many chunks from the same document, use prompt caching with a repeated user-level data prefix. Do not place document content in a system or developer message. Reusing the document tokens can reduce inference time and cost.

def contextualize_chunk(
    openai_client: OpenAI,
    full_document: str,
    chunk: str
) -> str:
    """
    Use OpenAI to generate context for a chunk within its document.
    Args:
        openai_client: Initialized OpenAI client
        full_document: The complete document text
        chunk: The specific chunk to contextualize
    Returns:
        The contextualized chunk (context prepended to original chunk)
    """
    prompt = f"""<document>
{full_document}
</document>
Here is the chunk we want to situate within the whole document:
<chunk>
{chunk}
</chunk>
Please give a short succinct context to situate this chunk within the overall document for the purposes of improving search retrieval of the chunk. If the document has a publication date, please include the date in your context. Answer only with the succinct context and nothing else."""
    response = openai_client.chat.completions.create(
        model="gpt-5.6-terra",
        messages=[{"role": "user", "content": prompt}],
        max_completion_tokens=256
    )
    context = response.choices[0].message.content.strip()
    # Combine context with original chunk
    return f"{context}\n\n---\n\n{chunk}"

Adding Chunks to Zep

Each contextualized chunk is added to the user’s graph with graph.add. The method returns an episode that you can use to track ingestion.

These examples submit independent chunks. To group chunks with document_id, use a pre-release v4 SDK. The current v3 SDKs do not include this field. See Documents.

def add_chunk_to_zep(
    zep_client: Zep,
    user_id: str,
    chunk_data: str
) -> dict:
    """
    Add a contextualized chunk to Zep's graph.
    Args:
        zep_client: Initialized Zep client
        user_id: The user ID to add data to
        chunk_data: The contextualized chunk text
    Returns:
        The episode response from Zep
    """
    episode = zep_client.graph.add(
        user_id=user_id,
        type="text",
        data=chunk_data,
    )
    return episode

Complete Ingestion Pipeline

Here’s how to put it all together:

import os
def ingest_document(
    openai_client: OpenAI,
    zep_client: Zep,
    document_path: str,
    user_id: str,
    chunk_size: int = 500,
    chunk_overlap: int = 50
) -> dict:
    """
    Ingest a document into Zep with contextualized retrieval.
    Args:
        openai_client: Initialized OpenAI client
        zep_client: Initialized Zep client
        document_path: Path to the text document
        user_id: Zep user ID to add the document to
        chunk_size: Maximum characters per chunk
        chunk_overlap: Character overlap between chunks
    Returns:
        Summary statistics of the ingestion
    """
    # Read document
    with open(document_path, 'r', encoding='utf-8') as f:
        full_document = f.read()
    # If document fits in a single request, add directly
    if len(full_document) <= 10000:
        episode = zep_client.graph.add(
            user_id=user_id,
            type="text",
            data=full_document,
        )
        return {"total_chunks": 1, "successful": 1, "episodes": [episode.uuid_]}
    # Chunk the document
    chunks = list(chunk_document(full_document, chunk_size, chunk_overlap))
    stats = {"total_chunks": len(chunks), "successful": 0, "episodes": []}
    for chunk_index, chunk_text in chunks:
        # Contextualize the chunk
        contextualized = contextualize_chunk(
            openai_client,
            full_document,
            chunk_text
        )
        # Validate size after contextualization
        if len(contextualized) > 10000:
            # Truncate context if needed
            excess = len(contextualized) - 10000
            contextualized = contextualized[excess:]
        # Add to Zep
        episode = add_chunk_to_zep(zep_client, user_id, contextualized)
        stats["successful"] += 1
        stats["episodes"].append(episode.uuid_)
    return stats

Usage Example

# Ensure the user exists
user_id = "user123"
zep_client.user.add(user_id=user_id)
# Ingest a document
stats = ingest_document(
    openai_client=openai_client,
    zep_client=zep_client,
    document_path="company_handbook.txt",
    user_id=user_id,
    chunk_size=500,
    chunk_overlap=50
)
print(f"Ingested {stats['successful']} of {stats['total_chunks']} chunks")

Best practices

  • Chunk size: Use 500 characters or less for optimal graph construction. Smaller chunks allow Zep to capture more granular entities and relationships.
  • Chunk overlap: 50 characters helps maintain continuity between chunks without excessive redundancy.
  • Small chunks produce better graphs: Zep can capture more entities and relationships from smaller, focused chunks. While the 10K character limit allows larger chunks, smaller chunks yield richer knowledge graphs.

Further Reading