Pipelines de memoria de agentes

Importante

Esta característica se encuentra en su versión beta.

Una canalización de memoria construye memoria a largo plazo a partir de las conversaciones pasadas de tu agente. Lee el historial de conversaciones de un almacén de sesiones y destila recuerdos a largo plazo, como hechos, preferencias y procedimientos, en un almacén de memoria. Tu agente entonces recuerda esos recuerdos en conversaciones posteriores de la misma manera que recuerda los recuerdos que él mismo guardó.

Un agente que guarda memorias durante una conversación las escribe una conversación a la vez, con solo el contexto de esa conversación. A lo largo de muchas conversaciones, su memoria recoge entradas superpuestas y contradictorias, y hechos que el agente no pensó en guardar no llegan a la memoria. Una tubería de memoria funciona a posteriori, con todas las conversaciones y la memoria existente de un actor a la vista, para poder capturar lo que el agente pasó por alto y mantener la memoria consolidada.

Utiliza una tubería de memoria cuando quieras:

  • Construye memoria a largo plazo a partir de conversaciones sin añadir herramientas para escribir memoria a tu agente.
  • Rellena la memoria con el historial de conversaciones que ya contiene el almacenamiento de sesión.
  • Mantén la memoria organizada con el tiempo fusionando entradas que se solapan y dividiendo las de gran tamaño.

Puedes usar una canalización de memoria sola o junto con herramientas de memoria que tu agente llame durante una conversación.

Nota

Las ejecuciones de la tubería de memoria utilizan precios generales de computación sin servidor para el trabajo y precios de Foundation Model Serving para llamadas a modelos.

Requirements

  • Un almacén de sesiones que guarda las conversaciones de tu agente y otro almacén de recuerdos para escribir recuerdos.
  • Acceso a ambos almacenes para la identidad que crea el pipeline, incluyendo permiso para crear tablas en la base de datos del almacén de memoria. Las ejecuciones de canalización se ejecutan con esa identidad.
  • Permiso para consultar el modelo que utiliza la canalización para la destilación. Sin él, las ejecuciones del pipeline fallan.
  • Para usar la CLI, instala databricks-agentbricks la versión 0.6.0 o superior. También puedes llamar directamente a la API REST desde cualquier lenguaje.

Cómo funcionan las canalizaciones de memoria

Una tubería de memoria es un recurso de memoria agente que conecta un almacén de sesión con un almacén de memoria. No está relacionado con los pipelines de Lakeflow. Un almacén de sesión puede alimentar más de un almacén de memoria, pero cada par de almacén de sesión y almacenamiento de memoria puede tener solo un pipeline.

  • Llamadas al modelo a través de Unity Gateway: La canalización invoca su modelo de destilación a través de Unity Gateway.
  • Tu elección de modelo: Eliges el modelo cuando creas una canalización y puedes cambiarlo más adelante. Si no eliges uno, la canalización usa un modelo por defecto.
  • Ejecuciones manuales o programadas en computación serverless: Tú mismo inicias las ejecuciones, o configuras la pipeline para que funcione con un horario aproximadamente cada 24 horas. En cualquier caso, cada ejecución se ejecuta como un trabajo en computación serverless, usando la identidad que creó el pipeline.

Cada ejecución de la canalización procesa cada actor en el almacenamiento de sesión por separado:

  1. Destilar: La canalización lee las nuevas conversaciones del actor y utiliza un LLM para extraer datos duraderos de ellas.
  2. Fusionar: La tubería deduplica los nuevos hechos y los fusiona en entradas de memoria basadas en temas bajo la misma clave actor_id en el almacén de memoria.
  3. Mantén: La pipeline reorganiza la memoria del actor. Fusiona entradas que se solapan y divide las que crecen demasiado.

Las ejecuciones de la tubería de memoria mantienen un punto de control para cada identificador del actor. Una ejecución procesa todas las sesiones de un actor desde la última ejecución exitosa de ese actor, por lo que la primera ejecución procesa el historial completo del actor. Si una ejecución falla para un actor, continúa con los otros actores, y la siguiente ejecución vuelve a intentar el actor fallido desde su punto de control.

El actor_id conecta las dos tiendas. Una tubería procesa solo actores que tienen sesiones en el almacén de sesiones, y escribe las memorias de cada actor bajo el mismo actor_id que usan las sesiones del actor. Por defecto, una canalización solo se ejecuta cuando inicias una ejecución. También puedes ejecutarlo de forma programada.

Nota

El mantenimiento se aplica a todas las entradas de memoria de un actor, incluidas las entradas que tu agente guardó directamente. Una tubería puede reescribir, fusionar, dividir o eliminar cualquier entrada dentro de un actor_id que procesa.

