Amazon Bedrock AgentCore ⑥ Memory ⑦ Observability 編

Posted at 2025-10-11

はじめに

Amazon Bedrock AgentCore の最終編です。

今回も触りながらをメインに書いていこうと思います。

⑥ Memory 編

AI エージェントに記憶機能を持たせるためのサービスで、2 種類のメモリを提供しています。

短期メモリと長期メモリ

短期メモリ	長期メモリ
会話を保存して、即時の状況を追跡 ...今の会話の流れを覚える（「さっき言った件だけど」が通じる	ユーザーの好み、意味的事実、要約などの抽出された洞察を保存し、セッション全体にわたって知識を保持 ...あなたの好みや特徴を覚える（次回会った時も覚えてる）

長期メモリには、3 つのタイプがある。

タイプ	説明	例
① ユーザー設定（User Preferences）	ユーザーの好みやパターンを学習して記憶	コーディングエージェントが、あなたが常にコメント付きのきれいなコードを書く、snake_case命名を好む、データ分析にはpandasを使う、などの傾向を学習。次回以降、これらの設定を自動的に適用。
② 意味的事実（Semantic Facts）	概念や知識を理解して記憶	「Pandasは表データを扱うPythonライブラリ」という知識を記憶。後日「表データに最適なライブラリは？」と聞かれたら、即座にPandasを提案。
③ 要約（Summary）	セッションの要約を生成して保存	「このやり取りで、データクリーニング関数を作成し、2つの構文エラーを修正し、線形回帰モデルをテストしました」といった要約を生成し、効率的に過去の活動を参照できるようにします。

前提

bedrock-agentcore をインストール済みであること(以下の前回の記事を参考にして下さい)

短期メモリを作成する

ドキュメントを参考に動かしてみます。

メモリを作成します。

my_bedrock_agentcore_memory_short1.py

# Memoryクライアントをインポート
from bedrock_agentcore.memory import MemoryClient

# Memoryクライアントの初期化
memory_client = MemoryClient(region_name="us-west-2")

