Pipelines für Agentenspeicher

Eine Speicherpipeline baut Langzeitgedächtnis aus den vergangenen Gesprächen Ihres Agenten auf. Es liest den Gesprächsverlauf aus einem Sitzungsspeicher und destilliert Langzeiterinnerungen wie Fakten, Präferenzen und Verfahren in einen Speicher für Erinnerungen. Dein Agent ruft diese Erinnerungen dann in späteren Gesprächen genauso ab, wie die Erinnerungen, die er selbst gespeichert hat.

Ein Agent, der während eines Gesprächs Erinnerungen speichert, schreibt sie jeweils ein Gespräch nach dem anderen, nur mit dem Kontext dieses einen Gesprächs. In vielen Gesprächen sammelt sein Gedächtnis überlappende und widersprüchliche Einträge, und Fakten, die der Agent nicht daran gedacht hat, zu speichern, gelangen nicht ins Gedächtnis. Eine Gedächtnispipeline arbeitet nachträglich, mit Blick auf alle Gespräche eines Akteurs und bestehende Gedächtnisinhalte, sodass sie erfassen kann, was der Agent verpasst hat, und das Gedächtnis konsolidieren kann.

Verwenden Sie eine Speicherpipeline, wenn Sie möchten:

• Baue Langzeitgedächtnis aus Gesprächen auf, ohne deinem Agenten Memory-Write-Tools hinzuzufügen.
• Fülle einen Speicher aus dem Konversationsverlauf auf, den ein Session-Store bereits enthält.
• Halte das Gedächtnis über die Zeit organisiert, indem du überlappende Einträge zusammenführst und übergroße Einträge aufteilst.

Du kannst eine Speicherpipeline allein oder zusammen mit Speicherwerkzeugen verwenden, die dein Agent während eines Gesprächs aufruft.

Requirements

• Ein Session-Store, der die Gespräche deines Agenten speichert, und einen Memory-Store, in den du Erinnerungen schreiben kannst.
• Zugriff auf beide Datenspeicher für die Identität, die die Pipeline erstellt, einschließlich der Berechtigung, Tabellen in der Datenbank des Memory Store zu erstellen. Pipeline-Läufe verwenden diese Identität.
• Berechtigung, das Modell abzufragen, das die Pipeline für die Distillation verwendet. Ohne es scheitern die Durchläufe der Pipeline.
• Um die CLI zu nutzen, installiere databricks-agentbricks 0.6.0 und höher. Du kannst die REST-API auch direkt aus jeder beliebigen Sprache aufrufen.

Wie Speicherpipelines funktionieren

Eine Speicherpipeline ist eine Agent-Speicherressource, die einen Sitzungsspeicher mit einem Speicherspeicher verbindet. Es hat nichts mit Lakeflow-Pipelines zu tun. Ein Session-Speicher kann mehr als einen Memory-Store versorgen, aber jedes Paar aus Session-Speicher und Memory-Store kann nur eine Pipeline haben.

• Modellaufrufe über Unity Gateway: Die Pipeline ruft ihr Destillationsmodell über Unity Gateway auf.
• Ihre Wahl des Modells: Sie wählen das Modell aus, wenn Sie eine Pipeline erstellen, und können es später ändern. Wenn du keinen wählst, verwendet die Pipeline ein Standardmodell.
• Manuelle oder geplante Läufe mit serverloser Rechenleistung: Du startest die Läufe selbst oder konfigurierst die Pipeline so, dass sie etwa alle 24 Stunden nach einem Zeitplan läuft. So oder so wird jeder Lauf als Job mit serverloser Rechenkapazität ausgeführt, wobei die Identität verwendet wird, die die Pipeline erstellt hat.

Jeder Durchlauf der Pipeline verarbeitet jeden Akteur im Session-Speicher separat:

1. Destillieren: Die Pipeline liest die neuen Gespräche des Akteurs und verwendet ein LLM, um daraus dauerhafte Fakten zu extrahieren.
2. Zusammenführen: Die Pipeline dedupliziert die neuen Fakten und führt sie in themenbasierte Speichereinträge unter demselben actor_id im Speicher zusammen.
3. Wartung: Die Pipeline organisiert das Gedächtnis des Akteurs neu. Es führt sich überlappende Einträge zusammen und teilt Einträge, die zu groß werden.

Speicherpipeline-Läufe speichern für jede Akteur-ID einen Checkpoint. Eine Ausführung verarbeitet alle Sitzungen eines Akteurs seit dessen letzter erfolgreicher Ausführung, sodass die erste Ausführung die vollständige Historie des Akteurs verarbeitet. Wenn ein Durchlauf für einen Akteur fehlschlägt, wird er mit den anderen Akteuren fortgesetzt, und der nächste Durchlauf versucht den fehlgeschlagenen Akteur von seinem Checkpoint aus.

Das actor_id verbindet die beiden Stores. Eine Pipeline verarbeitet nur Akteure, die Sitzungen im Sitzungsspeicher haben, und schreibt die Erinnerungen jedes Akteurs unter demselben actor_id, das die Sitzungen des Akteurs verwenden. Standardmäßig läuft eine Pipeline nur, wenn du einen Run startest. Du kannst es auch nach einem Zeitplan ausführen.

