# Chapter 03 — 统一数据模型 (SpatialMemory Schema)

> 本章目标：给出 **可直接复制运行** 的 Python schema、磁盘目录布局、JSON/USD 序列化规范，作为 PRISM 全系统的"数据宪法"。

---

## 3.1 设计原则

| 原则 | 含义 |
|------|------|
| **单一真理源** | 所有层共享 `SpatialMemory` 一个根对象；不允许"iPhone 数据库 + ZED 数据库"并立 |
| **自描述** | 每个节点带 `source`/`level`/`confidence`/`timestamps`，断网恢复后能自解释 |
| **可演化** | Schema 加字段不破坏旧数据；用 `schema_version` 标记 |
| **可序列化** | JSON 用于人读 + Git diff；HDF5/PLY/GLB 用于大数据；USD 用于与 Omniverse/Isaac 对接 |
| **可索引** | 关键查询（"附近的""相似的""最近见到的"）都有 O(log n) 索引支持 |

---

## 3.2 完整 Python Schema

```python
# spatial_memory/schema.py
"""PRISM Spatial Memory — canonical schema v1.0"""
from __future__ import annotations
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional, Literal, Any, Tuple
from enum import Enum
import numpy as np
import time
import uuid

SCHEMA_VERSION = "1.0.0"

# ────────────────────── 基础类型 ──────────────────────

@dataclass
class Pose:
    """SE(3) 位姿；统一在 `map` 帧；右手系，z 向上"""
    position: np.ndarray              # (3,) float32  [x,y,z]  单位 m
    quaternion: np.ndarray            # (4,) float32  [w,x,y,z]  单位四元数
    frame_id: str = "map"

    def to_matrix(self) -> np.ndarray:
        # 返回 4×4 齐次矩阵
        from scipy.spatial.transform import Rotation as R
        T = np.eye(4)
        T[:3, :3] = R.from_quat(
            [self.quaternion[1], self.quaternion[2],
             self.quaternion[3], self.quaternion[0]]).as_matrix()
        T[:3, 3] = self.position
        return T


class MemoryLevel(str, Enum):
    L1 = "L1"   # 感知缓冲
    L2 = "L2"   # 度量
    L3 = "L3"   # 拓扑
    L4 = "L4"   # 语义


class Source(str, Enum):
    IPHONE = "iphone"
    ZED2I  = "zed2i"
    VLM    = "vlm"
    FUSED  = "fused"
    HUMAN  = "human"   # 人工标注/纠正


# ────────────────────── 节点 ──────────────────────

@dataclass
class SpatialNode:
    """统一节点：可表示房间(L3)/家具(L4)/物品(L4)/路点(L2,可选)

    > ⚠️ v1.5 更新：本类新增 `keyframe_evidence: List[KeyframeEvidence]` 字段，
    >    用于保留 per-frame 独立"翻案证据"。完整动机、字段定义、存储预算与
    >    retention 策略详见 § 3.2.1（紧随本 Schema 块之后的新增小节）。
    >    旧字段语义不变；本注解仅作为升级提示，原 v1.0 字段全部保留。
    """
    # ── 标识 ──
    uid: str
    label: str
    level: MemoryLevel
    source: Source
    confidence: float = 1.0           # 0~1

    # ── 几何 ──
    pose: Optional[Pose] = None
    bbox_3d: Optional[np.ndarray] = None        # (8,3) OBB 顶点
    polygon_2d: Optional[np.ndarray] = None     # (N,2) 仅 L3 房间地面用
    mesh_uri: Optional[str] = None              # 相对 robot_memory/ 的路径
    gaussians_uri: Optional[str] = None

    # ── 语义 ──
    category: Optional[str] = None              # 'wall','furniture','small_item','room'
    attributes: Dict[str, Any] = field(default_factory=dict)
    # attributes 常用键：
    #   color, material, state(on/off/open/closed),
    #   mobile: bool,  fragile: bool,
    #   is_anchor: bool,   no_update_zone: bool
    clip_embedding: Optional[np.ndarray] = None   # (512,) float16

    # ── 时序 ──
    first_seen: float = field(default_factory=time.time)
    last_seen:  float = field(default_factory=time.time)
    observation_count: int = 1

    # ── 关联 ──
    parent_uid: Optional[str] = None              # L4 物品挂在哪件家具上
    parent_room: Optional[str] = None             # 反向：所属 L3 房间
    children: List[str] = field(default_factory=list)

    @staticmethod
    def new_uid(prefix: str = "n") -> str:
        return f"{prefix}_{uuid.uuid4().hex[:8]}"


# ────────────────────── 边 ──────────────────────

RelationT = Literal[
    "contains", "in",              # 容纳
    "on", "under",                 # 支撑
    "next_to", "front_of", "behind",  # 邻接
    "connects_to", "reachable_from",  # L3 通行
    "plugged_into", "inside_drawer",  # 特殊
]

@dataclass
class SpatialEdge:
    src_uid: str
    dst_uid: str
    relation: RelationT
    weight: float = 1.0                 # 距离/通行成本/置信度
    source: Source = Source.FUSED
    timestamp: float = field(default_factory=time.time)


# ────────────────────── 稠密层（不入节点）──────────────────────

@dataclass
class DenseLayerRefs:
    """L2 稠密表示：用文件 URI 引用，不放内存"""
    occupancy_grid_uri: Optional[str] = None   # .bt (OctoMap)
    tsdf_uri: Optional[str] = None             # .vbg / .npz
    global_mesh_uri: Optional[str] = None      # .glb
    global_3dgs_uri: Optional[str] = None      # .ply (gsplat 格式)
    prior_mask_uri: Optional[str] = None       # .npy (1=iPhone 保护)
    no_update_zone_uri: Optional[str] = None   # .npy (1=镜面)


# ────────────────────── 锚点 ──────────────────────

@dataclass
class Anchor:
    """用于 ZED 上线时与先验地图配准的锚点"""
    anchor_uid: str                   # 引用 SpatialNode.uid
    node_label: str                   # 'bed','tv',...
    is_mobile: bool                   # mobile=True 则不作 anchor
    clip_embedding: np.ndarray
    geometric_signature: Dict         # FPFH 直方图等几何特征
    last_validated: float


# ────────────────────── Delta（差异记忆）──────────────────────

@dataclass
class DeltaEvent:
    """ZED 在线发现的、与 LTM 不一致的事件"""
    event_id: str
    event_type: Literal["object_moved", "object_removed",
                        "object_added", "geometry_changed"]
    target_uid: Optional[str]         # 涉及的 LTM 节点（若有）
    new_pose: Optional[Pose] = None
    new_bbox: Optional[np.ndarray] = None
    evidence: List[str] = field(default_factory=list)  # 关键帧 ID 列表
    observation_count: int = 1
    first_observed: float = field(default_factory=time.time)
    last_observed:  float = field(default_factory=time.time)
    status: Literal["pending","confirmed","rejected","applied"] = "pending"


# ────────────────────── 根对象 ──────────────────────

@dataclass
class SpatialMemory:
    schema_version: str = SCHEMA_VERSION
    world_frame: str = "map"
    origin_description: str = "scan start point of iPhone session 1"
    gravity: np.ndarray = field(default_factory=lambda: np.array([0,0,-9.81]))

    nodes: Dict[str, SpatialNode] = field(default_factory=dict)
    edges: List[SpatialEdge] = field(default_factory=list)
    dense: DenseLayerRefs = field(default_factory=DenseLayerRefs)
    anchors: List[Anchor] = field(default_factory=list)
    deltas: List[DeltaEvent] = field(default_factory=list)

    # —— 索引（运行时构建，不序列化）——
    _kdtree: Any = None               # 空间近邻
    _faiss:  Any = None               # CLIP 向量

    # —— 便捷方法 ——
    def add_node(self, node: SpatialNode) -> None:
        assert node.uid not in self.nodes, f"duplicate uid {node.uid}"
        self.nodes[node.uid] = node

    def add_edge(self, edge: SpatialEdge) -> None:
        assert edge.src_uid in self.nodes
        assert edge.dst_uid in self.nodes
        self.edges.append(edge)

    def nodes_of_level(self, lvl: MemoryLevel) -> List[SpatialNode]:
        return [n for n in self.nodes.values() if n.level == lvl]

    def nodes_in_room(self, room_uid: str) -> List[SpatialNode]:
        return [n for n in self.nodes.values() if n.parent_room == room_uid]
```

