1
# mem0/memory/main.py - Memory 类核心初始化
2

3
class Memory(MemoryBase):
4
    def __init__(self, config: MemoryConfig = MemoryConfig()):
5
        self.config = config
6

7
        # 核心三件套
8
        self.embedding_model = EmbedderFactory.create(
9
            self.config.embedder.provider,
10
            self.config.embedder.config,
11
            self.config.vector_store.config,
12
        )
13
        self.vector_store = VectorStoreFactory.create(
14
            self.config.vector_store.provider, self.config.vector_store.config
15
        )
16
        self.llm = LlmFactory.create(self.config.llm.provider, self.config.llm.config)
17

18
        # 辅助组件
19
        self.db = SQLiteManager(self.config.history_db_path)
20

21
        # 可选重排序器
22
        self.reranker = None
23
        if config.reranker:
24
            self.reranker = RerankerFactory.create(
25
                config.reranker.provider, config.reranker.config
26
            )
27

28
        # 实体存储 — 懒初始化（首次使用时才创建）
29
        self._entity_store = None

1
# mem0/memory/main.py - V3 Pipeline Phase 0-2
2

3
def _add_to_vector_store(self, messages, metadata, filters, infer, prompt=None):
4
    # === V3 PHASED BATCH PIPELINE ===
5

6
    # Phase 0: Context gathering
7
    session_scope = _build_session_scope(filters)
8
    last_messages = self.db.get_last_messages(session_scope, limit=10)
9
    parsed_messages = parse_messages(messages)
10

11
    # Phase 1: Existing memory retrieval（用于去重和链接）
12
    search_filters = {k: v for k, v in filters.items()
13
                      if k in ("user_id", "agent_id", "run_id") and v}
14
    query_embedding = self.embedding_model.embed(parsed_messages, "search")
15
    existing_results = self.vector_store.search(
16
        query=parsed_messages, vectors=query_embedding,
17
        top_k=10, filters=search_filters,
18
    )
19

20
    # 🔑 关键：UUID → 整数映射（防止 LLM 幻觉）
21
    uuid_mapping = {}
22
    existing_memories = []
23
    for idx, mem in enumerate(existing_results):
24
        uuid_mapping[str(idx)] = mem.id
25
        existing_memories.append({"id": str(idx), "text": mem.payload.get("data", "")})
26

27
    # Phase 2: LLM extraction (单次调用)
28
    system_prompt = ADDITIVE_EXTRACTION_PROMPT
29
    user_prompt = generate_additive_extraction_prompt(
30
        existing_memories=existing_memories,
31
        new_messages=parsed_messages,
32
        last_k_messages=last_messages,
33
        custom_instructions=prompt or self.custom_instructions,
34
    )
35

36
    response = self.llm.generate_response(
37
        messages=[
38
            {"role": "system", "content": system_prompt},
39
            {"role": "user", "content": user_prompt},
40
        ],
41
        response_format={"type": "json_object"},
42
    )
43
    extracted_memories = json.loads(response).get("memory", [])

1
# mem0/memory/main.py - V3 Pipeline Phase 3-5
2

3
# Phase 3: Batch embed — 一次网络调用编码所有文本
4
mem_texts = [m.get("text", "") for m in extracted_memories if m.get("text")]
5
mem_embeddings_list = self.embedding_model.embed_batch(mem_texts, "add")
6
embed_map = dict(zip(mem_texts, mem_embeddings_list))
7

8
# Phase 4-5: CPU 处理 + 哈希去重
9
existing_hashes = {mem.payload.get("hash") for mem in existing_results
10
                   if mem.payload and mem.payload.get("hash")}
11
records = []
12
seen_hashes = set()
13

14
for mem in extracted_memories:
15
    text = mem.get("text")
16
    if not text or text not in embed_map:
17
        continue
18

19
    # 🔑 MD5 哈希去重：精确检测重复记忆
20
    mem_hash = hashlib.md5(text.encode()).hexdigest()
21
    if mem_hash in existing_hashes or mem_hash in seen_hashes:
22
        continue  # 跳过重复
23
    seen_hashes.add(mem_hash)
24

25
    # 预计算 BM25 词形还原文本（搜索时使用）
26
    text_lemmatized = lemmatize_for_bm25(text)
27

28
    memory_id = str(uuid.uuid4())
29
    mem_metadata = {
30
        "data": text,
31
        "text_lemmatized": text_lemmatized,
32
        "hash": mem_hash,
33
        "created_at": datetime.now(timezone.utc).isoformat(),
34
        **metadata,
35
    }
36
    records.append((memory_id, text, embed_map[text], mem_metadata))

1
# mem0/memory/main.py - Phase 7: Batch Entity Linking
2

3
# 7a: 全局去重 — 收集所有记忆中的唯一实体
4
all_texts = [r[1] for r in records]
5
all_entities = extract_entities_batch(all_texts)
6

7
global_entities = {}  # normalized_key -> (entity_type, text, set of memory_ids)
8
for idx, (memory_id, text, embedding, payload) in enumerate(records):
9
    entities = all_entities[idx]
10
    for entity_type, entity_text in entities:
11
        key = self._normalize_entity_text(entity_text)
12
        if key in global_entities:
13
            global_entities[key][2].add(memory_id)  # 关联更多记忆
14
        else:
15
            global_entities[key] = [entity_type, entity_text, {memory_id}]
16

17
# 7b: 批量嵌入所有唯一实体
18
entity_texts = [global_entities[k][1] for k in global_entities]
19
entity_embeddings = self.embedding_model.embed_batch(entity_texts, "add")
20

21
# 7c: 批量搜索是否已存在（相似度 >= 0.95 视为相同实体）
22
existing_matches = self.entity_store.search_batch(
23
    queries=valid_texts, vectors_list=valid_vectors,
24
    top_k=1, filters=search_filters,
25
)
26

