拆开 AI 记忆引擎的完整老底：Hindsight 数据架构全解

┌────────────────────────────────────────────────────────────────────┐
│                    HINDSIGHT 0.8.0 数据架构总览                        │
├────────────────────────────────────────────────────────────────────┤
│                                                                    │
│  WRITE PATH (retain)            READ PATH (recall)                 │
│  ┌──────────────┐               ┌──────────────┐                   │
│  │ POST /memories│               │ POST /memories│                  │
│  │              │               │  /recall      │                  │
│  └──────┬───────┘               └──────┬───────┘                   │
│         ▼                             ▼                            │
│  ┌──────────────┐              ┌──────────────────┐                │
│  │ chunk 分块    │              │ query embedding   │               │
│  │ delta 增量    │              │ (bge-m3, 1024维)  │               │
│  └──────┬───────┘              └──────┬───────────┘                │
│         ▼                             ▼                            │
│  ┌──────────────┐        ┌──────────────────────────────┐          │
│  │ LLM 事实抽取  │        │ 四通道并行检索                  │         │
│  │ experience/  │        │ ① semantic (HNSW)            │         │
│  │ world 双语    │        │ ② bm25 (tsvector/4后端)      │         │
│  └──────┬───────┘        │ ③ graph (实体扩展)            │         │
│         ▼                │ ④ temporal (时间约束)         │         │
│  ┌──────────────┐        └──────┬───────────────────────┘          │
│  │ 实体解析       │               ▼                            │
│  │ trigram 匹配  │        ┌──────────────┐                       │
│  └──────┬───────┘        │ RRF 融合       │                       │
│         ▼                │ (k=60/轮转)    │                       │
│  ┌──────────────┐        └──────┬───────┘                        │
│  │ embedding    │               ▼                                 │
│  │ 增强后编码    │        ┌──────────────┐                         │
│  └──────┬───────┘        │ Cross-Encoder │                        │
│         ▼                │ 精排          │                         │
│  ┌──────────────┐        └──────┬───────┘                        │
│  │ 写入 memory_ │               ▼                                 │
│  │ units + links│        ┌──────────────┐                         │
│  └──────┬───────┘        │ Token 截断    │                         │
│         ▼                └──────────────┘                         │
│  ┌──────────────┐                                                 │
│  │ consolidation│      MAINTENANCE PATH (async)                   │
│  │ 异步合并      │      ┌─────────────────────────┐               │
│  │ observation  │      │ graph_maintenance       │               │
│  └──────────────┘      │ relink 图谱自愈          │               │
│                        └─────────────────────────┘               │
└────────────────────────────────────────────────────────────────────┘

documents（根：原始文本 + retain 参数）
  └── chunks（分块：content_hash 去重）
        └── memory_units（事实：experience / world / observation）
              ├── unit_entities（事实 ↔ 实体 多对多）
              ├── memory_links（事实 ↔ 事实 图谱边：temporal/semantic/entity/causal）
              ├── entity_cooccurrences（实体共现计数）
              └── observation_history（observation 变更审计）

一次 retain 调用（一条 content）
    │
    ├─→ chunk（按 retain_chunk_size 切块，可能多块）
    ├─→ LLM 抽取 → experience fact（中文事实）
    ├─→ LLM 翻译 → world fact（英文事实，双语召回）
    ├─→ 实体解析 → entities（trigram 模糊匹配挂实体表）
    ├─→ 建图谱边 → memory_links（4 种边类型）
    └─→ 异步合并 → observation（去重精炼后的派生事实，带 source_memory_ids 追溯）

你调了一次 retain("Hindsight recall 延迟 201ms")
  │
  ├─ 写：chunk → DeepSeek 抽取 → pg_trgm 实体 → bge-m3 编码
  │      → INSERT experience + world → 建 3 种图谱边
  │      → 异步 consolidate → observation（合并后）
  │
  ├─ 读（后来某天）：recall("上周的 Hindsight 性能")
  │      → bge-m3 编码查询 → dateparser 解析"上周"
  │      → semantic HNSW + bm25 + graph + temporal 四路并行
  │      → RRF 融合 → MiniLM 精排 → 返回 observation（135-270ms）
  │
  ├─ 改：PATCH document tags → 传播到全部关联 facts
  │      → 旧 observation 失效 → 重新合并
  │
  └─ 删：DELETE document → 捕获 unit_ids → relink 受害者入队
         → CASCADE 删树 → stale sweep → 图谱后台自愈

 fact_type  | count
------------+-------
 observation |  1148
 experience  |   755
 world       |   709
(3 rows)

 link_type | count
-----------+-------
 temporal  | 19288
 semantic  |  3477
 caused_by |    96
(3 rows)

        Column          |           Type           | Nullable | Default
-------------------------+--------------------------+----------+-----------
 id                      | uuid                     | not null | gen_random_uuid()
 bank_id                 | text                     | not null |
 document_id             | text                     |          |
 text                    | text                     | not null |
 embedding               | vector(1024)             |          |
 context                 | text                     |          |
 event_date              | timestamp with time zone |          |
 occurred_start          | timestamp with time zone |          |
 occurred_end            | timestamp with time zone |          |
 mentioned_at            | timestamp with time zone |          |
 fact_type               | text                     | not null | 'world'
 access_count            | integer                  | not null | 0
 metadata                | jsonb                    | not null | '{}'
 created_at              | timestamp with time zone | not null | now()
 updated_at              | timestamp with time zone | not null | now()
 chunk_id                | text                     |          |
 tags                    | character varying[]      | not null | '{}'
 proof_count             | integer                  |          | 1
 source_memory_ids       | uuid[]                   |          | ARRAY[]::uuid[]
 consolidated_at         | timestamp with time zone |          |
 observation_scopes      | jsonb                    |          |
 text_signals            | text                     |          |
 search_vector           | tsvector                 |          |
 consolidation_failed_at | timestamp with time zone |          |

-- 语义通道：全量 + 部分 HNSW（按 fact_type 拆三个）
CREATE INDEX idx_memory_units_embedding_hnsw ON memory_units USING hnsw (embedding vector_cosine_ops);
CREATE INDEX idx_mu_emb_obsv ON memory_units USING hnsw (embedding vector_cosine_ops)
    WHERE fact_type = 'observation' AND bank_id = 'hermes';
CREATE INDEX idx_mu_emb_worl ON memory_units USING hnsw (embedding vector_cosine_ops)
    WHERE fact_type = 'world' AND bank_id = 'hermes';
