M-FLOW 深度学习笔记

学习时间：2026-04-22

项目地址：https://github.com/FlowElement-ai/m_flow

团队：心流元素（FlowElement.ai）

项目版本：v1.0+

---

M-FLOW 是一个认知记忆引擎（Cognitive Memory Engine），其核心理念是：

"RAG matches chunks. GraphRAG structures context. M-flow scores evidence paths."

RAG 匹配文本块。图RAG 构建结构。M-flow 评估证据路径。

核心创新点：

• 图是评分引擎（Graph is the scoring engine）
• 相关性不是分数，而是路径（Relevance is not a score. It's a path）
• 毫秒级检索（不依赖LLM）

---

M-FLOW 将知识组织为四层倒锥（Inverted Cone）结构，从抽象到具体：

                        ┌─────────────────┐
                        │   Episode (基座)  │  ← 最丰富的语义，召回着陆点
                        │  完整语义焦点     │
                        └────────┬────────┘
                                 │
                        ┌────────▼────────┐
                        │   Facet (中间层)  │  ← 横截面维度，一个角度/主题
                        │   事件截面       │
                        └────────┬────────┘
                                 │
                        ┌────────▼────────┐
                        │  FacetPoint (尖端)│ ← 原子断言，最小事实/三元组
                        │  最小原子事实    │
                        └────────┬────────┘
                                 │
                        ┌────────▼────────┐
                        │    Entity (贯穿) │  ← 横穿所有层级的锚线
                        │   命名实体       │
                        └─────────────────┘

传统知识图谱中，边只是类型标签（如 works_at、located_in）。M-FLOW 打破这一惯例：

edge = {
    "relationship_name": "has_facet",      # 关系类型
    "edge_text": "discussed deadline concerns in standup meeting"  # 自然语言语义
}

核心优势：边不再是被动连接器，而是主动语义过滤器。在代价传播过程中，系统不仅知道"两个节点之间存在连接"，还知道"这个连接与当前查询的相关性有多强"。

def compute_episode_bundles(index, best_by_id, edge_hit_map, config):
    # 取所有路径中的最小代价，而非平均
    episode_cost = min(all_path_costs)

设计哲学：一条强有力的证据链足以证明相关性。一个 Episode 可能包含十个 Facet，其中九个与查询无关。传统方法会取平均——被九个高代价路径拖累。Bundle Search 只看最佳路径。

if path_type == "direct_episode":
    cost += config.triplet_distance_penalty  # 额外惩罚

原因：Episode 摘要是一般性概括——它们对许多查询"看起来相关"。系统偏好：从精确锚点（FacetPoint、Entity）出发即使需要更多跳数。

---

flowchart TB
    subgraph Phase1["阶段一：锥尖广撒网"]
        Q["Query 查询"] --> |"嵌入"| V["7个向量集合并行搜索"]
        V --> |"top-100|top-k|top-50|..."| C["候选节点集合"]
    end
    
    subgraph Phase2["阶段二：投影到图中"]
        C --> |"entry points|锚点"| SG["子图构建"]
        SG --> |"扩展1跳"| G["连接拓扑结构"]
    end
    
    subgraph Phase3["阶段三：代价传播"]
        G --> |"路径代价计算"| PC["EpisodeBundle 评分"]
        PC --> |"最小代价"| R["Top-K Episodes"]
    end

查询同时在7个向量集合中搜索，每个集合返回最多100个候选：

锚点作为入口节点，系统提取周围子图并扩展1跳：

async def get_episodic_memory_fragment(relevant_ids_to_filter, config):
    memory_fragment = MemoryGraph()
    await memory_fragment.project_graph_from_db(
        graph_engine,
        node_properties_to_project=["type", "content", "summary"],
        edge_properties_to_project=["relationship_name", "edge_text"],
        relevant_ids_to_filter=relevant_ids_to_filter  # 仅包含候选节点
    )

核心公式：

Episode_Cost = min(
    Direct_Cost,
    Facet_Cost + Edge_Cost + Hop_Cost,
    FacetPoint_Cost + 2×Edge_Cost + 2×Hop_Cost,
    Entity_Cost + Edge_Cost + Hop_Cost
)

路径类型：

代价构成：

• 起点代价：锚点的向量距离（信号的尖锐程度）
• 边代价：每条边自身的向量距离 + 跳数惩罚
• 未命中惩罚：未匹配边的默认高代价

