开放式视频理解核心：实体导向记忆系统设计与实践

开放式视频理解一直比单段视频处理难，难在两个地方：一是视频没有固定结局，实体可能长时间反复出现；二是对象状态会变化，同一个球会变旧、位移、被遮挡，需要在持续输入中维护一份不断更新的“世界状态”。ReflectWorld 提出的解决思路是 Entity-oriented memory system，也就是以实体为中心组织记忆，而不是以帧或片段为中心。下面我会从零设计一个用于 open-ended video 的实体导向记忆系统，说明它的核心数据结构、存储方案、检索接口，以及实际工程里最常踩到的几个坑。

这套方案不依赖任何特定厂商的多模态推理服务，只使用常见开源组件。核心逻辑围绕一个思想展开：视频分析系统不应该只输出检测框和时间戳，而要把“实体是谁、实体现在怎么样、实体的历史轨迹是什么”作为一等公民来建模。理解了这一点，后续的存储设计、查询设计和模型选型都会变得更清晰。

1. 先把问题说清楚：为什么开放式视频需要实体记忆

1.1 单段视频处理和开放式视频的本质差异

一段短视频通常只有几秒到几分钟，所有对象几乎都在画面里出现并结束，分析任务可以一次处理完。开放式视频则不同，它指持续采集、没有预设结局的视频流，比如机器人视野、监控长视频、直播回放或剧情连贯的连续视频。

这种差异会直接改变系统设计目标：

在开放式视频里，同一个物理对象往往会被摄像头拍到很多次，但每次出现时外观、位置、姿态都可能变化。如果系统只保存检测结果，不做实体身份合并和状态更新，那么第二次出现时就会把它当成一个新对象，导致记忆碎片化，后续无法回答“这个物体什么时候出现过”这类基本问题。

1.2 实体导向记忆系统是什么

实体导向记忆系统的核心定义是：把视频中识别出的每个独立对象作为记忆主体，为每个主体维护状态、时间线、属性变化，以及与其他主体的关系。系统不直接回答“第120帧里有什么”，而是回答“那把红色椅子的颜色变化发生在哪些时间段”“这个人物和那个背包的共现关系是什么”。

一个实体可以包含以下记忆维度：

• 身份：稳定的实体 ID，比如person_1001、object_car_2233。
• 属性：当前和历史的属性值，比如颜色、位置、速度、类别。
• 状态：最新状态和状态有效期。
• 轨迹：一段时间内的位置序列或出现时间序列。
• 关系：与其他实体的空间关系和行为关系，比如“站在旁边”“正在使用”。

这种建模方式和传统“视频摘要”或“视频问答”不同。视频摘要仍然以帧或镜头为索引，而实体记忆把索引建立在实体上，所有查询都围绕实体展开。

1.3 适用场景和前置知识

实体导向记忆系统适合以下场景：

• 长时间行为分析：分析人物进入、离开、停留、交互的时间线。
• 视频生成控制：给生成模型提供一致的实体状态，避免每帧重新生成不同角色。
• 机器人和智能体感知：让体持续记录“我看到了什么”“发生了什么变化”。
• 内容库管理：对大量视频素材进行实体级索引，后续可以按实体搜索片段。

学习这套设计需要的基础知识并不高，主要是 Python、基本的 Redis 使用、关系型数据库表设计，以及调用一次预训练目标检测模型的经历。即使没有多模态模型基础，也可以先使用 YOLO 类别检测跑通整个链路，再逐步替换成更复杂的开放词汇模型。

2. 总体架构：从视频流到实体记忆要经过哪些层

2.1 分层设计

一个可行的 ReflectWorld 参考架构可以分成四层，每一层只解决一个大问题：

1. 感知层：读取视频流，按策略抽帧，处理图像分辨率、帧率、色彩空间。
2. 抽取层：对每一帧做目标检测、跟踪，或者用多模态模型生成实体描述和属性。
3. 记忆层：把抽取结果合并到实体记忆库中，负责实体身份识别、状态更新、时间线追加、关系维护。
4. 应用层：提供查询接口，支持按实体 ID、属性、时间范围、相似特征进行检索。

分层的主要原因是让模型迭代和存储迭代互不影响。今天用 YOLO 做人脸检测，明天换成 SAM 做分割，记忆层的接口可以保持不变。反过来，今天用 Redis 做状态存储，明天迁移到分布式数据库，抽取层也不需要改。