CREATE INDEX idx_mu_emb_expr ON memory_units USING hnsw (embedding vector_cosine_ops)
    WHERE fact_type = 'experience' AND bank_id = 'hermes';

-- 全文通道：GIN tsvector
CREATE INDEX idx_memory_units_text_search ON memory_units USING gin (search_vector);

-- 时间通道：3 个部分索引
CREATE INDEX idx_memory_units_bank_occurred_start ON memory_units
    (bank_id, fact_type, occurred_start) WHERE occurred_start IS NOT NULL;
CREATE INDEX idx_memory_units_bank_occurred_end ON memory_units
    (bank_id, fact_type, occurred_end) WHERE occurred_end IS NOT NULL;
CREATE INDEX idx_memory_units_bank_mentioned_at ON memory_units
    (bank_id, fact_type, mentioned_at) WHERE mentioned_at IS NOT NULL;

-- 类型/归属通道
CREATE INDEX idx_memory_units_bank_fact_type ON memory_units (bank_id, fact_type);
CREATE INDEX idx_memory_units_bank_date ON memory_units (bank_id, event_date DESC);
CREATE INDEX idx_memory_units_document_id ON memory_units (document_id);
CREATE INDEX idx_memory_units_chunk_id ON memory_units (chunk_id);

-- 合并通道
CREATE INDEX idx_memory_units_consolidation_failed ON memory_units
    (bank_id, consolidation_failed_at)
    WHERE consolidation_failed_at IS NOT NULL AND fact_type = ANY (ARRAY['experience','world']);

-- 双向索引（graph_maintenance 的 victim 查询依赖它）
CREATE INDEX idx_memory_links_from_type_weight ON memory_links (from_unit_id, link_type, weight DESC);
CREATE INDEX idx_memory_links_to_type_weight ON memory_links (to_unit_id, link_type, weight DESC);

    Column    |           Type           | Nullable | Default
--------------+--------------------------+----------+---------
 chunk_id     | text                     | not null |
 document_id  | text                     | not null |
 bank_id      | text                     | not null |
 chunk_index  | integer                  | not null |
 chunk_text   | text                     | not null |
 created_at   | timestamp with time zone | not null | now()
 content_hash | text                     |          |

Indexes:
    "pk_chunks" PRIMARY KEY, btree (chunk_id)
    "idx_chunks_bank_id" btree (bank_id)
    "idx_chunks_document_id" btree (document_id)

Foreign-key constraints:
    "chunks_document_fkey" FOREIGN KEY (document_id, bank_id)
        REFERENCES documents(id, bank_id) ON DELETE CASCADE
Referenced by:
    TABLE "memory_units" CONSTRAINT "memory_units_chunk_fkey"
        FOREIGN KEY (chunk_id) REFERENCES chunks(chunk_id) ON DELETE CASCADE

┌────────────────────────────────────────────────────────────────┐
│                    fact_type 三类事实分工                         │
├────────────────────────────────────────────────────────────────┤
│                                                                │
│  experience（原始经历）                                         │
│  ├─ 生产者：retain 时 LLM 从 content 抽取                        │
│  ├─ 内容：中文原始事实，如 "Hindsight v0.8.0 recall 延迟 201ms"  │
│  ├─ 特点：最接近原始输入，不精炼，可能重复                        │
│  ├─ 消费：作为 consolidation 的输入源                            │
│  └─ 生命周期：consolidated_at 标记后等待被合并                    │
│                                                                │
│  world（世界知识 / 双语版）                                      │
│  ├─ 生产者：LLM 自动把 experience 翻译成英文                     │
│  ├─ 内容："Hindsight v0.8.0 recall latency is 201ms"           │
│  ├─ 特点：与 experience 一一对应，实现双语召回                    │
│  ├─ 消费：召回时英文查询也能命中中文记忆                          │
│  └─ 生命周期：随 experience 一起被 consolidation 处理            │
│                                                                │
│  observation（观察 / 精炼知识）                                  │
│  ├─ 生产者：consolidation 引擎 LLM 合并多条源事实               │
│  ├─ 内容："Hindsight v0.8.0 召回性能 201ms，比 v0.7.1 快 3 倍"  │
│  ├─ 特点：去重后的共识知识，带 source_memory_ids 溯源             │
│  ├─ 消费：召回主通道（召回优先 observation）                     │
│  └─ 生命周期：可被更新/删除（LLM 裁决），可审计                   │
│                                                                │
└────────────────────────────────────────────────────────────────┘

-- consolidation 更新 observation 的时间边界（真实 SQL）
UPDATE memory_units
SET occurred_start = LEAST(occurred_start, COALESCE($6, occurred_start)),   -- 最早开始
    occurred_end   = GREATEST(occurred_end, COALESCE($7, occurred_end)),   -- 最晚结束
    mentioned_at   = GREATEST(mentioned_at, COALESCE($8, mentioned_at))    -- 最近提及
WHERE id = $5

# entity_resolver.py 策略选择（源码）
# "trigram" 用 pg_trgm GIN 索引只拉相似候选；"full" 全量扫描
# Oracle 后端用 UTL_MATCH
# 自动检测 pg_trgm 是否可用：
has_trgm = await conn.fetchval("SELECT EXISTS(SELECT 1 FROM pg_extension WHERE extname = 'pg_trgm')")
if not has_trgm:
    # fall back to 'full' entity lookup strategy

# 拉低阈值避免漏匹配（10k+ 实体的 bank 上 LIKE 全表扫描会超时）
await conn.execute("SET pg_trgm.similarity_threshold = 0.15")
# 用 % 操作符走 GIN 索引
# ... 执行候选查询 ...
await conn.execute("RESET pg_trgm.similarity_threshold")

LLM 抽取的实体名（"Hindsight", "CRUD"）
  │
  ├─ trigram 候选查询（% 相似度）
  │    ├─ 命中 → 复用已有 entity，mention_count++
  │    └─ 未命中 → 创建新 entity
  │
  └─ 写入 unit_entities（多对多）
       fact_id ↔ entity_id

