Zep.Batch (Zep v1.0.0)

Copy Markdown View Source

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

item()

@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

add(client_or_opts, batch_id, items, opts \\ [])

@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.

await(client_or_opts, batch_id, opts \\ [])

@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 (default 5_000, matching the "a few seconds" guidance in the Batch Ingestion guide for small batches - for large batches prefer the ingest.batch.completed webhook instead of polling)
  • :timeout - overall ms budget before giving up (default 300_000)

create(client_or_opts, opts \\ [])

@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.

delete(client_or_opts, batch_id)

@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.

get(client_or_opts, batch_id)

@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.

list(client_or_opts, opts \\ [])

@spec list(
  Zep.Client.t() | keyword(),
  keyword()
) :: {:ok, [Zep.Schemas.BatchJob.t()]} | {:error, term()}

Lists batch jobs in the project, optionally filtered by :status.

list_items(client_or_opts, batch_id, opts \\ [])

@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.

process(client_or_opts, batch_id)

@spec process(Zep.Client.t() | keyword(), String.t()) ::
  {:ok, term()} | {:error, term()}

Starts asynchronous processing of a batch. Zep returns immediately.