2.2 数据流设计

用文字描述的数据流如下：

视频输入流 -> 抽帧模块 -> 目标检测/多模态抽取 -> 跟踪与实体身份匹配 -> 实体状态更新器 -> 结构化记忆（Redis + SQLite） -> 向量特征索引（FAISS） -> 查询服务 API

这里最关键的一步是“跟踪与实体身份匹配”。检测模型只负责找到物体，跟踪器负责判断当前物体是否是之前出现过的实体。如果这一步做不好，记忆库里的实体数量会暴涨，状态更新会互相覆盖，后面所有查询都会失真。

2.3 项目目录结构参考

下面是一个适合中小型项目起步的目录结构。它把感知、抽取、记忆、查询四个部分拆成独立模块，便于单独调试。

reflectworld/ config/ config.yaml ingest/ video_scanner.py frame_sampler.py extract/ detector.py tracker.py descriptor.py memory/ entity.py memory_core.py vector_index.py storage/ redis_store.py sqlite_store.py query/ query_service.py tests/ test_memory_core.py requirements.txt

实际项目中，你不需要完全照搬目录，但至少要保持“抽取层不直接写查询代码”“记忆层不依赖具体检测模型”这两个边界。这样后续替换模型或存储引擎时，改动范围都能控制在一个包内。

3. 用最小可运行代码实现实体抽取与状态更新

3.1 依赖准备

先安装基础依赖。下面这份requirements.txt只包含演示需要的最小集合：

opencv-python==4.8.1.78 ultralytics==8.0.207 redis==5.0.1 faiss-cpu==1.7.4 numpy==1.24.4 pyyaml==6.0

注意：不同操作系统的 OpenCV 编译版本差异较大，如果安装失败，可以先只安装opencv-python-headless，在服务端环境更合适。ultralytics会自动下载 YOLO 模型权重，首次运行需要网络连接。

3.2 定义实体对象

实体对象是记忆系统的最小信息单元。在 Python 中，我会使用dataclass定义稳定的实体结构，避免直接使用字典导致字段名散落各处：

from dataclasses import dataclass, field from typing import Dict, List, Optional @dataclass class EntityObservation: entity_id: str category: str timestamp: float frame_id: int bbox: List[float] attributes: Dict[str, str] = field(default_factory=dict) @dataclass class EntityState: entity_id: str category: str last_seen: float latest_bbox: List[float] attributes: Dict[str, str] = field(default_factory=dict) history_count: int = 0

这里区分了两个概念：Observation是某一帧的即时观察，State是记忆库维护的合并状态。不要把两者混在一起。状态可能来自多次观察的累积，比如“最新位置”来自最新帧，而“最早出现时间”来自第一次出现。

3.3 从视频帧中产生观察

使用 YOLOv8 做目标检测的示例代码如下。这段代码负责读取视频、按间隔抽帧、检测目标、构造EntityObservation。这里的tracker_id只是临时跟踪 ID，它还不能直接作为实体 ID，因为跟踪器在目标离开画面后可能丢失身份。

import cv2 from ultralytics import YOLO model = YOLO("yolov8n.pt") def scan_video(video_path: str, sample_interval: int = 5): cap = cv2.VideoCapture(video_path) observations = [] frame_idx = 0 while True: ret, frame = cap.read() if not ret: break if frame_idx % sample_interval == 0: results = model(frame, verbose=False) timestamp = frame_idx / 30.0 for box in results[0].boxes: category = model.names[int(box.cls)] bbox = [round(v, 2) for v in box.xyxy[0].tolist()] temp_track_id = f"{category}_{frame_idx}_{int(box.id[0]) if box.id is not None else len(observations)}" obs = EntityObservation( entity_id=temp_track_id, category=category, timestamp=timestamp, frame_id=frame_idx, bbox=bbox, attributes={"confidence": round(float(box.conf[0]), 3)}, ) observations.append(obs) frame_idx += 1 cap.release() return observations

这段代码的问题也很明显：temp_track_id没有稳定性。想让实体 ID 稳定，需要依赖跟踪器，比如 ByteTrack、DeepSORT 或 BoT-SORT，而不是每帧重新生成 ID。在实际项目中，建议在检测结果之后接一个跟踪器，把同一段的检测框关联到同一个 track ID。

3.4 状态更新逻辑

