Create an Ingestion Pipeline

What is an ingestion pipeline?

An ingestion pipeline prepares and loads existing data into Zep in a defined order. It normalizes identities and timestamps, validates records, submits them to a graph, and monitors processing.

zep-ingest is a Python package for building these pipelines. Use it to import documents, conversations, structured records, known entities, and relationships.

This guide explains the package through a company knowledge graph example. The same workflow applies to customer data, research collections, product catalogs, support archives, and other domains.

The package ships runnable scripts and sample input data in its examples directory. Each script creates its own graph or user, ingests the bundled data, and ends with a search, so you can run a complete import before adapting one to your own sources.

Install and authenticate

zep-ingest depends on the Zep Cloud SDK, so one install brings in both packages:

$
pip install zep-ingest

LLM contextualization is optional. You only need an additional provider dependency when using LLMContextualizer to add document-level context to each chunk, covered at the end of the walkthrough:

$
pip install "zep-ingest[anthropic]"
$
# or
$
pip install "zep-ingest[openai]"

Every ingestion helper takes a Zep client as its first argument. Create one from your project API key before running any of the code below. The rest of this guide assumes this client:

1
import os
2
3
from zep_cloud.client import Zep
4
5
client = Zep(api_key=os.environ["ZEP_API_KEY"])

Example: Build a company knowledge graph

Imagine you are building an internal assistant that must answer questions such as:

• Who owns Customer Success?
• Which team maintains a product?
• What did the company decide about a launch?
• What do the handbook, Slack history, and meeting transcripts say about a policy?

The employee directory and org chart contain known entities and relationships. The handbook, Slack messages, meeting transcripts, and email contain source material for Zep to interpret.

Ingest them in this order:

1. Define canonical names and aliases.
2. Create the destination and set its ontology.
3. Seed people and teams from the employee directory with stable UUIDs.
4. Add reporting and ownership relationships from the org chart.
5. Prepare and preview the employee handbook using the canonical aliases.
6. Ingest the handbook and then the remaining documents and conversations in chronological order.

This gives later extraction an existing graph of typed, canonical entities to resolve against.

Choose how to ingest each source

Choose a feature according to what the source represents, not its file format alone. When your application already knows an entity or relationship, write it directly. When the source contains material Zep should interpret, ingest it as episodes or messages.

The directory and org chart should not go through extraction when they already contain the correct identities and relationships. Documents and conversations become episodes from which Zep extracts additional entities and facts.

Choose where the data lands

Every import targets either a named graph or a user graph, and that choice sets the data model for everything after it.

Use a user graph when the data belongs to one person’s memory. User graphs integrate directly with Zep’s context retrieval, so their data reaches an agent through that user’s context. Use a named graph for knowledge shared across users, such as a handbook, a product catalog, or an internal directory, which callers search explicitly. Users and user graphs and Create graph cover the difference in full.

Pass graph_id for a named graph or user_id for a user graph. Every path in the table above accepts either one, except ingest_thread_messages, which writes messages into threads and therefore takes user_id only.

This example uses a named graph, because company knowledge is not one user’s memory.

Build the pipeline step by step

1. Normalize aliases with AliasCanonicalizer

AliasCanonicalizer rewrites known aliases to one canonical entity name before Zep extracts entities. Use it when historical sources refer to the same entity in several ways.

The employee directory says the person’s full name is Sarah Brown and the official team name is Customer Success. Historical documents use several shorter forms. Define those mappings before loading documents and conversations.

1
ALIASES = {
2
    "Sarah Brown": ["Sarah B.", "S. Brown"],
3
    "Customer Success": ["Customer Success Team", "CS Team"],
4
}

Only include an alias when the mapping is valid across the source you are processing. The bare first name Sarah is deliberately absent: add it only when the source contains exactly one Sarah. The default safety guard does not catch that case. It rejects aliases shorter than three characters and a list of common words and word-like given names, but an ordinary first name passes.

2. Create the graph and set its ontology

The Zep ontology defines the entity and relationship types Zep should extract. This is a Zep Cloud SDK operation rather than a zep-ingest transform, but it must happen before the package submits data because ontology changes are not retroactive.