---

## 3.2.1 v1.5 新增：KeyframeEvidence（per-frame 独立证据）

### 动机

v1.4 之前 PRISM 的 L2 是**融合表示**——TSDF 把每帧深度加权累计到体素、OctoMap 把每条 ray 更新到八叉树占据概率。融合的代价是**累计误差不可回溯**：一旦 L2 在长走廊里漂了 30 cm，"第 t=12.3 s 那一帧 ZED 看到桌子在哪"就再也拿不回来——它已经被融进了几百万个体素的加权平均里。

借鉴 Lyra 2.0 § 3.2(a)"**3D 缓存绝不融合（never fuse）**"的设计，v1.5 让每个 L3 节点额外保留 ≤ 5 个独立 keyframe 作为"翻案证据"：当下游 L4 发现 bbox 估计与节点对不齐、或巩固期需要回溯某个语义改变是否真实发生时，可以**绕过已融合的 L2，直接从原始 keyframe 重新估计**。完整原则陈述与风险分析见 [`18_lyra_inspirations.md` § 18.3](18_lyra_inspirations.md)。

注意：这不是把 TSDF 扔掉，而是让 TSDF（路由 + 避障）与 keyframe 列表（翻案证据）**并存**——二者承担不同任务、对几何精度有不同容忍度。

