Extract memories from documents with MemoryETL

Extract long-term memories from documents, track processing, and inspect source evidence.

MemoryETL uses extract, transform, load (ETL) processing to turn text or Markdown documents into long-term memories. For example, a travel assistant can learn a user's plans from trip notes they upload before the conversation starts.

An extractor tells MemoryETL what to extract: a custom memory type and instructions for the model. Creating an extractor is an administrative task. Applications can then submit documents to it.

Use session extraction for conversation events. Use direct memory creation when your application already has the records it wants to save.

Prepare the service

MemoryETL is available in Redis Cloud and enabled in self-managed deployments. For a self-managed service's model, processing limits, and worker settings, see Configure document ingestion.

These examples continue from the quickstart. Load the same .env connection values and reuse its user ID. Expand an example to copy a complete program that connects for you.

Each run repeats every write in the program. Session-event writes and memory creates with a new idempotency key add new records, so run cleanup or change the identifiers before rerunning.

The examples save memories for quickstart-user. In your application, use the authenticated user ID that owns the document. For a self-managed service with ownership enforcement, use the authenticated client setup in delegation.

Create an extractor

Ask your operator to register a memory type and create a document extractor. Add the returned extractor ID to .env as EXTRACTOR_ID. Applications use their Data Plane credential to submit documents.

EXTRACTOR_ID='<extractor-id>'

Submit and track a document

Create trip-notes.md containing the following line. The examples create it if it's missing. The extractor publishes only information that its memory type captures, so use a document that matches your extractor.

Trip to Kyoto, April 3 to April 10. I am vegetarian and want to try local tofu dishes.

Only text/plain and text/markdown are accepted. Encode the UTF-8 bytes as base64 once before passing content to REST or an SDK. The SDK expects the encoded string, not a file object or raw text.

Each submission requires these fields:

Submit the file. The server returns a separate ingestion ID to track the run. Save it; you need it even if polling times out.

accepted = memory.client.create_document_ingestion(
    document_id="example-trip-notes",
    revision="1",
    content=base64.b64encode(Path("trip-notes.md").read_bytes()).decode("ascii"),
    media_type="text/markdown",
    owner_id=OWNER_ID,
    extractor_id=EXTRACTOR_ID,
    idempotency_key="example-trip-notes-revision-1",
)
ingestion_id = accepted.result.ingestion_id
print("Ingestion ID:", ingestion_id)

jq -n --rawfile text trip-notes.md --arg owner "$OWNER_ID" --arg extractor "$EXTRACTOR_ID" '{
  documentId: "example-trip-notes",
  revision: "1",
  content: ($text | @base64),
  mediaType: "text/markdown",
  ownerId: $owner,
  extractorId: $extractor,
  idempotencyKey: "example-trip-notes-revision-1"
}' > ingestion.json
ACCEPTED=$(curl --fail-with-body --silent --show-error \
  -X POST "$MEMORY_API/document-ingestions" \
  -H "$AUTH" -H 'Content-Type: application/json' --data-binary @ingestion.json)
INGESTION_ID=$(jq -er .ingestionId <<< "$ACCEPTED")
echo "Ingestion ID: $INGESTION_ID"

REST returns 202 Accepted; this confirms submission, not successful extraction. In Python, the submission content is under accepted.result.

Poll the ingestion until it reaches a terminal processing state. The examples check every 10 seconds for up to 15 minutes:

for _ in range(90):
    status = memory.client.get_document_ingestion(ingestion_id=ingestion_id).ingestion.status
    if status.processing.value in {"SUCCEEDED", "PARTIAL", "FAILED"}:
        print(status.model_dump_json(by_alias=True, exclude_none=True))
        break
    time.sleep(10)
else:
    raise TimeoutError(f"Continue polling ingestion {ingestion_id}")

STATE=''
for _ in $(seq 90); do
  RECORD=$(curl --fail-with-body --silent --show-error \
    "$MEMORY_API/document-ingestions/$INGESTION_ID" -H "$AUTH")
  STATE=$(jq -r .ingestion.status.processing <<< "$RECORD")
  case "$STATE" in
    SUCCEEDED|PARTIAL|FAILED) jq .ingestion.status <<< "$RECORD"; break ;;
  esac
  sleep 10
