zep-ingest

Bulk data ingestion pipeline for Zep. Owns everything upstream of the Zep API so getting unstructured and structured data into Context Graphs safely is a one-liner, with validation and preview warnings for the failure modes the package can detect before submission.

pip install zep-ingest
# optional LLM contextualization:
pip install "zep-ingest[anthropic]"   # or [openai]

# copy examples/example_ontology.py, adapt it
from example_ontology import EDGE_TYPES, ENTITY_TYPES
from zep_cloud.client import Zep
from zep_ingest import ingest_slack_export, ingest_documents, ingest_json_records

client = Zep(api_key="...")

# Setup is yours, once per graph: zep-ingest writes only into graphs that
# already exist and already carry their ontology (see Ontology below). The
# ontology is not retroactive, so set it before the first ingest.
graph_uuids = {}
for name in ("team_knowledge", "company_kb", "catalog"):
    graph = client.graph.create(name=name)
    graph_uuids[name] = graph.uuid_
    client.graph.set_ontology(graph.uuid_, entity_types=ENTITY_TYPES, edge_types=EDGE_TYPES)

ingest_slack_export(client, "slack-export.zip", graph_uuid=graph_uuids["team_knowledge"])
ingest_documents(client, "handbook/**/*.md", graph_uuid=graph_uuids["company_kb"])
ingest_json_records(client, "products.csv", graph_uuid=graph_uuids["catalog"], id_field="sku")

The v4 API addresses a graph, a user, and a thread by a server-generated UUID. A graph_id, a user_id, and a thread_id are names, not addresses. Every public parameter of this package therefore takes a UUID: graph_uuid and thread_uuid. The application creates the graph, the user, and the thread one time, stores each returned UUID, and passes the UUID here. The package does no identifier lookup at run time.

Why this package exists

The raw ingestion surface has sharp edges people hit constantly: a 10,000-character episode limit, DIY chunking, timestamps that silently default to ingestion time (corrupting fact timelines), entity aliases that never merge, a separate Batch API for bulk throughput, and rate limits. zep-ingest encodes all of those rules so you don't have to know them.

Graph setup stays outside the package: create the graph and set its ontology yourself, once, then ingest into it. See Ontology.

Choosing your path

Extraction vs explicit facts: narrative content (conversations, documents) goes through Zep's LLM extraction as episodes. When you already know the exact relationship ("Avery Brown is RESPONSIBLE for OPERATIONS-DASHBOARD"), assert it as a fact triple instead — no extraction variance, and pre-seeded canonical entities anchor later extraction.

User graph vs named graph: every episode one-liner takes graph_uuid=. A user graph has a UUID too: client.user.create(...) returns graph_uuid for that user, so business data and user data use the same parameter. A missing graph_uuid is a ConfigurationError before any API call. For a user's own conversations, use ingest_thread_messages — messages land on the user graph via threads (the same store thread.get_context() reads). Every message requires thread_uuid, role, name, and content (metadata and created_at are optional). The user and the thread must already exist: the package creates no user and no thread, and it resolves no name at run time. Messages over the 4,096-character thread-message limit are split at sentence boundaries, and per-thread order is preserved. method="auto" uses the Batch API and transparently falls back to sequential thread.add_messages only when the deployment doesn't serve the batch endpoint at all. Pass ignore_roles=["assistant"] to keep assistant turns as conversational context but exclude them from graph extraction (they stay in thread history); it applies on both the batch and sequential paths.

Reference time: the v4 API accepts a reference_time field on episodes, batch items, and thread messages. Episode.created_at and ThreadMessage.created_at are validated client-side and sent as reference_time on the batch and sequential paths alike.

Slack conversation types: a Slack export indexes its conversations in four files — channels.json (public channels), groups.json (private channels), dms.json (1:1 DMs) and mpims.json (group DMs). All four are read, but only public channels are ingested by default, so private conversations never start flowing into a graph unnoticed. Whatever the export contains and conversation_types= does not select is reported in result.warnings with counts and the exact parameter to pass — never skipped in silence. Any subset works:

# Public channels only — the default, e.g. a company-wide standalone graph:
ingest_slack_export(client, "export.zip", graph_uuid=company_wide_uuid)