### 完整 `KeyframeEvidence` Dataclass

```python
# spatial_memory/schema.py  (v1.5 新增,接在原 SpatialNode 之后)
@dataclass
class KeyframeEvidence:
    """
    每个 L3 节点保留的 per-frame 独立证据。
    一律 append-only,绝不被 L2 融合操作覆盖(immutable=True)。

    设计要点:
      - lazy load: depth_path / rgb_path 只存路径,原始张量留磁盘
      - 视觉特征预先算好 CLIP 嵌入并 inline(只有 1.5 KB,索引快)
      - visibility_score 是 retention 决策的唯一依据(见下文 retention 策略)
    """
    # ── 标识 ──
    kf_id:            str                       # 全局唯一,匹配 stm/keyframes/{kf_id}/
    ts:               float                     # 采集时间戳 (epoch sec)
    source_pipeline:  Literal["C_online", "A_offline", "D_consolidation"]
                                                # 该证据由哪条管线产生

    # ── 几何 ──
    T_cam_world:      np.ndarray                # (4,4) float32 相机→世界 SE(3)
    pose_uncert:      Optional[np.ndarray] = None  # (6,6) 协方差,可空

    # ── 内容引用(lazy load) ──
    rgb_path:         str = ""                  # 相对 robot_memory/ 的 JPEG 路径
    depth_path:       str = ""                  # 相对路径,uint16 PNG(mm 单位)
    intrinsics:       Optional[np.ndarray] = None  # (3,3) 该帧内参 K

    # ── 内联特征(小,索引友好) ──
    clip_emb:         Optional[np.ndarray] = None  # (768,) float16, ~1.5 KB

    # ── 路由元数据 ──
    routed_l3:        Optional[str] = None      # 当时路由到的 L3 节点 uid
    routed_l4:        List[str] = field(default_factory=list)  # 关联 L4 uid 列表
    visibility_score: float = 0.0               # [0,1] 该帧对节点的可见度/信息量
                                                # 用于 retention 时挑选最差帧替换

    # ── 不变性保护 ──
    fused_into_l2:    bool = False              # 是否曾参与 L2 TSDF 融合
    immutable:        bool = True               # 写入后禁止修改;违反者 raise
```