Prescribed types are prescriptive, not restrictive. They tell Zep which types to look for; they do not filter what is ingested. Entities that do not match a prescribed type are still ingested, typed under the ontology Zep learns from the data. Prescribing three types therefore does not reduce the graph to three types, and setting an ontology at all is optional.

This company graph needs people, teams, and leadership relationships. Create the graph and set that ontology before adding nodes, facts, or episodes.

zep-ingest writes only into a destination that already exists; it never creates one for you. Create a named graph with client.graph.create(graph_id=...), or a user graph with client.user.add(user_id=...), before any ingestion call targets it. See Create graph and Users and user graphs. This holds for ingest_thread_messages too: it raises ConfigurationError when the user does not exist, rather than creating a bare one that skips the profile Zep relies on to anchor that user’s identity. It does create the threads a backfill references, because those are containers for the import rather than a destination you provision.

1
from zep_cloud import EntityEdgeSourceTarget
2
from zep_cloud.external_clients.ontology import EdgeModel, EntityModel
3
4
5
class Person(EntityModel):
6
    """A named individual, such as an employee or customer contact."""
7
8
9
class Team(EntityModel):
10
    """A named organizational team or department."""
11
12
13
class Leads(EdgeModel):
14
    """A person leads, manages, or is responsible for a team."""
15
16
17
ONTOLOGY = {
18
    "entities": {
19
        "Person": Person,
20
        "Team": Team,
21
    },
22
    "edges": {
23
        "LEADS": (
24
            Leads,
25
            [EntityEdgeSourceTarget(source="Person", target="Team")],
26
        ),
27
    },
28
}
29
30
graph_id = "company-kb"
31
32
client.graph.create(graph_id=graph_id)
33
client.graph.set_ontology(
34
    entities=ONTOLOGY["entities"],
35
    edges=ONTOLOGY["edges"],
36
    graph_ids=[graph_id],
37
)

3. Add known entities with ingest_nodes

ingest_nodes directly creates or upserts entities through graph.add_nodes. Use it for canonical entities your application already knows, such as customers, employees, products, accounts, or teams. It bypasses entity extraction: the caller supplies the node’s name, type, attributes, and stable identity.

The directory already identifies Sarah Brown and the Customer Success team. Add them directly and persist their UUIDs so reruns upsert the same nodes.

Your source system does not need to use UUIDs already. If it identifies Sarah as EMP-1842, generate one UUIDv4 for that employee, store the EMP-1842 → UUID mapping in your application or import manifest, and reuse it on every run.

1
from zep_ingest import NodeItem, ingest_nodes
2
3
nodes = ingest_nodes(
4
    client,
5
    [
6
        NodeItem(
7
            uuid="c2b42bf8-5d3f-4c92-92a6-43d8854a1e5b",
8
            name="Sarah Brown",
9
            label="Person",
10
            summary="Customer success lead",
11
        ),
12
        NodeItem(
13
            uuid="79166282-3c4d-4d67-a6b6-4da1bd3a2f11",
14
            name="Customer Success",
15
            label="Team",
16
            summary="The customer success organization",
17
        ),
18
    ],
19
    graph_id=graph_id,
20
)
21
22
nodes.wait(timeout=600)
23
nodes.raise_for_status()

For direct node ingestion, the UUID is the upsert and deduplication key:

• Reusing a UUID updates the existing node.
• Sending a different UUID creates a different node, even when the name is identical.
• Omitting the UUID lets the API assign one, but ingest_nodes requires require_uuids=False because the operation is no longer safely repeatable.

4. Add known relationships with ingest_fact_triples

ingest_fact_triples writes a fact you already know between two entities through graph.add_fact_triple. Each triple contains a source entity, a target entity, the fact statement, the relationship type, and optionally when the fact became valid.

Use these fields to describe the relationship:

• source_node_name and target_node_name identify the two entities.
• fact is the human-readable assertion, such as Sarah Brown leads Customer Success.
• fact_name is the relationship type in SCREAMING_SNAKE_CASE, such as LEADS.
• source_node_uuid and target_node_uuid pin the relationship to existing nodes when their UUIDs are known.
• valid_at records when the relationship became true.