@dataclass
class EpisodeBundle:
    episode_id: str
    score: float
    best_path: str  # "direct_episode" | "facet" | "point" | "entity" | "facet_entity"
    best_support_id: Optional[str] = None
    best_facet_id: Optional[str] = None
    best_point_id: Optional[str] = None
    best_entity_id: Optional[str] = None
def compute_episode_bundles(index, best_by_id, edge_hit_map, config):
    # 构建关系索引
    relationship_index = build_relationship_index(memory_fragment)
    
    # 对每个 Episode 计算最佳路径代价
    for episode_id in relationship_index.episode_ids:
        paths = []
        
        # 路径1：直接命中（受惩罚）
        direct = direct_cost(episode_id) + config.triplet_distance_penalty
        
        # 路径2：经由 Facet
        for facet_id in facets_by_episode[episode_id]:
            via_facet = facet_cost[facet_id] + edge_cost(ep_facet_edge) + hop_cost
        
        # 路径3：经由 FacetPoint
        for facet_id in facets_by_episode[episode_id]:
            for point_id in points_by_facet[facet_id]:
                via_point = point_cost[point_id] + 2edge_cost + 2hop_cost
        
        # 取最小代价
        best = min(paths)

---

M-FLOW 是业内首个在知识图谱中实现指代消解的系统。

问题场景：

Turn 1: "Maria raised the deadline issue at Monday's standup."
Turn 2: "She said she wasn't told about the change."

无指代消解：Turn 2 不包含 Maria token，其 FacetPoint 无法锚定到 Entity Maria。后续查询"玛ria 对 deadline 说了什么？"只能找到 Turn 1。

有指代消解：Turn 2 解析为 "Maria said Maria wasn't told about the change."，现在可以锚定到 Entity Maria。

coreference/
├── coreference_module/      # 核心中文指代消解
│   ├── coreference.py       # 主解析器
│   ├── tokenizer.py         # 中文分词 + NER
│   ├── syntax_adapter.py    # HanLP/LTP 适配器
│   ├── time_normalizer.py   # 时间表达式归一化
│   ├── canonicalizer.py     # 实体规范化
│   └── ner_adapter.py       # NER 服务适配器
├── english_coreference/     # 英文指代消解
└── tests/                   # 测试套件（85个测试，100%通过）

from coreference_module import CoreferenceResolver, StreamCorefSession
resolver = CoreferenceResolver()
text = "小明去北京。他在那里工作。"
resolved, replacements = resolver.resolve_text(text)
print(resolved)
session = StreamCorefSession()
session.add_sentence("张三是医生。")
result = session.add_sentence("他很忙。")
print(result)  # 张三很忙。

---

Model Context Protocol (MCP) 是一种让 AI 助手与外部工具交互的标准协议。M-FLOW MCP Server 将记忆功能暴露给各种 IDE。

Cursor 配置：

{
  "mcpServers": {
    "m_flow": {
      "url": "http://localhost:8001/sse",
      "transport": "sse"
    }
  }
}

Claude Desktop 配置：

{
  "mcpServers": {
    "m_flow": {
      "command": "python",
      "args": ["-m", "src.server", "--transport", "stdio"],
      "cwd": "/path/to/mflow-main"
    }
  }
}

docker compose --profile mcp up -d
cd m_flow-mcp
pip install -e .
python src/server.py --transport sse --port 8000

---

主检索模式，基于 Bundle Search 算法：

class EpisodicRetriever(BaseGraphRetriever):
    async def get_triplets(self, query: str) -> List[Edge]:
        """使用 episodic_bundle_search 检索三元组"""
        return await episodic_bundle_search(query=query, config=self.config)
    
    async def get_context(self, query: str) -> Union[List[Edge], str]:
        """检索情景上下文"""
        # 支持 display_mode: summary / detail / highly_related_summary
        pass

class ProceduralRetriever(BaseGraphRetriever):
    async def retrieve(self, query: str) -> List[ProceduralMemory]:
        """提取可复用的抽象模式"""
        # 习惯、工作流、决策规则、命名约定等

