Hindsight 记忆数据增删改实操：70 个 API 端点的 CRUD 地图

import urllib.request, json

def api(method, path, body=None, timeout=180):
    url = "http://<your-server>:8888" + path
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method,
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req, timeout=timeout) as r:
        return json.loads(r.read().decode())

# 同步写入一条记忆
resp = api("POST", "/v1/default/banks/test_bank/memories", {
    "items": [{
        "content": "Hindsight CRUD 实验：单条写入测试，验证 retain 同步模式",
        "context": "test/crud",
        "tags": ["topic:code", "stage:process"],
        "metadata": {"experiment": "crud-20260821", "step": "1-create"},
    }],
    "async": False,   # 同步模式
})

print(resp)
# {'success': True, 'items_count': 1, 'async': False, 'operation_id': None,
#  'usage': {'input_tokens': 3029, 'output_tokens': 2628, 'total_tokens': 5657}}

resp = api("POST", "/v1/default/banks/test_bank/memories", {
    "items": [{"content": f"批量写入第 {i} 条", "context": "test/crud-batch",
               "tags": ["topic:code", "stage:process"],
               "metadata": {"idx": str(i)}} for i in range(10)],
    "async": True,    # 异步模式
})
print(resp)
# {'success': True, 'items_count': 10, 'async': True,
#  'operation_id': '24259077-8873-4e2e-be35-c568c62eafd5', 'usage': None}

import time
op_id = resp["operation_id"]
for i in range(12):
    time.sleep(5)
    st = api("GET", f"/v1/default/banks/test_bank/operations/{op_id}")
    print(f"t+{(i+1)*5}s status={st['status']}")
    if st["status"] == "completed":
        break
# t+5s  status=pending
# t+10s status=pending
# t+20s status=pending
# t+25s status=completed

HTTP 422: Input should be a valid string
{"type": "string_type", "loc": ["body", "items", 0, "metadata", "idx"], "msg": "..."}

items = api("GET", "/v1/default/banks/test_bank/memories/list?limit=50")
for m in items["items"][:5]:
    print(f"{m['id'][:8]} type={m.get('fact_type')} tags={m.get('tags')}")
# 999cf0be type=observation tags=['stage:process', 'topic:code']
# c34f1d81 type=world       tags=['stage:process', 'topic:code']

r = api("POST", "/v1/default/banks/test_bank/memories/recall",
        {"query": "Hindsight CRUD 批量写入实验", "limit": 5})
# hits=13, 端到端 281ms

# 找到 document
docs = api("GET", "/v1/default/banks/test_bank/documents")
doc_id = docs["documents"][0]["id"]

# 改 document tags
r = api("PATCH", f"/v1/default/banks/test_bank/documents/{doc_id}",
        {"tags": ["topic:code", "stage:decision", "tag:patched"]})
# {'success': True} 75ms

# 验证：document 的 tags 已更新
doc = api("GET", f"/v1/default/banks/test_bank/documents/{doc_id}")
# tags=['topic:code', 'stage:decision', 'tag:patched']

# 验证：关联的所有 memory units 的 tags 也被传播
items = api("GET", "/v1/default/banks/test_bank/memories/list?limit=50")
patched = [m for m in items["items"] if "tag:patched" in m.get("tags", [])]
# 10 条全部带上了新 tag！

r = api("POST", f"/v1/default/banks/test_bank/documents/{doc_id}/reprocess")
# 返回 {'success': True, 'operation_id': ..., 'items_count': ...}

r = api("DELETE", f"/v1/default/banks/test_bank/memories/{some_memory_id}")
# HTTP 405: {"detail": "Method Not Allowed"}  6ms

r = api("DELETE", f"/v1/default/banks/test_bank/documents/{doc_id}")
# {'success': True,
#  'message': "Document '90bd...' and 10 associated memory units deleted successfully",
#  'memory_units_deleted': 10}

# 只删 world 类型（英文翻译版）
r = api("DELETE", "/v1/default/banks/test_bank/memories?type=world")
# {'success': True}  30ms

# 清空整个 bank 的所有记忆
r = api("DELETE", "/v1/default/banks/test_bank/memories")
# {'success': True}  53ms

# 删除整个 bank（含文档、实体、图谱）
r = api("DELETE", "/v1/default/banks/test_bank")
# {'success': True, 'message': "Bank 'test_bank' and all associated data deleted successfully"}

import urllib.request, json, time

BASE = "http://<your-server>:8888"
BANK = "test_crud_demo"

def api(method, path, body=None, timeout=180):
    url = BASE + path
    data = json.dumps(body).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method,
                                 headers={"Content-Type": "application/json"})
    t0 = time.time()
    with urllib.request.urlopen(req, timeout=timeout) as r:
        raw = r.read().decode()
        return json.loads(raw), (time.time() - t0) * 1000

# 1. 增：异步批量（推荐）
resp, dt = api("POST", f"/v1/default/banks/{BANK}/memories", {
    "items": [{"content": f"CRUD 演示第 {i} 条：Hindsight 批量写入测试",
               "tags": ["topic:code", "stage:process"],
               "metadata": {"demo": "true", "idx": str(i)}} for i in range(10)],
    "async": True,
})
print(f"增: {dt:.0f}ms, op={resp['operation_id'][:8]}")
# 增: 54ms

# 2. 等异步完成（轮询 operation）
for _ in range(10):
    time.sleep(5)
    st, _ = api("GET", f"/v1/default/banks/{BANK}/operations/{resp['operation_id']}")
    if st["status"] == "completed":
        break

# 3. 查：list + recall
items, dt = api("GET", f"/v1/default/banks/{BANK}/memories/list?limit=50")
print(f"查 list: {dt:.0f}ms, {len(items['items'])} 条")
rec, dt = api("POST", f"/v1/default/banks/{BANK}/memories/recall",
              {"query": "Hindsight 批量写入测试", "limit": 5})
print(f"查 recall: {dt:.0f}ms, {len(rec['results'])} hits")

# 4. 改：PATCH document tags
docs, _ = api("GET", f"/v1/default/banks/{BANK}/documents")
doc_id = docs["documents"][0]["id"]
_, dt = api("PATCH", f"/v1/default/banks/{BANK}/documents/{doc_id}",
            {"tags": ["topic:code", "stage:decision", "demo:patched"]})
print(f"改: {dt:.0f}ms")

# 5. 删：DELETE document（级联）
del_resp, dt = api("DELETE", f"/v1/default/banks/{BANK}/documents/{doc_id}")
print(f"删: {dt:.0f}ms, {del_resp.get('memory_units_deleted')} 条级联删除")

# 6. 清理：删整个 bank
api("DELETE", f"/v1/default/banks/{BANK}")
print("清理完成")

增: 54ms
查 list: 56ms, 12 条
查 recall: 281ms, 13 hits
改: 75ms
删: 68ms, 10 条级联删除
清理完成