メインコンテンツまでスキップ
に最終更新

チュートリアル: 既存のエージェントにセッションとメモリを追加する

備考

ベータ版

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

Python で既にエージェントを構築している場合は、書き換えることなく、Databricks 上で永続的な会話履歴と長期メモリを提供できます。このチュートリアルでは、各会話を保持するセッションストア、会話全体で永続的な事実を保持するメモリストア、格納された会話をメモリに抽出するメモリパイプラインにエージェントを接続します。ステップは、エージェントがどこで実行されるかに関係なく、OpenAI Agents SDK、LangGraph、またはその他の Python フレームワークで構築されたエージェントを対象としています。

ヒント

エージェントがまだない場合は、代わりにAgent Bricksクイックスタートを起動してください。Agent Bricks CLI は、セッションとメモリが既に設定された新しいエージェントをスキャフォールドします。

この例では、サポートエージェントは顧客の会話を中断したところから再開し、後で別の会話を行うときに顧客の設定を記憶する必要があります。

セッションとメモリの関連性については、エージェントメモリとセッションを参照してください。

要件​

  • Python 3.10 以上で記述された既存のエージェント。フレームワーク固有のステップでは、OpenAI Agents SDKとLangGraphについて説明しています。その他のフレームワークのステップでは、AgentKit SDKを直接呼び出します。
  • Databricks認証。AgentKit SDKは、Databricks SDK for Pythonのdefault認証を使用します。ローカル開発の場合は、 databricks auth login --host <workspace-url> をランするか、 DATABRICKS_HOST および DATABRICKS_TOKEN 環境変数を設定します。
  • メモリパイプラインのランのステータスを確認するためのDatabricks CLI。
  • メモリパイプラインを作成するには、パイプラインが使用するモデルに対するクエリーアクセス許可、およびメモリパイプラインの要件にリストされている他のアクセス許可が必要です。

ステップ1：AgentKit SDKとAgent Bricks CLIをインストールする​

databricks-agentbricks パッケージには、以下が含まれます。

  • AgentKit SDK (databricks_agentkit) は、セッションおよびメモリストア用の Python クライアントです。
  • Agent Bricks CLI (agentbricks)。
  • セッションストアとメモリストアを各フレームワークの既存のセッションおよびツールインターフェイスに接続する、OpenAI Agents SDKおよびLangGraph用のアダプター。アダプターは、エージェントにすでにインストールされているフレームワークパッケージを使用します。

エージェントの環境にバージョン0.5.0以上をインストールします。

Bash
pip install --upgrade "databricks-agentbricks>=0.5.0"

ステップ2: セッションストアとメモリストアを作成する​

エージェントは、Databricksが管理する次の2つのストアを使用します：

  • セッションストア には会話の履歴が保持されます。その中の各セッションは1つの会話の記録であり、エージェントはそれを再ロードしてその会話が中断したところから再開します。
  • メモリストア は、設定や過去の決定など、ユーザーに関する永続的な事実を保持します。エージェントは、以前の別の会話で学習した内容を思い出すために検索します。

ストアはセキュリティ境界であり、ストアへのアクセス権を持つプリンシパルは、そのすべてのエントリを読み書きできます。テナントを互いに分離するには、テナントごとに別々のストアのペアを作成します。メモリのパーティション分割と保護を参照してください。

各ストアを一度作成し、すべての会話で再利用します。エージェントは両方のストアを名前で参照します。

Python
from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient

client = AgentKitClient(WorkspaceClient())
client.session_stores.create("support-agent-sessions", description="Support agent conversations")
client.memory_stores.create("support-agent-memory", description="Support agent long-term memory")

ストア名はワークスペース内で一意である必要があります。メモリストア名は3～56文字で、小文字で始まり、文字または数字で終わり、小文字、数字、およびハイフンのみを含める必要があります。セッションストア名は3～63文字である必要があります。

ステップ 3: アクター ID とセッション ID を選択する​