resolve_entities_batch(entities, bank_id, conn)
  │
  ├─ 检查 pg_trgm 是否可用（自动检测）
  │    ├─ 可用 → trigram 策略（GIN 索引加速）
  │    └─ 不可用 → full 策略（全量扫描，Oracle 用 UTL_MATCH）
  │
  ├─ trigram 策略细节：
  │    SET pg_trgm.similarity_threshold = 0.15
  │    对每个实体名 → 候选查询（% 相似度，走 GIN 索引）
  │    命中 → 返回已有 entity_id
  │    未命中 → 返回 None（待创建）
  │    RESET pg_trgm.similarity_threshold   ← 防泄漏到连接池
  │
  ├─ 创建新实体（幂等）
  │    INSERT ... ON CONFLICT DO NOTHING
  │
  └─ 写入 unit_entities（多对多）
       fact_id ↔ entity_id

def augment_texts_with_dates(facts, format_date_fn) -> list[str]:
    """Augment fact texts with readable dates for better temporal matching.

    This allows queries like "camping in June" to match facts that happened in June.
    """
    for fact in facts:
        fact_date = fact.occurred_start or fact.mentioned_at
        if fact_date is not None:
            readable_date = format_date_fn(fact_date)
            if fact.occurred_end and fact.occurred_end != fact.occurred_start:
                augmented_text = f"{fact.fact_text} (happened from {readable_date} to {readable_end})"
            else:
                augmented_text = f"{fact.fact_text} (happened in {readable_date})"
        else:
            augmented_text = fact.fact_text
        if fact.entities:
            augmented_text = f"{augmented_text} [{', '.join(fact.entities)}]"
        augmented_texts.append(augmented_text)

原始 fact：  "Hindsight v0.8.0 recall 延迟降低到 201ms"
增强后：      "Hindsight v0.8.0 recall 延迟降低到 201ms (happened in June 2026) [Hindsight, recall, v0.8.0]"

async def create_temporal_links_batch(conn, bank_id, unit_ids, ops=None):
    """Create temporal links between facts.
    Links facts that occurred close in time to each other."""

async def create_semantic_links_batch(conn, bank_id, unit_ids, embeddings, pre_computed_ann_links=None, ops=None):
    """Create semantic links between facts.
    Links facts that are semantically similar based on embeddings.
    When pre_computed_ann_links are provided (from Phase 1), they are used
    instead of running ANN queries inside the transaction."""

async def create_causal_links_batch(conn, bank_id, unit_ids, facts, ops=None):
    """Create causal links between facts.
    Links facts that have causal relationships (causes, enables, prevents)."""

- "world": About other people, external events, general knowledge, objective facts
- "assistant": First-person actions, experiences, or observations by the speaker/author
  (e.g., "I changed X", "I discovered Y", "I debugged Z"). Also includes interactions
  with the user (requests, recommendations).

Input: "Alice has 5 years of Kubernetes experience and holds CKA certification.
        She's been leading the infrastructure team since March. By the way,
        she prefers dark roast coffee."

Output:
1. what="Alice has 5 years Kubernetes experience, CKA certified",
   who="Alice", entities=["Alice", "Kubernetes", "CKA"]
2. what="Alice leads the infrastructure team since March",
   who="Alice", entities=["Alice", "infrastructure team"]
3. what="Alice prefers dark roast coffee",
   who="Alice", entities=["Alice"]

T+0ms     POST /memories（HTTP 到达）
T+0ms     retain_batch 分组（按 document_id）
T+10ms    chunk 分块 + delta 比对（SHA256 content_hash）
T+50ms    LLM 抽取（DeepSeek：fact + entities + 时间 + causal）
T+2s      英译 world（DeepSeek）
T+2.5s    实体解析（pg_trgm，0.15 阈值）
T+3s      embedding 增强 + bge-m3 编码
T+3.5s    INSERT memory_units（experience + world）
T+4s      挂 unit_entities
T+5s      temporal/semantic/causal 链接创建
T+6s      提交事务（同步模式在此返回，~18s 含 LLM 全部往返）
T+25s     consolidation 后台触发（异步模式）

retain_batch(contents_dicts, ...)
  │
  ├─ Phase 0: 预处理（_pre_resolve_phase1）
  │    bank profile → narrator → contents 转换 → 按 document_id 分组
  │
  ├─ Phase 1: 分块 + 增量比对（_try_delta_retain）
  │    chunk → SHA256 content_hash → 与已有 chunk 比对 → 只处理变化的
  │
  ├─ Phase 2: 抽取 + 编码（_extract_and_embed）
  │    LLM 抽取 fact + entities + 时间 → bge-m3 编码增强文本
  │
  └─ Phase 3: 写入 + 建边（_insert_facts_and_links）
       INSERT memory_units → 实体解析挂 unit_entities → 建 memory_links

# chunk_storage.py
chunk_id = f"{bank_id}_{document_id}_{chunk_index}"
# 例如: "hermes_7bdc5e7f-bdfc-4493-be75-1931257af76a_6"
# content_hash 用于增量更新：hash 变了才重新抽取

POST /memories（带 document_id）
  │
  ├─ _try_delta_retain（orchestrator.py）
  │    ├─ 读已有 document 的 chunks（按 document_id）
  │    ├─ 计算新内容的分块（chunk_text）
  │    ├─ 对每个 chunk 算 SHA256 content_hash
  │    ├─ 与已有 chunk 的 content_hash 比对
  │    ├─ 已存在 → 跳过（保留原 fact/实体/边）
  │    └─ 新增/变化 → 进入抽取管线
  │
  ├─ _delta_metadata_only（只更新元数据时）
  │    └─ 不重跑抽取，只更新 document 的 tags/metadata
  │
  └─ _streaming_retain_batch（大文档）
       └─ producer 内同样的 chunk_hash 比对（见 3.7）

def compute_temporal_links(new_units, candidates, time_window_hours=24):
    """Compute temporal links between new units and candidate neighbors."""
    for unit_id, unit_event_date in new_units.items():
        if unit_event_date is None:
            continue  # 无事件时间无法建 temporal 边
        # 24h 时间窗口（带 overflow 保护）
        time_lower = unit_event_date_norm - timedelta(hours=time_window_hours)
        time_upper = unit_event_date_norm + timedelta(hours=time_window_hours)
        # 窗口内候选，最多 10 个邻居
        matching_neighbors = [...][:10]
        for recent_id, recent_event_date in matching_neighbors:
            time_diff_hours = abs((unit - recent).total_seconds() / 3600)
            weight = max(0.3, 1.0 - (time_diff_hours / time_window_hours))
            links.append((unit_id, recent_id, "temporal", weight, None))
    return _cap_links_per_unit(links)

weight = max(0.3, 1 - time_diff_hours / 24)