# DMs only, into their own graph:
ingest_slack_export(
    client,
    "export.zip",
    graph_uuid=dm_history_uuid,
    conversation_types=["dm", "group_dm"],
)

channels= composes with it: it filters by name within the selected types, matching a channel name, an mpdm-… slug, or a DM id (D01ABC234) or its resolved member label. Naming a conversation whose type isn't selected is a ConfigurationError that says which type to add, rather than a silently empty run. DMs and group DMs are labeled by their resolved members ("Avery Brown, Blake Carter") instead of the opaque id or slug Slack names their folder with — raw ids degrade entity extraction — and every episode carries its conversation_type in metadata for filtering at search time.

Slack names: messages store author ids, so every speaker, @mention, and DM label is resolved through the export's users.json roster, preferring profile.real_name over profile.display_name (then the username, then the raw id). Slack's own precedence is the opposite, but it optimizes for how a name reads in a chat client: a display name is often a short handle ("morgan") that Zep cannot merge with the same person written in full ("Morgan Lee") in an email or document, which silently splits one person into two nodes. Authors whose roster entry has no real_name are counted in result.warnings. When your roster is thin, formatter= receives each SlackMessage — including its raw user_id — so you can substitute names from your own directory:

ingest_slack_export(
    client,
    "export.zip",
    graph_uuid=team_knowledge_uuid,
    formatter=lambda m: f"{DIRECTORY.get(m.user_id, m.sender)}: {m.text}",
)

Batch vs sequential: the Batch API is the high-throughput submission path (10k items/batch by default, up to the API's 50k cap). method="auto" tries batch and transparently falls back to sequential graph.episode.add calls with rate-limit-aware pacing in exactly one case: the deployment has no batch endpoint to call (HTTP 404 — an older server, a self-hosted or Community deployment, or a base URL that doesn't route /batches). Authorization and quota errors are raised as errors instead of quietly downgrading the run. Every source here ingests fine without the Batch API — pass method="sequential" to take that path deliberately. Sequential does not mean "wait until this file has finished extracting before submitting the next." It only means each item is sent with graph.episode.add (or thread.add_messages) instead of the Batch API. Submit every file into the graph first; wait() is opt-in. On the default Batch path it monitors batch.get on the last batch; on sequential fallback it polls the last-submitted episode's processed flag (see Check data ingestion status).

The pipeline

Loader  →  Transforms (chunk / contextualize / canonicalize)  →  LimitGuard  →  Submitter

Each stage is a small protocol (Loader, Transform, Submitter, LLMClient) — a new source, prep step, submission path, or LLM provider is one small class. Everything is lazy generators: a 500k-message export never sits in memory.

from zep_ingest import Pipeline, TextFileLoader, TextChunker, LLMContextualizer, AliasCanonicalizer
from zep_ingest.llm.anthropic import AnthropicLLM

pipeline = Pipeline(
    TextFileLoader("handbook/**/*.md"),
    transforms=[
        AliasCanonicalizer({"ROBOT-202": ["PROTOTYPE-202"]}),
        TextChunker(chunk_size=500, overlap=50),
        LLMContextualizer(AnthropicLLM()),
    ],
)
report = pipeline.preview()  # NO Zep API calls: inspect episodes + warnings first
result = pipeline.run(client, graph_uuid=company_kb_uuid)
result.wait()  # opt-in; batch.get tail or last-submitted episode; timeout scales with item count

preview() shows the transformed episodes and validation warnings (including missing timestamps, oversize splits, and alias rewrites) before a Zep API call. It lazily inspects at most 10 transformed episodes by default and clearly marks its warning counts as sample-scoped. Use preview(limit=None) when you need an exhaustive preflight of the entire stream.

Temporal correctness (the silent backfill killer)

If an episode has no created_at, Zep silently uses the ingestion time to date extracted facts. Zep resolves contradictions by "latest valid_at wins" — so a timestamp-less backfill doesn't just lose history, it makes fact invalidation pick wrong survivors, permanently.

Loaders preserve source timestamps when available (for example Slack ts or a configured record date field). result.warnings counts every episode missing one because run() validates the entire stream before submission. The default preview() reports on its 10-episode sample; use preview(limit=None) to count every missing timestamp before ingestion. Filesystem mtime requires explicit use_file_mtime=True. For document corpora with publication dates, the contextualizer's default prompt also asks the LLM to include the date in each chunk's context.