done
case "$STATE" in
  SUCCEEDED|PARTIAL|FAILED) ;;
  *) echo "Continue polling ingestion $INGESTION_ID" >&2; exit 1 ;;
esac

List the IDs of the memories the run published. Pass nextCursor as cursor to read the next page:

output_ids = []
cursor = None
while True:
    page = memory.client.list_document_ingestion_outputs(
        ingestion_id=ingestion_id, limit=100, cursor=cursor
    )
    output_ids.extend(page.items)
    cursor = page.next_cursor
    if not cursor:
        break
print("Output memory IDs:", output_ids)
if status.processing.value != "SUCCEEDED":
    raise RuntimeError(f"Inspect ingestion {ingestion_id}: {status.processing.value}")
if not output_ids:
    raise SystemExit("The extractor published no memories from this document.")

OUTPUT_IDS=()
CURSOR=''
while :; do
  PAGE=$(curl --fail-with-body --silent --show-error --get \
    "$MEMORY_API/document-ingestions/$INGESTION_ID/outputs" -H "$AUTH" \
    --data-urlencode 'limit=100' ${CURSOR:+--data-urlencode "cursor=$CURSOR"})
  while IFS= read -r id; do OUTPUT_IDS+=("$id"); done < <(jq -r '.items[]' <<< "$PAGE")
  CURSOR=$(jq -r '.nextCursor // empty' <<< "$PAGE")
  [ -n "$CURSOR" ] || break