记忆状态更新的核心逻辑是：如果实体已存在，则合并属性、追加历史；如果不存在，则创建新实体。下面是一个极简内存版MemoryCore，用字典模拟状态存储，方便理解主流程：

class MemoryCore: def __init__(self): self.states = {} def update(self, observation: EntityObservation): entity_id = observation.entity_id if entity_id not in self.states: self.states[entity_id] = EntityState( entity_id=entity_id, category=observation.category, last_seen=observation.timestamp, latest_bbox=observation.bbox, attributes=dict(observation.attributes), history_count=1, ) else: state = self.states[entity_id] state.last_seen = max(state.last_seen, observation.timestamp) state.latest_bbox = observation.bbox state.attributes.update(observation.attributes) state.history_count += 1 def get_state(self, entity_id: str): return self.states.get(entity_id)

这个实现的核心是“合并属性时使用更新而不是覆盖”。如果观察中只有置信度，不使用update直接整体赋值，就会丢掉之前的属性信息。另一个细节是last_seen使用max而不是无条件赋值，这能避免乱序帧把时间倒退。

3.5 为什么不能用帧 ID 作主键

很多新手会把实体主键直接设为frame_id + category，这会导致同一个物体在不同帧中被当成不同实体。实体 ID 的本质是“世界对象的稳定标识”，它应该跨越帧和片段存在。只有跟踪算法或重识别模型才能提供这种稳定身份。

如果你刚开始做原型，可以先接受“过拟合”的实体 ID，但要在系统里预留一个reid_engine接口，后续接入更好的身份匹配。这个接口建议接收两个实体的视觉特征和时空信息，返回是否为同一实体的置信度。

4. 记忆存储设计：表、字段、索引如何配合

4.1 存储组件选型

实体记忆系统会同时面对三种数据，因此不能只用一种存储。实践中我会这样区分职责：

这三个组件在演示项目里可以都跑在单机，生产环境再拆分。先不要引入过重的分布式系统，否则排查问题时很难分清是算法问题还是基础设施问题。

4.2 Redis 数据结构设计

在 Redis 中，每个实体可以用一个 Hash 保存最新状态，用一个 ZSet 保存时间线。示例设计如下：

实体状态 Hash: key: entity:{entity_id} field: category, latest_bbox, last_seen, history_count, ... value: 对应属性值 实体出现时间 ZSet: key: entity:{entity_id}:timeline member: "{frame_id}:{event_type}" score: 时间戳

使用 ZSet 保存时间线的好处是按时间范围切片非常容易，比如查询第 10 秒到第 20 秒之间实体出现了多少次。使用 Hash 保存状态时，字段的更新使用HSET或HMSET，适合高频部分更新。

4.3 SQLite 事件表设计

事件日志用于保存不能丢失的、不可变的明细数据。设计一个entity_events表，每次观察都插一条记录：