字段从 [`18_lyra_inspirations.md` § 18.3](18_lyra_inspirations.md) 的 `KeyframeEvidence`
扩展而来：保留 Lyra 风格的 append-only 与 lazy-load 语义，并补齐了 PRISM
专用的 `rgb_path / intrinsics / visibility_score / source_pipeline` 字段以贴合
"L3 节点级别保留"这一更具体的工程目标。

### `SpatialNode` 字段扩展示例

在原 `SpatialNode` 定义（§ 3.2 中段，第 72 行起）的末尾新增**一个字段**，其他字段全部保留：

```python
@dataclass
class SpatialNode:
    # ... 原 v1.0 所有字段保持不变(uid/label/level/source/confidence/
    #     pose/bbox_3d/.../parent_room/children) ...

    # ── v1.5 新增 ──
    keyframe_evidence: List[KeyframeEvidence] = field(default_factory=list)
    # 最多保留 5 条,append-only;满后用 visibility_score 替换最差(见下)
```

序列化时 `keyframe_evidence` 走与 `clip_embedding` 同样的"大向量外链"策略：
JSON 中仅留 `kf_id + rgb_path + depth_path + visibility_score + ts` 等元数据，
`clip_emb` 落 `embeddings/{kf_id}.clip.npy`。

### 存储预算估算

按典型酒店楼层（500 L3+L4 节点、每节点保留 5 帧）做单帧 + 总量两级估算：

| 项 | 编码 | 单帧大小 | 备注 |
|----|------|----------|------|
| RGB | 320×240 JPEG q=85 | ~30 KB | ZED/iPhone 下采样后足够做事后翻案 |
| Depth | 320×240 uint16 PNG | ~50 KB | mm 单位,压缩率 ~3× |
| CLIP 嵌入 | 768D float16 | 1.5 KB | inline 进 JSON 索引 |
| 内参 + 位姿 + meta | JSON | ~0.5 KB | T_cam_world + intrinsics + pose_uncert |
| **单帧合计** | — | **~82 KB** | 取整 ≈ 80 KB |

| 维度 | 数值 | 总量 |
|------|------|------|
| 每节点保留帧数 | 5 | 5 × 80 KB ≈ **400 KB / 节点** |
| 节点数（典型一层） | 500 | 500 × 400 KB ≈ **200 MB / 楼层** |
| 节点数（大型场馆 5 层） | 2500 | 2500 × 400 KB ≈ **1.0 GB / 整馆** |
| 旁通存储（NVMe SSD） | — | 1 GB 完全可接受（snapshots/ 同盘） |

结论：**单层 ~200 MB，整馆 ~1 GB**，相比 `dense/octomap.bt`（典型 50–200 MB）
与 `dense/3dgs.ply`（典型 100 MB–1 GB / 房间）属于同量级，**没有引入新数量
级的存储瓶颈**。若磁盘吃紧，可把 `rgb_path` 进一步压到 160×120 JPEG
（~12 KB / 帧）把整馆压到 ~250 MB。

### Retention 策略（满 5 帧后如何替换）

`keyframe_evidence` 容量上限默认 **5 帧/节点**。新证据写入时若已满，按以下规则替换：

```python
def admit_evidence(node: SpatialNode, new_kf: KeyframeEvidence,
                   capacity: int = 5) -> None:
    """append-only 语义下的 admission control:
       capacity 未满则直接 append;满则用 visibility_score 替换最差帧。
       注意:被替换的 KeyframeEvidence 在 Pipeline D 巩固期归档到
       robot_memory/snapshots/ 而非硬删,保留可追溯性。
    """
    if len(node.keyframe_evidence) < capacity:
        node.keyframe_evidence.append(new_kf)
        return
    # 已满:挑当前 visibility_score 最低的一帧
    worst_idx = min(range(capacity),
                    key=lambda i: node.keyframe_evidence[i].visibility_score)
    if new_kf.visibility_score > node.keyframe_evidence[worst_idx].visibility_score:
        archive_to_snapshot(node.keyframe_evidence[worst_idx])  # 归档,不硬删
        node.keyframe_evidence[worst_idx] = new_kf
    # 否则新帧也不如最差帧好,直接丢弃(由 Pipeline D 决定是否进 snapshots/)
```