done
echo "Output memory IDs: ${OUTPUT_IDS[*]:-}"
[ "$STATE" = SUCCEEDED ] || { echo "Inspect ingestion $INGESTION_ID: $STATE" >&2; exit 1; }
[ ${#OUTPUT_IDS[@]} -gt 0 ] || { echo "The extractor published no memories from this document." >&2; exit 1; }

Interpret results and recover

MemoryETL processes a document in two stages. First, each chunk is extracted into candidate memories. These candidates are staged, not published. Then a final reconciliation step merges duplicates and continued observations across chunks and publishes the result. No memories are searchable until reconciliation finishes. Publication is incremental, so a run that stops during publication can leave some memories.

completedChunks, totalChunks, extractorId, and extractorRevision help identify what ran. The chunk counts cover source chunks only; reconciliation isn't counted. cleanup is independent: PENDING, DELETED, or FAILED describes temporary source cleanup. A successful extraction with failed cleanup needs operator attention; it is not a reason to submit the document again. Extracted memories and provenance can remain after source bytes are deleted.

Reuse the same idempotency key and original request after a timeout or lost submission response. Changing the input under the same key conflicts. Reusing an immutable document revision with different content also conflicts. Use a new revision for changed content; do not change the bytes merely to force a retry of an existing run.

The service bounds decoded bytes, estimated input tokens, chunks, output records, queued bytes, retries, execution time, and source/checkpoint retention. The reference self-managed configuration allows 8,000 input tokens per extraction request; reconciliation has a separate budget that defaults to 32,000 tokens per request. Inspect the deployment settings and retentionDeadline. Idempotency and status retention are bounded; retain your logical document IDs, revisions, ingestion IDs, and recovery decisions in your application. A new revision does not promise automatic removal of every memory from an older revision.

Extracted records return documentManaged: true. They appear in ordinary long-term-memory searches unless you set excludeDocumentManaged: true. Their text and attributes are managed by document ingestion: generic overwrite and PATCH requests cannot replace them. The exception is a PATCH that changes only the namespace: it moves the memory, keeps its content, and keeps its current expiration. Document chunk spans use UTF-8 byte offsets; do not apply those offsets to a character-indexed string.

With ingestion webhooks, the submission and the reconciled candidates can also be reviewed before they change data. Check hook decisions separately from ingestion processing. Receiver failure and explicit filtering have different outcomes.

Trace memories to their sources

Provenance connects a memory to document or session inputs used during extraction. A sourceId identifies one revisioned origin in the store; an ingestionId identifies a particular document run. Session evidence lists input event IDs. Document evidence identifies ingestions, chunks, and optional source spans. These describe the input supplied to extraction, not a guarantee that each sentence in the resulting memory is a quoted fact.

The examples take the first output memory ID. They follow its source pages and each source's independent evidence pages, then list memories that the same document source produced for quickstart-user:

memory_id = output_ids[0]
source_cursor = None
while True:
    links = memory.client.list_memory_source_links(
        memory_id=memory_id, limit=100, cursor=source_cursor
    )
    for source in links.items:
        print(source.model_dump_json(by_alias=True, exclude_none=True))
        evidence_cursor = source.evidence_next_cursor
        while evidence_cursor:
            more = memory.client.list_memory_source_evidence(
                memory_id=memory_id, source_id=source.source_id,
                limit=100, cursor=evidence_cursor,
            )
            print(more.source.model_dump_json(by_alias=True, exclude_none=True))
            evidence_cursor = more.source.evidence_next_cursor
        if source.kind.value != "DOCUMENT":
            continue  # reverse lookup covers document sources
        related_cursor = None
        while True:
            related = memory.client.list_source_memories(
                source_id=source.source_id, owner_id=OWNER_ID,
                limit=100, cursor=related_cursor,
            )
            for record in related.items:
                print("From the same source:", record.memory_id, record.text)
            related_cursor = related.next_cursor
            if not related_cursor:
                break
    source_cursor = links.next_cursor
    if not source_cursor:
        break

MEMORY_ID=${OUTPUT_IDS[0]}
SOURCE_CURSOR=''
while :; do
  LINKS=$(curl --fail-with-body --silent --show-error --get \
    "$MEMORY_API/memories/$MEMORY_ID/sources" -H "$AUTH" \
    --data-urlencode 'limit=100' ${SOURCE_CURSOR:+--data-urlencode "cursor=$SOURCE_CURSOR"})
  jq '.items' <<< "$LINKS"
  while IFS= read -r SOURCE; do
    SOURCE_ID=$(jq -r .sourceId <<< "$SOURCE")
    EVIDENCE_CURSOR=$(jq -r '.evidenceNextCursor // empty' <<< "$SOURCE")
    while [ -n "$EVIDENCE_CURSOR" ]; do
      MORE=$(curl --fail-with-body --silent --show-error --get \
        "$MEMORY_API/memories/$MEMORY_ID/sources/$SOURCE_ID/evidence" -H "$AUTH" \
        --data-urlencode 'limit=100' --data-urlencode "cursor=$EVIDENCE_CURSOR")
      jq .source <<< "$MORE"
      EVIDENCE_CURSOR=$(jq -r '.source.evidenceNextCursor // empty' <<< "$MORE")
    done
    [ "$(jq -r .kind <<< "$SOURCE")" = DOCUMENT ] || continue  # reverse lookup covers document sources
    RELATED_CURSOR=''
    while :; do
      RELATED=$(curl --fail-with-body --silent --show-error --get \
        "$MEMORY_API/sources/$SOURCE_ID/memories" -H "$AUTH" \
        --data-urlencode "ownerId=$OWNER_ID" --data-urlencode 'limit=100' \
        ${RELATED_CURSOR:+--data-urlencode "cursor=$RELATED_CURSOR"})
      jq '.items[] | {memoryId, text}' <<< "$RELATED"
      RELATED_CURSOR=$(jq -r '.nextCursor // empty' <<< "$RELATED")
      [ -n "$RELATED_CURSOR" ] || break
    done
  done < <(jq -c '.items[]' <<< "$LINKS")
  SOURCE_CURSOR=$(jq -r '.nextCursor // empty' <<< "$LINKS")
  [ -n "$SOURCE_CURSOR" ] || break
done

The evidence response wraps source; it does not return a top-level items list. The source object contains evidence for documents or sessionEvidence for sessions. Newer evidence can also include evidenceId, schemaVersion, capturedAt, and producer. Handle their absence on legacy data. capturedAt is when extraction input was prepared, not when the original event occurred. A retried delivery retains its evidence identity.

For reverse lookup, use GET /iris/data/v1/agent-memory/stores/{storeId}/sources/{sourceId}/memories with ownerId, or list_source_memories. Reverse lookup covers document sources; a session source returns 404. Follow its nextCursor independently. Keep every token with the exact query and endpoint that issued it. Tokens expire; restart a query when one is invalid, and URL-encode tokens in REST calls.

A source link does not grant access to the original document, session, or another owner's memories. The service checks source and record access separately; sharing an extracted memory does not share the private document it came from.