エージェントがストアに対して行うすべての呼び出しには、次の 2 つの識別子が含まれます。

  • actor_id は、誰の会話とメモリかを識別します。両方のストアがこれを使用しており、メモリパイプラインは各アクターのメモリをそのアクターのセッションと同じ actor_id の下に書き込むため、エージェントは両者が一致する場合にのみ抽出されたメモリを呼び出します。
  • session_id は1つの会話を識別するため、フォローアップリクエストによって会話が再開されます。

このチュートリアルでは、test-userをactor_idとして、conversation-1をsession_idとして使用します。以下のステップでは、handle_message関数にそれらをuser_idおよびconversation_idとして渡します。

本番運用では、サインインユーザーの検証済みIDなど、信頼できるリクエストコンテキストからアプリケーションコード内でactor_idを設定し、モデルやユーザーのメッセージにその値を選択させないでください。チーム間でメモリを共有したり、テナントを分離したりするには、メモリを分割して保護するをご覧ください。session_idには、アプリの既存の会話IDまたはスレッドIDを使用するか、会話の起動時にUUIDを生成します。

ステップ4：セッションストアに会話を保存する​

会話履歴とは、ユーザーのメッセージ、エージェントの返信、およびツール呼び出しと結果など、1つの会話の順序付けられた記録です。セッションストアは各会話の履歴を保存するため、エージェントは再起動後でも、次のリクエストで履歴を再ロードできます。エージェントのインプロセス会話履歴を、ステップ 2 で作成したセッションストアに置き換えます。

フレームワークアダプターは各フレームワークがセッション状態に既に使っているインターフェースにプラグインされるため、エージェントのモデル、ツール、プロンプトは変更されません。

OpenAIエージェントSDKは、Sessionオブジェクトを通じて会話履歴の読み取りと書き込みを行います。session_store()は、管理されたセッションストアに支えられたSessionを返します。それをRunner.runに渡します：

Python
from agents import Agent, Runner
from databricks_agentkit.openai import session_store

SESSION_STORE = "support-agent-sessions"

# Your existing agent, unchanged.
agent = Agent(name="Support agent", instructions="You help customers with their Databricks workspaces.")

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    session = session_store(conversation_id, actor=user_id, store=SESSION_STORE)
    result = await Runner.run(agent, message, session=session)
    return result.final_output

会話の最初のメッセージで、アダプターはセッションを作成します。すべてのランで、SDKはセッションから会話の履歴をロードし、新しいユーザーメッセージ、モデル出力、およびツールアイテムを追加します。

セッションストアに切り替えても、エージェントが他の場所に保存した会話は移動しません。既存の会話を引き継ぐには、その項目を新しいセッションに追加します。

ステップ 5: エージェントにメモリを与える​

セッション履歴は、1回の会話を対象としています。エージェントが過去の会話で学習した内容を使用するには、エージェントをメモリストアに接続します。2つのパターンがあり、それらを組み合わせることができます。

  • メモリツール: モデルは、いつメモリを呼び出し、いつ新しい事実を保存するかを決定します。
  • 各ランの前に呼び出す： アプリケーションは、メモリ内で受信メッセージを検索し、その結果をエージェントのコンテキストに追加します。そのため、エージェントはツールを呼び出すことなく関連するメモリを確認できます。

メモリパイプラインがメモリを書き込む場合（ステップ6を参照）、エージェントはメモリを呼び出すだけで済みます。メモリを保存するツールはスキップできます。

次の例では、エージェントの要求ごとのコピーで両方のパターンを使用しています。

  • memory_tools() ユーザーのメモリを検索する recall ツールと、事実を保存する remember ツールを返します。ツールは1つのアクターにバインドされているため、モデルは他のユーザーのメモリを読み書きできません。
  • recalled_facts() ランの前に着信メッセージのメモリを検索し、その結果をエージェントの指示に追加します。
Python
from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient
from databricks_agentkit.openai import memory_tools, session_store

MEMORY_STORE = "support-agent-memory"
memory_store = AgentKitClient(WorkspaceClient()).memory_stores.get(MEMORY_STORE)

def recalled_facts(user_id: str, message: str) -> str:
    results = memory_store.search(actor_id=user_id, query=message, limit=5)
    return "\n".join(f"- {r.memory.content}" for r in results)

