Agent memory pipelines

Important

This feature is in Beta.

A memory pipeline builds long-term memory from your agent's past conversations. It reads conversation history from a session store, and distills long-term memories, such as facts, preferences, and procedures, into a memory store. Your agent then recalls those memories in later conversations the same way it recalls memories it saved itself.

An agent that saves memories during a conversation writes them one conversation at a time, with only that conversation's context. Over many conversations, its memory collects overlapping and conflicting entries, and facts that the agent didn't think to save don't reach memory. A memory pipeline works after the fact, with all of an actor's conversations and existing memory in view, so it can capture what the agent missed and keep memory consolidated.

Use a memory pipeline when you want to:

  • Build long-term memory from conversations without adding memory-writing tools to your agent.
  • Backfill memory from conversation history that a session store already holds.
  • Keep memory organized over time by merging overlapping entries and splitting oversized ones.

You can use a memory pipeline alone or alongside memory tools that your agent calls during a conversation.

Note

Memory pipeline runs use serverless general compute pricing for the job and Foundation Model Serving pricing for model calls.

Requirements

  • A session store that holds your agent's conversations, and a memory store to write memories to.
  • Access to both stores for the identity that creates the pipeline, including permission to create tables in the memory store's database. Pipeline runs use that identity.
  • Permission to query the model that the pipeline uses for distillation.
  • To use the CLI, install databricks-agentbricks 0.5.0 and above. You can also call the REST API directly from any language.

How memory pipelines work

A memory pipeline is an agent memory resource that connects one session store to one memory store. It isn't related to Lakeflow pipelines. A session store can feed more than one memory store, but each pair of session store and memory store can have only one pipeline.

  • Model calls through Unity Gateway: The pipeline calls its distillation model through Unity Gateway.
  • Your choice of model: You choose the model when you create a pipeline and can change it later. If you don't choose one, the pipeline uses a default model.
  • Manual or scheduled runs on serverless compute: You start runs yourself, or configure the pipeline to run on a schedule about every 24 hours. Either way, each run executes as a job on serverless compute, using the identity that created the pipeline.

Each run of the pipeline processes every actor in the session store separately:

  1. Distill: The pipeline reads the actor's new conversations and uses an LLM to extract durable facts from them.
  2. Merge: The pipeline deduplicates the new facts and merges them into topic-based memory entries under the same actor_id in the memory store.
  3. Maintain: The pipeline reorganizes the actor's memory. It merges entries that overlap and splits entries that grow too large.

Memory pipeline runs keep a checkpoint for each actor ID. A run processes all of an actor's sessions since that actor's last successful run, so the first run processes the actor's full history. If a run fails for one actor, it continues with the other actors, and the next run retries the failed actor from its checkpoint.

The actor_id links the two stores. A pipeline processes only actors that have sessions in the session store, and it writes each actor's memories under the same actor_id that the actor's sessions use. By default, a pipeline runs only when you start a run. You can also run it on a schedule.

Note

Maintenance applies to all of an actor's memory entries, including entries your agent saved directly. A pipeline can rewrite, merge, split, or delete any entry under an actor_id it processes.

What a pipeline keeps

A pipeline extracts individual, durable facts from conversations. It's designed to keep:

  • Facts that the user states explicitly and preferences that they express, without inferring beyond what the conversation says.
  • Corrections and explicit feedback, such as which source to use for revenue figures.
  • Durable domain knowledge, such as authoritative sources, metric definitions, terminology, and team conventions, as well as workflows and ongoing projects.

It's designed to skip:

  • Values from live sources that change over time, such as a table's current row count. It records how to find or compute the value instead.
  • Details that the agent can look up with its tools, such as a table's schema.
  • Secrets, such as passwords, API keys, and tokens.

Get started

These examples create a memory pipeline for a support agent's session and memory stores, start a run, and check its result. Identify each store by its ID, the identifier you chose when you created the store, or by its resource name, such as memory-stores/support-agent-memory. Choose the client that fits your project.