def compute_adaptive_context(collection_stats, config):
    """
    根据查询动态调整权重
    - 节点类型 vs 边类型集合分组
    - 按置信度比例分配权重
    """
    if entity_confidence > facet_confidence:
        weights["entity_paths"] *= 1.5  # 增强实体路径影响

---

LoCoMo-10（Aligned, Top-K=10）：

M-FLOW:      ████████████████████████████████████ 81.8%
Cognee:      █████████████████████████████████ 79.4%
Zep Cloud:   █████████████████████████████ 73.4%
Supermemory: ██████████████████████ 64.4%
Mem0:        ████████████ 50.4%

LongMemEval：

M-FLOW:      ████████████████████████████████████████ 89%
Supermemory: ██████████████████████████████ 74%
Mem0:        ████████████████████████████ 71%
Zep Cloud:   ████████████████████ 61%
Cognee:      ██████████████████ 57%

---

m_flow/
├── m_flow/                    # 核心引擎
│   ├── retrieval/             # 检索模块
│   │   ├── episodic/          # 情景记忆检索
│   │   │   ├── bundle_search.py        # Bundle Search 主流程
│   │   │   ├── bundle_scorer.py        # EpisodeBundle 评分
│   │   │   ├── memory_fragment.py      # 子图投影
│   │   │   ├── adaptive_scoring.py     # 自适应评分
│   │   │   ├── exact_match_bonus.py    # 精确匹配奖励
│   │   │   └── output_assembler.py     # 输出组装
│   │   ├── procedural/       # 程序性记忆
│   │   ├── lexical/          # 词法检索
│   │   └── cypher/           # 图查询
│   ├── knowledge/           # 知识组织
│   │   ├── extraction/       # 知识抽取
│   │   ├── summarization/    # 摘要生成
│   │   └── graph_ops/        # 图操作
│   ├── coreference/          # 指代消解
│   ├── storage/              # 存储后端
│   └── adapters/             # 适配器
├── m_flow-mcp/               # MCP 服务器
├── coreference/              # 指代消解独立模块
└── docs/                     # 文档
    └── RETRIEVAL_ARCHITECTURE.md

async def episodic_bundle_search(query, config):
    # Step 1: 查询预处理
    preprocessed = preprocess_query(query, config)
    
    # Step 2: 7路向量搜索
    node_distances, edge_distances = await vector_search(
        preprocessed.vector_query, 
        config, 
        effective_wide_search_top_k
    )
    
    # Step 2.5: 自适应评分上下文
    adaptive_context = compute_adaptive_context(collection_stats, config)
    
    # Step 3: 精确匹配奖励
    apply_exact_match_bonus(query, preprocessed, node_distances, config)
    
    # Step 4: 两阶段投影
    memory_fragment = await two_phase_projection(node_distances, config)
    
    # Step 5: 节点距离写回
    apply_node_distances(memory_fragment, best_by_id, config)
    
    # Step 6: 边距离映射
    memory_fragment.map_vector_distances_to_graph_edges(edge_distances)
    
    # Step 7: 关系索引构建
    index = build_relationship_index(memory_fragment)
    
    # Step 8: Bundle 评分
    bundles = compute_episode_bundles(index, best_by_id, edge_hit_map, config)
    
    # Step 9: 时间奖励（可选）
    apply_time_bonus(bundles, memory_fragment, time_info, config)
    
    # Step 10: 排序取 Top-K
    top_bundles = heapq.nsmallest(config.top_k, bundles)
    
    # Step 11: 输出组装
    return assemble_output_edges(top_bundles, index, ...)

---

1. 图即评分引擎 - 向量搜索只是撒网入口 - 最终相关性由图传播决定

2. 相关性即路径 - 不是平坦排名 - 而是最强证据链

3. 一条强路径足够 - 单个关联足以触发完整记忆 - 类比人类记忆机制

性能提升 = 多粒度路由（30%）+ 图传播（40%）+ 指代消解（15%）+ 语义边（15%）

---

• M-FLOW GitHub
• m-flow.ai
• flowelement.ai
• Retrieval Architecture 文档
• MCP Server 文档
• Coreference 模块文档
• Benchmark 数据集

学习感悟：M-FLOW 的创新在于将"相关性"从静态的相似度分数转变为动态的路径代价。这不仅是一个技术实现，更是对人类记忆机制的计算模拟——我们回忆某件事，往往是因为一个关联足够强烈，而不是所有关联都指向它。