【AWS】Bedrock AgentCore Memoryを使って会話履歴が保存されるように実装してみた！

どうもこんにちは。2025年年末にこの記事を書いています...(笑)

今回は、Amazon Bedrock AgentCoreの機能の1つである、Memoryについて検証をしてみたので、紹介しようと思います。

検証には、以下の記事を参考にさせていただきました。

こちらも参考にしています。

Bedrock AgentCore Memoryってなに？

ほとんど以下のスライドに記載されていますが、Memoryには2種類あります。
簡単に言えば、短期記憶と長期記憶です。

Short-term Memory

Short-term Memoryの特徴は以下です。

• セッション中の会話履歴を保持できる
• 最大365日保存できる
• リアルタイムでアクセス可能
• 生データが直接保存される
• Actor IDとSession IDでデータを整理されている

Long-term Memory

Long-term Memoryの特徴は以下です。

• セッションを跨いで保存される
• Short-term Memoryから、重要な情報を自動的に抽出・統合する
• 抽出されたデータはベクトルデータとして保存される
• 取得するときは、セマンティック検索によって検索され、効率的に取得される

大きな特徴

エンジニアとして、Bedrock AgentCore Memoryの最もありがたいところは、DBに会話履歴を保存するテーブルを用意する必要がないというところです。そのため、テーブル設計やマイグレーションを行う必要がないのがいちばんの強みです。

また、4種類のAIエージェントが理解しやすい形式でデータが保存されるため、ニュアンスがブレることなく会話を記憶してくれます。

データの保存形式 = 戦略

• 要約
• ユーザプリファレンス
• セマンティック
• エピソード

ただし、デメリットもあり、画面へ表示するデータが存在する場合、aws-sdkで取得してレンダリングする必要があり、画面表示に時間がかかってしまうというのがデメリットかと思っています。（非同期で実装すればほぼ問題ないものだろうか？）

※ 上記スライドは、弊社システム開発部の勉強会で使用したスライドです。

Memoryの作成方法

AWSマネジメントコンソールからの作成方法を示します。

1. Bedrock AgentCoreのMemoryトップページへ遷移する

以下の画面へ遷移していればOKです。

2. 新規作成

「メモリを作成」をクリックします。すると、新規作成画面へ遷移するので、必要な情報を入力していきます。

2-1. メモリの詳細

メモリの詳細を入力していきます。

項目	説明
メモリ 名前	「どのエージェントに紐づけるのか」「用途」などを意識して命名するのが良い
短期メモリ (生のイベント) の有効期限	エージェントの短期記憶の期限です(デフォルトは90日)
追加設定/メモリの説明	「用途」などを記載しておくのが良い

2-2. 長期メモリ抽出戦略

長期メモリを設定する場合、以下を設定します。

2-2-1. 組み込み戦略

項目	説明
要約	会話のサマリーを生成して保存
セマンティックメモリ	事実や知識の抽出・保存
ユーザプリファレンス	ユーザーの好み・傾向を記録
エピソード	一連のインタラクションをエピソードとして記録・保存

各項目にチェックを入れると、以下の項目の入力欄が表示されます。

項目	説明
戦略 名前	戦略の名前をつけます（例: semantic_knowledge）
名前空間	記憶を保存する名前空間を指定します（例: /app/{actorId}/{sessionId}/summary） ※ エピソード戦略の場合は、2種類の名前空間を指定する必要があります

2-2-2. Built-in strategy with override

今回は触れません。

2-2-3. Self-managed strategy

今回は触れません。

3. 作成実行

「メモリを作成」ボタンをクリックすると、メモリの作成が開始されます。

参考

メモリの作成をスクリプトで実行したい方は以下を参考にしてください。

import os
import time
from bedrock_agentcore.memory import MemoryClient

"""
# AgentCore Memoryについて

## 管理レベル
- Memory ID: メモリインスタンス全体を識別
- Actor ID: 個々のユーザーを識別・データ分離
- Session ID: 学習セッションごとの管理・サマリー生成

## 戦略
- SUMMARY: インタラクションを要約して、重要なコンテキストと主要なインサイトを保持
- SEMANTIC: コンテキストに依拠しない形式で、生の会話から一般的な事実に関する知識、概念、意味を抽出
- USER_PREFERENCE: 生の会話からユーザーの行動パターンを抽出
- EPISODIC: 生の会話からユーザーの行動パターンを抽出
"""

BEDROCK_AGENT_AWS_REGION = os.getenv("BEDROCK_AGENT_AWS_REGION", "ap-northeast-1")
client = MemoryClient(region_name=BEDROCK_AGENT_AWS_REGION)

# タイムスタンプを取得
timestamp = int(time.time())
memory_name = f"SampleAppAgentMemory_{timestamp}"

# 戦略の定義
strategies=[
  {
    "summaryMemoryStrategy": {
      "name": "conversation_summary",
      "namespaces": ["/sample_app/users/{actorId}/{sessionId}/summary"]
    }
  },
  {
    "semanticMemoryStrategy": {
      "name": "semantic_knowledge",
      "namespaces": ["/sample_app/knowledge"]
    }
  },
  {
    "userPreferenceMemoryStrategy": {
      "name": "user_preferences",
      "namespaces": ["/sample_app/users/{actorId}/preferences"]
    }
  },
  # {
  #   "episodicMemoryStrategy": {
  #     "name": "user_episodes",
  #     "namespaces": ["/sample_app/users/{actorId}/episodes"]
  #   }
  # }
]

