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

備考

ベータ版

この機能はベータ版です。

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

会話中にメモリを保存するエージェントは、一度に1つの会話について、その会話のコンテキストのみを使ってメモリを書き込みます。多くの会話を経るうちに、メモリには重複した項目や矛盾する項目が蓄積され、エージェントが保存しようと思わなかった事実はメモリに記録されません。メモリパイプラインは、後からアクターのすべての会話と既存のメモリを確認できる状態で動作するため、エージェントが見落とした情報を取り込み、メモリを統合できます。

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

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

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

注記

メモリパイプラインのランでは、ジョブにServerless汎用コンピュートの価格が適用され、モデル呼び出しには基盤モデルサービングの価格が適用されます。

要件

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

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

メモリパイプラインは、1つのセッションストアと1つのメモリストアを接続するエージェントメモリリソースです。LakeFlow Pipelinesとは関連していません。1つのセッションストアから複数のメモリストアにデータを供給できますが、セッションストアとメモリストアの各ペアに設定できるパイプラインは1つだけです。

• Unity Gateway経由のモデル呼び出し： パイプラインは、Unity Gatewayを介して蒸留モデルを呼び出します。
• モデルの選択： パイプラインの作成時にモデルを選択し、後で変更できます。モデルを選択しない場合、パイプラインはdefault モデルを使用します。
• Serverlessコンピュートでの手動またはスケジュール実行： ランを自分で起動するか、約24時間ごとに実行されるようにパイプラインを構成します。どちらの場合も、各ランはパイプラインを作成したIDを使用して、Serverlessコンピュート上でジョブとして実行されます。

パイプラインの各ランでは、セッションストア内のすべてのアクターが個別に処理されます：

1. 蒸留： パイプラインはアクターの新しい会話を読み取り、LLMを使用してそこから長期的に保持すべき事実を抽出します。
2. Merge: パイプラインは、新しいファクトの重複を排除し、それらをメモリストア内の同じ actor_id の下にあるトピックベースのメモリエントリにマージします。
3. 維持： パイプラインは、アクターのメモリを再編成します。重複するエントリーを統合し、大きくなりすぎたエントリーを分割します。

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

actor_idが2つのストアをLink。パイプラインは、セッションストアにセッションがあるアクターのみを処理し、各アクターのメモリを、そのアクターのセッションが使用するものと同じactor_idの配下に書き込みます。defaultでは、パイプラインはランを起動したときにのみ実行されます。また、スケジュールに従って実行することもできます。

注記

アクターのすべてのメモリエントリにメンテナンスが適用されます。これには、エージェントが直接保存したエントリも含まれます。パイプラインは、処理するactor_id内の任意のエントリを書き換えたり、Mergeしたり、分割または削除したりできます。

パイプラインに保持されるもの

パイプラインは、会話から個々の永続的な事実を抽出します。次の内容を保持するように設計されています：

• 会話の内容を超えて推測することなく、ユーザーが明示的に述べた事実や表明した好み。
• 売上高の数値にどのソースを使用するかなどの修正や明示的なフィードバック。
• 信頼できるソース、メトリクスの定義、用語、チームの慣習などの持続的なドメイン知識に加え、ワークフローや進行中のプロジェクト

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

• テーブルの現在の行数など、時間の経過とともに変化するライブソースの値。代わりに、その値の検索方法またはコンピュート方法を記録します。
• テーブルのスキーマなど、エージェントがツールを使って参照できる詳細情報。
• パスワード、APIキー、トークンなどのシークレット。

起動する

これらの例では、サポートエージェントのセッションストアとメモリストア用のメモリパイプラインを作成し、ランを起動して、その結果を確認します。各ストアは、ID（ストアの作成時に選択した識別子）、またはmemory-stores/support-agent-memoryのようなリソース名で識別します。プロジェクトに適したクライアントを選択します。

Agent Bricks CLI

The Agent Bricks CLIは、保存済みのDatabricks CLIプロファイルを使用して認証します。

1. Agent Bricks CLIをインストールするか、0.5.0以降にアップグレードします：

   Bash

   pip install --upgrade databricks-agentbricks

2. support-agent-sessionsセッションストアの内容を抽出してsupport-agent-memoryメモリストアに格納するパイプラインを作成します。--modelはオプションです。省略した場合、パイプラインではdefaultモデルが使用されます：

   Bash

   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. パイプラインのランを起動する：

   Bash

   agentbricks memory pipeline run <pipeline-id>

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

4. CLIはランの結果を報告しません。確認するには、DATABRICKS_HOSTにワークスペースのホスト名を設定し、OAuthトークンを生成して、REST APIを呼び出します：

   Bash

   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トークンを生成します：

   Bash

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

2. support-agent-sessionsからメモリを抽出し、support-agent-memoryに保存するパイプラインを作成します。modelはオプションです。省略すると、パイプラインはdefaultモデルを使用します：

   Bash

   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. パイプラインのランを起動する：

   Bash

   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. ランの結果を確認します：

   Bash

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

