开放式视频理解一直比单段视频处理难难在两个地方一是视频没有固定结局实体可能长时间反复出现二是对象状态会变化同一个球会变旧、位移、被遮挡需要在持续输入中维护一份不断更新的“世界状态”。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 参考架构可以分成四层每一层只解决一个大问题感知层读取视频流按策略抽帧处理图像分辨率、帧率、色彩空间。抽取层对每一帧做目标检测、跟踪或者用多模态模型生成实体描述和属性。记忆层把抽取结果合并到实体记忆库中负责实体身份识别、状态更新、时间线追加、关系维护。应用层提供查询接口支持按实体 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-python4.8.1.78 ultralytics8.0.207 redis5.0.1 faiss-cpu1.7.4 numpy1.24.4 pyyaml6.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_factorydict) dataclass class EntityState: entity_id: str category: str last_seen: float latest_bbox: List[float] attributes: Dict[str, str] field(default_factorydict) 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, verboseFalse) 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_idtemp_track_id, categorycategory, timestamptimestamp, frame_idframe_idx, bboxbbox, 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_identity_id, categoryobservation.category, last_seenobservation.timestamp, latest_bboxobservation.bbox, attributesdict(observation.attributes), history_count1, ) 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 存储组件选型实体记忆系统会同时面对三种数据因此不能只用一种存储。实践中我会这样区分职责存储组件负责数据选型原因Redis实体最新状态、属性 KV、时间线 ZSet读写延迟低适合高频状态更新SQLite / PostgreSQL事件日志、轨迹表、关系表事务能力强支持复杂条件查询FAISS / 向量数据库实体视觉特征向量支持相似实体检索、重识别召回这三个组件在演示项目里可以都跑在单机生产环境再拆分。先不要引入过重的分布式系统否则排查问题时很难分清是算法问题还是基础设施问题。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, dim512): 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_k5): 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.yaml6.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_count1 [2025-01-01 10:00:06] entity person_1001 state updated, history_count2如果日志中实体 ID 一直变化说明跟踪或 ID 分配逻辑有问题。需要立刻停止后检查tracker_id的稳定性。6.3 用几个问题验证记忆是否正确验证阶段建议设计一组“记忆测试题”测试问题查询接口预期结果当前有哪几个实体处于活跃状态get_recent_active_entities返回最近出现的实体 ID 列表实体 person_1001 的当前坐标是多少get_entity_state返回最新bboxperson_1001 在过去 10 秒内出现过几次get_entity_timeline返回事件列表count1找到与色块 A 最相似的实体find_similar返回色块 A 自身或类似物体如果这些测试都不能稳定通过说明记忆链路还没有真正打通不要急着接复杂模型。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应该逐帧更新记忆后释放引用。下面是问题现象与处理方案的速查表问题现象常见原因检查方式处理建议实体 ID 漂移跟踪器未接入或跟踪丢失检查连续帧 track ID接入 ByteTrack加入重识别状态被旧时间覆盖乱序帧、并发写打印时间戳和更新顺序比较时间戳再更新使用原子脚本向量召回不准特征模型不匹配检查类别过滤和距离值先按类别过滤升级特征模型内存暴涨观察列表未释放监控进程内存改为流式处理事件无法回溯只存最新状态不存事件检查事件表记录数建事件表追加写入8. 生产环境落地的建议和扩展方向8.1 从演示到生产要补哪些能力演示代码里的MemoryCore是内存字典适合理解原理但不适合长期运行。生产环境至少还需要补齐下面几项配置外置化不要把 Redis 地址、模型路径写死在代码里至少使用环境变量或配置中心。日志和监控记录实体更新数量、检测延迟、错误率并设置告警。权限和安全模型服务接口不要暴露在公网存储层需要最小权限账号。回滚方案模型或配置变更后要能快速回到上一个可用版本。数据备份Redis 和 SQLite 都需要定期备份事件表尤其重要。对于机器人或监控场景还需要设计“记忆持久化”和“记忆清理”。不是所有实体都需要永久保存过期实体可以归档到冷存储避免活跃状态库无限增长。8.2 模型层演进从闭集检测到开放文本描述YOLO 只能识别固定类别而开放世界视频需要发现不在预定义列表里的实体。下一步可以接入多模态模型让检测和描述变成开放文本。比如使用 Grounding DINO 做开放词汇检测使用 CLIP 对检测区域生成文本属性再把这些文本属性写入实体记忆的attributes字段。在多模态模型接入后实体 ID 的稳定性会更容易维护因为可以结合文本描述和视觉特征判断两个观察是否属于同一实体。但这个方向的代价是推理延迟更高生产环境需要设置异步队列避免视频扫描被模型推理阻塞。8.3 可复用清单实体记忆系统上线检查清单每次上线或升级前建议按这份清单逐项检查[ ] 实体 ID 是否稳定是否经过跟踪器关联和重识别验证。[ ] 状态更新是否按时间戳比较是否是原子操作。[ ] 事件表是否只追加不更新是否有按实体和时间创建的索引。[ ] 向量索引维度与特征模型输出维度是否一致。[ ] 实体状态和事件日志是否有备份和恢复方案。[ ] 视频抽帧频率是否与业务需求匹配是否控制了内存上限。[ ] 查询接口是否做了候选集过滤是否限制了返回数量。[ ] 是否有日志记录实体更新次数、延迟和异常。[ ] 生产配置是否通过环境变量或配置中心管理不硬编码。[ ] 模型服务是否具备灰度发布和回滚能力。做完这些检查实体导向记忆系统才算是真正具备上线条件。ReflectWorld 代表的并不是某一个固定算法而是一种把视频分析从“帧图集”升级到“实体世界状态”的设计思路。实际落地时优先级应当是先保证实体身份稳定再丰富状态属性和关系最后再考虑大规模并发和模型升级。先把最小闭环跑通再逐步扩展这套系统会在开放视频任务中成为非常实用的基础设施。