# Memoryの作成
memory = memory_client.create_memory(
    name="CustomerSupportAgentMemory",
    description="Memory for customer support conversations",

my_bedrock_agentcore_memory_short1.py

作成されたリソースを AWS マネージメントコンソールから確認してみます。

既存のメモリリソースを一覧表示し、短期メモリに対して会話を書き込み、読み込んでみます。

my_bedrock_agentcore_memory_short2.py

# Memoryクライアント をインポートする
from bedrock_agentcore.memory import MemoryClient

# Memoryクライアントの初期化
memory_client = MemoryClient(region_name="us-west-2")

# メモリの一覧を取得して表示
for memory in memory_client.list_memories():
    print(f"Memory Arn: {memory.get('arn')}")
    print(f"Memory ID: {memory.get('id')}")
    print("--------------------------------------------------------------------")

# イベントを作成
memory_client.create_event(
    memory_id=memory.get("id"), # create_memoryまたはlist_memoriesから取得したID
    actor_id="User84",  # アクターの識別子（エージェントまたはエンドユーザー）
    session_id="OrderSupportSession1", # 特定のリクエスト/会話の一意のID
    messages=[
        ("こんにちは、注文番号#12345に問題があります", "USER"),
        ("申し訳ございません。ご注文を確認いたします。", "ASSISTANT"),
        ("lookup_order(order_id='12345')", "TOOL"),
        ("ご注文は3日前に発送されています。具体的にどのような問題がございますか？", "ASSISTANT"),
        ("その前に - メールアドレスも変更したいのですが", "USER"),
        (
            "もちろんです！両方お手伝いいたします。まずメールアドレスの更新から始めましょう。新しいメールアドレスは何ですか？",
            "ASSISTANT",
        ),
        ("newemail@example.com", "USER"),
        ("update_customer_email(old='old@example.com', new='newemail@example.com')", "TOOL"),
        ("メールアドレスを更新しました！それでは、ご注文の問題について教えてください。", "ASSISTANT"),
        ("荷物が破損して届きました", "USER"),
    ],
)

# 最近の会話イベントを取得して表示
conversations = memory_client.list_events(
    memory_id=memory.get("id"),
    actor_id="User84",
    session_id="OrderSupportSession1",
    max_results=5,
)

# 会話イベントを表示
for e in conversations:
    print("------------------------------------------------")
    print(f"Event ID: {e['eventId']}")
    print(f"Timestamp: {e['eventTimestamp']}")

    for p in e.get("payload", []):
        conv = p.get("conversational", {})
        role = conv.get("role")
        text = conv.get("content", {}).get("text")
        print(f"{role}: {text}")

my_bedrock_agentcore_memory_short2.py

長期メモリを作成する

メモリを作成します。

my_bedrock_agentcore_memory_long1.py

# Memoryクライアントをインポートする
from bedrock_agentcore.memory import MemoryClient

# Memoryクライアントの初期化
memory_client = MemoryClient(region_name="us-west-2")

# Memoryの作成
memory = memory_client.create_memory_and_wait(
    name="MyAgentMemory",
    strategies=[{
        "userPreferenceMemoryStrategy": {
            "name": "UserPreference", # メモリ戦略の名前
            "namespaces": ["/users/{actorId}"] # メモリの名前空間
        }
    }]
)

python my_bedrock_agentcore_memory_long1.py

作成されたリソースを AWS マネージメントコンソールから確認してみます。

エージェントが長期記憶から情報を取得できるか Strands Agents を使って実行してみます。

my_bedrock_agentcore_memory_long2.py

# Memoryクライアントをインポートする
from bedrock_agentcore.memory import MemoryClient
from strands import Agent
from strands_tools.agent_core_memory import AgentCoreMemoryToolProvider
import time

# Memoryクライアントの初期化
memory_client = MemoryClient(region_name="us-west-2")

# 既存のメモリを取得
memories = memory_client.list_memories()
memory_id = None
for mem in memories:
    mem_id = mem.get("id")
    # IDにMyAgentMemoryが含まれているか、nameがMyAgentMemoryのものを探す
    if mem_id and "MyAgentMemory" in mem_id:
        memory_id = mem_id
        break
    elif mem.get("name") == "MyAgentMemory":
        memory_id = mem_id
        break

if memory_id is None:
    print("エラー: MyAgentMemoryが見つかりません。先にメモリを作成して下さい。")
    print(f"利用可能なメモリ: {memories}")
    exit(1)

print(f"memory_id: {memory_id}")

# AgentCoreMemoryToolProviderの初期化
strands_provider = AgentCoreMemoryToolProvider(
    memory_id=memory_id,
    actor_id="CaliforniaPerson",
    session_id="TalkingAboutFood",
    namespace="/users/CaliforniaPerson",
    region="us-west-2"
)
agent = Agent(tools=strands_provider.tools)

agent("私はベジタリアンで、静かな雰囲気のレストランが好きです。")
agent("イタリアン料理が食べたい気分です。")
agent("中価格帯でダウンタウンにある店が良いです。")
agent("私はアーバインに住んでいます。")

print("\n30秒待機中...")
time.sleep(30)

# ToolProviderを再初期化して、同じメモリを使用する
print("\n長期記憶から情報を取得中...")
agent("何が食べたかったか思い出せないのですが、覚えていますか？")

python my_bedrock_agentcore_memory_long2.py

⑦ Observability 編

Observability は Amazon Bedrock AgentCore が提供する、AI エージェントのトレース、デバッグ、本番環境でのパフォーマンス監視を支援する機能です。

特徴

Amazon CloudWatch を活用したダッシュボードを通じて、セッション数、レイテンシ、処理時間、トークン使用量、エラー率などの主要メトリクスをリアルタイムで可視化
AgentCoreOpenTelemetry(OTEL)互換形式でテレメトリデータを出力するため、既存の監視・可視化スタックと簡単に統合可能
デフォルトで、エージェント、ゲートウェイリソース、メモリリソース向けの主要な組み込みメトリクスを出力

イメージ

① Runtime 編をやった時に確認しとけといったところですが、少しだけイメージを貼っておきます。

最後に

2025/10/11 現在これを検証している最中はオレゴンリージョンでやっていましたが、東京リージョンにもきているようです。

企業内で導入するにあたっては、ひとりができてもといったところがあるので割と勉強コストは高いのかなと思いました。

次は個別に動作検証をしていったので複数のコンポーネントを組み合わせて検証をしていけたらなと思います。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up