Qué conserva un pipeline

Un pipeline extrae hechos individuales y duraderos de las conversaciones. Está diseñado para mantener:

  • Hechos que el usuario expone explícitamente y preferencias que expresa, sin inferir más allá de lo que dice la conversación.
  • Correcciones y comentarios explícitos, como qué fuente usar para las cifras de ingresos.
  • Conocimiento duradero del dominio, como fuentes autorizadas, definiciones métricas, terminología y convenciones de equipo, así como flujos de trabajo y proyectos en curso.

Está diseñado para saltar:

  • Valores de fuentes activas que cambian con el tiempo, como el recuento actual de filas de una tabla. Registra cómo encontrar o calcular el valor en su lugar.
  • Detalles que el agente puede consultar con sus herramientas, como el esquema de una tabla.
  • Secretos, como contraseñas, claves API y tokens.

Primeros pasos

Estos ejemplos crean una tubería de memoria para la sesión y los almacenes de memoria de un agente de soporte, inician una ejecución y comprueban su resultado. Identifica cada tienda por su ID, el identificador que elegiste al crear la tienda o por su nombre de recurso, como memory-stores/support-agent-memory. Elige el cliente que se adapte a tu proyecto.

Ladrillos del agente

La CLI de Databricks utiliza tu perfil guardado de la CLI de Databricks para autenticarse.

  1. Instala la CLI de Agent Bricks, o actualízala a la 0.6.0 o superior:

    pip install --upgrade databricks-agentbricks

  2. Crea una canalización que condensa el support-agent-sessions almacén de sesiones en el support-agent-memory almacén de memoria. --model es opcional. Si lo omites, la canalización utiliza un modelo por defecto:

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

    El comando devuelve el nombre del recurso de la canalización, en la forma memory-pipelines/<pipeline-id>.

  3. Inicia una ejecución del pipeline:

    agentbricks memory pipeline run <pipeline-id>

    El comando devuelve el nombre del recurso de la ejecución, en la forma memory-pipelines/<pipeline-id>/runs/<run-id>, y su estado RUNNING.

  4. El CLI no informa del resultado de una ejecución. Para comprobarlo, configura DATABRICKS_HOST con el nombre de host de tu espacio de trabajo, genera un token OAuth y llama a la API REST:

    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}"

API DE REST

La CLI llama a la API REST bajo /api/2.0/agents/memory-pipelines. Llámalo directamente en cualquier idioma.

  1. Genera un token OAuth con la CLI de Databricks:

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

  2. Crea una canalización que extrae memorias de support-agent-sessions y las guarda en support-agent-memory. dreamer_policy.model es opcional. Si lo omites, la canalización utiliza un modelo por defecto:

    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": {"model": "system.ai.claude-sonnet-5"}}'

    El campo de name la respuesta contiene el nombre del recurso de la canalización, en la forma memory-pipelines/<pipeline-id>.

  3. Inicia una ejecución del 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 '{}'

    El campo de name la respuesta contiene el nombre del recurso de la ejecución, en la forma memory-pipelines/<pipeline-id>/runs/<run-id>.

  4. Consulta el resultado de la ejecución:

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

Después de que una ejecución se complete correctamente, tu agente puede recuperar los nuevos recuerdos con las operaciones de lista y búsqueda del almacén de recuerdos. Ver memoria del agente.

Supervisión de ejecuciones de canalización

La state de una ejecución es una de las siguientes:

State Meaning
RUNNING La ejecución está pendiente o en curso.
SUCCEEDED La ejecución procesó con éxito a todos los actores.
FAILED La ejecución terminó con un error para uno o más actores. El error campo describe la falla.
CANCELLED La carrera fue cancelada antes de completarse.

Guiar la destilación con instrucciones

Las instrucciones indican a la canalización qué mantener y cómo organizarla. La canalización los aplica en cada etapa. Utilízalos para orientación general, como:

  • Los temas en los que centrarse, como las preferencias de los clientes y los detalles de la cuenta.
  • El contenido que se debe omitir, como pasos puntuales para solucionar problemas.
  • Las rutas bajo las cuales organizar las entradas.

Sin instrucciones de layout, la pipeline sigue la disposición de las entradas existentes de la tienda, o deriva una disposición basada en temas cuando la tienda está vacía. No renombra las entradas existentes para que coincidan con un nuevo diseño. Para cambiar una entrada específica, edítala con la operación de actualización del memory store en lugar de con instrucciones.

