Check Data Ingestion Status

Data added to Zep is processed asynchronously and can take a few seconds to a few minutes to finish processing. This recipe shows how to check whether data upload operations are finished processing.

Zep provides these methods for checking data ingestion status:

• Task polling: Use client.task.get() to check the status of clone operations, direct node additions, and fact triple additions
• Episode polling: Use graph.episode.get() to check individual episode processing status
• Batch status: Use batch.get() and batch.list_items() for Batch API imports

For tracking large historical ingestions, see the Batch API, which has its own progress reporting via batch.get and per-item status via batch.list_items.

Submit many episodes, poll once

When you add several episodes to one graph at once — for example during a backfill — submit every episode without polling between adds. To know when the import is retrievable, poll the episode that is ingested last — see the table below for which episode that is on each path.

graph.add and single-message thread.add_messages

Plain graph.add calls (text or JSON, no document_id) dispatch one episode per submit in submission order. After every submit returns, poll the last-submitted episode’s processed status.

Single-message thread.add_messages calls follow the same rule between groups. Poll the last-submitted message.

Scale your poll timeout and interval with the total episode count in that graph. As a starting point, allow several seconds per episode — a 100-episode import often needs minutes, not seconds.

A processed: true on that episode means extraction has reached that point. It does not guarantee every episode succeeded — check for failed episodes before you treat the import as complete.

Do not mix Batch API imports with graph.add into the same graph and expect one poll to cover both paths. They are not globally serialized with each other.

Multi-message thread.add_messages

When you pass multiple messages in one request, those messages run in request array order, not created_at order. Poll the last message in the last request.

Batch API and zep-ingest

Batch items are ingested in submission order (sequence_index). Monitor the batch with batch.get, batch.list_items, or IngestResult.wait() — not individual episode polling. See Tracking progress and Monitor zep-ingest with IngestResult below.

zep-ingest uses the Batch API on most deployments. Submit every source without waiting between them, then call wait() once (or monitor with batch.get). Use failed_items() to inspect per-item failures.

Monitor zep-ingest with IngestResult

zep-ingest returns an IngestResult for 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()

The result records the identifiers needed to continue monitoring in another process:

1
print(result.batch_ids)
2
print(result.episode_uuids)
3
print(result.task_ids)

Persist the appropriate identifiers and reconstruct the result later:

1
from zep_ingest import IngestResult
2
3
batch_result = IngestResult.from_batch_ids(client, saved_batch_ids)
4
batch_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 in the result until they reach a terminal state. When Zep accepts a write but returns no handle for it, the result counts those items in untracked_items, status reports untracked, and wait() raises IngestUntrackedError rather than polling indefinitely. The submission succeeded in that case; only server-side extraction cannot be tracked, so confirm the data with a read of your own.

Search indexing can take additional time. search_when_ready() retries until a 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.

Checking Operation Status with Task Polling

When using operations that return a task_id, you can poll for completion status using client.task.get(). The following operations return a task_id:

• graph.clone() - Graph cloning operations
• graph.add_nodes() - Direct node additions and upserts
• graph.add_fact_triple() - Custom fact/node triplet additions

The pattern is the same for each operation: capture the task_id returned by the operation, then poll client.task.get(task_id=task_id) until status is succeeded or failed.

Checking Individual Episode Status with Episode Polling

Use graph.episode.get() to check whether one episode has finished processing. When you submit multiple episodes to one graph, poll the last-submitted episode — see Submit many episodes, poll once.

First, let’s create a user:

1
import os
2
import uuid
3
import time
4
from dotenv import find_dotenv, load_dotenv
5
from zep_cloud.client import Zep
6
7
load_dotenv(dotenv_path=find_dotenv())
8
9
client = Zep(api_key=os.environ.get("ZEP_API_KEY"))
10
uuid_value = uuid.uuid4().hex[:4]
11
user_id = "-" + uuid_value
12
client.user.add(
13
    user_id=user_id,
14
    first_name = "John",
15
    last_name = "Doe",
16
    email="[email protected]"
17
)

Now, let’s add some data and immediately try to search for that data; because data added to Zep is processed asynchronously and can take a few seconds to a few minutes to finish processing, our search results do not have the data we just added:

1
episode = client.graph.add(
2
    user_id=user_id,
3
    type="text", 
4
    data="The user is an avid fan of Eric Clapton"
5
)
6
7
search_results = client.graph.search(
8
    user_id=user_id,
9
    query="Eric Clapton",
10
    scope="nodes",
11
    limit=1,
12
    reranker="cross_encoder",
13
)
14
15
print(search_results.nodes)

None

We can check the status of the episode to see when it has finished processing, using the episode returned from the graph.add method and the graph.episode.get method:

1
while True:
2
    episode = client.graph.episode.get(
3
        uuid_=episode.uuid_,
4
    )
5
    if episode.processed:
6
        print("Episode processed successfully")
7
        break
8
    print("Waiting for episode to process...")
9
    time.sleep(10)

Waiting for episode to process...
Waiting for episode to process...
Waiting for episode to process...
Waiting for episode to process...
Waiting for episode to process...
Episode processed successfully

Now that the episode has finished processing, we can search for the data we just added, and this time we get a result:

1
search_results = client.graph.search(
2
    user_id=user_id,
3
    query="Eric Clapton",
4
    scope="nodes",
5
    limit=1,
6
    reranker="cross_encoder",
7
)
8
9
print(search_results.nodes)

[EntityNode(attributes={'category': 'Music', 'labels': ['Entity', 'Preference']}, created_at='2025-04-05T00:17:59.66565Z', labels=['Entity', 'Preference'], name='Eric Clapton', summary='The user is an avid fan of Eric Clapton.', uuid_='98808054-38ad-4cba-ba07-acd5f7a12bc0', graph_id='6961b53f-df05-48bb-9b8d-b2702dd72045')]