The Batch API - the recommended way to load large historical datasets
(backfills, document collections, archived conversations, migrations)
into Context Graphs. Replaces the deprecated
Zep.Thread.add_messages_batch/4 and (for large jobs) Zep.Graph.add_batch/3.
A batch follows a three-step lifecycle:
{:ok, batch} = Zep.Batch.create(client, metadata: %{description: "Support backfill"})
{:ok, _} =
Zep.Batch.add(client, batch.batch_id, [
%{type: :graph_episode, user_id: "alice", data: "Alice upgraded to Pro.", data_type: :text},
%{type: :thread_message, thread_id: "alice-support-42", content: "Dashboard won't load.", role: "user", name: "Alice"}
])
{:ok, _} = Zep.Batch.process(client, batch.batch_id)
{:ok, final} = Zep.Batch.await(client, batch.batch_id)Limits
A single batch holds up to 50,000 items total; each add/3 call accepts
up to 500 items (call it repeatedly against the same batch_id to load
more before calling process/2).
Terminal statuses
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 record persists until deleted. Before
process/2 is called, a batch sits in "draft".
For batches with thousands of items, prefer subscribing to the
ingest.batch.completed webhook over polling await/3 - see the Batch
Ingestion guide.
Summary
Functions
Adds up to 500 items to an existing (draft) batch.
Polls get/2 until the batch reaches a terminal status
(["succeeded", "partial", "failed", "invalid"]), or :timeout elapses.
Creates a new, empty draft batch. Add items to it with add/3.
Deletes a batch. Only batches that haven't been processed yet ("draft")
can be deleted - once processed, a batch cannot be removed.
Fetches a batch's summary: top-level :status plus a nested :progress
with total_items / succeeded_items / failed_items / etc.
Lists batch jobs in the project, optionally filtered by :status.
Lists the individual items within a batch, each with its own status
("pending", "queued", "processing", "succeeded", "failed",
"skipped") and :error message when applicable.
Starts asynchronous processing of a batch. Zep returns immediately.
Types
@type item() :: %{ :type => :graph_episode | :thread_message, optional(:user_id) => String.t(), optional(:graph_id) => String.t(), optional(:thread_id) => String.t(), optional(:data) => term(), optional(:data_type) => :text | :json | :message, optional(:content) => String.t(), optional(:role) => String.t(), optional(:name) => String.t(), optional(:created_at) => String.t() }
Functions
@spec add(Zep.Client.t() | keyword(), String.t(), [item()], keyword()) :: {:ok, term()} | {:error, term()}
Adds up to 500 items to an existing (draft) batch.
Call this repeatedly against the same batch_id to load more than
500 items before calling process/2.
@spec await(Zep.Client.t() | keyword(), String.t(), keyword()) :: {:ok, Zep.Schemas.BatchJob.t()} | {:error, :timeout | term()}
Polls get/2 until the batch reaches a terminal status
(["succeeded", "partial", "failed", "invalid"]), or :timeout elapses.
Options
:poll_interval- ms between polls (default5_000, matching the "a few seconds" guidance in the Batch Ingestion guide for small batches - for large batches prefer theingest.batch.completedwebhook instead of polling):timeout- overall ms budget before giving up (default300_000)
@spec create( Zep.Client.t() | keyword(), keyword() ) :: {:ok, Zep.Schemas.BatchJob.t()} | {:error, term()}
Creates a new, empty draft batch. Add items to it with add/3.
@spec delete(Zep.Client.t() | keyword(), String.t()) :: {:ok, term()} | {:error, term()}
Deletes a batch. Only batches that haven't been processed yet ("draft")
can be deleted - once processed, a batch cannot be removed.
@spec get(Zep.Client.t() | keyword(), String.t()) :: {:ok, Zep.Schemas.BatchJob.t()} | {:error, term()}
Fetches a batch's summary: top-level :status plus a nested :progress
with total_items / succeeded_items / failed_items / etc.
@spec list( Zep.Client.t() | keyword(), keyword() ) :: {:ok, [Zep.Schemas.BatchJob.t()]} | {:error, term()}
Lists batch jobs in the project, optionally filtered by :status.
@spec list_items(Zep.Client.t() | keyword(), String.t(), keyword()) :: {:ok, [Zep.Schemas.BatchItem.t()]} | {:error, term()}
Lists the individual items within a batch, each with its own status
("pending", "queued", "processing", "succeeded", "failed",
"skipped") and :error message when applicable.
@spec process(Zep.Client.t() | keyword(), String.t()) :: {:ok, term()} | {:error, term()}
Starts asynchronous processing of a batch. Zep returns immediately.