Puedes establecer instrucciones cuando crees una canalización y cambiarlas después. La siguiente ejecución usa las nuevas instrucciones.

Ladrillos del agente

Para establecer instrucciones al crear una canalización, pasa --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/."

Para cargar instrucciones más largas desde un archivo UTF-8, pasa --instructions @<path-to-file> en vez de eso.

Para cambiar las instrucciones de un pipeline existente, ejecuta lo siguiente:

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

API DE REST

En la API REST, el objeto dreamer_policy de una tubería mantiene sus configuraciones de destilación: instructions, trigger y model. Para establecer instrucciones al crear una canalización, inclúyelas en 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/."}}'

Para cambiar las instrucciones de un pipeline existente, actualiza 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."}}'

Ejecutar una canalización de forma programada

Una pipeline programada se ejecuta automáticamente aproximadamente cada 24 horas. Puedes configurar el trigger cuando crees una pipeline, o cambiarlo más adelante. El programador de tareas elige cuándo comienza la primera ejecución programada. Para destilar las conversaciones existentes de inmediato, inicia una ejecución manualmente.

Ladrillos del agente

Para crear un pipeline programado, pasa el argumento --trigger scheduled. El valor predeterminado es --trigger manual:

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

Para cambiar el desencadenador de un pipeline existente, ejecuta lo siguiente:

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

Si tienes un pipeline programado, puedes pausarlo:

agentbricks memory pipeline pause <pipeline-id>

Para reiniciar las ejecuciones programadas, reanúdalas:

agentbricks memory pipeline resume <pipeline-id>

API DE REST

Para crear una canalización programada, fije dreamer_policy.trigger en 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"}}'

Para cambiar el desencadenador de un pipeline existente, actualice dreamer_policy.trigger a SCHEDULED o 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"}}'

También puedes iniciar manualmente una ejecución de un pipeline programado, siempre que no haya otra ejecución en curso.

Administración de canalizaciones

Los siguientes comandos enumeran, inspeccionan, actualizan y eliminan pipelines. Cada comando CLI acepta el ID de la canalización o el nombre completo memory-pipelines/<pipeline-id> de su recurso.

tarea CLI de Agent Bricks API DE REST
Enumerar canalizaciones agentbricks memory pipeline list GET /api/2.0/agents/memory-pipelines
Consigue un pipeline agentbricks memory pipeline get <pipeline-id> GET /api/2.0/agents/memory-pipelines/<pipeline-id>
Actualizar una canalización agentbricks memory pipeline update <pipeline-id> con --display-name, --model, --trigger, o --instructions PATCH /api/2.0/agents/memory-pipelines/<pipeline-id> con un update_mask de display_name, dreamer_policy.model, dreamer_policy.trigger, o dreamer_policy.instructions
Eliminación de una canalización agentbricks memory pipeline delete <pipeline-id> DELETE /api/2.0/agents/memory-pipelines/<pipeline-id>

Si no configuras un nombre para mostrar, la canalización obtiene uno según los nombres de su almacenamiento de sesión y memoria. Para volver una tubería al modelo por defecto, actualice su modelo a un valor vacío, como --model "". Los cambios en el modelo y la instrucción se aplican a la siguiente ejecución.

Eliminar una canalización elimina su trabajo subyacente y detiene futuras ejecuciones. Una ejecución que ya está en progreso continúa hasta su finalización. Las entradas de memoria que el pipeline ya escribió permanecen en el almacén de memoria.

Limitations

  • El almacenamiento de sesiones y el almacenamiento de memoria se establecen cuando creas una canalización. Para cambiar cualquiera de los dos, elimina el pipeline y crea uno nuevo. Una nueva canalización no reutiliza los puntos de control de la antigua, así que su primera ejecución procesa el historial completo de cada actor, incluso para el mismo par de almacenes.
  • Si una tubería ya conecta el mismo almacén de sesiones y el mismo almacén de memoria, crear otra falla con ALREADY_EXISTS.
  • Un pipeline tiene una ejecución en curso a la vez. Iniciar una ejecución mientras otra está en progreso falla con un error ABORTED.
  • Las canalizaciones programadas se ejecutan aproximadamente cada 24 horas. No puedes establecer un intervalo diferente ni un horario cron.
  • Una canalización solo procesa a los actores que tienen sesiones en el almacén de sesiones. Las entradas de memoria para actores sin sesiones permanecen sin cambios.
  • Si el almacén de sesiones tiene más de unos 1.200 actores con ID de actor UUID, la ejecución falla. El límite es de 48 KiB de IDs de actores serializados, por lo que IDs más cortos permiten más actores.

Recursos adicionales

  • Se actualizó por última vez el