`visibility_score ∈ [0, 1]` 的计算遵循 **geometry-aware retrieval**
（Lyra 2.0 § 3.2(b)）：综合 (a) 该帧对节点 OBB 的覆盖面积比例、
(b) 视角与已有保留帧的角度差异、(c) 深度有效像素占比，三者加权平均。
直觉是"保留信息互补、视角多样、深度可信的 5 帧"，而不是"最近 5 帧"。

回链：本小节落实了 [`18_lyra_inspirations.md` § 18.3](18_lyra_inspirations.md)
"原则二：Per-frame 独立 keyframe 证据"在数据 schema 上的承接。
[`13_evaluation.md`](13_evaluation.md) 后续需新增"翻案命中率 / 翻案误报率"
两项指标来闭环验证本字段的工程价值。

---

## 3.3 序列化规范

### 3.3.1 JSON（人读 + Git 友好）

```python
# spatial_memory/io_json.py
import json
import numpy as np
from .schema import SpatialMemory, SpatialNode, SpatialEdge

def _np_encoder(o):
    if isinstance(o, np.ndarray):
        return {"__ndarray__": True, "dtype": str(o.dtype),
                "shape": list(o.shape), "data": o.flatten().tolist()}
    if isinstance(o, (np.float32, np.float16)):
        return float(o)
    raise TypeError(f"non-serializable: {type(o)}")

def _np_decoder(d):
    if d.get("__ndarray__"):
        return np.array(d["data"], dtype=d["dtype"]).reshape(d["shape"])
    return d

def save(mem: SpatialMemory, path: str) -> None:
    from dataclasses import asdict
    blob = asdict(mem)
    # 大向量不入 JSON：替换为 .npy 外链
    for uid, node in blob["nodes"].items():
        if node["clip_embedding"] is not None:
            np.save(f"{path}.{uid}.clip.npy", node["clip_embedding"])
            node["clip_embedding"] = {"__npy__": f"{uid}.clip.npy"}
    with open(path, "w") as f:
        json.dump(blob, f, indent=2, default=_np_encoder)

def load(path: str) -> SpatialMemory:
    with open(path) as f:
        blob = json.load(f, object_hook=_np_decoder)
    # 反向恢复 dataclass
    ...
    return SpatialMemory(**blob)
```

### 3.3.2 USD（与 Omniverse / Isaac Sim / Polycam 互通）

每个 L4 节点 → 一个 USD `Xform` prim；几何挂在子 prim：

```
/World/Hotel
    /Room_301        (Xform, custom attr: room_label="Bedroom")
        /Bed_301     (Xform, ref=bed.usd)
        /TV_301      (Xform, ref=tv.usd)
    /Hallway_3F
        ...
```

转换器：

```python
# spatial_memory/io_usd.py
from pxr import Usd, UsdGeom, Gf

def memory_to_usd(mem: SpatialMemory, usd_path: str):
    stage = Usd.Stage.CreateNew(usd_path)
    world = UsdGeom.Xform.Define(stage, "/World")

    # L3 房间作为父 Xform
    for room in mem.nodes_of_level(MemoryLevel.L3):
        room_xform = UsdGeom.Xform.Define(stage, f"/World/{room.uid}")
        room_xform.GetPrim().CreateAttribute(
            "prism:label", Sdf.ValueTypeNames.String).Set(room.label)

        for obj in mem.nodes_in_room(room.uid):
            obj_xform = UsdGeom.Xform.Define(
                stage, f"/World/{room.uid}/{obj.uid}")
            T = obj.pose.to_matrix()
            obj_xform.AddTransformOp().Set(Gf.Matrix4d(T.tolist()))
            if obj.mesh_uri:
                obj_xform.GetPrim().GetReferences().AddReference(obj.mesh_uri)

    stage.GetRootLayer().Save()
```

