【第2回】Amazon Bedrock AgentCore 入門 — ツール定義とアクショングループを使ったエージェントの拡張

前回は Amazon Bedrock AgentCore の概要と、Runtime を使ってエージェントをデプロイするところまで紹介しました（第1回はこちら）。今回は AgentCore の目玉機能のひとつである Memory を掘り下げていきます。

正直、Memory の存在を知ったとき「Bedrock Agents の会話履歴管理とどう違うの？」ってなりました。調べてみると、単なる履歴保持じゃなくてかなりちゃんとした仕組みになっていたので、まとめておきます。

この記事でわかること

AgentCore Memory の短期メモリ・長期メモリの違い
長期メモリの3つのストラテジー（Semantic、Summary、UserPreference）の使い分け
Pythonでのメモリ作成・会話記録・記憶取得の実装方法
メモリを使ったエージェント実装のパターン
使う際の注意点とハマりポイント

この記事でわかること

AgentCore Memory とは
長期メモリの3つのストラテジー
実装：Pythonでの使い方
取得した記憶をエージェントのプロンプトに組み込む
ハマりポイントと注意点
まとめ

AgentCore Memory とは

AgentCore Memory は、AIエージェントに「記憶」を持たせるためのマネージドサービスです。短期的なワーキングメモリ（セッション内の会話コンテキストを即座に捉える）と長期的なインテリジェントメモリ（セッションをまたいで持続的なインサイトや好みを保存する）の両方を提供します。

AgentCore Memory はワンオフの会話を、ユーザーと AI エージェントの継続的で進化する関係性に変えます。「口座番号は何ですか？」と何度も聞いたり、「貝類アレルギーがあります」という情報を忘れたりすることなく、エージェントはコンテキストを保持し、インタラクションから学習できます。

つまり、こういうことです：

短期メモリ：セッション内の会話の流れを保持する。単純なコンテキスト管理
長期メモリ：会話から重要な情報を自動抽出して、セッションをまたいで永続化する

長期メモリのキモは「自動抽出」の部分で、会話ログをそのまま保存するのではなく、LLM が情報を整理・圧縮してから保存してくれます。これのおかげでコンテキストウィンドウを圧迫しないらしいです。

長期メモリの3つのストラテジー

長期メモリを使うには、メモリ作成時に「どういう種類の情報を抽出するか」をストラテジーとして指定します。抽出プロセスは3種類のビルトインメモリストラテジーをサポートしています：Semantic（ファクトや知識を抽出）、Summary（会話のサマリーを生成）、UserPreference（ユーザーの好みを抽出）です。

それぞれの使い分けイメージ：

Semantic：「ユーザーはAWS認定試験を受ける予定」「注文番号は12345」みたいな事実ベースの情報
Summary：「このセッションでは在庫問題について話した」みたいな会話全体のサマリー
UserPreference：「辛いものが好き」「ダークモードを使っている」みたいなユーザーの好み

複数のストラテジーを同時に指定することも可能で、それぞれ独立して並列処理されます。並列処理アーキテクチャにより、複数のメモリストラテジーが独立して処理されるため、異なるメモリタイプを互いにブロックすることなく同時に処理できます。

余談ですが、最初に「ストラテジー」という言葉を見たとき、なんか将棋の戦型みたいでかっこいいと思った。それだけです。

実装：Pythonでの使い方

まずは SDK のインストールから。

pip install bedrock-agentcore

ステップ1：メモリリソースを作成する

短期メモリだけ使う場合（ストラテジーなし）はシンプルです。

from bedrock_agentcore.memory import MemoryClient

client = MemoryClient(region_name="us-west-2")

memory = client.create_memory(
    name="MyAgentMemory",
    description="Memory for customer support conversations",
)

memory_id = memory.get("id")
print(f"Memory ID: {memory_id}")

長期メモリも使いたい場合はストラテジーを指定します。ストラテジーの有効化には少し時間がかかるため、create_memory_and_wait を使うと ACTIVE になるまで待ってくれます。

from bedrock_agentcore.memory import MemoryClient

client = MemoryClient(region_name="us-west-2")

memory = client.create_memory_and_wait(
    name="MyAgentMemoryWithStrategy",
    strategies=[
        {
            "semanticMemoryStrategy": {
                "name": "customer-facts",
                "namespaceTemplates": ["/users/{actorId}/facts/"],
            }
        },
        {
            "summaryMemoryStrategy": {
                "name": "session-summary",
                "namespaceTemplates": ["/users/{actorId}/{sessionId}/summary/"],
            }
        },
        {
            "userPreferenceMemoryStrategy": {
                "name": "user-preferences",
                "namespaceTemplates": ["/users/{actorId}/preferences/"],
            }
        },
    ]
)

memory_id = memory.get("id")

namespaceTemplates の {actorId}、{sessionId} はそれぞれ実行時に動的に置換されます。ここのパス設計は取得時の検索精度にも影響するので、ある程度考えておいたほうがよさそうです。

ステップ2：会話を記録する

MemorySessionManager を使ってセッションを作り、会話ターンを追加します。

from bedrock_agentcore.memory import MemorySessionManager
from bedrock_agentcore.memory.constants import ConversationalMessage, MessageRole

session_manager = MemorySessionManager(
    memory_id=memory_id,
    region_name="us-west-2"
)

session = session_manager.create_memory_session(
    actor_id="user-001",     # ユーザーや Agent の識別子
    session_id="session-abc" # 会話セッションの識別子
)