# Memoryインスタンスの作成
# create_memory_and_wait を使用して作成が完了するまで待機
print(f"メモリーを作成中... (名前: {memory_name})")
memory = client.create_memory_and_wait(
  name=memory_name,
  strategies=strategies,
  event_expiry_days=30  # イベントの保存期間（日数）
)

memory_id = memory['id']
print(f"メモリーを作成しました。Memory ID: {memory_id}")

# 各戦略のIDを確認
print("\n作成された戦略:")
for strategy in memory.get('strategies', []):
  strategy_type = strategy.get('type', 'unknown')
  strategy_id = strategy.get('strategyId', 'unknown')
  strategy_name = strategy.get('name', 'unknown')
  namespaces = strategy.get('namespaces', [])
  print(f"- 戦略タイプ: {strategy_type}")
  print(f"  戦略ID: {strategy_id}")
  print(f"  戦略名: {strategy_name}")
  if namespaces:
    print(f"  Namespace: {namespaces[0]}")

# .envファイルにMEMORY_IDを更新
print("\n.envファイルを更新中...")
try:
  with open('.env', 'r') as f:
    lines = f.readlines()

  # MEMORY_ID行を探して更新
  updated = False
  for i, line in enumerate(lines):
    if line.startswith('MEMORY_ID='):
      lines[i] = f'MEMORY_ID={memory_id}\n'
      updated = True
      break

  # MEMORY_ID行が見つからない場合は追加
  if not updated:
    # Bedrock Agent Configuration セクションの後に追加
    for i, line in enumerate(lines):
      if 'BEDROCK_AGENT_AWS_REGION' in line:
        lines.insert(i + 1, f'MEMORY_ID={memory_id}\n')
        updated = True
        break

  # ファイルに書き戻し
  with open('.env', 'w') as f:
    f.writelines(lines)

  print(f"✅ .envファイルにMEMORY_ID={memory_id}を設定しました")
except Exception as e:
  print(f"⚠️  .envファイルの更新に失敗しました: {e}")
  print(f"   手動で以下を.envファイルに追加してください:")
  print(f"   MEMORY_ID={memory_id}")

AIエージェントと連携する方法

0. 前提条件

• Amazon Bedrock AgentCore Runtimeなどで、AIエージェントを実装していることを前提としています（私はagent.pyというファイルでstrands agentsを使用してAIエージェントを構築しています。）

1. メモリ保存関数を作成

メモリを保存するための処理を記述します。

def save_memory(memory_id: str, actor_id: str, session_id: str, user_message: str, assistant_message: str):
    """会話をメモリに保存"""
    try:
        memory_client.create_event(
            memory_id=memory_id,
            actor_id=actor_id,
            session_id=session_id,
            messages=[
                (user_message, "USER"),
                (assistant_message, "ASSISTANT")
            ]
        )
        print("✅ 会話をメモリに保存しました")
    except Exception as e:
        print(f"❌ メモリ保存エラー: {e}")

この関数は、エージェントからのレスポンスが返ってきてから呼び出します。

2. メモリ呼び出し関数を作成

保存しておいた記憶を呼び出すための関数を作成します。

def load_memory(memory_id: str, actor_id: str, session_id: str, max_results: int = 10):
    """過去の会話履歴を取得"""
    try:
        events = memory_client.list_events(
            memory_id=memory_id,
            actor_id=actor_id,
            session_id=session_id,
            max_results=max_results
        )
        # 時系列順に直す
        sorted_events = sorted(events, key=lambda x: x['eventTimestamp'])
        # StrandsAgentsで認識可能な形式に直す
        history_messages = []
        for event in sorted_events:
            for item in event.get('payload', []):
                if 'conversational' in item:
                    conv = item['conversational']
                    history_messages.append({
                        "role": conv['role'].lower(),
                        "content": [{"text": conv['content']['text']}]
                    })

        return history_messages
    except Exception as e:
        print(f"⚠️ 履歴取得エラー: {e}")
        return []

3. エージェントへのリクエストを送信する前に会話履歴を読み込む

# 会話履歴の読み込み
actor_id = "xxxxxxxxxxx"
session_id = "yyyyyyyyyyy"
MEMORY_ID = os.getenv("MEMORY_ID", "None")

try:
    history_messages = load_memory(MEMORY_ID, actor_id, session_id, max_results=10)
    if history_messages:
        print(f"✅ {len(history_messages)}件の過去メッセージを取得")
        # 履歴がある場合は、エージェントのメッセージリストに設定
        agent.messages = history_messages
except Exception as e:
    print(f"⚠️ 会話履歴の読み込みに失敗しました: {e}")
    history_messages = []


# エージェント呼び出し関数（既に実装済）
res = await agent.invoke_async(user_message)
result = res.message['content'][0]['text']

4. レスポンスが帰ってきたら、保存する

# エージェント呼び出し関数（既に実装済）
res = await agent.invoke_async(user_message)
result = res.message['content'][0]['text']

# 会話を保存
save_memory(MEMORY_ID, actor_id, session_id, user_message, agent.get_last_assistant_message()
)

これで、会話履歴の保存ができるようになりました！

まとめ

今回は、Bedrock AgentCore Memoryを使用して、会話履歴を保存する方法について紹介をしました。

次の記事で、Rails製のWebアプリケーションでDBを使用せずにAIエージェントチャットを実装できるのか、検証してみたいと思います。