Amazon Bedrock AgentCore Memory 実践編：短期記憶と要約による長期記憶を試す

from bedrock_agentcore.memory import MemoryClient

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

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

# The memory_id will be used in following operations
print(f"Memory ID: {memory.get('id')}")
print(f"Memory: {memory}") 

from bedrock_agentcore.memory import MemoryClient

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

for memory in client.list_memories():
    print(f"Memory Arn: {memory.get('arn')}")
    print(f"Memory ID: {memory.get('id')}")
    print("--------------------------------------------------------------------")


client.create_event(
    memory_id=memory.get("id"), # This is the id from create_memory or list_memories
    actor_id="User84",  # This is the identifier of the actor, could be an agent or end-user.
    session_id="OrderSupportSession1", #Unique id for a particular request/conversation.
    messages=[
        ("Hi, I'm having trouble with my order #12345", "USER"),
        ("I'm sorry to hear that. Let me look up your order.", "ASSISTANT"),
        ("lookup_order(order_id='12345')", "TOOL"),
        ("I see your order was shipped 3 days ago. What specific issue are you experiencing?", "ASSISTANT"),
        ("Actually, before that - I also want to change my email address", "USER"),
        (
            "Of course! I can help with both. Let's start with updating your email. What's your new email?",
            "ASSISTANT",
        ),
        ("newemail@example.com", "USER"),
        ("update_customer_email(old='old@example.com', new='newemail@example.com')", "TOOL"),
        ("Email updated successfully! Now, about your order issue?", "ASSISTANT"),
        ("The package arrived damaged", "USER"),
    ],
)

conversations = client.list_events(
    memory_id=memory.get("id"),
    actor_id="User84",
    session_id="OrderSupportSession1",
    max_results=5,
)

for ev in conversations:
    print("------------------------------------------------")
    print(f"Event ID: {ev['eventId']}")
    print(f"Timestamp: {ev['eventTimestamp']}")

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

from bedrock_agentcore.memory import MemoryClient

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

memory = client.create_memory_and_wait(
    name="longmemory",
    strategies=[{
        "summaryMemoryStrategy": {
            # Name of the extraction model/strategy
            "name": "SessionSummarizer",
            # Organize facts by session ID for easy retrieval
            # Example: "summaries/User84/session123" contains summary of session123
            "namespaces": ["/summaries/{actorId}/{sessionId}"]
        }
    }]
) 

# longconversation.py
from bedrock_agentcore.memory import MemoryClient
import time
import json
import sys

# ====== 設定 ======
REGION = "us-west-2"
MEMORY_ID = "longmemory-5xFyxF30b8"  # 既存のメモリIDに置き換え可
ACTOR_ID = "User84"
SESSION_ID = "OrderSupportSession1"
NAMESPACE = "/summaries/User84/OrderSupportSession1"  # SummaryMemoryStrategy のテンプレに合わせる

client = MemoryClient(region_name=REGION)

# ====== 会話イベントを投入 ======
try:
    event = client.create_event(
        memory_id=MEMORY_ID,         # create_memory / list_memories の ID
        actor_id=ACTOR_ID,           # エージェント or エンドユーザー識別子
        session_id=SESSION_ID,       # 会話セッションのユニークID
        messages=[
            ("Hi, I'm having trouble with my order #12345", "USER"),
            ("I'm sorry to hear that. Let me look up your order.", "ASSISTANT"),
            ("lookup_order(order_id='12345')", "TOOL"),
            ("I see your order was shipped 3 days ago. What specific issue are you experiencing?", "ASSISTANT"),
            ("Actually, before that - I also want to change my email address", "USER"),
            (
                "Of course! I can help with both. Let's start with updating your email. What's your new email?",
                "ASSISTANT",
            ),
            ("newemail@example.com", "USER"),
            ("update_customer_email(old='old@example.com', new='newemail@example.com')", "TOOL"),
            ("Email updated successfully! Now, about your order issue?", "ASSISTANT"),
            ("The package arrived damaged", "USER"),
        ],
    )
    print(f"[OK] Event created: {event.get('eventId') if isinstance(event, dict) else 'created'}")
except Exception as e:
    print(f"[ERR] create_event failed: {e}", file=sys.stderr)
    sys.exit(1)

# ====== 抽出（Long-term Summary）は非同期。少し待つ ======
print("[INFO] Waiting 60s for summary extraction...")
time.sleep(60)

# ====== 取得ユーティリティ ======
def _extract_items(res):
    """SDKの戻りがlist/dictいずれでも配列部分を取り出す"""
    if isinstance(res, list):
        return res
    if isinstance(res, dict):
        return res.get("memories") or res.get("items") or res.get("results") or []
    return []

def _get_text(m):
    if not isinstance(m, dict):
        return None

    # content.text
    txt = (m.get("content") or {}).get("text")
    if txt:
        return txt

    # conversational.content.text
    conv = m.get("conversational") or {}
    txt = (conv.get("content") or {}).get("text")
    if txt:
        return txt

    # summary / text / value
    for k in ("summary", "text", "value"):
        v = m.get(k)
        if isinstance(v, str) and v.strip():
            return v

    # document.text
    doc = m.get("document") or {}
    if isinstance(doc, dict):
        txt = doc.get("text")
        if txt:
            return txt

    return None

# ====== サマリーをポーリングして取得 ======
deadline = time.time() + 180  # 最大3分待つ
poll_interval = 5
items = []

print(f"[INFO] Retrieving summaries from namespace: {NAMESPACE}")
while time.time() < deadline:
    try:
        res = client.retrieve_memories(
            memory_id=MEMORY_ID,
            namespace=NAMESPACE,
            query="summarize the support issue"
        )
    except Exception as e:
        print(f"[WARN] retrieve_memories failed, retrying: {e}")
        time.sleep(poll_interval)
        continue

    items = _extract_items(res)
    if items:
        # 必要なら RAW を見たいときにコメントアウト解除
        # print("---- RAW ----")
        # print(json.dumps(res, indent=2, default=str))
        break

    time.sleep(poll_interval)

# ====== 出力 ======
if not items:
    print("[INFO] No summary extracted yet.")
else:
    print("\n---- SUMMARIES ----")
    for i, m in enumerate(items, 1):
        text = _get_text(m) or "[no text field]"
        score = m.get("score") or m.get("similarity")
        ns = m.get("namespace") or (m.get("metadata") or {}).get("namespace")
        print(f"[{i}] {text}")
        if score is not None or ns:
            print(f"      meta: score={score} namespace={ns}")