Was eine Pipeline beibehält

Eine Pipeline gewinnt individuelle, dauerhafte Fakten aus Gesprächen. Sie ist konzipiert, Folgendes beizubehalten

• Tatsachen, die der Nutzer explizit angibt, und Präferenzen, die er äußert, ohne über das hinauszugehen, was das Gespräch sagt.
• Korrekturen und explizite Rückmeldungen, wie zum Beispiel, welche Quelle für die Umsatzzahlen verwendet werden sollte.
• Beständiges Fachwissen, wie autoritative Quellen, metrische Definitionen, Terminologie und Teamkonventionen sowie Arbeitsabläufe und laufende Projekte.

Es ist darauf ausgelegt, Folgendes zu überspringen:

• Werte von Live-Quellen, die sich im Laufe der Zeit ändern, wie zum Beispiel die aktuelle Zeilenanzahl einer Tabelle. Stattdessen wird aufgezeichnet, wie man den Wert findet oder berechnet.
• Details, die der Agent mit seinen Werkzeugen nachschlagen kann, wie zum Beispiel das Schema einer Tabelle.
• Geheimnisse wie Passwörter, API-Schlüssel und Token.

Get started

Diese Beispiele erstellen eine Speicherpipeline für die Session- und Memory Stores eines Support-Agenten, starten einen Run und überprüfen dessen Ergebnis. Identifizieren Sie jeden Store anhand seiner ID, des von Ihnen gewählten Identifikators, den Sie bei der Erstellung des Stores gewählt haben, oder anhand seines Ressourcennamens, wie zum Beispiel memory-stores/support-agent-memory. Wählen Sie den Kunden, der zu Ihrem Projekt passt.

Agent Bricks

Die Agent Bricks CLI verwendet Ihr gespeichertes Databricks CLI-Profil zur Authentifizierung.

1. Installiere die Agent Bricks-CLI oder upgrade sie auf 0.6.0 und höher:

   pip install --upgrade databricks-agentbricks
   

2. Erstelle eine Pipeline, die den support-agent-sessions-Sitzungsspeicher in den support-agent-memory-Speicherstore überführt. --model ist optional. Wenn du es weglässt, verwendet die Pipeline ein Standardmodell:

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

   Der Befehl gibt den Ressourcennamen der Pipeline in der Form memory-pipelines/<pipeline-id>zurück.

3. Starte einen Durchlauf der Pipeline:

   agentbricks memory pipeline run <pipeline-id>
   

   Der Befehl liefert den Ressourcennamen des Laufs in der Form memory-pipelines/<pipeline-id>/runs/<run-id>, und seinen RUNNING Zustand.

4. Die CLI meldet das Ergebnis eines Durchlaufs nicht. Um es zu überprüfen, setzen Sie DATABRICKS_HOST auf den Hostnamen Ihres Arbeitsbereichs, generieren Sie ein OAuth-Token und rufen Sie die REST-API auf:

   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

Die CLI ruft die REST-API unter /api/2.0/agents/memory-pipelines. Rufen Sie es direkt aus jeder Programmiersprache auf.

1. Generiere ein OAuth-Token mit der Databricks-CLI:

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

2. Erstelle eine Pipeline, die Erinnerungen aus support-agent-sessions destilliert und in support-agent-memory speichert. dreamer_policy.model ist optional. Wenn du es weglässt, verwendet die Pipeline ein Standardmodell:

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

   Das name Feld der Antwort enthält den Ressourcennamen der Pipeline in der Form memory-pipelines/<pipeline-id>.

3. Starte einen Durchlauf der 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 '{}'
   

   Das name Feld der Antwort enthält den Ressourcennamen des Laufs in der Form memory-pipelines/<pipeline-id>/runs/<run-id>.

4. Überprüfen Sie das Ergebnis des Laufs:

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

Nach einem erfolgreichen Durchlauf kann dein Agent die neuen Erinnerungen mit den Listen- und Suchoperationen des Memory-Stores abrufen. Siehe Agent-Arbeitsspeicher.

Pipeline-Läufe überwachen

Das state eines Runs ist eines der folgenden:

Destillation mit Anweisungen steuern

Die Anweisungen geben der Pipeline vor, was die Pipeline behalten und wie es organisiert werden soll. Die Pipeline setzt sie in jeder Phase ein. Nutzen Sie sie zur allgemeinen Orientierung, wie zum Beispiel:

• Die Themen, auf die man sich konzentrieren sollte, wie Kundenpräferenzen und Kontodetails.
• Die Inhalte, die man überspringen kann, wie zum Beispiel einmalige Schritte zur Problembehebung.
• Die Pfade, unter denen Einträge organisiert werden.