# 会話を追加
session.add_turns(
    messages=[
        ConversationalMessage("こんにちは、注文番号12345の件で困っています。", MessageRole.USER),
        ConversationalMessage("承知しました。注文を確認しますね。", MessageRole.ASSISTANT),
        ConversationalMessage("3日前に発送済みとなっています。何かお困りの点はありますか？", MessageRole.ASSISTANT),
        ConversationalMessage("まだ届いていないんです。配送先はアメリカです。", MessageRole.USER),
    ]
)

actor_id と session_id の使い分けは、actor_id がユーザー固有の識別子（複数セッションをまたいで共通）、session_id が1回の会話単位というイメージです。長期メモリの namespace パスでこの2つが使えるので、「このユーザーの全セッションから検索」とか「このセッションだけのサマリーを取得」みたいな粒度の制御ができます。

ステップ3：長期メモリから記憶を取得する

ストラテジーによる抽出は非同期で行われます。抽出と統合の処理は環境や会話量にもよりますが、数十秒から1分以上かかることがあるため、記録直後に取得しようとしても空で返ってくることがあります。テストするときは少し待つ必要があります。

import time

# 抽出完了を待つ（本番では非同期処理で対応するのが現実的）
time.sleep(60)

# セマンティック検索で関連する記憶を取得
memory_records = session.search_long_term_memories(
    query="配送に関する問題",
    namespace_path="/",   # ルートから全体を検索
    top_k=3
)

for record in memory_records:
    print(record)

セマンティック検索による取得は状況によってはかなり速く返ってくるようですが、レイテンシは環境依存です。書き込みより読み込みはかなり速いです。

特定の namespace を絞り込んで取得したい場合は namespace_path を具体的に指定します。

# 特定ユーザーの記憶だけ取得
memory_records = session.search_long_term_memories(
    query="ユーザーの好み",
    namespace_path="/users/user-001/",
    top_k=5
)

また、特定のレコードを直接取得したり、一覧を取得する方法もあります。

# namespace配下のレコード一覧を取得
records = session.list_long_term_memory_records(
    namespace_prefix="/users/user-001/",
    max_results=20
)

# IDを指定して特定のレコードを取得
record = session.get_memory_record("record-id-xxx")
print(record.content)

補足：低レベルAPIでの操作

Python SDK を使わず、boto3 の低レベル API で操作することもできます。

import boto3

control_client = boto3.client("bedrock-agentcore-control", region_name="us-west-2")

# メモリ一覧を取得
response = control_client.list_memories()
memories = response["memories"]
print(memories)

AWS SDK を使うと、AgentCore Python SDK と同じ結果を達成できますが、AgentCore Python SDK がサポートしていない操作も可能です。Python 以外の言語を使っている場合も AWS SDK を使う必要があります。チームで複数言語が混在している場合はこちらのほうが現実的かもしれません。

取得した記憶をエージェントのプロンプトに組み込む

メモリを取得しても、それをどうエージェントに渡すかは自前で実装する必要があります。基本的な流れはこうです：

ユーザーの入力を受け取る
過去の関連記憶を search_long_term_memories で取得
取得した記憶をプロンプトに埋め込んで LLM に渡す
LLM の回答と今回の会話を add_turns で記録する

from bedrock_agentcore.memory import MemorySessionManager
from bedrock_agentcore.memory.constants import ConversationalMessage, MessageRole
import boto3
import json

def chat_with_memory(user_input: str, actor_id: str, session_id: str, memory_id: str):
    session_manager = MemorySessionManager(memory_id=memory_id, region_name="us-west-2")
    session = session_manager.create_memory_session(actor_id=actor_id, session_id=session_id)

    # 関連する長期記憶を取得
    past_memories = session.search_long_term_memories(
        query=user_input,
        namespace_path=f"/users/{actor_id}/",
        top_k=3
    )

    # プロンプトに記憶を埋め込む
    memory_context = "\n".join([str(m) for m in past_memories]) if past_memories else "なし"
    system_prompt = f"""あなたは親切なサポートエージェントです。
以下はこのユーザーに関する過去の記憶です：
{memory_context}
---
この情報を参考にして応答してください。"""

    bedrock = boto3.client("bedrock-runtime", region_name="us-west-2")
    response = bedrock.converse(
        modelId="anthropic.claude-3-5-sonnet-20241022-v2:0",
        system=[{"text": system_prompt}],
        messages=[{"role": "user", "content": [{"text": user_input}]}]
    )

    assistant_reply = response["output"]["message"]["content"][0]["text"]

    # 今回の会話を記録
    session.add_turns(messages=[
        ConversationalMessage(user_input, MessageRole.USER),
        ConversationalMessage(assistant_reply, MessageRole.ASSISTANT),
    ])

    return assistant_reply

もっといい方法がある気がしますが、とりあえずこれで動きます。Strands Framework を使うと memory フックをより宣言的に書けるようなので、そちらも試してみたいと思っています。

ハマりポイントと注意点

使ってみて気になったことをいくつか。

長期メモリの抽出は即時ではない：add_turns 直後に search_long_term_memories を呼んでも何も返ってこないことがあります。テスト時は time.sleep(60) などで待つ必要があります
ストラテジーを後から追加しても過去分は対象外：新たに追加されたストラテジーがアクティブになった後に保存されたイベントからのみ抽出されます。ストラテジーを追加する前に保存された会話は長期メモリに現れません。これは知らないとハマります
リージョンはまだ限定的：AgentCore の対応状況はリージョンによって異なるため、使用前に対応リージョンを確認しておくのが安全です
Memory ID の管理：作成したメモリリソースのIDは環境変数などで管理するのが定石です。コード内にハードコードしてしまいがちなので注意

※この記事にはプロモーションが含まれます

ちなみに、お名前.com レンタルサーバー（WordPressに特化した高速レンタルサーバー。月額990円〜、独自ドメイン実質0円）も気になっています。お名前.com レンタルサーバー