两件事同时发生（diff=0h）  → weight = 1.0
间隔 12 小时                → weight = 0.5
间隔 24 小时（窗口边界）     → weight = 0.3（下限）
间隔超过 24 小时            → 不建边

async def _streaming_retain_batch(...):
    """
    Process a large document in streaming mini-batches to bound memory usage.

    Instead of extracting facts from ALL chunks at once (which can OOM for 17k+
    chunk documents), this splits the pre-chunked content into batches of
    ``chunk_batch_size`` chunks.  Each mini-batch goes through the full
    extract -> embed -> Phase 1/2/3 pipeline and commits to the DB before the
    next batch starts, so memory is released between batches.
    """

async def _llm_producer() -> None:
    async def _extract_one(global_idx, chunk_text) -> None:
        # ... 构建 RetainContent，调 _extract_and_embed（LLM + embedding）...
        await chunk_queue.put((global_idx, content, extracted, processed, chunk_meta, usage))
        # Memory: release the chunk text from the shared list now that it's
        # been extracted and queued. The queued RetainContent holds its own copy.
        all_pre_chunks[global_idx] = ""   # ← 关键：提取完立即释放内存

    for i, chunk_text in enumerate(all_pre_chunks):
        chunk_hash = chunk_storage.compute_chunk_hash(chunk_text)
        if chunk_hash in existing_chunk_hashes:
            all_pre_chunks[i] = ""        # 已提交的 chunk 也不需要保留
            skipped_total += 1
            continue
        tasks.append(asyncio.create_task(_extract_one(i, chunk_text)))

# chunk_index_offset 的注释：
# Without a per-document offset each sub-batch would restart chunk_index at 0,
# so their chunk_ids collide and later sub-batches overwrite earlier chunks —
# leaving only one sub-batch's worth of chunks/memories behind (issue #1888).

{"input_tokens": 3029, "output_tokens": 2628, "total_tokens": 5657, "cached_tokens": 0}

t+5s  status=pending
t+10s status=pending
t+20s status=pending
t+25s status=completed

POST /v1/default/banks/test_crud_20260821/memories
{
  "items": [{
    "content": "Hindsight CRUD 实验：单条写入测试，验证 retain 同步模式",
    "context": "test/crud",
    "tags": ["topic:code", "stage:process"],
    "metadata": {"experiment": "crud-20260821", "step": "1-create"}
  }],
  "async": false
}

{
  "success": true,
  "bank_id": "test_crud_20260821",
  "items_count": 1,
  "async": false,
  "operation_id": null,
  "usage": {
    "input_tokens": 3029,
    "output_tokens": 2628,
    "total_tokens": 5657,
    "cached_tokens": 0
  }
}

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

async def insert_facts_batch(conn, bank_id, facts, ...):
    # Convert tags to JSON string for proper batch insertion
    # (PostgreSQL unnest doesn't handle 2D arrays well)
    ...
    # unnest (PG) vs row-by-row (Oracle) transparently

INSERT INTO documents (id, bank_id, original_text, content_hash, retain_params, tags, created_at, updated_at)
VALUES ($1, $2, ...)
ON CONFLICT (id, bank_id) DO UPDATE
-- 重新摄取文档时 delete + insert，created_at 保持首次时间，updated_at 用 NOW()

recall("Hindsight recall architecture")
  │
  ├─ Stage 0: query embedding（bge-m3, 84ms）
  │    query → 1024 维向量
  │
  ├─ Stage 1: 四通道并行检索（~60-350ms）
  │    ① semantic（HNSW 语义）② bm25（关键词）③ graph（实体扩展）④ temporal（时间）
  │    3 个 fact_type × 通道 = 多路 UNION ALL
  │
  ├─ Stage 2: RRF 融合（2ms）
  │    k=60 倒数排名融合 / interleave 轮转
  │
  ├─ Stage 3: Cross-Encoder 精排（3ms）
  │    MiniLM-L-6-v2 query+doc 联合打分
  │
  └─ Stage 4: Token 截断（4ms）
       贪心选取直到 max_tokens 上限

def build_semantic_arm(self, *, table, cols, fact_type, embedding_param,
                       bank_id_param, fetch_limit, min_similarity,
                       tags_clause="", groups_clause="", extra_where=""):
    return (
        f"(SELECT {cols},"
        f"        1 - (embedding <=> {embedding_param}::vector) AS similarity,"
        f"        NULL::float AS bm25_score,"
        f"        'semantic' AS source"
        f" FROM {table}"
        f" WHERE bank_id = {bank_id_param}"
        f"   AND fact_type = '{fact_type}'"
        f"   AND embedding IS NOT NULL"
        f"   AND (1 - (embedding <=> {embedding_param}::vector)) >= {min_similarity}"
        f"   {tags_clause}{groups_clause}{extra_where}"
        f" ORDER BY embedding <=> {embedding_param}::vector"
        f" LIMIT {fetch_limit})"
    )

QueryAnalyzer (抽象基类)
├── DateparserQueryAnalyzer   → 规则优先，dateparser 兜底（200+ 语言）
└── TransformerQueryAnalyzer  → 规则优先，T5 小模型兜底（80M 参数）

def load(self) -> None:
    """Triggers the real initialization cost (regex tables, timezone data) at
    load time so the first actual recall doesn't pay the cold-start penalty."""
    if self._search_dates is None:
        from dateparser.search import search_dates
        self._search_dates = search_dates
        self._search_dates("today")  # 假调用触发懒加载

# Last week patterns（英西意法德）
if re.search(
    r"\b(last\s+week|la\s+semana\s+pasada|la\s+settimana\s+scorsa|la\s+semaine\s+derni[eè]re|letzte\s+woche)\b",
    query, re.IGNORECASE):
    start = reference_date - timedelta(days=reference_date.weekday() + 7)
    return constraint(start, start + timedelta(days=6))

try:
    results = self._search_dates(query, settings=settings)
except Exception as e:
    logger.warning("dateparser raised %s on query (treating as no temporal constraint): %s", ...)
    return QueryAnalysis(temporal_constraint=None)

prompt = f"""Today is {reference_date}. Extract date range or "none".

June 2024 = 2024-06-01 to 2024-06-30
yesterday = {yesterday} to {yesterday}
last Saturday = {last_saturday} to {last_saturday}
what is the weather = none
{query} ="""

EXPLAIN (ANALYZE) SELECT id FROM memory_units
WHERE bank_id='hermes' AND fact_type='experience'
  AND embedding IS NOT NULL