The org chart states that Sarah leads Customer Success. Add that relationship directly and pin each endpoint to its seeded UUID.

1
from zep_ingest import FactTriple, ingest_fact_triples
2
3
facts = ingest_fact_triples(
4
    client,
5
    [
6
        FactTriple(
7
            fact="Sarah Brown leads customer success",
8
            fact_name="LEADS",
9
            source_node_name="Sarah Brown",
10
            source_node_uuid="c2b42bf8-5d3f-4c92-92a6-43d8854a1e5b",
11
            target_node_name="Customer Success",
12
            target_node_uuid="79166282-3c4d-4d67-a6b6-4da1bd3a2f11",
13
            valid_at="2025-01-15T00:00:00Z",
14
        ),
15
    ],
16
    graph_id=graph_id,
17
)
18
19
facts.wait(timeout=600)
20
facts.raise_for_status()

As in step 3, wait() raises IngestUntrackedError if Zep returned no completion handle for a submitted triple.

See Manually updating the graph for the underlying node and fact-triple behavior.

5. Build and preview an episode Pipeline

Pipeline connects a source Loader to ordered transforms, validates the complete output, and selects a submitter. Use preview() to inspect the transformed episodes and warnings without making Zep API calls.

The episode pipeline is:

Loader → transforms → LimitGuard → Submitter

The handbook is source material Zep should interpret, so ingest it as episodes. Apply the company alias map before chunking. Pipeline.preview() runs the loader, transforms, and validation without making any Zep API calls.

TextFileLoader raises when its glob matches no files, so create the input before building the pipeline. A minimal handbook/customer-success.md that exercises the alias map:

# Customer success
The Customer Success Team owns renewals and escalations. Sarah B. leads the
team. Escalations that involve a billing dispute are routed to Finance within
one business day.

For a larger ready-made input, the package’s examples/data directory contains a sample handbook, Slack export, transcripts, email, and chat history.

1
from zep_ingest import (
2
    AliasCanonicalizer,
3
    Pipeline,
4
    TextChunker,
5
    TextFileLoader,
6
)
7
8
pipeline = Pipeline(
9
    TextFileLoader("handbook/**/*.md", created_at="2025-06-01T00:00:00Z"),
10
    transforms=[
11
        AliasCanonicalizer(ALIASES),
12
        TextChunker(chunk_size=500, overlap=50),
13
    ],
14
)
15
16
preview = pipeline.preview(limit=20)
17
18
for episode in preview.episodes:
19
    print(episode.data_type, episode.created_at, episode.data[:120])
20
21
for warning in preview.warnings:
22
    print(warning)

A sampled preview validates at most limit episodes, and its warning counts cover only that sample. Before a production import, run pipeline.preview(limit=None) for an exhaustive pass over the whole stream, so a problem that first appears in episode 5,000 is not missed by a 20-episode check.

6. Ingest documents and conversations

The source-specific helpers package common loaders and transforms into one-line ingestion functions. Use them for material Zep should interpret, such as documents, conversations, transcripts, email, and structured business records.

When the preview is correct, run the same pipeline against the graph that already contains the ontology, canonical nodes, and known relationships:

1
result = pipeline.run(
2
    client,
3
    graph_id=graph_id,
4
)
5
result.wait(timeout=3600)
6
7
print(result.method, result.status)
8
for warning in result.warnings:
9
    print(warning)

The pipeline validates and spools the full transformed stream before it submits the first episode. Large imports can spill to a temporary file instead of remaining in memory.

Once the handbook import behaves as expected, ingest the remaining company sources. Reuse the same alias map and preserve each source’s original timestamps:

1
from zep_ingest import ingest_emails, ingest_slack_export, ingest_transcripts
2
3
slack = ingest_slack_export(
4
    client,
5
    "exports/slack.zip",
6
    graph_id=graph_id,
7
    aliases=ALIASES,
8
)
9
slack.wait()
10
11
meetings = ingest_transcripts(
12
    client,
13
    "exports/meetings/**/*.vtt",
14
    graph_id=graph_id,
15
    aliases=ALIASES,
16
)
17
meetings.wait()
18
19
email = ingest_emails(
20
    client,
21
    "exports/email/**/*.eml",
22
    graph_id=graph_id,
23
    aliases=ALIASES,
24
)
25
email.wait()