### 3.3.3 二进制（生产部署）

为了在 Jetson 上加载快，关键张量走二进制：

| 数据 | 格式 | 工具 |
|------|------|------|
| 点云 | `.ply` / `.las` | Open3D |
| TSDF | `.vbg` | Open3D VoxelBlockGrid |
| 3DGS | `.ply` (gsplat 标准布局) | gsplat / nerfstudio |
| OctoMap | `.bt` | octomap-cpp |
| Mesh | `.glb` (gltf 2.0) | trimesh |
| CLIP 向量库 | `.faiss` | Faiss |

JSON 仅存**索引和元数据**，数据走文件引用。

---

## 3.4 磁盘目录布局（落盘规范）

```
robot_memory/
├── manifest.json                # 总入口，含 schema_version & 各文件 SHA256
├── ltm/                         # 长期记忆（iPhone 主导，写入后近乎只读）
│   ├── spatial_memory.json      # SpatialMemory 主体（不含大向量）
│   ├── embeddings/
│   │   └── {uid}.clip.npy       # 每节点 CLIP 向量
│   ├── meshes/
│   │   ├── room_301.glb
│   │   └── bed_301.glb
│   ├── pointcloud/
│   │   └── room_301.ply
│   ├── dense/
│   │   ├── octomap.bt
│   │   ├── tsdf.vbg
│   │   ├── 3dgs.ply
│   │   ├── prior_mask.npy
│   │   └── no_update_zone.npy
│   ├── anchors.json
│   └── usd/
│       └── hotel.usdz           # 给 Omniverse / 仿真用
│
├── stm/                         # 短期/工作记忆（运行时，环形覆盖）
│   ├── current_pose.txt         # 单行最新位姿
│   ├── trajectory.tum           # TUM 格式轨迹（追加写）
│   ├── keyframes/
│   │   └── {ts}_{idx}/
│   │       ├── rgb.jpg
│   │       ├── depth.png
│   │       └── meta.json
│   ├── live_octomap.bt          # 实时局部
│   └── live_tsdf.vbg
│
├── delta/                       # 差异记忆（待巩固）
│   ├── pending.jsonl            # 一行一个 DeltaEvent
│   ├── confirmed.jsonl
│   └── rejected.jsonl
│
├── snapshots/                   # 版本化历史 LTM
│   ├── 2026-05-16_v1.tar.zst
│   └── 2026-06-01_v2.tar.zst
│
└── logs/
    ├── relocalize.log
    ├── consolidation.log
    └── metrics.parquet          # 评测指标时间序列
```

`manifest.json` 示例：

```json
{
  "schema_version": "1.0.0",
  "world_frame": "map",
  "created_at": "2026-05-16T10:00:00+08:00",
  "scene_name": "Hotel-Demo-Floor3",
  "ltm_version": 5,
  "files": {
    "ltm/spatial_memory.json": {
      "sha256": "ab12...",
      "size_bytes": 1048576
    },
    "ltm/dense/octomap.bt": { "sha256": "...", "size_bytes": 52428800 }
  }
}
```

---

## 3.5 数据库映射（生产部署用 Neo4j）

原型期可用 JSON + NetworkX，生产期建议把 L3+L4 进 Neo4j（节点 + 关系），稠密数据仍走文件：

```cypher
// 创建房间节点
MERGE (r:Room {uid: 'room_301'})
SET r.label='Bedroom',
    r.center=point({x:1.2, y:3.4, z:0.0}),
    r.confidence=0.95;

// 创建家具节点 + 关系
MERGE (b:Furniture {uid: 'bed_301'})
SET b.label='bed', b.mobile=false,
    b.pose_x=2.0, b.pose_y=3.5, b.pose_z=0.3;

MERGE (r)-[:CONTAINS]->(b);

// 小物品 on 家具
MERGE (rc:Item {uid: 'remote_xyz'})
SET rc.label='remote_control', rc.mobile=true;
MERGE (rc)-[:ON {confidence:0.85}]->(b);
```