ORDER BY embedding <=> (SELECT embedding FROM memory_units WHERE fact_type='experience' LIMIT 1)::vector
LIMIT 10;

QUERY PLAN
────────────────────────────────────────────────────────────────
 Limit  (cost=432.17..432.19 rows=10) (actual rows=10.00 loops=1)
   Buffers: shared hit=5186
   InitPlan 1
     ->  Limit  (cost=0.00..0.55 rows=1) (actual rows=1.00 loops=1)
           ->  Seq Scan on memory_units memory_units_1
                 Filter: (fact_type = 'experience')
   ->  Sort  (cost=431.61..433.50 rows=755)   ← 全量排序！
         Sort Key: ((memory_units.embedding <=> ...))
         Sort Method: top-N heapsort  Memory: 26kB
         Buffers: shared hit=5186
         ->  Bitmap Heap Scan on memory_units
               Recheck Cond: (fact_type = 'experience')
               Buffers: shared hit=5183
               ->  Bitmap Index Scan on idx_memory_units_fact_type
Planning Time: 2.343 ms
Execution Time: 8.694 ms

-- 参数化：$1 是预计算的 query embedding，planner 知道是常量
ORDER BY embedding <=> $1::vector   -- 直接走 HNSW，1-3ms
LIMIT 500

semantic/observation: 198 hits, top cosine=0.67
bm25/observation:      0 hits   （纯语义查询无关键词命中）
graph/observation:    58 hits, entity score=3
semantic/world:      131 hits, top cosine=0.62
bm25/world:           33 hits
graph/world:          16 hits
semantic/experience: 164 hits, top cosine=0.59
bm25/experience:      35 hits
graph/experience:     59 hits, entity score=3
────────────────────
合计 694 raw hits

Bi-Encoder（Stage 1）：query 和 doc 各自独立编码 → 余弦
Cross-Encoder（Stage 3）：[CLS] query [SEP] doc [SEP] → 联合打分

recency_boost      = 1 + 0.2 × (recency - 0.5)       # 新鲜度：±10%
temporal_boost     = 1 + 0.2 × (temporal - 0.5)      # 时间约束：±10%
proof_count_boost  = 1 + 0.1 × (proof_norm - 0.5)    # 证据强度：±5%
combined_score = CE_normalized × recency_boost × temporal_boost × proof_count_boost

输入：按 combined_score 降序的候选列表
预算：max_tokens（2048/4096）

算法（贪心）：
  selected = []
  used_tokens = 0
  for candidate in sorted_candidates:      # 按 combined_score 降序
      t = count_tokens(candidate.text)
      if used_tokens + t > max_tokens:
          break                             # 触顶停止
      selected.append(candidate)
      used_tokens += t
  return selected

实测：24 条选中，1974/2048 tokens 使用

{
  "id": "7a376883-50ba-409b-8db1-88df2bb290a2",
  "text": "Hindsight CRUD 实验：批量写入第 2 条，测试 async 模式吞吐与 operation 跟踪",
  "type": "experience",
  "entities": ["Hindsight", "CRUD"],
  "context": "test/crud-batch",
  "occurred_start": "2026-08-21T00:00:00+00:00",
  "occurred_end": "2026-08-21T23:59:59+00:00",
  "mentioned_at": "2026-08-21T14:32:48.838700+00:00",
  "document_id": "90bd443d-1e1d-4e90-8231-cd648da061a2",
  "metadata": {"idx": "2", "step": "4-batch", "experiment": "crud-20260821"},
  "chunk_id": "test_crud_20260821_90bd443d-1e1d-4e90-8231-cd648da061a2_2",
  "tags": ["stage:process", "topic:code"],
  "source_fact_ids": []
}

上篇（首次部署快照）              本篇（当前实例）
generate_query_embedding  84ms  ┐
parallel_retrieval       352ms  ├─ 预热后 dateparser ≈ 0
  └─ temporal extraction  305ms │   规则优先拦截高频表达
rrf_merge                  2ms  ┘   语义+BM25 合并 UNION ALL
reranking                  3ms
token_filtering            4ms
─────────────────────────────────────────────
总计                     472ms      总计 135-270ms

score(d) = Σ_i 1 / (k + rank_i(d))

k = 60（源码默认值）

RRF_score = 1/(60+12) + 1/(60+18) + 1/(60+70)
          = 0.0139 + 0.0128 + 0.0077
          = 0.0344

取 semantic #1 → bm25 #1 → graph #1 → temporal #1
→ semantic #2 → bm25 #2 → graph #2 → temporal #2 → ...
（去重，直到全部排完）

recency_boost      = 1 + 0.2 × (recency - 0.5)       # 新鲜度：±10%
temporal_boost     = 1 + 0.2 × (temporal - 0.5)      # 时间约束：±10%
proof_count_boost  = 1 + 0.1 × (proof_norm - 0.5)    # 证据强度：±5%

combined_score = CE_normalized × recency_boost × temporal_boost × proof_count_boost

{
  'cross_encoder_score_normalized': 0.997,  # 联合语义得分，主导权重
  'recency': 0.983,                         # 近期事实，新鲜度加成
  'temporal': 0.500,                        # 无时间约束，中性
  'rrf_score': 0.033,                       # RRF 得分已被 CE 压制
  'combined_score': 1.093                   # 最终得分
}

class _CreateAction(BaseModel):
    """创建新 observation"""
    text: str
    source_fact_ids: list[str]   # 引用的 NEW FACTS 的 UUID
    reason: str = ""             # LLM 的一句话理由（诊断用）

class _UpdateAction(BaseModel):
    """更新已有 observation"""
    text: str
    observation_id: str          # 已有 observation 的 UUID
    source_fact_ids: list[str]
    reason: str = ""

class _DeleteAction(BaseModel):
    """删除已有 observation（被新事实否定/覆盖）"""
    observation_id: str
    reason: str = ""

新事实（experience/world）
  │
  ├─ 召回相关 observation（并行，只读）
  │
  ├─ 有匹配的已有 observation？
  │    ├─ 是 → 是同一事件/实体/facet？
  │    │     ├─ 是 → UPDATE（挂载新证据，LEAST/GREATEST 合并时间）
  │    │     └─ 否 → CREATE（新 observation）
  │    └─ 否 → CREATE
  │
  ├─ 有被新事实否定的 observation？
  │    └─ 是 → DELETE（仅当 restated identically 或 truly meaningless）
  │
  └─ 输出：_ConsolidationBatchResponse（creates/updates/deletes）