Start with the oldest historical source and move forward. Within each source, keep events in their original order and retain their real created_at values.

Optional: add document context with an LLM

LLMContextualizer prepends document-level context to each chunk so that a chunk read on its own still identifies what it is about. It costs one LLM call per chunk, so weigh it against the size of your import.

LLMContextualizer is provider-independent. The package includes:

• AnthropicLLM for the Anthropic API;
• OpenAILLM for the OpenAI API;
• OpenAICompatibleLLM for any provider, proxy, or local model that exposes an OpenAI-compatible /chat/completions endpoint.

The openai extra installs the client library used by both OpenAILLM and OpenAICompatibleLLM. It does not restrict you to OpenAI models. For a compatible provider, supply its model name and full base URL. Include the API-version path required by that provider, such as /v1 or /v1beta/openai/.

For example, configure a Gemini model through Gemini’s OpenAI-compatible endpoint:

1
from zep_ingest.llm.openai import OpenAICompatibleLLM
2
3
gemini_llm = OpenAICompatibleLLM(
4
    model=os.environ["GEMINI_MODEL"],
5
    base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
6
    api_key=os.environ["GEMINI_API_KEY"],
7
)

Pass the adapter to a document helper:

1
from zep_ingest import ingest_documents
2
3
result = ingest_documents(
4
    client,
5
    "documents/**/*.md",
6
    graph_id=graph_id,
7
    llm=gemini_llm,
8
)
9
result.wait()

The same adapter works with OpenAI-compatible endpoints from providers and gateways such as LiteLLM, Ollama, vLLM, OpenRouter, Together, or Groq. If a provider does not offer that interface, pass a custom client that implements complete(prompt: str) -> str.

See Gemini’s OpenAI compatibility documentation for its current endpoint and model support.

Choose a submission method

The package submitter controls how episode pipelines reach Zep. auto chooses the best available path, while batch and sequential let you require a particular submission behavior.

Episode pipelines accept method="auto", method="batch", or method="sequential".

Use auto unless you need to require a specific path. Falling back is narrower than it sounds: it happens only when the deployment does not serve the Batch API at all. Any other refusal, such as a rejected key or an exhausted quota, is raised as it is, because sequential submission would meet the same refusal one call at a time. When the fallback does happen, the result reports it in warnings.

If Batch becomes unavailable after part of a multi-batch import has already been submitted, the package stops instead of resubmitting the complete source sequentially and creating duplicate episodes. See Recover from a partial import for what to do next.

Backfill thread messages

ingest_thread_messages writes historical application conversations to threads on a user graph. It requires the user to already exist, creates any missing threads, preserves the order of messages within each thread, and splits messages that exceed the thread-message limit.

1
from zep_ingest import ingest_thread_messages
2
3
result = ingest_thread_messages(
4
    client,
5
    "chat-history.jsonl",
6
    user_id="user-123",
7
    method="auto",
8
)

Each row must use exactly these fields, and no others:

1
{"thread_id": "support-4417", "role": "user", "name": "Sarah Brown", "content": "The renewal quote has the wrong term length.", "created_at": "2025-03-01T10:00:00Z", "metadata": {"channel": "web"}}

thread_id, role, name, content, and created_at are all required; only metadata is optional. Carrying the speaker type, speaker name, and original timestamp on every row is what lets a backfill reconstruct both the conversation and its fact validity timeline. role must be one of user, assistant, system, function, tool, or norole; a source-system value such as human is rejected. created_at is RFC3339, for example 2025-03-01T10:00:00Z.

Any other column is rejected outright rather than ignored, because silently dropping a misspelled field produces an import that looks complete and is not. Real chat exports usually carry extra columns such as an internal message ID or a channel name, so strip or rename them, and map source roles to the values above, before ingesting.

Thread IDs are unique across a Zep project. Use thread_id_suffix when an imported ID might collide with an existing thread. A suffix namespaces the import; it does not make reruns idempotent.

