エージェントメモリパイプライン

メモリーパイプラインは、エージェントの過去の会話から長期記憶を構築します。 セッションストアから会話履歴を読み取り、事実、好み、手順などの長期記憶を抽出してメモリストアに保存します。 エージェントはその後の会話で、自分が保存した記憶と同じようにその記憶を思い出します。

会話中に記憶を保存するエージェントは、その会話の文脈だけを踏まえて、記憶を会話ごとに1件ずつ書き込みます。 多くの会話の中で、その記憶は重複し矛盾する記録を集め、エージェントが保存しようと思わなかった事実は記憶に届かない。 メモリパイプラインは事後に機能し、アクターのすべての会話と既存のメモリを視認できるため、エージェントが見落としたものをキャプチャし、メモリを統合できます。

次の場合は、メモリ パイプラインを使用します:

• メモリ書き込みツールをエージェントに追加せずに、会話から長期メモリを構築します。
• セッションストアがすでに保持している会話履歴からメモリをバックフィルします。
• 重なり合う項目を統合したり、大きすぎる項目を分割して記憶を整理しましょう。

メモリパイプラインは単独でも、会話中にエージェントが呼び出す メモリツール と併用することも可能です。

Requirements

• エージェントの会話を保存する セッションストア と、メモリーを書き込むための メモリストア です。
• パイプラインを作成するアイデンティティには、両方のストアへのアクセス権と、メモリストアのデータベース内でテーブルを作成する権限が必要です。 パイプラインの実行はそのIDを使用します。
• パイプラインが蒸留に使用するモデルをクエリする権限。
• CLI を使うには、databricks-agentbricks 0.5.0 以降をインストールしてください。 また、どの言語からでも REST API を直接呼び出すこともできます。

メモリパイプラインの仕組み

メモリパイプラインは、1つのセッションストアを1つのメモリストアに接続するエージェントメモリリソースです。 これはLakeflowパイプラインとは関係ありません。 セッションストアは複数のメモリストアにデータを供給できますが、セッションストアとメモリストアの各ペアはパイプラインを1つしか持てません。

• Unity Gatewayを通じたモデル呼び出し: パイプラインは Unity Gatewayを通じて蒸留モデルを呼び出します。
• モデルの選択:パイプラインを作成するときにモデルを選び、後で変更できます。 もし選ばなければ、パイプラインはデフォルトのモデルを使用します。
• サーバーレスコンピュートでの手動またはスケジュールされた実行: 自分で実行を開始するか、パイプラインを約24時間ごとにスケジュールで実行するように設定します。 いずれにせよ、各実行はサーバーレスコンピュート上でジョブとして実行され、パイプラインを作成したアイデンティティを使用します。

パイプラインの各実行はセッションストア内のすべてのアクターを個別に処理します:

1. 抽出: パイプラインはアクターの新しい会話を読み取り、LLM を使ってそこから永続的な情報を抽出します。
2. マージ: パイプラインは新しい情報の重複を排除し、メモリ ストア内の同じ actor_id の下でトピックベースのメモリ エントリにマージします。
3. 維持: パイプラインはアクターのメモリを再編成します。 重複するエントリを統合し、大きくなりすぎるエントリを分割します。

メモリパイプラインの実行は、各アクターIDごとにチェックポイントを保持します。 ランはアクターの最後の成功した実行以降のすべてのセッションを処理するため、最初のランでアクターの全履歴を処理します。 あるアクターで実行が失敗した場合、他のアクターで継続し、次の実行で失敗したアクターをチェックポイントから再試行します。

この actor_id は両店舗を結んでいます。 パイプラインはセッションストアにセッションを持つアクターのみを処理し、各アクターのメモリをそのアクターのセッションが使う同じ actor_id に書き込みます。 デフォルトでは、パイプラインは実行を開始したときにのみ実行されます。 スケジュール で実行することもできます。

パイプラインが保持する内容

パイプラインは会話から個々の永続的な情報を抽出します。 以下を保持するよう設計されています:

• ユーザーが明示的に述べた事実や表明した好みであり、会話の内容を超えた推測を含まないものです。
• 収益の数字にどの情報源を使うかなどの訂正や明確なフィードバック。
• 権限のあるソース、メトリックの定義、用語、チームの規則などの永続的なドメイン ナレッジ、およびワークフローや進行中のプロジェクト。

スキップするように設計されています:

• テーブルの現在の行数など、時間とともに変化するライブソースからの値を指す。 代わりに、値の見つけ方や計算方法を記録しています。
• エージェントがツールで調べられる詳細、例えばテーブルのスキーマなどです。
• パスワード、APIキー、トークンなどの秘密情報です。

概要

これらの例は、サポートエージェントのセッションとメモリストアのためのメモリパイプラインを作成し、実行を開始し、その結果を確認します。 各ストアはID、ストア作成時に選択した識別子、またはリソース名( memory-stores/support-agent-memoryなど)で識別します。 自分のプロジェクトに合ったクライアントを選びましょう。

エージェント・ブリックス

Agent Bricks CLIは保存したDatabricks CLIプロファイルを使って認証します。

1. Agent Bricks CLIをインストールするか、0.5.0以上にアップグレードしてください:

   pip install --upgrade databricks-agentbricks
   

2. support-agent-sessionsセッションストアをsupport-agent-memoryメモリストアに集約するパイプラインを作成しましょう。 --model はオプションです。 これを省略すると、パイプラインはデフォルトのモデルを使用します:

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

   コマンドはパイプラインのリソース名を memory-pipelines/<pipeline-id>の形で返します。