async def handle_message(user_id: str, conversation_id: str, message: str) -> str:
    instructions = agent.instructions  # Your agent's existing instructions, as a string.
    if facts := recalled_facts(user_id, message):  # Recall before the run.
        instructions += f"\n\nWhat you know about this user:\n{facts}"
    user_agent = agent.clone(
        instructions=instructions,
        tools=[*agent.tools, *memory_tools(user_id, store=MEMORY_STORE)],  # Memory tools.
    )
    session = session_store(conversation_id, actor=user_id, store=SESSION_STORE)
    result = await Runner.run(user_agent, message, session=session)
    return result.final_output

呼び出された事実は入力ではなく指示に入り、セッションには保存されません。いずれか1つのパターンのみを使用するには、もう一方を削除します。memory_tools()行を削除するか、recalled_facts()呼び出しを削除します。ツールを維持する場合は、エージェントの指示を更新して使用するタイミングを伝えます。例えば、「ユーザーが回答する前に、メモリを使用してユーザーの好みを検索する」などです。"ユーザーが共有する永続的な事実を保存するには、記憶を使用します。"

ステップ 6: メモリパイプラインを作成する​

メモリパイプラインは、セッションストア内の会話を読み取り、各アクターの永続的な事実、好み、決定をメモリストアに抽出します。また、時間が経過しても重複するエントリをMergeするため、エージェントのメモリは追加のエージェントコードなしで統合されたままになります。

  1. 2つのストアを接続するパイプラインを作成します：

    Bash
    agentbricks memory pipeline create \
      --session-store support-agent-sessions \
      --memory-store support-agent-memory \
      --trigger scheduled \
      --model system.ai.claude-sonnet-5 \
      --instructions "Focus on durable customer preferences and account details. Ignore one-off troubleshooting steps."
    • --trigger scheduled 約24時間ごとに自動的にパイプラインを実行する。これがない場合、パイプラインはdefault Triggerである manual を使用し、ランを起動したときにのみランします。スケジュールに基づいてパイプラインをランするを参照してください。
    • --model 記憶を抽出するモデルを設定します。クエリーの実行権限がある任意のモデルを使用できます。--modelを省略すると、パイプラインはDatabricksが選択したdefaultのモデルを使用します。
    • --instructions はオプションであり、パイプラインが保持するものを制御します。指示による蒸留の制御を参照してください。

    コマンドは、パイプラインのリソース名を memory-pipelines/<pipeline-id> の形式で返します。オプションごとに、 agentbricks memory pipeline create --help を実行するか、 Agent Bricks CLI コマンド リファレンスを参照してください。

  2. スケジューラは、最初のスケジュールされたランを起動するタイミングを選択します。セッションストアに既に存在する会話を抽出するには、今すぐ実行を起動してください：

    Bash
    agentbricks memory pipeline run <pipeline-id>

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

  3. REST APIでランの結果を確認します。DATABRICKS_HOST をワークスペースのホスト名に設定してから、以下を実行します：

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

    ランのstateがSUCCEEDEDの場合、抽出されたメモリはメモリストアにあります。

パイプラインのランでは、エージェントのIDではなく、パイプラインを作成したIDが使用されます。パイプラインは、エージェントがメモリツールで保存したエントリを含め、処理する actor_id の下にあるエントリを書き換え、Merge、または削除できます。REST API、ランの状態、制限事項については、エージェントのメモリパイプラインを参照してください。

LangGraphエージェントのトランスクリプトを記録する​

メモリパイプラインは、ユーザーとアシスタントのメッセージのみを抽出します。LangGraph チェックポインタはシリアル化されたチェックポイントを格納するため、チェックポインタのセッションストアを読み取るパイプラインは抽出するメッセージを見つけられません。LangGraph エージェントでパイプラインを使用するには、各ターンのメッセージを別のセッションストアに追加し、そのストアにパイプラインを作成します。

Python
transcripts = client.session_stores.create("support-agent-transcripts")  # Create once.

def start_conversation(user_id: str, conversation_id: str) -> None:
    transcripts.add(actor_id=user_id, session_id=conversation_id)