CREATE TABLE IF NOT EXISTS entity_events ( event_id INTEGER PRIMARY KEY AUTOINCREMENT, entity_id TEXT NOT NULL, category TEXT NOT NULL, timestamp REAL NOT NULL, frame_id INTEGER NOT NULL, bbox_x1 REAL, bbox_y1 REAL, bbox_x2 REAL, bbox_y2 REAL, attributes TEXT, created_at TEXT DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_entity_events_entity_time ON entity_events(entity_id, timestamp);

这里的关键决定是：entity_events只追加，不更新。状态可以覆盖，事件不能覆盖。一旦需要回溯“某个实体在某个时间点当时的位置”，只有事件表能给出可靠答案。

4.4 向量索引设计

实体视觉特征使用 FAISS 保存。构造索引时，需要保证每个向量都绑定实体 ID，否则检索出向量后无法对应记忆状态：

import numpy as np import faiss class VectorIndex: def __init__(self, dim=512): self.index = faiss.IndexFlatIP(dim) self.entity_ids = [] def add(self, entity_id: str, feature: np.ndarray): self.index.add(feature.reshape(1, -1)) self.entity_ids.append(entity_id) def search(self, feature: np.ndarray, top_k: int = 5): distances, indices = self.index.search(feature.reshape(1, -1), top_k) results = [] for dist, idx in zip(distances[0], indices[0]): if idx == -1: continue results.append({ "entity_id": self.entity_ids[idx], "distance": float(dist), }) return results

实际使用中要小心entity_ids与 FAISS 索引的顺序一致性。一旦删除实体，索引和列表都要同时维护，否则会出现错位。生产环境建议使用支持过滤和持久化的向量数据库，比如 Milvus 或 Qdrant。

5. 检索层设计：实体查询和跨时间查询怎么写

5.1 检索类型划分

实体记忆系统的查询需求大致可以分成四类：

• 状态查询：“实体当前在哪里，状态是什么。”
• 历史查询：“实体在过去 5 分钟内的轨迹是怎样的。”
• 相似查询：“找到最像这个实体的其他实体。”
• 关系查询：“这个实体和哪些实体同时出现过。”

不同类型对应不同存储组件。状态查询走 Redis，历史查询走 SQLite 或事件表，相似查询走向量索引，关系查询需要额外的共现计数表。

5.2 查询接口示例

下面是一个QueryService的关键实现，它封装了不同查询的读取逻辑：

class QueryService: def __init__(self, redis_store, sqlite_store, vector_index): self.redis = redis_store self.sqlite = sqlite_store self.vector_index = vector_index def get_entity_state(self, entity_id: str): return self.redis.get_state(entity_id) def get_entity_timeline(self, entity_id: str, start: float, end: float): return self.sqlite.query_events(entity_id, start, end) def find_similar(self, feature, top_k=5): return self.vector_index.search(feature, top_k) def get_recent_active_entities(self, minutes: int = 5): return self.redis.get_recent_entities(minutes)

查询接口尽量保持“薄”，不要在这里写复杂的业务规则。真正的业务规则，比如“只有置信度大于 0.5 的检测才进入记忆”，应该在抽取层就过滤掉，而不是在查询层处理。

5.3 查询逻辑中的过滤顺序

当混合使用向量检索和结构化过滤时，推荐顺序是：先用结构化条件缩小候选集，再对候选向量做相似计算。例如查询“类别为椅子且最近 10 分钟出现过的最相似实体”，不要对全量向量库做搜索，而是先从 Redis 拿最近 10 分钟活跃的实体 ID，再在向量索引中限制候选范围。

这样做的原因很简单：向量搜索计算量远大于 Redis Key 查询。先过滤可以显著降低延迟，也避免向量召回结果受无关实体干扰。

6. 端到端运行与验证

6.1 准备测试视频和环境配置

用一段 30 秒左右的简单视频做验证。如果手边没有视频，可以使用 OpenCV 生成一段包含移动色块的合成视频，这样实体行为可控，方便检查记忆是否正确。

在config/config.yaml中保存基础参数：

video: path: "./data/sample.mp4" sample_interval: 5 fps: 30 extract: detector: "yolov8n.pt" confidence_threshold: 0.5 memory: redis_host: "localhost" redis_port: 6379 sqlite_path: "./data/memory.db" embedding_dim: 512

然后运行主入口脚本，把扫描、抽取、状态更新串起来：

python -m reflectworld.ingest.video_scanner --config config/config.yaml

6.2 运行流程和预期输出

一次正确运行的流程应该包括以下日志：

[2025-01-01 10:00:01] frame 0 sampled [2025-01-01 10:00:01] detected object person_1001 at [120, 80, 300, 400] [2025-01-01 10:00:02] entity person_1001 state updated, history_count=1 [2025-01-01 10:00:06] entity person_1001 state updated, history_count=2

如果日志中实体 ID 一直变化，说明跟踪或 ID 分配逻辑有问题。需要立刻停止后检查tracker_id的稳定性。

6.3 用几个问题验证记忆是否正确

验证阶段建议设计一组“记忆测试题”：

如果这些测试都不能稳定通过，说明记忆链路还没有真正打通，不要急着接复杂模型。

7. 常见问题和排查路径

7.1 实体 ID 漂移导致同一物体被当成不同实体

现象：日志中同一个物体的实体 ID 不断变化，比如person_5下一次变成person_9。

可能原因：跟踪器没有跨帧关联，或实体 ID 分配逻辑只依赖帧索引和类别，没有结合视觉特征。

检查方式：

• 看连续帧检测框的 IoU 是否交叠。
• 看跟踪器的输出 track ID 是否稳定。
• 检查MemoryCore.update是否过滤了临时 ID。

处理建议：接入 ByteTrack 等跟踪器，让同一目标的检测框共享 track ID。对于长时间遮挡后重新出现的实体，还需要加入重识别模块，或使用视觉向量距离判断是否与历史实体匹配。

7.2 状态更新乱序和覆盖问题

现象：实体最新位置忽前忽后，历史事件被覆盖。

可能原因：视频流时间戳不稳定，或者多个消费者并发更新同一实体状态。

检查方式：

• 打印同一实体在相邻两次更新时的timestamp。
• 检查 Redis 是否存在并发写覆盖。

处理建议：状态更新前先比较timestamp，只接受时间更新的观察。并发场景下使用 Redis Lua 脚本实现比较并更新，避免原子性问题。

7.3 向量检索召回不准确

现象：查询相似实体时，返回结果与肉眼预期差距很大。

可能原因：特征提取模型不适合当前物体类别，或者向量维度与索引维度不匹配，或者候选集只包含少量相似样本。

检查方式：

• 打印查询向量和候选向量的距离值。
• 验证新增实体后向量索引是否同步更新。
• 查看候选集是否被过多无关类别占据。

处理建议：先按类别过滤，再在类别内做向量搜索。如果还是不准，可以替换视觉特征提取模型，比如从小模型换到 CLIP 或 DINOv2。

7.4 抽帧太密导致内存暴涨

现象：视频处理速度越来越慢，内存持续升高。

可能原因：每帧都做检测，且observations列表一直没有清空。

处理建议：使用生成器或消费者模型，抽帧、检测、更新记忆分别在独立线程中处理。不要在内存中保存所有Observation，应该逐帧更新记忆后释放引用。

下面是问题现象与处理方案的速查表：

8. 生产环境落地的建议和扩展方向

8.1 从演示到生产要补哪些能力

演示代码里的MemoryCore是内存字典，适合理解原理，但不适合长期运行。生产环境至少还需要补齐下面几项：

• 配置外置化：不要把 Redis 地址、模型路径写死在代码里，至少使用环境变量或配置中心。
• 日志和监控：记录实体更新数量、检测延迟、错误率，并设置告警。
• 权限和安全：模型服务接口不要暴露在公网，存储层需要最小权限账号。
• 回滚方案：模型或配置变更后，要能快速回到上一个可用版本。
• 数据备份：Redis 和 SQLite 都需要定期备份，事件表尤其重要。

对于机器人或监控场景，还需要设计“记忆持久化”和“记忆清理”。不是所有实体都需要永久保存，过期实体可以归档到冷存储，避免活跃状态库无限增长。

8.2 模型层演进：从闭集检测到开放文本描述

YOLO 只能识别固定类别，而开放世界视频需要发现不在预定义列表里的实体。下一步可以接入多模态模型，让检测和描述变成开放文本。比如使用 Grounding DINO 做开放词汇检测，使用 CLIP 对检测区域生成文本属性，再把这些文本属性写入实体记忆的attributes字段。

在多模态模型接入后，实体 ID 的稳定性会更容易维护，因为可以结合文本描述和视觉特征判断两个观察是否属于同一实体。但这个方向的代价是推理延迟更高，生产环境需要设置异步队列，避免视频扫描被模型推理阻塞。

8.3 可复用清单：实体记忆系统上线检查清单

每次上线或升级前，建议按这份清单逐项检查：

• [ ] 实体 ID 是否稳定，是否经过跟踪器关联和重识别验证。
• [ ] 状态更新是否按时间戳比较，是否是原子操作。
• [ ] 事件表是否只追加不更新，是否有按实体和时间创建的索引。
• [ ] 向量索引维度与特征模型输出维度是否一致。
• [ ] 实体状态和事件日志是否有备份和恢复方案。
• [ ] 视频抽帧频率是否与业务需求匹配，是否控制了内存上限。
• [ ] 查询接口是否做了候选集过滤，是否限制了返回数量。
• [ ] 是否有日志记录实体更新次数、延迟和异常。
• [ ] 生产配置是否通过环境变量或配置中心管理，不硬编码。
• [ ] 模型服务是否具备灰度发布和回滚能力。

做完这些检查，实体导向记忆系统才算是真正具备上线条件。ReflectWorld 代表的并不是某一个固定算法，而是一种把视频分析从“帧图集”升级到“实体世界状态”的设计思路。实际落地时，优先级应当是先保证实体身份稳定，再丰富状态属性和关系，最后再考虑大规模并发和模型升级。先把最小闭环跑通，再逐步扩展，这套系统会在开放视频任务中成为非常实用的基础设施。