3. パイプラインの実行を開始:

   agentbricks memory pipeline run <pipeline-id>
   

   コマンドは、実行のリソース名 (memory-pipelines/<pipeline-id>/runs/<run-id> の形式) と RUNNING の状態を返します。

4. CLI は実行の結果を報告しません。 確認するには、 DATABRICKS_HOST をワークスペースのホスト名に設定し、OAuthトークンを生成し、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

CLI は /api/2.0/agents/memory-pipelines で REST API を呼び出します。 どの言語からでも直接呼んでください。

1. Databricks CLIを使ってOAuthトークンを生成します:

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

2. support-agent-sessionsから記憶を抽出し、それをsupport-agent-memoryに保存するパイプラインを作りましょう。 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"}'
   

   レスポンスの name フィールドには、パイプラインのリソース名が memory-pipelines/<pipeline-id>という形で入っています。

3. パイプラインの実行を開始:

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

   レスポンスの name フィールドには、 memory-pipelines/<pipeline-id>/runs/<run-id>という形でランのリソース名が保持されています。

4. 実行結果を確認してください:

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

実行が成功すると、エージェントはメモリストアのリストや検索操作で新しいメモリを呼び戻すことができます。 「エージェント メモリ」を参照してください。

パイプライン実行をモニタリングする

ランの state は以下のいずれかです:

指示に従って抽出を制御する

指示は、何を残し、どのように整理するかをパイプラインに伝えます。 パイプラインはあらゆる段階でそれらを適用します。 以下のような高レベルのガイダンスで活用してください:

• 顧客の好みやアカウントの詳細など、注目すべきトピック。
• 一度きりのトラブルシューティングステップなど、スキップできる内容。
• エントリを整理するためのパス。

レイアウト指示がない場合、パイプラインはストアの既存エントリのレイアウトに従うか、ストアが空のときにトピックベースのレイアウトを導出します。 既存のエントリを新しいレイアウトに合わせて名前変更することはありません。 特定のエントリを変更するには、命令ではなくメモリストアの更新操作で編集してください。

パイプラインを作成するときに指示を設定し、後で変更することができます。 次の実行では新しい指示を使用します。

エージェント・ブリックス

パイプラインを作成する際に指示を設定するには、--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/."

UTF-8ファイルから長い命令を読み込むには、代わりに --instructions @<path-to-file> を渡してください。

既存のパイプラインの手順を変更するには、以下を実行します。

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

REST API

REST API では、パイプラインの dreamer_policy オブジェクトが抽出用の設定として instructions と trigger を保持します。 パイプラインを作成する際に指示を設定するには、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/."}}'

既存のパイプラインの指示を変更するには、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."}}'

スケジュールに従ってパイプラインを実行する

スケジュールされたパイプラインは約24時間ごとに自動的に稼働します。 パイプラインを作成するときにトリガーを設定することも、後で変更することもできます。 スケジューラーは、最初のスケジュールされた実行の開始時を選びます。 既存の会話をすぐに抽出するには、手動で実行を開始します。

エージェント・ブリックス

スケジュールされたパイプラインを作成するには、--trigger scheduled を渡します。 既定値は --trigger manual です。

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

既存のパイプラインのトリガーを変更するには、以下を実行します。

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

REST API

スケジュールパイプラインを作成するには、 dreamer_policy.trigger を 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"}}'

既存のパイプラインのトリガーを変更するには、 dreamer_policy.trigger を SCHEDULED または 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"}}'

また、他の実行が進行中でなければ、スケジュールされたパイプラインの実行を手動で開始することも可能です。

パイプラインの管理

以下のコマンドは、パイプラインのリスト、検査、更新、削除を行います。 各CLIコマンドはパイプラインIDまたはその完全な memory-pipelines/<pipeline-id> リソース名のいずれかを受け付けます。

表示名を設定しなければ、パイプラインはセッションストア名とメモリストア名に基づいて表示名を割り当てます。 パイプラインを既定のモデルに戻すには、そのモデルを --model "" のような空の値に更新します。 モデルと指示の変更は次の実行に適用されます。

パイプラインを削除すると関連ジョブが削除され、今後の実行も停止します。 すでに進行中のランは完了まで続きます。 パイプラインがすでに書き込んだメモリエントリはメモリストアに残ります。

Limitations

• セッションストアとメモリストアはパイプラインを作成するときに固定されます。 どちらか一方を変更するには、パイプラインを削除して新しいパイプラインを作成します。
• パイプラインが既に同じセッション ストアとメモリ ストアを接続している場合、別のパイプラインを作成しようとすると、ALREADY_EXISTS で失敗します。
• パイプラインでは同時に進行中の実行は1つだけです。 別のランが進行中のときにランを開始すると、ABORTED エラーが出て失敗します。
• スケジュールされたパイプラインは約24時間ごとに稼働しています。 異なる間隔やクロンのスケジュールを設定することはできません。
• パイプラインはセッションストアにセッションを持つアクターのみを処理します。 セッションを持たないアクターのメモリエントリは変更されません。
• セッションストアにUUIDアクターIDを持つアクターが約1,200個以上いる場合、実行は失敗します。 シリアライズされたアクターIDの上限は48 KiBなので、短いIDほど、より多くのアクターを含められます。

その他のリソース

• エージェント メモリ
• エージェント セッション
• エージェントメモリとセッション
• エージェントAPI参照