MFlow03-数据模型解析

MFlow03-数据模型解析

从生活故事到代码实现的完整思考路径
源码地址：https://github.com/FlowElement-ai/m_flow

📖 第一部分：用生活故事理解数据结构

故事：侦探事务所的记忆管理系统

想象你经营一家侦探事务所，每天处理大量案件。你需要建立一个记忆系统，让新侦探能快速找到历史经验。

🎭 场景一：一个完整案件（Episode）

案件："周一项目会议争论" ├─ 时间：2024年1月15日 ├─ 参与者：Maria、张经理、李工 ├─ 经过：讨论数据库选型，Maria和张经理发生争执 └─ 结果：暂定PostgreSQL，两周后再议

这个案件就是一个Episode（事件包）—— 它是记忆的最高层单元，包含一个完整的故事。

🔍 场景二：案件的多个角度（Facet）

当新侦探问：“发生了什么？” 你不会简单复述，而是从不同角度组织信息：

周一项目会议争论（Episode） ├─ Facet 1: 决策讨论 │ ├─ 内容：选型PostgreSQL vs MySQL │ └─ 关键点：性能优先 ├─ Facet 2: 人际冲突 │ ├─ 内容：Maria对截止日期不满 │ └─ 关键点：沟通不畅 ├─ Facet 3: 技术评估 │ ├─ 内容：P99延迟要求<500ms │ └─ 关键点：PostgreSQL更合适 └─ Facet 4: 后续计划 ├─ 内容：两周后最终决定 └─ 关键点：需要更多数据

每个Facet（主题）是案件的一个维度或切面。

📍 场景三：具体的事实细节（FacetPoint）

新侦探追问：“P99延迟要求是什么？”

Facet: 技术评估 └─ FacetPoint: "P99延迟目标必须低于500毫秒" ├─ 证据来源：会议纪要第3页 └─ 相关实体：PostgreSQL、性能指标

FacetPoint是原子级的事实点—— 最精确的记忆锚点。

👤 场景四：跨越案件的人（Entity）

一个月后，又有一个关于Maria的案子：

案件A："周一项目会议争论" └─ involves_entity → Maria（角色：后端负责人） 案件B："周五冲刺回顾" └─ involves_entity → Maria（角色：提出问题的人）

通过Entity（实体），你能跨案件追踪同一个人：

Maria（Entity） ├─ same_entity_as → Maria（案件A中的描述） └─ same_entity_as → Maria（案件B中的描述）

这样查询"Maria参与的所有事件"时，系统能把相关案件都找出来。

🎯 第二部分：从0到1的设计推导

现在让我们像设计师一样，从零开始推导这个数据结构。

步骤1：最简单的记录 ❌

# 想法：直接存储文本memory={"text":"周一开会讨论数据库，Maria生气了"}

问题：无法检索，无法组织，无法关联。

步骤2：分类管理 ❌

# 想法：像图书馆一样分类memory={"category":"技术会议","title":"数据库选型讨论","content":"..."}

问题：

• 一个会议可能同时是"技术会议"和"人际冲突"
• 静态分类无法应对复杂场景
• 无法表达关系（Maria参加了会议）

步骤3：引入"事件包"概念 ✅

洞察：人类记忆是以事件为单位组织的。

# 设计：Episode（事件包）classEpisode:name:str# "周一数据库选型会议"summary:str# "讨论PostgreSQL vs MySQL，Maria因截止日期问题不满"status:str# "open" | "closed"

为什么这样设计：

• summary是可检索的字段（会被向量化）
• status表示事件是否已完结
• 一个 Episode 是一个完整的语义单元

步骤4：事件需要分解为多个角度 ✅

问题：一个事件包含多个维度（决策、冲突、技术、计划），如何组织？

传统方案：用子标题

summary=""" ## 决策 选型PostgreSQL ## 冲突 Maria不满 ## 技术 P99 < 500ms """

缺点：

• 结构化信息丢失
• 无法精确检索到"技术评估"这个维度
• LLM容易混淆不同主题

M-flow方案：引入 Facet

classFacet:name:str# "技术评估"facet_type:str# "metric" | "decision" | "risk" ...search_text:str# "性能目标讨论"（简短，用于检索）description:str# "详细描述..."

Episode 和 Facet 的关系：

classEpisode:has_facet:List[tuple[Edge,Facet]]# 一个Episode包含多个Facet

关键设计：tuple[Edge, Facet]

• Edge携带关系的语义（edge_text: “涉及技术评估”）
• 这让关系本身可检索！

步骤5：主题需要更细粒度的锚点 ✅