class _DedupDecision(BaseModel): ...      # 去重决策模型
async def _dedup_adjudicate(...): ...     # 裁决：新 fact 是否与已有 observation 重复
async def _dedup_reconcile_create(...): ... # 重复 → 合并进已有
async def _dedup_reconcile_update(...): ... # 重复 → 更新已有

UPDATE memory_units
SET text = $1,
    embedding = $2::vector,
    source_memory_ids = $3,
    proof_count = $4,
    tags = $9,
    updated_at = now(),
    occurred_start = LEAST(occurred_start, COALESCE($6, occurred_start)),   # 最早开始
    occurred_end = GREATEST(occurred_end, COALESCE($7, occurred_end)),      # 最晚结束
    mentioned_at = GREATEST(mentioned_at, COALESCE($8, mentioned_at))       # 最近提及
WHERE id = $5

live_source_memory_ids = await _filter_live_source_memories(conn, bank_id, source_memory_ids)
if not live_source_memory_ids:
    # Update skipped: all source memories were deleted concurrently
    return

写入 5 条高度相似事实（async 批量）
  → 等待写入完成（operation completed）
  → 记录 fact_type 分布
  → POST /banks/{id}/consolidate 手动触发
  → 轮询 operation 状态
  → 60s 后复查 fact_type 分布

POST /v1/default/banks/test_arch_20260821/consolidate
→ 200 {"operation_id": "67b7e48f-6994-4823-bae8-d93dcaceb2b0", "deduplicated": true}

GET /v1/default/banks/test_arch_20260821/operations/67b7e48f...
{
  "status": "pending",          // pending → completed
  "operation_type": "consolidation",
  "created_at": "2026-08-21T15:00:04+00:00",
  "completed_at": null,
  "error_message": null,
  "retry_count": 0,
  "result_metadata": {}
}

max_obs = config.max_observations_per_scope
if max_obs > 0 and fact_tags:
    current_count = await _count_observations_for_scope(conn, bank_id, fact_tags)
    remaining = max(max_obs - current_count, 0)
    if remaining == 0:
        # 到 observation 上限，只允许 updates/deletes（不允许 creates）

consolidation 引擎完整流程
───────────────────────────

  后台定时任务（每 5 分钟）
  │   banks_needing_consolidation() → 找有待处理 fact 的 bank
  ▼
run_consolidation_job
  │   _count_unconsolidated() → 统计未合并的 experience/world
  │
  ├─ 按 tag group 分组（_process_tag_group）
  │    同标签的事实一起处理
  │
  ├─ 分批（_process_memory_batch）
  │    ① 每条 fact 并行召回相关 observation（只读）
  │    ② union 去重
  │    ③ 单次 LLM 调用裁决 create/update/delete
  │    ④ 顺序执行动作（写串行）
  │
  ├─ 执行动作
  │    _execute_create_action → INSERT observation
  │    _execute_update_action → UPDATE + LEAST/GREATEST + 历史
  │    _execute_delete_action → DELETE observation
  │
  ├─ _append_observation_history（审计）
  │
  └─ _trigger_mental_model_refreshes（触发心智模型刷新）

# 所有表都有 bank_id 列，检索 SQL 强制过滤
WHERE bank_id = $2  # 每个查询都带

# memory_engine.py 的 _LLM_REQUEST_COLUMNS
"id, bank_id, operation, scope, trace_id, span_id, parent_span_id, "
"provider, model, status, started_at, ended_at, duration_ms, "
"input_tokens, output_tokens, cached_tokens, total_tokens, "
"input, output, error, llm_info, metadata"

reflect 输入：
  ├─ 记忆库（bank）
  ├─ 查询（query）
  └─ 行为档案（disposition：skepticism / literalism / empathy，1-5 分）

reflect 流程：
  ├─ ① recall 检索相关记忆
  ├─ ② LLM 基于记忆 + 行为档案生成响应
  ├─ ③ 形成/强化新意见（opinion）
  └─ ④ 更新意见网络

{
  "skepticism": 3,   // 怀疑度 1-5：越高越质疑新信息
  "literalism": 3,   // 字面理解度 1-5：越高越按字面理解
  "empathy": 3       // 同理心 1-5：越高越考虑情感
}

docker ps -a
NAMES         STATUS
hindsight     Up 9 seconds                ← 刚被拉起来
vllm          Exited (255) 44 hours ago   ← vllm 44 小时前挂了！

docker inspect hindsight --format 'RestartCount={{.RestartCount}}'
RestartCount=10384   ← 崩溃重启了 10384 次！

docker logs hindsight --tail 25
openai.APIConnectionError: Connection error.
ERROR:    Application startup failed. Exiting.

# ① 先启 vllm，等 health 200（约 80s）
docker start vllm
curl -s http://localhost:8000/health   # 等 200

# ② vllm 就绪后 hindsight 自动恢复（restart=always 拉起）
curl -s http://localhost:8888/health   # 等 200

# ③ 真实 recall 验证（不只是 health 200）
POST /v1/default/banks/hermes/memories/recall
# 命中 40+ 条 → 服务真正可用

# 服务健康（health 200 ≠ 检索可用，必须 recall 验证）
curl -s http://localhost:8888/health
curl -s -X POST http://localhost:8888/v1/default/banks/hermes/memories/recall \
  -H 'Content-Type: application/json' -d '{"query":"test","limit":1}'

# 容器状态（看 RestartCount 是否暴涨）
docker inspect hindsight --format 'RestartCount={{.RestartCount}}'
docker inspect vllm --format 'RestartCount={{.RestartCount}}'

# LLM key 有效性（fact extraction 失败的常见根因）
docker inspect hindsight --format '{{json .Config.Env}}' | grep -o 'LLM_API_KEY=...'

# 磁盘（bge-m3 模型 4.3G 在 NVMe，根分区别放模型）
df -h / /app

async def update_document(self, document_id, bank_id, *, tags=None, request_context):
    """Update mutable fields on a document without re-processing its content.

    Tag changes propagate to all associated memory units and trigger observation
    invalidation + re-consolidation (same semantics as delete_document):
    - Observations referencing the document's memory units are deleted.
    - The document's own units and any co-source memories from other documents
      have consolidated_at reset so they are re-consolidated under the new tags.
    """
    # ① UPDATE documents SET tags（75ms）
    # ② UPDATE memory_units SET tags WHERE document_id（级联传播）
    # ③ 找到引用这些事实的 observation，删除（invalidation）
    # ④ 重置 consolidated_at → 触发重新 consolidation

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

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