LLM Agent 查询时可以直接发 Cypher：
```cypher
MATCH (i:Item {label:'remote_control'})-[:ON]->(f)<-[:CONTAINS]-(r:Room)
RETURN i, f, r;
```

---

## 3.6 版本化与迁移

```python
# spatial_memory/migrate.py
def migrate(blob: dict) -> dict:
    v = blob.get("schema_version", "0.0.0")
    if v == "1.0.0":
        return blob
    if v == "0.9.0":
        # 0.9 → 1.0：把 'class' 字段重命名为 'category'
        for node in blob["nodes"].values():
            node["category"] = node.pop("class", None)
        blob["schema_version"] = "1.0.0"
        return migrate(blob)
    raise ValueError(f"Unsupported schema version: {v}")
```

每次 schema 升级 → 写一个迁移函数 + 在 `snapshots/` 留备份。

---

## 3.7 一致性校验

写完任何节点都跑：

```python
def validate(mem: SpatialMemory) -> List[str]:
    errs = []
    for uid, node in mem.nodes.items():
        if uid != node.uid:
            errs.append(f"uid mismatch: {uid} vs {node.uid}")
        if node.level == MemoryLevel.L4 and node.parent_room is None:
            errs.append(f"L4 node {uid} missing parent_room")
        if node.confidence < 0 or node.confidence > 1:
            errs.append(f"{uid} confidence out of [0,1]")
    for edge in mem.edges:
        if edge.src_uid not in mem.nodes:
            errs.append(f"dangling edge src: {edge.src_uid}")
        if edge.dst_uid not in mem.nodes:
            errs.append(f"dangling edge dst: {edge.dst_uid}")
    return errs
```

CI 里跑 `validate()` 防止 schema 退化。

---

## 3.8 API 设计原则（给上层 Agent）

不要让 Agent 直接访问 `mem.nodes`；提供高层方法：

```python
class SpatialMemoryAPI:
    def find(self, text: str, top_k: int = 5) -> List[SpatialNode]: ...
    def locate(self, query: SpatialNode) -> Pose: ...
    def neighbors(self, uid: str, radius: float = 1.0) -> List[SpatialNode]: ...
    def path_rooms(self, src_room: str, dst_room: str) -> List[str]: ...
    def changes_since(self, t: float) -> List[DeltaEvent]: ...
    def describe(self, uid: str) -> str:
        """生成自然语言描述供 LLM 消化"""
        node = self.nodes[uid]
        room = self.nodes[node.parent_room]
        return f"A {node.attributes.get('color','')} {node.label} " \
               f"in {room.label}, last seen {ago(node.last_seen)}."
```

→ Agent 看不见底层格式变更，只用 API。

---

## 3.9 本章小结

| 关键约定 | 一句话 |
|----------|--------|
| **根对象** | `SpatialMemory`（含 nodes/edges/dense/anchors/deltas） |
| **节点** | `SpatialNode`（带 level + source + confidence + 时间戳） |
| **边** | `SpatialEdge`（含 relation + weight） |
| **稠密数据** | 不入节点，走文件 URI |
| **磁盘** | 三主目录：`ltm/` `stm/` `delta/` + `snapshots/` |
| **数据库** | 原型 NetworkX/JSON，生产 Neo4j |
| **序列化** | JSON 主，二进制大数据走外链；USD 用于仿真桥接 |
| **版本** | `schema_version` + 迁移函数 |

读完本章你应能：
- ✅ 把 `schema.py` 复制进项目就开始写代码
- ✅ 知道一份 LTM 在磁盘上长什么样
- ✅ 给 LLM Agent 提供安全的 API

下一章 [`04_pipeline_A_iphone_offline.md`](04_pipeline_A_iphone_offline.md) 用这套 schema 实现"iPhone 扫描 → SpatialMemory 灌入"的完整管线。

---

**章节版本**：v1.0
**估计阅读时间**：15 分钟
**关键收获**：拿到可立即使用的 dataclass + 磁盘布局