Chunking

The 10k-character limit is handled twice over:

• TextChunker implements Zep's own cookbook: paragraph-boundary splitting with sentence fallback, 500-char chunks, 50-char overlap. Smaller chunks yield richer graphs — 500 is the documented optimum, 10k is only the hard cap.
• LimitGuard is always appended to every pipeline as a safety net: nothing ever reaches the API oversized, split boundary-aware per data type (lines for message, paragraphs for text, top-level structure for json).
• LLMContextualizer adds contextual retrieval (the technique the docs recommend): an LLM situates each chunk within its source document before ingestion. An LLM failure never aborts a run — the raw chunk is kept and a warning recorded.

Bring any LLM

The contextualizer talks to LLMs through a one-method protocol — anything with complete(prompt: str) -> str works, so you can inject your own client for any provider, proxy, or local model:

class MyLLM:
    def complete(self, prompt: str) -> str: ...


ingest_documents(client, "docs/**/*.md", graph_uuid=kb_uuid, llm=MyLLM())

Three adapters ship as conveniences (mirroring the pattern Graphiti — Zep's own engine — uses for its LLM clients):

from zep_ingest.llm.openai import OpenAILLM, OpenAICompatibleLLM
from zep_ingest.llm.anthropic import AnthropicLLM

OpenAILLM()  # OpenAI (pip install "zep-ingest[openai]")
AnthropicLLM()  # Anthropic (pip install "zep-ingest[anthropic]")
# The universal connector — any OpenAI-compatible /chat/completions endpoint:
OpenAICompatibleLLM(
    model="llama3.1:70b",  # LiteLLM, Ollama, vLLM,
    base_url="http://localhost:11434/v1",  # OpenRouter, Together, ...
    api_key="ollama",
)

OpenAICompatibleLLM is the docs-recommended route to "any provider": LiteLLM alone proxies 100+ models behind this interface, with no extra dependency here beyond the openai package.

Entity canonicalization

Zep merges entities by the names it sees in text; semantic aliases ("PROTOTYPE-202" vs "ROBOT-202") stay separate nodes. The supported fix is canonicalizing before ingestion:

AliasCanonicalizer({"ROBOT-202": ["PROTOTYPE-202", "Picker X1"]})  # rewrite
AliasCanonicalizer(
    {"ROBOT-202": ["PROTOTYPE-202"]}, mode="annotate"
)  # "PROTOTYPE-202 (also known as ROBOT-202)"

Ambiguous aliases are a data-corruption hazard (alias "Will" must not rewrite "he will go"). The exported DEFAULT_RISKY_WORDS guard is enabled by default and rejects risky or very short aliases; extend it with | {"your", "words"} or pass an empty set to opt out explicitly. Matching is case-sensitive and word-boundary by default, URLs/code spans are never touched, the transform is idempotent, and per-alias replacement counts surface as warnings so you see "will → Will Hughes: 4,213 replacements" in preview() — not after your graph is poisoned.

Structured data (JSON): shape it before you ingest

Zep extracts best from JSON that is small, flat, self-contained, and about one entity, and poorly from large, deeply nested, or multi-entity records. ingest_json_records ingests each record as you provide it — one json episode per record — so shaping the data for good extraction is your responsibility (see the JSON best-practices in the Zep docs). LimitGuard still guarantees no episode exceeds the API size limit — it top-level-splits an oversize record as a safety net, with a warning — but it does not restructure your data.

JsonRecordsLoader (and the ingest_json_records one-liner) helps you attach meaning to opaque records: map your own fields onto the canonical id/name/description identity keys, lift a timestamp field into created_at, tag records with a record_type, and promote fields to episode metadata.

Every record episode is stamped with source_type and file_name provenance, which spends 2 of the API's 10 metadata keys — so metadata_fields may name at most 8 fields of your own, and an over-budget list is rejected up front (a ConfigurationError before any file is read, not an error mid-run). Naming source_type or file_name there does not override the provenance: those fields are skipped with a warning, and the record still reaches the graph whole — that field included — in the episode body.

Ontology: set it before you ingest

Two facts drive everything here:

1. The ontology is not retroactive. Data ingested before set_ontology is never re-typed; the only fix is re-ingesting. So set it on the graph once, before your first ingest run:

   from zep_cloud import EdgeSourceTarget, EdgeType, EntityProperty, EntityType
   
   client.graph.set_ontology(
       graph_uuid,  # the UUID that graph.create or user.create returned
       entity_types=[
           EntityType(
               name="Person",
               description="A named individual.",
               properties=[EntityProperty(name="role", type="Text", description="Job role.")],
           ),
       ],
       edge_types=[
           EdgeType(
               name="WORKS_AT",
               description="A person works at an organization.",
               source_targets=[EdgeSourceTarget(source="Person", target="Organization")],
           ),
       ],
   )
   

   The ontology is a property of the graph, not of an ingest run, so zep-ingest has no ontology= parameter — running several one-liners against one graph would otherwise re-declare it per call, and set_ontology replaces the whole ontology of the graph, so the last call would silently win. Configure the graph first; then ingest into it.

2. The ontology guides classification; it is not enforced. Extraction reuses a declared type when it confidently matches and derives a new name otherwise. Your levers: richer type descriptions (enumerate synonym verbs), wider source→target signatures, and graph custom instructions. Custom entity and edge types are capped per scope (10 of each and 10 fields per type at time of writing, but the cap depends on your plan — check the Zep docs or your project's limits).

Typing quality is therefore an iteration loop: sample-ingest, inspect the node labels and edge type names in the Zep dashboard (a long tail of derived types like OWNS/LEADS or untyped entities tells you which description or signature to widen), refine, and re-ingest into a fresh graph. Start simple (few generic types), add precision incrementally.

Default types are user-graph-only. Zep's default ontology (User, Assistant, Preference, Location, Event, Object, Topic, Organization, Document + LOCATED_AT/OCCURRED_AT edges) is applied only to user graphs — named (standalone) graphs carry no default types. The rule therefore differs by graph kind:

• Named graphs (the destination for most one-liners here): nothing is typed unless you declare it. Declare every entity and edge type you rely on — including ones that reuse a default's name like Location or Organization — or those entities stay untyped and any custom edge whose signature needs them never applies. This is exactly why examples/example_ontology.py declares Location and LOCATED_AT itself.
• User graphs: your custom types are additive to the defaults; a same-name declaration overrides how that type classifies, and the defaults can be turned off with user.create(disable_default_ontology=True).

Either way, avoid the reserved field names (uuid, name, graph_id, name_embedding, summary, created_at). graph_id stays reserved as a property name on a node, although v4 addresses the graph itself by UUID.

Don't start from a blank page: examples/example_ontology.py ships a starter ontology (Person / Organization / Project / Product / Location + RESPONSIBLE / WORKS_AT / SUPPLIES / SELLS / CUSTOMER_OF / LOCATED_AT) built with those levers — every example applies it with client.graph.set_ontology(...) before ingesting. Copy the file and adapt the types to your domain. The examples themselves are self-contained and re-runnable: each creates a fresh graph, sets that ontology, and ingests bundled sample data with zero arguments — see examples/README.md.

Seeding a graph from scratch

The full lifecycle, in the order that works (see examples/fact_triples_example.py for a named graph and examples/user_graph_example.py for a user graph):

1. Create the graph: client.graph.create(name=...) (or client.user.create(...) for a user graph, which returns the user's graph_uuid). Keep the returned UUID: it addresses the graph in every later call. The ingestion package writes only into existing graphs — it never creates them.
2. Set the ontology before any data flows: client.graph.set_ontology(graph_uuid, entity_types=..., edge_types=...). It is not retroactive, and the ingestion package never sets it for you.
3. Optionally seed canonical entities with ingest_nodes and keep result.node_uuids (Zep assigns them; do not supply a client UUID). Use client.graph.node.update with those UUIDs for later edits — graph.node.add always creates new nodes.
4. Optionally connect fact triples to those entities by pinning endpoints with source_node_uuid/target_node_uuid. Extraction dedups against the existing graph, so known entities anchor resolution.
5. Ingest the corpus with real created_at timestamps and alias canonicalization. Submit every source for a graph before waiting. Then, if you need extraction to finish, block on the bound result with result.wait() (or combined.wait() after IngestResult.combine).

Many files, one graph

Do not wait for one file (or one graph) to finish processing before submitting the next. Create destinations and set ontology, then enqueue everything.

from zep_ingest import ConcatLoader, ingest, ingest_json_records, ingest_nodes

# Same loader, several files — submitted in this order, one wait:
result = ingest_json_records(
    client,
    ["data/issues.jsonl", "data/prs.jsonl", "data/jira.jsonl"],
    graph_uuid=engineering_uuid,
)
result.wait()  # batch.get on the last batch (default path); timeout scales with items_submitted

# Mixed sources, still one submit stream:
from zep_ingest import JsonRecordsLoader, TextFileLoader

result = ingest(
    client,
    ConcatLoader(
        [
            JsonRecordsLoader(["data/issues.jsonl", "data/prs.jsonl"]),
            TextFileLoader("data/runbooks/**/*.md"),
        ]
    ),
    graph_uuid=engineering_uuid,
)
result.wait()

# Already-separate calls: prefer separate wait() for seeding vs episodes.
nodes = ingest_nodes(client, node_items, graph_uuid=engineering_uuid)
nodes.wait()
docs = ingest_json_records(
    client, ["data/issues.jsonl", "data/prs.jsonl"], graph_uuid=engineering_uuid
)
docs.wait()

If you are not polling, stop after submit — there is nothing else to do.

Fact triples

from zep_ingest import FactTriple, ingest_fact_triples

result = ingest_fact_triples(
    client,
    [
        FactTriple(
            fact="Ana Azimova is responsible for GTM analytics",
            fact_name="RESPONSIBLE",
            source_node_name="Ana Azimova",
            source_node_labels=["Person"],  # ties the node to a declared type
            target_node_name="GTM analytics",
            target_node_labels=["Project"],
            # Optional: pin endpoints to UUIDs from ingest_nodes / a prior read.
            # source_node_uuid=..., target_node_uuid=...,
            valid_at="2024-06-15T00:00:00Z",
        ),
    ],
    graph_uuid=org_uuid,
)
result.wait()
# Zep assigns the fact UUID; it is in the task result as edge_uuid after completion.
# Parallel to the submitted triples — failed tasks leave None in that slot.
result.edge_uuids

Triples skip extraction entirely, so nodes they create are untyped unless you label them — pass source_node_labels/target_node_labels (one declared entity type each) or the declared ontology never touches a triples-only graph. Do not supply a fact_uuid; Zep owns fact identity.

Every documented limit (fact ≤250 chars, names ≤50, summaries ≤500, SCREAMING_SNAKE_CASE fact_name, string attribute and metadata keys whose values are scalars or arrays of scalars, ≤10 metadata keys) is validated client-side at construction — a clear Python error naming the field, not an HTTP 400 three hours into a run. Also accepts a JSONL or JSON object or array path whose columns match the field names. Sequential only (the Batch API doesn't take triples).

Direct node seeding

When you have canonical entities to create up front — before any extraction, and without relationships — ingest_nodes adds them directly via client.graph.node.add:

from zep_ingest import NodeItem, ingest_nodes

result = ingest_nodes(
    client,
    [
        NodeItem(name="Ana Azimova", label="Person"),
        NodeItem(name="GTM analytics", label="Project"),
    ],
    graph_uuid=org_uuid,
)
# Zep assigns each node's UUID — keep them for updates and fact-triple pinning.
result.node_uuids

Zep assigns node UUIDs; do not supply them (a client uuid is rejected). result.node_uuids is parallel to the submitted list — each success carries the assigned UUID, and a failed batch leaves None in those slots so a later success cannot shift under zip. Direct graph.node.add calls create new nodes each time (no name or UUID upsert) — use client.graph.node.update with a returned UUID to rename, replace a summary, edit attributes, or clear an entity type. Up to 100 nodes per request, every documented limit (name ≤50, summary ≤500, label ≤100, ≤10 attributes, ≤10 metadata keys, each value a scalar or an array of scalars) validated client-side at construction. Sequential only (the Batch API doesn't take direct nodes).

Monitoring a run

result = ingest_slack_export(client, "export.zip", graph_uuid=graph_uuid)
result.status  # queued | processing | untracked | succeeded | partial | failed | canceled
result.wait()
result.failed_items()  # Batch API item records and/or submission AddErrors
result.warnings  # everything the pipeline noticed
result.raise_for_status()  # opt-in strictness

Every ingest call returns as soon as the data is submitted; result.wait() is the only thing that blocks. Bind the result first, as above — don't chain ingest_slack_export(...).wait(). The chain works (wait() returns self), but when it raises — a timeout, or a submission the API left untracked — nothing was ever bound, and batch_ids / task_ids are the only handles for resuming or diagnosing that run.

What wait() polls (aligned with Check data ingestion status):

Do not mix Batch API and sequential graph.episode.add into the same graph and expect one wait() to cover both. Seed nodes/triples first with their own wait(), then ingest episodes.

Ingestion is asynchronous — a just-added fact is not instantly retrievable, even after wait(): search indexing lands a few seconds after processing. search_when_ready owns that gap so scripts don't hand-roll poll loops:

from zep_ingest import search_when_ready

items = search_when_ready(client, "who runs the pilot?", graph_uuid=graph_uuid)

Partial failures never crash a run: pages/episodes that keep failing are recorded as AddErrors (indices and API messages only — never episode content) and the run continues. batch_ids / episode_uuids / task_ids are the resume handles; node_uuids / edge_uuids record identities Zep assigned on ingest_nodes and completed ingest_fact_triples tasks (None slots mark failures so later successes stay zip-aligned). Task IDs come from ingest_nodes, ingest_fact_triples, and thread.add_messages, which returns the task that tracks extraction in v4.

If the API accepts a task-backed submission without returning a completion handle, the result reports status == "untracked" instead of claiming success. wait() raises IngestUntrackedError immediately in that state; use search_when_ready or an application-specific read to verify availability.

Checking status later — if you never call wait(), persist result.batch_ids and reconstruct in another process:

from zep_ingest import IngestResult

result = IngestResult.from_batch_ids(client, ["batch-id-1"])
result.refresh()  # status reads cached state, so fetch it first
result.status
result.wait()  # or skip refresh() entirely — wait() polls for you

# Task-backed ingestion can be resumed the same way:
result = IngestResult.from_task_ids(client, ["task-id-1"])
result.wait()

Security notes

• LLMContextualizer sends document content to your LLM. The prompt marks the content as data-not-instructions, strips the tag vocabulary from inputs so hostile text can't break the prompt structure, and length-caps + sanitizes the LLM's output — but if you ingest untrusted content, remember that graph content ultimately reaches your agents' prompts; sanitize upstream where that matters.
• Alias maps are validated (no control characters, sane lengths) since they often come from config files.
• Stored AddError records and package-generated warnings never add content from the episodes you submitted. They do carry the API's own response body verbatim, so a failure reads exactly as it would from a direct SDK call — worth knowing if you ship failed_items() output to a log aggregator.

What this package does NOT fix

Retrieval-side behavior is out of scope: survivor selection under contradiction (strictly latest-valid_at), confidence/authority weighting, and as-of search. The ingestion-side mitigations — correct created_at, canonical names, and source_type episode metadata you can filter on at search time — are all supported here.

Extending

class MyLoader:  # any source
    def load(self) -> Iterator[Episode]: ...


class MyTransform:  # any prep step; optional .warnings list
    def apply(self, episodes: Iterable[Episode]) -> Iterator[Episode]: ...


class MyLLM:  # any LLM provider
    def complete(self, prompt: str) -> str: ...

Planned as future work: PDF/Drive loaders, LLM date extraction for undated corpora, idempotency/resume manifests (re-running the same export today duplicates episodes), per-item destination routing, an async API, and a CLI.

Development

make install   # uv sync --extra dev
make all       # format + lint + type-check + test

Before a release, run the production smoke script against a throwaway graph (requires ZEP_API_KEY and ZEP_API_URL):

keybank run zep-prod -- env ZEP_API_URL=https://api.getzep.com \
  uv run python scripts/release_smoke_prod.py

Without KeyBank, export a production ZEP_API_KEY of your own and run the same script:

ZEP_API_URL=https://api.getzep.com uv run python scripts/release_smoke_prod.py

Live integration tests run only when ZEP_API_KEY is set. See SETUP.md for account setup.