[アップデート] AgentCore Memory の長期記憶でカスタムメタデータフィルタを使ってみた

# プロジェクト初期化
uv init --python 3.13

# 依存パッケージの追加（uv add すると仮想環境とロックファイルは自動で整います）
uv add boto3

import boto3

control = boto3.client("bedrock-agentcore-control", region_name="us-east-1")

res = control.create_memory(
    name="MetadataFilterBlogMinimal",
    eventExpiryDuration=30,
    memoryStrategies=[
        {"userPreferenceMemoryStrategy": {
            "name": "BlogUserPreferenceMinimal",
            "namespaces": ["/users/{actorId}/preferences"],
        }},
    ],
)
print(res["memory"]["id"])

uv run create_memory_minimal.py

MetadataFilterBlogMinimal-XXXXXXXXXX

import boto3

control = boto3.client("bedrock-agentcore-control", region_name="us-east-1")

destination_schema_entry = {
    "key": "destination",
    "type": "STRING",
    "extractionConfig": {
        "llmExtractionConfig": {
            "definition": "Travel destination as a lowercase city code.",
            "llmExtractionInstruction": "LATEST_VALUE",
            "validation": {
                "stringValidation": {"allowedValues": ["kyoto", "tokyo"]},
            },
        }
    },
}

res = control.create_memory(
    name="MetadataFilterBlogIndexed",
    eventExpiryDuration=30,
    indexedKeys=[{"key": "destination", "type": "STRING"}],
    memoryStrategies=[
        {"userPreferenceMemoryStrategy": {
            "name": "BlogUserPreferenceIndexed",
            "namespaces": ["/users/{actorId}/preferences"],
            "memoryRecordSchema": {"metadataSchema": [destination_schema_entry]},
        }},
    ],
)
print(res["memory"]["id"])

uv run create_memory_indexed.py

MetadataFilterBlogIndexed-XXXXXXXXXX

import os
import boto3
from datetime import datetime, timezone

MEMORY_ID = os.environ["MEMORY_ID"]
ACTOR_ID = "blog-actor-001"
SESSION_ID = "blog-session-001"

client = boto3.client("bedrock-agentcore", region_name="us-east-1")

events = [
    ("旅行の宿は和室が好きです。京都の和菓子も巡りたい。",
     {"category": "travel", "destination": "kyoto", "priority": "high"}),
    ("出張で東京に行くときは新幹線のグリーン車を選びがちです。",
     {"category": "travel", "destination": "tokyo", "priority": "medium"}),
    ("コーヒーは深煎りのブラックが好みです。",
     {"category": "food", "topic": "coffee", "priority": "low"}),
    ("ホテルは静かなクラブフロアが好きで、朝食は和食派です。",
     {"category": "travel", "destination": "kyoto", "priority": "high"}),
]

for text, md in events:
    ev = client.create_event(
        memoryId=MEMORY_ID, actorId=ACTOR_ID, sessionId=SESSION_ID,
        eventTimestamp=datetime.now(timezone.utc),
        payload=[{"conversational": {"content": {"text": text}, "role": "USER"}}],
        metadata={k: {"stringValue": v} for k, v in md.items()},
    )
    print("created", ev["event"]["eventId"])

MEMORY_ID=MetadataFilterBlogMinimal-XXXXXXXXXX uv run create_events.py
MEMORY_ID=MetadataFilterBlogIndexed-XXXXXXXXXX uv run create_events.py

created 0000001XXXXXXXXXXXX#xxxxxxxx
created 0000001XXXXXXXXXXXX#xxxxxxxx
created 0000001XXXXXXXXXXXX#xxxxxxxx
created 0000001XXXXXXXXXXXX#xxxxxxxx

import boto3
from botocore.exceptions import ClientError

# 構成 A（indexedKeys なし）の Memory ID を指定
MEMORY_ID = "MetadataFilterBlogMinimal-XXXXXXXXXX"
ACTOR_ID = "blog-actor-001"

client = boto3.client("bedrock-agentcore", region_name="us-east-1")
ns = f"/users/{ACTOR_ID}/preferences"

try:
    res = client.retrieve_memory_records(
        memoryId=MEMORY_ID, namespace=ns,
        searchCriteria={
            "searchQuery": "好み", "topK": 10,
            "metadataFilters": [
                {"left": {"metadataKey": "destination"},
                 "operator": "EQUALS_TO",
                 "right": {"metadataValue": {"stringValue": "kyoto"}}},
            ],
        },
        maxResults=10,
    )
    print("hits =", len(res.get("memoryRecordSummaries", [])))