# 验证：关联的所有 memory units 的 tags 也被传播
items = api("GET", f"/banks/{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"/banks/{BANK}/documents/{doc_id}/reprocess")
# 返回 {'success': True, 'operation_id': ..., 'items_count': ...}

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

async def delete_document(self, document_id, bank_id, *, request_context):
    async with conn.transaction():
        # ① 先捕获要删的 unit_ids（否则级联后就查不到了）
        unit_rows = await conn.fetch(
            "SELECT id FROM memory_units WHERE document_id = $1 ...")
        unit_ids = [str(row["id"]) for row in unit_rows]

        # ② 先入队 relink 受害者（必须在级联前！）
        if unit_ids:
            await enqueue_relink_victims(conn, bank_id, unit_ids, ops=backend.ops)

        # ③ 删文档（CASCADE 删 memory_units + links）
        deleted = await conn.fetchval(
            "DELETE FROM documents WHERE id = $1 AND bank_id = $2 RETURNING id")

        # ④ 删除后清理 stale observations（在删除之后跑！）
        if unit_ids:
            invalidated_obs = await self._delete_stale_observations_for_memories(
                conn, bank_id, unit_ids)
    # ⑤ 事务外：触发 consolidation + graph maintenance

# ① 有 observation 被失效 → 触发重新合并
if invalidated_obs > 0:
    if config.enable_auto_consolidation:
        await self.submit_async_consolidation(bank_id=bank_id, ...)

# ② 有任何 unit 被删 → 触发图谱维护
if unit_ids:
    try:
        await self.submit_async_graph_maintenance(bank_id=bank_id, ...)
    except Exception as e:
        logger.warning(f"Failed to submit graph maintenance ...: {e}")

# 删除前 graph edges
GET /banks/{id}/graph → 2038 条边

# DELETE document（5 条 memory_units）
DELETE /banks/{id}/documents/{doc_id}
# {'success': True, 'message': "Document '...' and 5 associated memory units deleted successfully"}  32ms

# 删除后 graph edges
GET /banks/{id}/graph → 1809 条边（-229 条！）

DELETE /banks/{id}/memories?type=world
# {'success': True}  30ms
# 之后 list 中 world 全部消失，experience/observation 保留

DELETE /banks/{id}/memories/{mid}/observations
# {'success': True, 'deleted_count': 0}  11ms

{
  "success": true,
  "message": "Document '90bd443d-1e1d-4e90-8231-cd648da061a2' and 10 associated memory units deleted successfully",
  "document_id": "90bd443d-1e1d-4e90-8231-cd648da061a2",
  "memory_units_deleted": 10
}

DELETE /v1/default/banks/test_bank/documents/nonexistent-id
→ 404 {"detail": "Document not found"}（10ms）

-- Entity 扩展：通过共享实体找到候选（简化版）
SELECT DISTINCT mu.*, COUNT(DISTINCT ue.entity_id) AS graph_score
FROM unit_entities ue
JOIN unit_entities seed_ue ON ue.entity_id = seed_ue.entity_id
JOIN memory_units mu ON ue.unit_id = mu.id
WHERE seed_ue.unit_id = ANY($1::uuid[])   -- 语义种子
GROUP BY mu.id
ORDER BY graph_score DESC;

victim_rows = await conn.fetch(
    f"""
    SELECT DISTINCT from_unit_id
    FROM memory_links
    WHERE to_unit_id = ANY($1::uuid[])
      AND bank_id = $2
      AND link_type IN ('temporal', 'semantic')
    """,
    deleted_uuids, bank_id,
)

graph_maintenance_queue
  └── run_graph_maintenance_job（后台）
        └── _relink_batch（分批重连）
              ├─ 重新计算受害者的出边
              ├─ 找新邻居（语义相似/时间邻近/共享实体）
              └─ UPDATE memory_links

时间线
  │
  ├─ T0  用户 DELETE /documents/{id}
  │
  ├─ T1  事务开始
  │       ① 捕获 unit_ids
  │       ② enqueue_relink_victims（受害者入队）
  │       ③ DELETE documents（CASCADE）
  │       ④ stale observation sweep
  ├─ T2  事务提交（主删除完成，API 返回 32ms）
  │
  ├─ T3  后台：submit_async_consolidation（如果有 observation 失效）
  ├─ T4  后台：submit_async_graph_maintenance（图谱维护）
  │        └─ _relink_batch 重连受害者
  │
  └─ T5  图谱最终一致（后台收敛完成）

"""
Hindsight 四条管道端到端演示
用法: python3 hindsight_journey.py
前置: Hindsight 0.8.0 已部署（见 11.1 环境搭建）
"""
import urllib.request, json, time
from collections import Counter

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

def api(method, path, body=None, timeout=180):
    """统一 API 调用（写路径/读路径/删路径共用）"""
    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

def wait_operation(op_id, timeout_s=120):
    """合并路径/异步写路径的任务轮询"""
    for _ in range(int(timeout_s / 5)):
        time.sleep(5)
        st, _ = api("GET", f"/v1/default/banks/{BANK}/operations/{op_id}")
        if st["status"] == "completed":
            return True
        if st["status"] in ("failed", "error"):
            return False
    return False

def fact_distribution():
    """查 fact_type 分布（观察 experience/world/observation 变化）"""
    r, _ = api("GET", f"/v1/default/banks/{BANK}/memories/list?limit=500")
    items = r.get("items", [])
    return dict(Counter(m.get("fact_type", "?") for m in items)), len(items)

print("=" * 60)
print("PHASE 1: 写路径（异步批量 → operation 跟踪）")
print("=" * 60)
items = [
    {"content": f"用户在第 {i} 天部署了 Hindsight 记忆系统，配置了 DeepSeek 和 bge-m3，"
                 f"记录了召回延迟约 200ms，比旧系统快 3 倍",
     "tags": ["topic:code", "stage:process"],
     "metadata": {"journey": "write", "day": str(i)}}
    for i in range(5)
]
resp, dt = api("POST", f"/v1/default/banks/{BANK}/memories", {"items": items, "async": True})
op_id = resp.get("operation_id")
print(f"[写] HTTP {dt:.0f}ms 返回, operation={op_id[:8]}")
wait_operation(op_id)
dist_before, total_before = fact_distribution()
print(f"[写] 完成后: total={total_before}, types={dist_before}")

print()
print("=" * 60)
print("PHASE 2: 读路径（四通道检索 → RRF → CE）")
print("=" * 60)
queries = [
    ("语义", "Hindsight 部署 召回延迟"),
    ("时间", "Hindsight 部署 last week"),
    ("标签", "Hindsight 部署", ["topic:code"]),
]
for label, q, *tags in queries:
    body = {"query": q, "limit": 5}
    if tags:
        body["tags"] = tags[0]
    r, dt = api("POST", f"/v1/default/banks/{BANK}/memories/recall", body)
    hits = len(r.get("results", []))
    print(f"[读] {label}: {dt:.0f}ms, hits={hits}")

print()
print("=" * 60)
print("PHASE 3: 合并路径（手动触发 consolidation → observation 派生）")
print("=" * 60)
resp, _ = api("POST", f"/v1/default/banks/{BANK}/consolidate")
print(f"[合] 触发: {json.dumps(resp, ensure_ascii=False)[:100]}")
# consolidation 是后台任务，等它收敛
time.sleep(60)
dist_after, total_after = fact_distribution()
print(f"[合] 60s 后: total={total_after}, types={dist_after}")
print(f"[合] observation 变化: {dist_before.get('observation', 0)} → {dist_after.get('observation', 0)}")

print()
print("=" * 60)
print("PHASE 4: 删路径（document 级联 → 图谱收缩）")
print("=" * 60)
graph, _ = api("GET", f"/v1/default/banks/{BANK}/graph")
edges_before = len(graph.get("edges", []))
docs, _ = api("GET", f"/v1/default/banks/{BANK}/documents")
if docs.get("documents"):
    doc_id = docs["documents"][0]["id"]
    del_resp, dt = api("DELETE", f"/v1/default/banks/{BANK}/documents/{doc_id}")
    print(f"[删] DELETE doc: {dt:.0f}ms, 级联删 {del_resp.get('memory_units_deleted')} 条事实")
    graph2, _ = api("GET", f"/v1/default/banks/{BANK}/graph")
    edges_after = len(graph2.get("edges", []))
    print(f"[删] 图谱边: {edges_before} → {edges_after} ({- (edges_before - edges_after)} 条)")

print()
print("=" * 60)
print("PHASE 5: 清理（删整个 test bank）")
print("=" * 60)
clean, dt = api("DELETE", f"/v1/default/banks/{BANK}")
print(f"[清] {json.dumps(clean, ensure_ascii=False)[:120]} ({dt:.0f}ms)")

print()
print("全部四条管道执行完毕 ✅")

# ① 启动 vLLM（bge-m3 embedding，端口 8000）
docker run -d --name vllm --network host --device nvidia.com/gpu=all --restart=always \
  -e HF_ENDPOINT=https://hf-mirror.com \
  ghcr.1ms.run/yuyirobotlab/vllm-orin:0.19.0 \
  --model /app/hf-cache/bge-m3-models --runner pooling --port 8000 \
  --gpu-memory-utilization 0.3 --enforce-eager --max-model-len 8192 \
  --served-model-name BAAI/bge-m3
# 等 /health 200（约 60-80s）

# ② 启动 Hindsight 0.8.0-slim（端口 8888）
docker run -d --name hindsight --network host --restart=always \
  -v DATA_DIR:容器内pg数据目录 \
  --env-file /app/hindsight/env.list \
  ghcr.1ms.run/vectorize-io/hindsight-api:0.8.0-slim
# 等 /health 200（约 38s）

HINDSIGHT_API_LLM_PROVIDER=deepseek
HINDSIGHT_API_LLM_MODEL=deepseek-v4-flash
HINDSIGHT_API_LLM_BASE_URL=https://api.deepseek.com/v1
HINDSIGHT_API_EMBEDDINGS_PROVIDER=openai
HINDSIGHT_API_EMBEDDINGS_OPENAI_BASE_URL=http://localhost:8000/v1
HINDSIGHT_API_EMBEDDINGS_OPENAI_MODEL=BAAI/bge-m3
HINDSIGHT_API_RERANKER_PROVIDER=rrf
HINDSIGHT_API_RETAIN_EVERY_N_TURNS=20
HINDSIGHT_API_HOST=0.0.0.0
HINDSIGHT_API_PORT=8888

import urllib.request, json, time

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

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:
        return json.loads(r.read().decode()), (time.time()-t0)*1000

# 同步单条（预期 ~18s，返回 usage）
resp, dt = api("POST", f"/v1/default/banks/{BANK}/memories", {
    "items": [{"content": "Hindsight 复现实验：同步写入测试",
               "tags": ["topic:code", "stage:process"],
               "metadata": {"repro": "sync"}}],
    "async": False,
})
print(f"sync: {dt:.0f}ms usage={resp.get('usage')}")

# 异步批量 10 条（预期 HTTP ~20-60ms，后台 ~25s）
resp, dt = api("POST", f"/v1/default/banks/{BANK}/memories", {
    "items": [{"content": f"批量写入第 {i} 条", "tags": ["topic:code", "stage:process"],
               "metadata": {"repro": "batch", "i": str(i)}} for i in range(10)],
    "async": True,
})
op_id = resp["operation_id"]
print(f"async batch: {dt:.0f}ms op={op_id[:8]}")
for _ in range(10):
    time.sleep(5)
    st, _ = api("GET", f"/v1/default/banks/{BANK}/operations/{op_id}")
    if st["status"] == "completed":
        print("后台完成")
        break

# 语义召回（预期 150-270ms）
r, dt = api("POST", f"/v1/default/banks/{BANK}/memories/recall",
            {"query": "Hindsight 记忆引擎 存储与检索设计", "limit": 5})
print(f"recall: {dt:.0f}ms hits={len(r.get('results', []))}")

# 时间约束（预期比无约束快或相近）
r, dt = api("POST", f"/v1/default/banks/{BANK}/memories/recall",
            {"query": "Hindsight 记忆引擎 last week", "limit": 5})
print(f"recall+last week: {dt:.0f}ms")

# tags 硬过滤（预期无关 tag 命中 0）
r, dt = api("POST", f"/v1/default/banks/{BANK}/memories/recall",
            {"query": "Hindsight 记忆引擎", "limit": 50, "tags": ["topic:business"]})
print(f"recall+无关tag: hits={len(r.get('results', []))}（预期 0）")