问题：用户问"P99延迟目标是什么？"

这个问题太精确，无法匹配 Facet.search_text（"性能目标讨论"太宽泛）。

解决方案：引入 FacetPoint

classFacetPoint:name:str# "P99延迟目标"search_text:str# "P99延迟必须低于500毫秒"description:str# "详细解释..."

Facet 和 FacetPoint 的关系：

classFacet:has_point:List[tuple[Edge,FacetPoint]]# 一个Facet包含多个FacetPoint

为什么需要三层：

• Episode：回答"发生了什么？"
• Facet：回答"哪个方面？"
• FacetPoint：回答"具体是什么？"

这就是锥形图的物理结构！

步骤6：跨事件关联问题 ✅

问题：如何找到"Maria参与的所有事件"？

传统方案：文本搜索"Maria"

• ❌ 会漏掉用同义词提到的 Maria（如"后端负责人"）

M-flow方案：引入 Entity

classEntity:name:str# "Maria"canonical_name:str# "maria"（规范化，用于跨文档匹配）description:str# "后端负责人，负责性能优化"

Episode 和 Entity 的关系：

classEpisode:involves_entity:List[tuple[Edge,Entity]]# 一个Episode涉及多个Entity

Entity 之间的关联：

classEntity:same_entity_as:List[tuple[Edge,Entity]]# 跨Episode的实体关联

示例：

Episode A: involves_entity → Maria（描述：后端负责人） Episode B: involves_entity → Maria（描述：提出问题的人） ↓ same_entity_as（通过 canonical_name 自动关联）

这样查询"Maria"时，两个Episode都会被找到。

步骤7：基类抽象 ✅

观察：Episode、Facet、FacetPoint、Entity 都有共同属性：

• id：唯一标识
• type：类型名称
• created_at：创建时间
• metadata：索引配置

设计模式：继承 + 模板方法

classMemoryNode(BaseModel):"""所有图节点的基类"""id:UUIDtype:str# 自动填充为类名version:intmetadata:dict# {"index_fields": ["字段名"]}created_at:intupdated_at:int@classmethoddefextract_index_text(cls,node):"""拼接索引字段，用于向量化"""# 实现...

子类继承：

classEpisode(MemoryNode):name:strsummary:strmetadata:dict={"index_fields":["summary"]}# 只索引summary

好处：

• 统一的字段管理
• 统一的向量化逻辑
• 统一的序列化/反序列化

步骤8：关系本身也有语义 ✅

创新设计：Edge 不只是连接符，它携带可检索的语义！

classEdge:edge_text:str# "讨论了" / "涉及" / "基于"weight:float# 权重（可选）relationship_type:str# 关系类型（可选）

使用方式：

episode.has_facet=[(Edge(edge_text="重点讨论了"),Facet(name="技术评估",search_text="性能目标讨论"))]

检索时：Edge.edge_text 也会被向量化！

为什么重要：

• 查询"会议争论了什么？" → 匹配 edge_text=“争论了”
• 查询"会议决定了什么？" → 匹配 edge_text=“决定了”
• 关系本身参与相关性评分！

🔗 第三部分：调用链路分析

场景A：用户调用memorize()将文档转化为记忆

用户代码 ↓ m_flow.api.v1.memorize.memorize() 【业务核心】接收文档，协调整个处理流程 ↓ m_flow.pipeline.execute_workflow() 【业务核心】执行多阶段处理管线 ↓ m_flow.pipeline.tasks.Stage 【业务核心】定义5个处理阶段 ├─ Stage1: 文本分块 ├─ Stage2: 信息提取（LLM） ├─ Stage3: 构建Episode ├─ Stage4: 构建Facet └─ Stage5: 构建FacetPoint和Entity ↓ m_flow.memory.episodic.write_episodic_memories() 【业务核心】将提取的数据转化为Episode/Facet/FacetPoint/Entity对象 ↓ m_flow.storage.persist_memory_nodes() 【适配层-可延后学习】统一持久化入口 ↓ m_flow.adapters.graph.graph_db_interface.write_nodes() 【适配层-可延后学习】图数据库适配器接口 ↓ m_flow.adapters.graph.kuzu.adapter.write_nodes() 【适配层-可延后学习】Kuzu数据库实现 ↓ m_flow.adapters.vector.vector_db_interface.upsert() 【适配层-可延后学习】向量数据库适配器接口 ↓ m_flow.adapters.vector.chroma.adapter.upsert() 【适配层-可延后学习】Chroma向量数据库实现

关键方法说明：