Ohne Layout-Anweisungen folgt die Pipeline dem Layout der bestehenden Einträge des Stores oder leitet ein themenbasiertes Layout ab, wenn der Store leer ist. Bestehende Einträge werden nicht umbenannt, um zu einem neuen Layout zu passen. Um einen bestimmten Eintrag zu ändern, bearbeiten Sie ihn mithilfe der Aktualisierungsoperation des Speichers anstatt durch Anweisungen.

Du kannst Anweisungen beim Erstellen einer Pipeline festlegen und sie später ändern. Der nächste Durchlauf verwendet die neuen Anweisungen.

Agent Bricks

Um beim Erstellen einer Pipeline Anweisungen festzulegen, übergeben Sie --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/."

Um längere Befehle aus einer UTF-8-Datei zu laden, geben Sie stattdessen --instructions @<path-to-file> an.

Um die Anweisungen einer bestehenden Pipeline zu ändern, führen Sie Folgendes aus:

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

REST API

In der REST-API hält das dreamer_policy Objekt einer Pipeline seine Destillationseinstellungen: instructions, trigger, und model. Um Anweisungen beim Erstellen einer Pipeline festzulegen, fügen Sie sie in dreamer_policy.instructions ein:

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/."}}'

Um die Anweisungen einer bestehenden Pipeline zu ändern, aktualisieren Sie 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."}}'

Eine Pipeline nach einem Zeitplan ausführen

Eine geplante Pipeline läuft automatisch etwa alle 24 Stunden. Du kannst den Trigger beim Erstellen einer Pipeline festlegen oder später ändern. Der Scheduler wählt, wann der erste geplante Lauf beginnt. Um bestehende Gespräche sofort zusammenzufassen, starten Sie manuell einen Durchlauf.

Agent Bricks

Um eine geplante Pipeline zu erstellen, geben Sie --trigger scheduled an. Der Standardwert ist --trigger manual:

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

Um den Trigger einer bestehenden Pipeline zu ändern, führen Sie Folgendes aus:

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

Wenn Sie eine geplante Pipeline haben, können Sie sie pausieren:

agentbricks memory pipeline pause <pipeline-id>

Um geplante Ausführungen neu zu starten, setzen Sie ihn fort:

agentbricks memory pipeline resume <pipeline-id>

REST API

Um eine geplante Pipeline zu erstellen, setzen Sie dreamer_policy.trigger auf 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"}}'

Um den Trigger einer bestehenden Pipeline zu ändern, aktualisieren Sie dreamer_policy.trigger auf SCHEDULED oder 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"}}'

Du kannst auch manuell einen Lauf einer geplanten Pipeline starten, solange kein weiterer Lauf läuft.

Verwalten von Pipelines

Die folgenden Befehle listen Pipelines auf, inspizieren, aktualisieren und löschen. Jeder CLI-Befehl akzeptiert entweder die Pipeline-ID oder den vollständigen memory-pipelines/<pipeline-id> Ressourcennamen.

Wenn Sie keinen Anzeigenamen festlegen, erhält die Pipeline einen basierend auf ihren Session- und Memory-Store-Namen. Um eine Pipeline wieder auf das Standardmodell zurückzuschalten, aktualisieren Sie ihr Modell auf einen leeren Wert, wie beispielsweise --model "". Modell- und Befehlsänderungen gelten für den nächsten Durchlauf.

Das Löschen einer Pipeline löscht ihren Backing-Job und stoppt zukünftige Durchläufe. Eine bereits laufende Ausführung läuft bis zum Abschluss weiter. Speichereinträge, die die Pipeline bereits in den Speicher geschrieben hat, bleiben im Speicher.

Einschränkungen

• Der Session Store und der Memory Store sind festgelegt, wenn du eine Pipeline erstellst. Um eine der beiden zu ändern, lösche die Pipeline und erstelle eine neue. Eine neue Pipeline verwendet die Checkpoints der alten Pipeline nicht wieder, sodass ihr erster Lauf die vollständige Historie jedes Akteurs verarbeitet, selbst für dasselbe Speicherpaar.
• Wenn eine Pipeline bereits denselben Sitzungsspeicher und Memory-Store verbindet, schlägt das Erstellen einer weiteren mit ALREADY_EXISTS fehl.
• Eine Pipeline hat jeweils einen laufenden Durchlauf. Das Starten eines Durchlaufs, während ein anderer läuft, schlägt mit einem ABORTED Fehler fehl.
• Planmäßige Pipelines laufen etwa alle 24 Stunden. Du kannst kein anderes Intervall oder einen Cron-Plan festlegen.
• Eine Pipeline verarbeitet nur Actors, die Sitzungen im Sitzungsspeicher haben. Speichereinträge für Akteure ohne Sitzungen bleiben unverändert.
• Wenn der Session Store mehr als etwa 1.200 Aktoren mit UUID-Aktor-IDs hat, schlägt der Run fehl. Das Limit liegt bei 48 KiB serialisierter Aktor-IDs, sodass kürzere IDs mehr Aktoren erlauben.

Weitere Ressourcen

• Agentspeicher
• Agent-Sitzungen
• Agent-Speicher und Sitzungen
• Agent-API-Referenz