ランが成功すると、エージェントはメモリストアのリストおよび検索操作を使用して新しいメモリを呼び出すことができます。エージェントメモリを参照してください。

パイプラインのランを監視する

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

状態	意味
RUNNING	ランは保留中または進行中です。
SUCCEEDED	ランでは、すべてのアクターが正常に処理されました。
FAILED	1つ以上のアクターでエラーが発生し、ランが終了しました。errorフィールドは失敗の内容を説明します。
CANCELLED	ランは完了する前にキャンセルされました。

状態	意味
RUNNING	ランは保留中または進行中です。
SUCCEEDED	ランでは、すべてのアクターが正常に処理されました。
FAILED	1つ以上のアクターでエラーが発生し、ランが終了しました。errorフィールドは失敗の内容を説明します。
CANCELLED	ランは完了する前にキャンセルされました。

指示を使って蒸留を制御する

手順では、パイプラインに保持する内容とその整理方法を指定します。パイプラインは、各ステージでそれらを適用します。次のようなハイレベルのガイダンスに使用します：

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

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

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

Agent Bricks CLI

パイプラインの作成時に指示を設定するには、--instructionsを渡します：

Bash

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>を渡します。

既存のパイプラインの指示を変更するには、次を実行します：

Bash

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

REST API

REST APIでは、パイプラインのdreamer_policyオブジェクトに蒸留設定が格納されています：instructionsとtrigger。パイプラインの作成時に指示を設定するには、dreamer_policy.instructionsに含めます：

Bash

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を更新します：

Bash

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時間ごとに自動的に実行されます。パイプラインの作成時にトリガーを設定することも、後から変更することもできます。スケジューラは、最初のスケジュールされたランの起動時刻を選択します。既存の会話をすぐに蒸留するには、ランを手動で起動します。

Agent Bricks CLI

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

Bash

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

既存のパイプラインのTriggerを変更するには、次のコマンドをランします：

Bash

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

REST API

スケジュールされたパイプラインを作成するには、dreamer_policy.triggerをSCHEDULEDに設定します：

Bash

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

既存のパイプラインのTriggerを変更するには、dreamer_policy.triggerをSCHEDULEDまたはMANUAL_ONLYに更新します：

Bash

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>リソース名のいずれかを指定できます。

タスク	Agent Bricks CLI	REST API
パイプラインの一覧表示	agentbricks memory pipeline list	GET /api/2.0/agents/memory-pipelines
パイプラインを取得する	agentbricks memory pipeline get <pipeline-id>	GET /api/2.0/agents/memory-pipelines/<pipeline-id>
パイプラインを更新する	agentbricks memory pipeline update <pipeline-id> --display-name、--model、--trigger、または --instructions	PATCH /api/2.0/agents/memory-pipelines/<pipeline-id> update_maskの値としてdisplay_name、model、dreamer_policy.trigger、または dreamer_policy.instructions
パイプラインを削除	agentbricks memory pipeline delete <pipeline-id>	DELETE /api/2.0/agents/memory-pipelines/<pipeline-id>

タスク	Agent Bricks CLI	REST API
パイプラインの一覧表示	agentbricks memory pipeline list	GET /api/2.0/agents/memory-pipelines
パイプラインを取得する	agentbricks memory pipeline get <pipeline-id>	GET /api/2.0/agents/memory-pipelines/<pipeline-id>
パイプラインを更新する	agentbricks memory pipeline update <pipeline-id> --display-name、--model、--trigger、または --instructions	PATCH /api/2.0/agents/memory-pipelines/<pipeline-id> update_maskの値としてdisplay_name、model、dreamer_policy.trigger、または dreamer_policy.instructions
パイプラインを削除	agentbricks memory pipeline delete <pipeline-id>	DELETE /api/2.0/agents/memory-pipelines/<pipeline-id>

表示名を設定しない場合、パイプラインにはセッションストア名とメモリストア名に基づく名前が付けられます。パイプラインをdefaultモデルに戻すには、モデルを--model ""などの空の値に更新します。モデルと指示の変更は、次回のランに適用されます。

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

制限事項

• セッションストアとメモリストアは、パイプラインの作成時に固定されます。どちらかを変更するには、パイプラインを削除して新しく作成します。
• 同じセッションストアとメモリストアにすでに接続されているパイプラインがある場合、別のパイプラインを作成するとALREADY_EXISTSで失敗します
• パイプラインでは、一度に1つのランのみ進行中になります。別のランが進行中にランを開始すると、ABORTEDエラーが発生します。
• スケジュールされたパイプラインは、約24時間ごとに実行されます。別の間隔やcronスケジュールは設定できません。
• パイプラインは、セッションストアにセッションがあるアクターのみを処理します。セッションのないアクターのメモリエントリーは変更されません。
• セッションストア内のアクター数が約1,200を超え、アクターIDがUUIDの場合、ランは失敗します。シリアル化されたアクターIDの上限は48 KiBであるため、IDが短いほど、より多くのアクターを登録できます。

その他のリソース

• エージェントメモリ
• エージェントセッション
• エージェントのメモリとセッション
• エージェントAPIリファレンス