场景B：用户调用search()检索记忆

用户代码 ↓ m_flow.api.v1.search.search() 【业务核心】接收查询文本，协调检索流程 ↓ m_flow.search.methods.search() 【业务核心】处理权限检查，选择检索模式 ↓ m_flow.search.methods.no_access_control_search() 【业务核心】执行无权限控制的检索（或其他模式） ↓ m_flow.search.operations.get_recall_mode_tools() 【业务核心】根据RecallMode选择检索策略 ├─ VECTOR: 纯向量搜索 ├─ GRAPH: 纯图遍历 ├─ HYBRID: 混合搜索 └─ TRIPLET_COMPLETION: M-flow特有的锥形图检索 ↓ m_flow.adapters.graph.get_graph_provider() 【适配层-可延后学习】获取图数据库实例 ↓ m_flow.adapters.vector.get_vector_provider() 【适配层-可延后学习】获取向量数据库实例 ↓ m_flow.search.operations.execute_triplet_search() 【业务核心】执行M-flow核心检索算法 ├─ 向量搜索找到锚点（Entity/FacetPoint/Facet/Episode） ├─ 图遍历传播成本（沿着边） ├─ 计算每个Episode的最低路径成本 └─ 返回排序后的Episode列表 ↓ m_flow.search.utils.prepare_search_result() 【业务核心】格式化检索结果 ↓ 返回给用户

关键方法说明：

📊 第四部分：数据流转图

1️⃣ 写入流程（memorize）

输入：原始文档 │ ▼ ┌─────────────────────────────────────────────────────┐ │ Stage 1: 文本分块 (TextChunker) │ │ 输入：长文档 │ │ 输出：List[TextChunk] - 按语义分割的文本块 │ └─────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ Stage 2: LLM信息提取 (extract_graph) │ │ 输入：List[TextChunk] │ │ 输出：FragmentDigest - 提取的结构化信息 │ │ ├─ summaries: 摘要 │ │ ├─ entities: 实体列表 │ │ ├─ sections: 分段信息 │ │ └─ time_ranges: 时间范围 │ └─────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ Stage 3: 构建Episode (episode_builder) │ │ 输入：FragmentDigest │ │ 输出：Episode对象列表 │ │ ├─ name: "数据库选型会议" │ │ ├─ summary: "讨论了PostgreSQL vs MySQL..." │ │ ├─ has_facet: [] │ │ └─ involves_entity: [] │ └─────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ Stage 4: 构建Facet (episode_builder) │ │ 输入：FragmentDigest.sections │ │ 输出：Facet对象列表 │ │ ├─ search_text: "性能目标讨论" │ │ ├─ facet_type: "metric" │ │ └─ anchor_text: "详细描述..." │ │ 同时构建Episode→Facet的边（携带edge_text） │ └─────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ Stage 5: 构建FacetPoint和Entity │ │ 输入：FragmentDigest的细粒度信息 │ │ 输出：FacetPoint对象列表 + Entity对象列表 │ │ FacetPoint: │ │ ├─ search_text: "P99延迟必须低于500ms" │ │ └─ supported_by: ContentFragment │ │ Entity: │ │ ├─ name: "Maria" │ │ ├─ canonical_name: "maria" │ │ └─ description: "后端负责人" │ └─────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ 数据持久化 (persist_memory_nodes) │ │ 输入：Episode + Facet + FacetPoint + Entity + Edge │ │ 输出：写入图数据库 + 向量数据库 │ │ │ │ 图数据库存储： │ │ ├─ 节点：Episode, Facet, FacetPoint, Entity │ │ ├─ 边：has_facet, involves_entity, has_point │ │ └─ 边属性：edge_text（可检索的语义） │ │ │ │ 向量数据库存储： │ │ ├─ Episode_summary集合 │ │ ├─ Facet_search_text集合 │ │ ├─ Facet_anchor_text集合 │ │ ├─ FacetPoint_search_text集合 │ │ ├─ Entity_name集合 │ │ └─ Edge_edge_text集合 │ └─────────────────────────────────────────────────────┘

2️⃣ 检索流程（search）