Agent Bricks

The Agent Bricks CLI uses your saved Databricks CLI profile to authenticate.

  1. Install the Agent Bricks CLI, or upgrade it to 0.5.0 and above:

    pip install --upgrade databricks-agentbricks

  2. Create a pipeline that distills the support-agent-sessions session store into the support-agent-memory memory store. --model is optional. If you omit it, the pipeline uses a default model:

    agentbricks memory pipeline create \
  --session-store support-agent-sessions \
  --memory-store support-agent-memory \
  --model system.ai.claude-sonnet-5

    The command returns the pipeline's resource name, in the form memory-pipelines/<pipeline-id>.

  3. Start a run of the pipeline:

    agentbricks memory pipeline run <pipeline-id>

    The command returns the run's resource name, in the form memory-pipelines/<pipeline-id>/runs/<run-id>, and its RUNNING state.

  4. The CLI doesn't report a run's result. To check it, set DATABRICKS_HOST to your workspace hostname, generate an OAuth token, and call the REST API:

    export DATABRICKS_TOKEN=$(databricks auth token --profile <profile> | jq -r .access_token)
curl "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines/<pipeline-id>/runs/<run-id>" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}"

REST API

The CLI calls the REST API under /api/2.0/agents/memory-pipelines. Call it directly from any language.

  1. Generate an OAuth token with the Databricks CLI:

    databricks auth login --host ${DATABRICKS_HOST}
export DATABRICKS_TOKEN=$(databricks auth token | jq -r .access_token)

  2. Create a pipeline that distills memories from support-agent-sessions and saves them to support-agent-memory. model is optional. If you omit it, the pipeline uses a default model:

    curl -X POST "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"session_store": "support-agent-sessions", "memory_store": "support-agent-memory", "model": "system.ai.claude-sonnet-5"}'

    The response's name field holds the pipeline's resource name, in the form memory-pipelines/<pipeline-id>.

  3. Start a run of the pipeline:

    curl -X POST "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines/<pipeline-id>/run" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{}'

    The response's name field holds the run's resource name, in the form memory-pipelines/<pipeline-id>/runs/<run-id>.

  4. Check the run's result:

    curl "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines/<pipeline-id>/runs/<run-id>" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}"

After a run succeeds, your agent can recall the new memories with the memory store's list and search operations. See Managed agent memory.

Monitor pipeline runs

A run's state is one of the following:

State Meaning
RUNNING The run is pending or in progress.
SUCCEEDED The run processed every actor successfully.
FAILED The run finished with an error for one or more actors. The error field describes the failure.
CANCELLED The run was cancelled before it completed.

Steer distillation with instructions

Instructions tell the pipeline what to keep and how to organize it. The pipeline applies them at every stage. Use them for high-level guidance, such as:

  • The topics to focus on, such as customer preferences and account details.
  • The content to skip, such as one-off troubleshooting steps.
  • The paths to organize entries under.

Without layout instructions, the pipeline follows the layout of the store's existing entries, or derives a topic-based layout when the store is empty. It doesn't rename existing entries to match a new layout. To change a specific entry, edit it with the memory store's update operation instead of with instructions.

You can set instructions when you create a pipeline and change them later. The next run uses the new instructions.

Agent Bricks

To set instructions when you create a pipeline, pass --instructions:

agentbricks memory pipeline create \
  --session-store support-agent-sessions \
  --memory-store support-agent-memory \
  --instructions "Focus on durable customer preferences and account details. Ignore one-off troubleshooting steps. Organize entries under /preferences/ and /account/."

To load longer instructions from a UTF-8 file, pass --instructions @<path-to-file> instead.

To change the instructions of an existing pipeline, run the following:

agentbricks memory pipeline update <pipeline-id> \
  --instructions "Keep only durable customer preferences."

REST API