except ClientError as e:
    print(type(e).__name__, ":", e)

uv run retrieve_minimal.py

botocore.errorfactory.ValidationException: Filter key 'destination' is not a valid filter key

import boto3

# 構成 B（indexedKeys + metadataSchema を宣言した Memory）の ID を指定
MEMORY_ID = "MetadataFilterBlogIndexed-XXXXXXXXXX"
ACTOR_ID = "blog-actor-001"

client = boto3.client("bedrock-agentcore", region_name="us-east-1")
ns = f"/users/{ACTOR_ID}/preferences"

res = client.list_memory_records(
    memoryId=MEMORY_ID, namespace=ns, maxResults=20,
)

for r in res.get("memoryRecordSummaries", []):
    print(r)

uv run list_records.py

{
  "memoryRecordId": "mem-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx",
  "content": {
    "text": "{\"context\":\"ユーザーが明示的に旅行の宿は和室が好きだと述べています。\",\"preference\":\"旅行の宿は和室が好き\",\"categories\":[\"旅行\",\"宿泊\"]}"
  },
  "namespaces": ["/users/blog-user-xxxxxxxxx/preferences"],
  "metadata": {
    "x-amz-agentcore-memory-recordType": {"stringValue": "BASE"},
    "destination": {"stringValue": "kyoto"},
    "x-amz-agentcore-memory-createdAt": {"stringValue": "Wed May 06 05:08:15 UTC 2026"},
    "x-amz-agentcore-memory-updatedAt": {"stringValue": "Wed May 06 05:08:15 UTC 2026"}
  }
}

import boto3
import json

# 構成 B の Memory ID を指定（list_records.py と同じ ID）
MEMORY_ID = "MetadataFilterBlogIndexed-XXXXXXXXXX"
ACTOR_ID = "blog-actor-001"

client = boto3.client("bedrock-agentcore", region_name="us-east-1")
ns = f"/users/{ACTOR_ID}/preferences"

res = client.retrieve_memory_records(
    memoryId=MEMORY_ID, namespace=ns,
    searchCriteria={
        "searchQuery": "旅行", "topK": 10,
        "metadataFilters": [
            {"left": {"metadataKey": "destination"},
             "operator": "EQUALS_TO",
             "right": {"metadataValue": {"stringValue": "kyoto"}}},
        ],
    },
    maxResults=10,
)

print("hits =", len(res.get("memoryRecordSummaries", [])))
for r in res.get("memoryRecordSummaries", []):
    pref = json.loads(r["content"]["text"])["preference"]
    dest = r["metadata"]["destination"]["stringValue"]
    print(f"  {pref} / destination = {dest}")

uv run retrieve_indexed.py

hits = 6
  旅行の宿は和室が好き / destination = kyoto
  京都の和菓子巡りに興味がある / destination = kyoto
  ホテルは静かなクラブフロアが好き / destination = kyoto
  コーヒーは深煎りのブラックが好き / destination = kyoto
  朝食は和食派 / destination = kyoto
  コーヒー豆はエチオピア産が好み / destination = kyoto

import boto3
import time
from datetime import datetime, timezone

# 構成 B（indexedKeys + metadataSchema を宣言した Memory）の ID を指定
MEMORY_ID = "MetadataFilterBlogIndexed-XXXXXXXXXX"
ACTOR_ID = "blog-actor-coffee"
SESSION_ID = "blog-session-coffee"

client = boto3.client("bedrock-agentcore", region_name="us-east-1")

events = [
    ("コーヒーは深煎りのブラックが好きです。", {"category": "food", "topic": "coffee"}),
    ("豆はエチオピア産が好みです。", {"category": "food", "topic": "coffee"}),
    ("朝はマグカップ一杯から始めることが多いです。", {"category": "food", "topic": "coffee"}),
]

for text, md in events:
    client.create_event(
        memoryId=MEMORY_ID, actorId=ACTOR_ID, sessionId=SESSION_ID,
        eventTimestamp=datetime.now(timezone.utc),
        payload=[{"conversational": {"content": {"text": text}, "role": "USER"}}],
        metadata={k: {"stringValue": v} for k, v in md.items()},
    )

# LLM 抽出はバックグラウンドで走るので待機（環境次第で 1〜2 分ほど）
time.sleep(120)

ns = f"/users/{ACTOR_ID}/preferences"
res = client.list_memory_records(memoryId=MEMORY_ID, namespace=ns, maxResults=20)

records = res.get("memoryRecordSummaries", [])
with_destination = [r for r in records if "destination" in (r.get("metadata") or {})]
values = [r["metadata"]["destination"]["stringValue"] for r in with_destination]

print(f"records={len(records)}, records_with_destination={len(with_destination)}")
print(f"  destination values observed: {values}")

uv run probe_destination.py

records=2, records_with_destination=2
  destination values observed: ['kyoto']

Only output 'kyoto' or 'tokyo' when the user explicitly mentions a travel destination matching that city. Otherwise output 'unknown'.

import boto3
import time
from datetime import datetime, timezone

REGION = "us-east-1"
control = boto3.client("bedrock-agentcore-control", region_name=REGION)
data = boto3.client("bedrock-agentcore", region_name=REGION)

# 検証する 3 パターン
PATTERNS = [
    ("BaselineLATESTVALUE",
     ["kyoto", "tokyo"],
     "LATEST_VALUE"),
    ("WithUnknown",
     ["kyoto", "tokyo", "unknown"],
     "LATEST_VALUE"),
    ("WithUnknownAndInstruction",
     ["kyoto", "tokyo", "unknown"],
     "Only output 'kyoto' or 'tokyo' when the user explicitly mentions a travel destination matching that city. Otherwise output 'unknown'."),
]

EVENTS = [
    "コーヒーは深煎りのブラックが好きです。",
    "豆はエチオピア産が好みです。",
    "朝はマグカップ一杯から始めることが多いです。",
]

def create_memory(label, allowed_values, instruction):
    schema_entry = {
        "key": "destination",
        "type": "STRING",
        "extractionConfig": {
            "llmExtractionConfig": {
                "definition": "Travel destination as a lowercase city code.",
                "llmExtractionInstruction": instruction,
                "validation": {"stringValidation": {"allowedValues": allowed_values}},
            }
        },
    }
    res = control.create_memory(
        name=f"ProbeUnknown{label}",
        eventExpiryDuration=30,
        indexedKeys=[{"key": "destination", "type": "STRING"}],
        memoryStrategies=[{
            "userPreferenceMemoryStrategy": {
                "name": f"ProbeStrat{label}",
                "namespaces": ["/users/{actorId}/preferences"],
                "memoryRecordSchema": {"metadataSchema": [schema_entry]},
            }
        }],
    )
    return res["memory"]["id"]

def probe(memory_id, label):
    actor = f"probe-actor-{label}"
    session = f"probe-session-{label}"
    for text in EVENTS:
        data.create_event(
            memoryId=memory_id, actorId=actor, sessionId=session,
            eventTimestamp=datetime.now(timezone.utc),
            payload=[{"conversational": {"content": {"text": text}, "role": "USER"}}],
        )

    # LLM 抽出待ち
    time.sleep(120)

    ns = f"/users/{actor}/preferences"
    res = data.list_memory_records(memoryId=memory_id, namespace=ns, maxResults=20)
    values = [
        r["metadata"]["destination"]["stringValue"]
        for r in res.get("memoryRecordSummaries", [])
        if "destination" in (r.get("metadata") or {})
    ]
    print(f"[{label}] destination values: {values}")

if __name__ == "__main__":
    memories = []
    for label, allowed, instruction in PATTERNS:
        mid = create_memory(label, allowed, instruction)
        print(f"created {label}: {mid}")
        memories.append((mid, label))

    # Memory が ACTIVE になるまで待機（実測で 2〜3 分）
    print("Memory ACTIVE 待ち")
    time.sleep(180)

    for mid, label in memories:
        probe(mid, label)

uv run probe_destination_with_unknown.py

[BaselineLATESTVALUE]       destination values: ['tokyo', 'tokyo', 'tokyo']
[WithUnknown]               destination values: ['unknown', 'unknown', 'unknown']
[WithUnknownAndInstruction] destination values: ['unknown', 'unknown']

Choose exactly one of the following lowercase city codes that the user explicitly mentions as a travel destination: kyoto, tokyo, osaka. If the user does not mention any of these cities as a travel destination, do not produce a value for this field.

[A: LATEST_VALUE のみ]   records=3 destination_values=['kyoto', 'kyoto', 'tokyo']
[B: カスタム指示]        records=4 destination_values=['kyoto', 'tokyo']
[C: 選択肢を明示]        records=4 destination_values=['tokyo', 'kyoto']