27
# 7d: 分流 — 已存在的更新 linked_memory_ids，新的批量插入
28
for j, key in enumerate(valid_keys):
29
    entity_type, entity_text, memory_ids = global_entities[key]
30
    match = existing_matches[j][0] if existing_matches[j] else None
31

32
    if match and match.score >= 0.95:
33
        # 更新现有实体：追加新的 memory_id
34
        linked = set(match.payload.get("linked_memory_ids", []))
35
        linked |= memory_ids
36
        self.entity_store.update(vector_id=match.id,
37
                                 payload={"linked_memory_ids": sorted(linked)})
38
    else:
39
        # 新实体：准备批量插入
40
        to_insert.append({
41
            "data": entity_text,
42
            "entity_type": entity_type,
43
            "linked_memory_ids": sorted(memory_ids),
44
        })

1
# mem0/memory/main.py - _search_vector_store 核心 9 步
2

3
def _search_vector_store(self, query, filters, limit, threshold=0.1, explain=False):
4
    # Step 1: 预处理 — 词形还原 + 实体提取
5
    query_lemmatized = lemmatize_for_bm25(query)
6
    query_entities = extract_entities(query)
7

8
    # Step 2: 嵌入查询
9
    embeddings = self.embedding_model.embed(query, "search")
10

11
    # Step 3: 语义搜索（4 倍 over-fetch 确保候选充足）
12
    internal_limit = max(limit * 4, 60)
13
    semantic_results = self.vector_store.search(
14
        query=query, vectors=embeddings, top_k=internal_limit, filters=filters
15
    )
16

17
    # Step 4: BM25 关键词搜索
18
    keyword_results = self.vector_store.keyword_search(
19
        query=query_lemmatized, top_k=internal_limit, filters=filters
20
    )
21

22
    # Step 5: BM25 分数归一化（Sigmoid）
23
    bm25_scores = {}
24
    if keyword_results is not None:
25
        midpoint, steepness = get_bm25_params(query, lemmatized=query_lemmatized)
26
        for mem in keyword_results:
27
            mem_id = str(mem.id)
28
            if mem.score > 0:
29
                bm25_scores[mem_id] = normalize_bm25(mem.score, midpoint, steepness)
30

31
    # Step 6: 实体提升分数计算
32
    entity_boosts = {}
33
    if query_entities:
34
        entity_boosts = self._compute_entity_boosts(query_entities, filters)
35

36
    # Step 7-8: Score & Rank (见下文详解)
37
    scored_results = score_and_rank(
38
        semantic_results=candidates,
39
        bm25_scores=bm25_scores,
40
        entity_boosts=entity_boosts,
41
        threshold=threshold,
42
        top_k=limit,
43
        explain=explain,
44
    )
45
    return scored_results

1
# mem0/utils/scoring.py - 查询长度自适应 BM25 参数
2

3
def get_bm25_params(query: str, *, lemmatized=None) -> tuple:
4
    """长查询产生更高的 BM25 原始分，需要更大的 midpoint 来平衡"""
5
    if lemmatized is None:
6
        lemmatized = lemmatize_for_bm25(query)
7
    num_terms = len(lemmatized.split()) if lemmatized else 1
8

9
    if num_terms <= 3:     return 5.0, 0.7   # 短查询：低阈值，陡峭曲线
10
    elif num_terms <= 6:   return 7.0, 0.6   # 中等查询
11
    elif num_terms <= 9:   return 9.0, 0.5   # 较长查询
12
    elif num_terms <= 15:  return 10.0, 0.5  # 长查询
13
    else:                  return 12.0, 0.5  # 超长查询：高阈值
14

15
def normalize_bm25(raw_score: float, midpoint: float, steepness: float) -> float:
16
    """Logistic sigmoid: 在 midpoint 处输出 0.5"""
17
    return 1.0 / (1.0 + math.exp(-steepness * (raw_score - midpoint)))

1
# mem0/utils/scoring.py - 核心评分逻辑
2

3
ENTITY_BOOST_WEIGHT = 0.5
4

5
def score_and_rank(semantic_results, bm25_scores, entity_boosts,
6
                   threshold, top_k, explain=False):
7
    """
8
    核心公式：combined = (semantic + bm25 + entity_boost) / max_possible
9

10
    max_possible 自适应：
11
      - 仅语义:           max_possible = 1.0
12
      - 语义 + BM25:      max_possible = 2.0
13
      - 语义 + BM25 + 实体: max_possible = 2.5
14
      - 语义 + 实体:       max_possible = 1.5
15
    """
16
    has_bm25 = bool(bm25_scores)
17
    has_entity = bool(entity_boosts)
18

19
    # 🔑 自适应分母：只对实际存在的信号求和
20
    max_possible = 1.0
21
    if has_bm25:   max_possible += 1.0
22
    if has_entity: max_possible += ENTITY_BOOST_WEIGHT  # 0.5
23

24
    scored = []
25
    for result in semantic_results:
26
        semantic_score = result.get("score", 0.0)
27

28
        # ⚡ Threshold 门控：语义分不够直接淘汰
29
        if semantic_score < threshold:
30
            continue
31

32
        mem_id = str(result["id"])
33
        bm25_score = bm25_scores.get(mem_id, 0.0)
34
        entity_boost = entity_boosts.get(mem_id, 0.0)
35

36
        raw_combined = semantic_score + bm25_score + entity_boost
37
        combined = min(raw_combined / max_possible, 1.0)  # 归一化到 [0,1]
38

39
        scored.append({"id": mem_id, "score": combined, "payload": result["payload"]})
40

41
    scored.sort(key=lambda x: x["score"], reverse=True)
42
    return scored[:top_k]