In the REST API, a pipeline's dreamer_policy object holds its distillation settings: instructions and trigger. To set instructions when you create a pipeline, include them in dreamer_policy.instructions:

curl -X POST "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"session_store": "support-agent-sessions", "memory_store": "support-agent-memory", "dreamer_policy": {"instructions": "Focus on durable customer preferences and account details. Ignore one-off troubleshooting steps. Organize entries under /preferences/ and /account/."}}'

To change the instructions of an existing pipeline, update dreamer_policy.instructions:

curl -X PATCH "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines/<pipeline-id>?update_mask=dreamer_policy.instructions" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"dreamer_policy": {"instructions": "Keep only durable customer preferences."}}'

Run a pipeline on a schedule

A scheduled pipeline runs automatically about every 24 hours. You can set the trigger when you create a pipeline, or change it later. The scheduler chooses when the first scheduled run starts. To distill existing conversations right away, start a run manually.

Agent Bricks

To create a scheduled pipeline, pass --trigger scheduled. The default is --trigger manual:

agentbricks memory pipeline create \
  --session-store support-agent-sessions \
  --memory-store support-agent-memory \
  --trigger scheduled

To change the trigger of an existing pipeline, run the following:

agentbricks memory pipeline update <pipeline-id> --trigger scheduled

REST API

To create a scheduled pipeline, set dreamer_policy.trigger to SCHEDULED:

curl -X POST "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"session_store": "support-agent-sessions", "memory_store": "support-agent-memory", "dreamer_policy": {"trigger": "SCHEDULED"}}'

To change the trigger of an existing pipeline, update dreamer_policy.trigger to SCHEDULED or MANUAL_ONLY:

curl -X PATCH "https://${DATABRICKS_HOST}/api/2.0/agents/memory-pipelines/<pipeline-id>?update_mask=dreamer_policy.trigger" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" -H "Content-Type: application/json" \
  -d '{"dreamer_policy": {"trigger": "SCHEDULED"}}'

You can also start a run of a scheduled pipeline manually, as long as no other run is in progress.

Manage pipelines

The following commands list, inspect, update, and delete pipelines. Each CLI command accepts either the pipeline ID or its full memory-pipelines/<pipeline-id> resource name.

Task Agent Bricks CLI REST API
List pipelines agentbricks memory pipeline list GET /api/2.0/agents/memory-pipelines
Get a pipeline agentbricks memory pipeline get <pipeline-id> GET /api/2.0/agents/memory-pipelines/<pipeline-id>
Update a pipeline agentbricks memory pipeline update <pipeline-id> with --display-name, --model, --trigger, or --instructions PATCH /api/2.0/agents/memory-pipelines/<pipeline-id> with an update_mask of display_name, model, dreamer_policy.trigger, or dreamer_policy.instructions
Delete a pipeline agentbricks memory pipeline delete <pipeline-id> DELETE /api/2.0/agents/memory-pipelines/<pipeline-id>

If you don't set a display name, the pipeline gets one based on its session store and memory store names. To switch a pipeline back to the default model, update its model to an empty value, such as --model "". Model and instruction changes apply to the next run.

Deleting a pipeline deletes its backing job and stops future runs. A run that is already in progress continues to completion. Memory entries that the pipeline already wrote stay in the memory store.

Limitations

  • The session store and memory store are fixed when you create a pipeline. To change either one, delete the pipeline and create a new one.
  • If a pipeline already connects the same session store and memory store, creating another one fails with ALREADY_EXISTS.
  • A pipeline has one run in progress at a time. Starting a run while another is in progress fails with an ABORTED error.
  • Scheduled pipelines run about every 24 hours. You can't set a different interval or a cron schedule.
  • A pipeline processes only actors that have sessions in the session store. Memory entries for actors without sessions are left unchanged.
  • If the session store has more than about 1,200 actors with UUID actor IDs, the run fails. The limit is 48 KiB of serialized actor IDs, so shorter IDs allow more actors.

Additional resources

  • Востаннє оновлено: