# Extract memories from documents with MemoryETL ```json metadata { "schema_version": 2, "title": "Extract memories from documents with MemoryETL", "description": "Extract long-term memories from documents, track processing, and inspect source evidence.", "categories": ["docs","develop","ai"], "tableOfContents": {"sections":[{"id":"prepare-the-service","title":"Prepare the service"},{"id":"create-an-extractor","title":"Create an extractor"},{"id":"submit-and-track-a-document","title":"Submit and track a document"},{"id":"interpret-results-and-recover","title":"Interpret results and recover"},{"id":"trace-memories-to-their-sources","title":"Trace memories to their sources"}]} , "codeExamples": [{"codetabsId":"iris_agent_memory_memory_etl-stepmemory-etl-submit","difficulty":"beginner","id":"memory-etl-submit","languages":[{"clientId":"iris-python","clientName":"Python (Iris)","id":"Python (Iris)","langId":"python","panelId":"panel_Python (Iris)_iris_agent_memory_memory_etl-stepmemory-etl-submit"},{"clientId":"iris-curl","clientName":"cURL (Iris)","id":"cURL (Iris)","langId":"bash","panelId":"panel_cURL (Iris)_iris_agent_memory_memory_etl-stepmemory-etl-submit"}]},{"codetabsId":"iris_agent_memory_memory_etl-stepmemory-etl-status","difficulty":"beginner","id":"memory-etl-status","languages":[{"clientId":"iris-python","clientName":"Python (Iris)","id":"Python (Iris)","langId":"python","panelId":"panel_Python (Iris)_iris_agent_memory_memory_etl-stepmemory-etl-status"},{"clientId":"iris-curl","clientName":"cURL (Iris)","id":"cURL (Iris)","langId":"bash","panelId":"panel_cURL (Iris)_iris_agent_memory_memory_etl-stepmemory-etl-status"}]},{"codetabsId":"iris_agent_memory_memory_etl-stepmemory-etl-outputs","difficulty":"beginner","id":"memory-etl-outputs","languages":[{"clientId":"iris-python","clientName":"Python (Iris)","id":"Python (Iris)","langId":"python","panelId":"panel_Python (Iris)_iris_agent_memory_memory_etl-stepmemory-etl-outputs"},{"clientId":"iris-curl","clientName":"cURL (Iris)","id":"cURL (Iris)","langId":"bash","panelId":"panel_cURL (Iris)_iris_agent_memory_memory_etl-stepmemory-etl-outputs"}]},{"codetabsId":"iris_agent_memory_memory_etl-stepmemory-etl-provenance","difficulty":"beginner","id":"memory-etl-provenance","languages":[{"clientId":"iris-python","clientName":"Python (Iris)","id":"Python (Iris)","langId":"python","panelId":"panel_Python (Iris)_iris_agent_memory_memory_etl-stepmemory-etl-provenance"},{"clientId":"iris-curl","clientName":"cURL (Iris)","id":"cURL (Iris)","langId":"bash","panelId":"panel_cURL (Iris)_iris_agent_memory_memory_etl-stepmemory-etl-provenance"}]}] } ``` ## Code examples These examples connect to Iris services using SDKs or HTTP requests with curl. Follow one language throughout each walkthrough. 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](https://redis.io/docs/latest/develop/ai/context-engine/agent-memory/sessions) for conversation events. Use [direct memory creation](https://redis.io/docs/latest/develop/ai/context-engine/agent-memory/long-term-memory#create-a-custom-memory-directly) 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](https://redis.io/docs/latest/operate/iris/agent-memory/self-managed/configuration#enable-document-ingestion). These examples continue from the [quickstart](https://redis.io/docs/latest/develop/ai/context-engine/agent-memory/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](https://redis.io/docs/latest/operate/iris/agent-memory/self-managed/authentication#delegate-a-user-with-an-agent-key). ## Create an extractor Ask your operator to [register a memory type and create a document extractor](https://redis.io/docs/latest/operate/iris/agent-memory/self-managed/api-examples#create-an-extractor). Add the returned extractor ID to `.env` as `EXTRACTOR_ID`. Applications use their Data Plane credential to submit documents. ```dotenv 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. ```text 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: | Field | Purpose | Example | |:------|:--------|:--------| | `documentId` | Your ID for the logical document. | `example-trip-notes` | | `revision` | An immutable, case-sensitive version. Use a new value when the content changes. | `1` | | `content` | The file's UTF-8 bytes, base64-encoded once. | The examples encode `trip-notes.md`. | | `mediaType` | The document's content type. | `text/plain` or `text/markdown` | | `ownerId` | The authenticated user the memories belong to. | `quickstart-user` | | `extractorId` | The extractor's returned `id`. | `$EXTRACTOR_ID` | | `idempotencyKey` | A key to reuse when retrying the same submission. | `example-trip-notes-revision-1` | Submit the file. The server returns a separate ingestion ID to track the run. Save it; you need it even if polling times out. **Available in:** Python (Iris), cURL (Iris) ##### Python (Iris) ```python 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) ``` ##### cURL (Iris) ```bash 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: **Available in:** Python (Iris), cURL (Iris) ##### Python (Iris) ```python 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}") ``` ##### cURL (Iris) ```bash 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: **Available in:** Python (Iris), cURL (Iris) ##### Python (Iris) ```python 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.") ``` ##### cURL (Iris) ```bash 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. | Processing state | Meaning and action | |:-----------------|:-------------------| | `QUEUED` or `RUNNING` | Keep polling the saved ingestion ID. Do not create another submission to check progress. `completedChunks` can equal `totalChunks` while reconciliation is still running. | | `SUCCEEDED` | Processing finished. Inspect `createdCount` and `filteredCount`; success can produce no memories. | | `PARTIAL` | Processing stopped after some memories were published. Inspect `error`, counts, and available output IDs. Published memories remain. | | `FAILED` | Inspect `error` and available outputs. Do not assume that every prior write was rolled back. | `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](https://redis.io/docs/latest/develop/ai/context-engine/agent-memory/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. | Request | Python method | REST endpoint | |:--------|:--------------|:--------------| | Memory → sources | `list_memory_source_links` | `GET .../memories/{memoryId}/sources` | | Memory/source → more evidence | `list_memory_source_evidence` | `GET .../memories/{memoryId}/sources/{sourceId}/evidence` | | Source → memories for an owner | `list_source_memories` | `GET .../sources/{sourceId}/memories` | 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`: **Available in:** Python (Iris), cURL (Iris) ##### Python (Iris) ```python 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 ``` ##### cURL (Iris) ```bash 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.