def record_turn(conversation_id: str, message: str, reply: str) -> None:
    transcripts.get(conversation_id).append_items(
        [
            {"type": "message", "role": "user", "content": message},
            {"type": "message", "role": "assistant", "content": reply},
        ]
    )

会話が開始された時、最初の ainvoke の前に start_conversation を呼び出し、各 ainvoke の後に record_turn を呼び出します。パイプラインを作成するときに --session-store support-agent-transcripts を渡します。あるいは、パイプラインをスキップして、エージェントがメモリツールでメモリを保存できるようにします。

ステップ 7: デプロイされたエージェントにアクセス権を付与する​

エージェントをローカルでランすると、独自の資格情報が使用されます。デプロイされたエージェントは通常、Service Principalとして実行されます。これには、読み取りまたは書き込みを行うすべてのストアへのアクセスが必要です。AgentKit SDKでアクセス権を付与し、Service Principalのアプリケーション（クライアント）IDを渡します：

Python
from databricks.sdk import WorkspaceClient
from databricks_agentkit import AgentKitClient

w = WorkspaceClient()
client = AgentKitClient(w)

# For an agent deployed as a Databricks app, look up the app's service principal.
sp_client_id = w.apps.get("support-agent").service_principal_client_id

client.session_stores.get("support-agent-sessions").grant_permission(sp_client_id)
client.memory_stores.get("support-agent-memory").grant_permission(sp_client_id)

grant_permission defaultで書き込みアクセス権を付与します。読み取り専用アクセス権を付与するには、permission="READ" を渡します。LangGraphエージェントのトランスクリプトストアを作成した場合は、それへのアクセス権も付与します。メモリパイプラインはエージェントのService Principalに対する権限を必要としません。これは、パイプラインの実行がパイプラインを作成したIDを使用するためです。

ステップ8: エージェントをエンドツーエンドでテストする​

  1. user_id="test-user"とconversation_id="conversation-1"でhandle_messageを呼び出し、「電話よりEメールがよく、太平洋タイムゾーンにいます」といった永続的な好みを共有します。他のフレームワークのステップに従った場合は、最初にstart_conversation("test-user", "conversation-1")を呼び出してください。同じ会話でフォローアップメッセージを送信し、エージェントが以前のコンテキストを保持していることを確認します。

  2. セッションストアに会話が保持されていることを確認します：

    Bash
    agentbricks sessions items list --store support-agent-sessions --session-id conversation-1

  3. メモリパイプラインの実行を起動し、ステップ6と同様に、stateがSUCCEEDEDになるまで待ちます。

  4. メモリストアにテストユーザーの設定が保持されていることを確認します。

    Bash
    agentbricks memory entries search --store support-agent-memory --actor-id test-user --query "contact preferences"

  5. test-userについてconversation_id="conversation-2"との新しい会話を起動し、「どのように連絡すればよいですか？」と質問します。エージェントはメモリから回答します。

トラブルシューティング​

  • PERMISSION_DENIED (HTTP 403)をデプロイ済みエージェントから取得しました。 エージェントのService Principalがストアにアクセスできません。ステップ 7 のように、アクセス権を付与してください。
  • パイプラインのランは成功しますが、エントリは書き込まれません。 セッションストアに、パイプラインが読み取る形式でユーザーメッセージとアシスタントメッセージが保持されていることを確認します。パイプラインは、ツール呼び出し、ツールの結果、および LangGraph チェックポイントをスキップします。ランは、そのアクターの最後の成功したラン以降のアクターのセッションのみを処理します。
  • エージェントは蒸留されたメモリを想起しません。 セッションで使用されているものと同じactor_idを使って、エージェントがメモリを検索することを確認します。
  • ランの開始がABORTEDで失敗しました。 パイプラインの別のランが進行中です。終了するまで待ち、新しいランを起動してください。

クリーンアップ​

このチュートリアルで作成したリソースを削除するには、以下を実行してください。ストアを削除すると、その中のすべてが削除されます。

Bash
agentbricks memory pipeline delete <pipeline-id>
agentbricks sessions stores delete support-agent-sessions
agentbricks memory stores delete support-agent-memory

次のステップ​

このページの見出し