输入：用户查询 "为什么Maria在周一站会上生气？" │ ▼ ┌─────────────────────────────────────────────────────┐ │ 步骤1: 宽网向量搜索 (wide_search) │ │ 在7个向量集合中同时搜索： │ │ ├─ Episode_summary │ │ ├─ Facet_search_text │ │ ├─ Facet_anchor_text │ │ ├─ FacetPoint_search_text │ │ ├─ Entity_name │ │ ├─ Entity_canonical_name │ │ └─ Edge_edge_text │ │ 每个集合返回top-100候选 │ └─────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ 步骤2: 投影到知识图 (project_to_graph) │ │ 输入：向量命中的节点ID列表 │ │ 操作： │ │ ├─ 从图数据库获取这些节点的完整信息 │ │ ├─ 获取相邻节点（1-hop邻居） │ │ └─ 获取连接边（包括edge_text） │ │ 输出：局部子图（锚点 + 邻居 + 边） │ └─────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ 步骤3: 成本传播 (cost_propagation) 【核心算法】 │ │ │ │ 对子图中的每个Episode： │ │ ┌─────────────────────────────────────────┐ │ │ │ 找到所有从锚点到该Episode的路径 │ │ │ │ │ │ │ │ 路径示例： │ │ │ │ 1. Entity → Episode │ │ │ │ 起始成本: 0.1（Entity.name匹配度高） │ │ │ │ 边成本: 0.3（edge_text相关性中等） │ │ │ │ 跳数惩罚: 0.2 │ │ │ │ 总成本: 0.6 │ │ │ │ │ │ │ │ 2. FacetPoint → Facet → Episode │ │ │ │ 起始成本: 0.05（精确匹配） │ │ │ │ 边1成本: 0.1（"属于"边） │ │ │ │ 边2成本: 0.2（edge_text匹配） │ │ │ │ 跳数惩罚: 0.4（2跳） │ │ │ │ 总成本: 0.75 │ │ │ │ │ │ │ │ Episode得分 = MIN(所有路径成本) = 0.6 │ │ │ └─────────────────────────────────────────┘ │ │ │ │ 关键设计： │ │ ✓ 取最小成本（一条强路径足够） │ │ ✓ 直接命中Episode会被惩罚（防止泛化匹配） │ │ ✓ 边的edge_text参与成本计算 │ └─────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────┐ │ 步骤4: 排序和组装 (rank_and_assemble) │ │ ├─ 按成本排序Episode │ │ ├─ 选择top-K │ │ └─ 根据display_mode组装输出： │ │ ├─ summary: 只返回Episode.summary │ │ ├─ detail: 返回Episode+Facet+Entity │ │ └─ highly_related: 只返回匹配的Facet相关段落 │ └─────────────────────────────────────────────────────┘ │ ▼ 输出：List[SearchResult]

🎨 第五部分：设计模式和架构思想

1️⃣继承 + 模板方法模式

体现：所有节点继承自MemoryNode

MemoryNode(抽象基类)├─ Episode ├─ Facet ├─ FacetPoint ├─ Entity ├─ EntityType └─ Procedure

好处：

• 统一字段管理（id, created_at, metadata）
• 统一向量化逻辑（extract_index_text）
• 统一序列化/反序列化（Pydantic BaseModel）

2️⃣组合模式（Composite Pattern）

体现：Episode → Facet → FacetPoint 的层级结构

Episode ├─ has_facet:List[Facet]│ └─ has_point:List[FacetPoint]└─ involves_entity:List[Entity]

好处：

• 树形结构的统一处理
• 递归遍历（从Episode找到所有FacetPoint）
• 灵活扩展（可以添加新的层级）

3️⃣适配器模式

体现：图数据库和向量数据库的适配层

GraphProvider(接口)├─ KuzuAdapter ├─ Neo4jAdapter └─ NeptuneAdapter VectorProvider(接口)├─ ChromaAdapter ├─ PgVectorAdapter └─ OpenSearchAdapter

好处：

• 解耦业务逻辑和具体存储
• 切换数据库不需要修改业务代码
• 支持多种存储后端

4️⃣工厂模式

体现：get_graph_provider(),get_vector_provider()

defget_graph_provider():backend=config.GR_BACKEND# "kuzu" | "neo4j" | ...ifbackend=="kuzu":returnKuzuAdapter()elifbackend=="neo4j":returnNeo4jAdapter()# ...

好处：

• 根据配置动态创建适配器
• 隐藏实现细节
• 统一入口

5️⃣策略模式

体现：RecallMode检索策略

classRecallMode(Enum):VECTOR="vector"# 纯向量GRAPH="graph"# 纯图HYBRID="hybrid"# 混合TRIPLET_COMPLETION="triplet"# M-flow特有defget_recall_mode_tools(mode):ifmode==RecallMode.VECTOR:returnVectorSearchStrategy()elifmode==RecallMode.TRIPLET_COMPLETION:returnTripletSearchStrategy()# ...

好处：

• 运行时切换检索算法
• 每种策略独立实现
• 易于添加新策略