On the sequential path, messages_per_call sets how many messages go in each thread.add_messages call. It defaults to 30, which is also the maximum; a larger value is rejected when the arguments are checked, before the user and threads are touched.

Monitor ingestion with IngestResult

IngestResult gives every helper a consistent interface for processing status, errors, and monitoring identifiers across Batch, episode, and task-backed operations.

1
result.status
2
result.refresh()
3
result.wait(timeout=3600)
4
result.failed_items()
5
result.raise_for_status()
6
7
print(result.batch_ids)
8
print(result.episode_uuids)
9
print(result.task_ids)
10
print(result.untracked_items)

status aggregates every handle in the result and reports the least-complete one: failed, partial, canceled, untracked, processing, queued, or succeeded.

Persist the returned IDs if another process will monitor the import:

1
from zep_ingest import IngestResult
2
3
result = IngestResult.from_batch_ids(client, saved_batch_ids)
4
result.wait(timeout=3600)
5
6
task_result = IngestResult.from_task_ids(client, saved_task_ids)
7
task_result.wait(timeout=3600)

wait() polls the Batch, episode, or task handles returned by the operation until they reach a terminal state.

Some accepted writes come back without a completion handle. When that happens the result counts them in untracked_items, status reports untracked, and wait() raises IngestUntrackedError immediately rather than polling forever. The submission itself succeeded; only server-side extraction is untrackable. Catch the error and confirm the data landed with your own read, such as search_when_ready:

1
from zep_ingest import IngestUntrackedError, search_when_ready
2
3
try:
4
    result.wait(timeout=3600)
5
except IngestUntrackedError:
6
    search_when_ready(client, "Who leads customer success?", graph_id=graph_id)

Sequential thread ingestion tracks a task handle when thread.add_messages returns one, and counts the messages as untracked when it does not.

Search indexing can take additional time after processing succeeds. Use search_when_ready when a script must ingest and then immediately check retrieval:

1
from zep_ingest import search_when_ready
2
3
search_results = search_when_ready(
4
    client,
5
    "Who leads customer success?",
6
    graph_id="company-kb",
7
    scope="edges",
8
    timeout=120,
9
)

search_when_ready retries until the query returns any result or reaches the timeout. It does not verify that a specific imported record produced the result. Use a query unique to the imported data when checking indexing readiness. An empty result after the timeout is returned normally.

For lower-level polling and webhook options, see Check data ingestion status.

Recover from a partial import

The package stops on an unrecoverable submission error rather than resubmitting a source from the beginning, which would duplicate everything already accepted. It does not resume, so recovery is a procedure you run, not a flag you pass. It depends on the import manifest you recorded at submit time.

1. Establish what was accepted. Reconstruct the result from the identifiers in your manifest with IngestResult.from_batch_ids or IngestResult.from_task_ids, then call refresh() and read status.
2. List what failed. failed_items() returns the submission errors the package recorded and, for a Batch import, the per-item failures Zep reported. Each entry identifies the item, so you can map it back to a source record through your manifest.
3. Rebuild the input from the records your manifest does not show as accepted. Do not rerun the original source: reingesting an accepted record creates a second episode for it, because Zep does not deduplicate episodes by content.
4. Submit that remainder as a normal import, and record its identifiers in the manifest as well.

If the manifest is incomplete and you cannot tell which records landed, treat the destination as unusable rather than reingesting over it. Delete the affected data and start again. Deleting data from the graph describes what deleting an episode, node, or thread removes.

Current limitations

• Rerunning the same export creates new episodes. zep-ingest does not yet provide a source manifest, content-hash deduplication, or resume checkpoint.
• batch_ids, episode_uuids, and task_ids let you resume status monitoring. They do not resume parsing or submission from the middle of a source.
• Accepted writes are sometimes returned without a completion handle. The result counts them in untracked_items and wait() raises IngestUntrackedError, so confirming those items requires your own read.
• LLMContextualizer sends document content to the LLM client you provide. Confirm that the provider and deployment meet your data-handling requirements.
• Retrieval ranking, authority weighting, and contradiction policy are outside the package’s scope.

Review Prepare Data for Ingestion before running a production import.