Batch ingestion

The Batch API is the recommended way to load large historical datasets — backfills, document collections, archived conversations, migrations from another system — into your Context Graphs.

Why use the Batch API

Calling graph.add or thread.add_messages once per item works for live data but becomes hard to manage at scale. Compared to issuing those calls one at a time, the Batch API gives you:

• Faster processing. It ingests large datasets faster than the same operations sent one at a time.
• No interference with live traffic. It is designed not to slow the real-time graph.add and thread.add_messages ingestion serving your agents, so a large backfill can run alongside production.
• Progress you can watch. Monitor each batch’s status, item counts, and errors in the batch dashboard, or poll programmatically.
• One batch instead of many calls. Group items into a single batch — splitting across batches when needed (see Batch limits) — and hand it off to Zep to process as one job.

How batches work

A batch follows a three-step lifecycle:

Items in a batch are grouped by destination graph and processed in the order they were added. Episodes and messages added through the Batch API are priced the same as those added through graph.add or thread.add_messages.

Batch limits

• A single batch can contain up to 50,000 items.
• Each call to batch.add accepts up to 500 items.

To ingest more than 500 items, make multiple batch.add calls against the same batch ID before calling batch.process.

Quickstart

The example below creates a batch, adds a mix of graph episodes and thread messages, starts processing, and polls until the batch finishes.

1
import time
2
from zep_cloud.client import Zep
3
from zep_cloud import BatchAddItem
4
5
client = Zep(api_key=API_KEY)
6
7
# 1. Create the batch
8
batch = client.batch.create(
9
    metadata={"description": "Customer support backfill"},
10
)
11
batch_id = batch.batch_id
12
13
# 2. Add items to the batch
14
items = [
15
    BatchAddItem(
16
        type="graph_episode",
17
        user_id="alice",
18
        data="Alice signed up for the Pro plan on 2024-06-15.",
19
        data_type="text",
20
    ),
21
    BatchAddItem(
22
        type="graph_episode",
23
        graph_id="company_kb",
24
        data="Refund policy: orders may be refunded within 30 days of purchase.",
25
        data_type="text",
26
    ),
27
    BatchAddItem(
28
        type="thread_message",
29
        thread_id="alice_support_thread_42",
30
        content="My dashboard isn't loading.",
31
        role="user",
32
        name="Alice",
33
    ),
34
]
35
36
client.batch.add(batch_id=batch_id, items=items)
37
38
# 3. Start processing
39
client.batch.process(batch_id=batch_id)
40
41
# 4. Poll until the batch finishes
42
TERMINAL_STATUSES = ("succeeded", "partial", "failed", "invalid")
43
while True:
44
    summary = client.batch.get(batch_id=batch_id)
45
    if summary.status in TERMINAL_STATUSES:
46
        break
47
    print(f"Status: {summary.status} ({summary.progress.percent_complete:.0f}%)")
48
    time.sleep(5)
49
50
print(f"Final status: {summary.status}")

Adding items to a batch

Each item in a batch is one of two types:

• graph_episode — equivalent to a single graph.add call. Targets a graph by graph_id or a user graph by user_id.
• thread_message — equivalent to one message inside a thread.add_messages call. Targets a thread by thread_id.

The fields below mirror the equivalent fields on graph.add and thread.add_messages. See Adding business data and Adding messages for the underlying semantics.

Common fields

Graph episode fields (type: "graph_episode")

Thread message fields (type: "thread_message")

Setting timestamps on batch items

Pass created_at on each item to give Zep accurate temporal information for historical data. This is important for backfills — Zep uses these timestamps in its fact invalidation process to determine the valid_at and invalid_at values on extracted facts (edges).

The created_at value should be in RFC3339 format (e.g., "2024-06-15T10:30:00Z").

1
from zep_cloud import BatchAddItem
2
3
items = [
4
    BatchAddItem(
5
        type="graph_episode",
6
        user_id="alice",
7
        data="Alice joined the engineering team as a senior developer.",
8
        data_type="text",
9
        created_at="2024-06-15T10:30:00Z",
10
    ),
11
    BatchAddItem(
12
        type="graph_episode",
13
        user_id="alice",
14
        data="Alice was promoted to tech lead of the engineering team.",
15
        data_type="text",
16
        created_at="2024-09-01T09:00:00Z",
17
    ),
18
]
19
20
client.batch.add(batch_id=batch_id, items=items)

Tracking progress

Two methods report on a running or completed batch:

• batch.get(batch_id) returns a summary of the whole batch, including a progress object with counts for total_items, queued_items, processing_items, succeeded_items, failed_items, skipped_items, and percent_complete. Before batch.process is called the batch is in draft and the progress counts are unpopulated; once processing starts the counts begin to update.
• batch.list_items(batch_id) returns each item with its individual status (pending, queued, processing, succeeded, failed, skipped).

When polling batch.get, a few-second interval (e.g., 5 seconds) is appropriate for small batches. For batches with thousands of items or more, polling becomes impractical — subscribe to the ingest.batch.completed webhook instead to be notified when a batch reaches a terminal state. The payload includes the batch_id so you can match it back to the batch you submitted.

Batch statuses

The status field on BatchSummary is one of:

Once a batch reaches succeeded, partial, or failed, no further state changes occur. invalid is also non-progressing — the batch never starts processing, but the state persists until you delete the batch. When polling, exit on any of succeeded, partial, failed, or invalid.

Per-item statuses

The status field on each BatchItemDetail is one of:

1
summary = client.batch.get(batch_id=batch_id)
2
print(f"Status: {summary.status}")
3
print(f"Progress: {summary.progress.succeeded_items}/{summary.progress.total_items}")
4
5
# Inspect individual items
6
items = client.batch.list_items(batch_id=batch_id, limit=50)
7
for item in items.items:
8
    print(item.item_id, item.status)

Listing and managing batches

Use batch.list to enumerate batches in your project, optionally filtered by status. Use batch.delete to remove a batch that has not yet been processed — once a batch has been processed, it cannot be deleted.

1
# List recent batches
2
result = client.batch.list(limit=20)
3
for b in result.batches:
4
    print(b.batch_id, b.status, b.item_count)
5
6
# List only batches that are still being processed
7
result = client.batch.list(status="processing")
8
9
# Delete a draft batch
10
client.batch.delete(batch_id=batch_id)

Viewing batches in the dashboard

The Zep web dashboard provides a batches view showing all batches in your project, their status, item counts, and processing progress. Click into a batch to inspect its individual items and any errors.

Deprecated batch methods

The following methods are deprecated and no longer recommended. Use the Batch API described above for all new ingestion work.

The deprecated methods continue to work but will be removed in a future release.