6️⃣边的语义化（Semantic Edge）

创新设计：Edge 不是简单的连接符，而是携带可检索语义的"一等公民"

Edge(edge_text="重点讨论了",weight=0.8)

为什么重要：

• 查询"争论了什么" → 匹配 edge_text=“争论了”
• 查询"决定了什么" → 匹配 edge_text=“决定了”
• 边参与相关性评分，过滤无关路径

这是M-flow区别于传统图数据库的核心创新之一！

7️⃣锥形图拓扑（Inverted Cone Topology）

架构思想：从精确到抽象的倒金字塔

Entity FacetPoint ← 尖端：最精确的匹配点 │ │ └─────┬─────┘ │ Facet ← 中层：主题维度 │ Episode ← 底座：完整事件包

检索逻辑：

1. 向量搜索在尖端找到精确锚点
2. 图遍历沿着边向下传播到底座
3. 计算路径成本，返回最相关的Episode

为什么这样设计：

• 不同粒度的查询自然路由
• 精确问题命中FacetPoint，宽泛问题命中Episode
• 跨文档关联通过Entity实现

8️⃣最小成本路径（Minimum Cost Path）

设计哲学：一条强证据链足够证明相关性

episode_score=MIN(all_path_costs)# 而不是episode_score=AVG(all_path_costs)# ❌

为什么重要：

• 模仿人类记忆（一个联想触发回忆）
• 防止无关Facet拉低相关性
• 即使Episode有10个Facet，只要1个相关就应被检索

📌 总结：核心数据模型的关系

MemoryNode (基类) │ ├─ Episode (事件包) - 最顶层记忆单元 │ ├─ has_facet → List[Facet] │ ├─ involves_entity → List[Entity] │ └─ includes_chunk → List[ContentFragment] │ ├─ Facet (主题维度) - 事件的一个角度 │ ├─ search_text (索引字段) │ ├─ anchor_text (索引字段) │ └─ has_point → List[FacetPoint] │ ├─ FacetPoint (事实点) - 最细粒度的记忆 │ └─ search_text (索引字段) │ ├─ Entity (实体) - 跨事件的人/物 │ ├─ name (索引字段) │ ├─ canonical_name (索引字段) │ └─ same_entity_as → List[Entity] │ ├─ EntityType (实体类型) - 实体的分类标签 │ └─ name (索引字段) │ └─ Procedure (程序性记忆) - 方法/步骤/流程 ├─ summary (索引字段) ├─ has_context_point → List[ProcedureContextPoint] └─ has_key_point → List[ProcedureStepPoint] Edge (边) - 携带语义的关系 ├─ edge_text (索引字段！) ← 核心创新 ├─ weight └─ relationship_type

🎯 学习建议

如果你想深入理解数据模型，按这个顺序阅读：

1. 基础：m_flow/core/models/MemoryNode.py（30分钟）

   • 理解基类的设计
   • 理解extract_index_text的逻辑

2. 核心节点：m_flow/core/domain/models/Episode.py（1小时）

   • 理解Episode的结构
   • 理解has_facet和involves_entity的设计

3. 主题层：m_flow/core/domain/models/Facet.py（30分钟）

   • 理解search_text和anchor_text的区别
   • 理解索引字段的配置

4. 事实层：m_flow/core/domain/models/FacetPoint.py（20分钟）

   • 理解最细粒度的记忆单元

5. 实体层：m_flow/core/domain/models/Entity.py（30分钟）

   • 理解canonical_name和same_entity_as的设计

6. 边的语义：m_flow/core/models/Edge.py（15分钟）

   • 理解为什么Edge需要edge_text

7. 写入流程：m_flow/memory/episodic/write_episodic_memories.py（2小时）

   • 理解如何从LLM输出构建Episode
   • 理解如何建立边的关系

可以暂时忽略的部分：

• ❌m_flow/adapters/- 适配器层（可延后学习）
• ❌m_flow/storage/- 持久化细节（可延后学习）
• ❌m_flow/pipeline/tasks/- 具体的LLM提示词（可延后学习）

最后的提醒：

数据模型是M-flow的骨架，理解它需要：

1. ✅ 从生活场景入手（侦探事务所的故事）
2. ✅ 理解设计者的推导思路（从0到1）
3. ✅ 追踪调用链路（看数据如何流动）
4. ✅ 识别设计模式（继承、组合、适配器）
5. ✅ 理解核心创新（边的语义化、锥形图、最小成本路径）

不要一开始就陷入细节，先理解为什么这样设计，再深入怎么实现！🚀