先看类里管并发的私有属性（unified_memory.py:161）：

# unified_memory.py:161
_config: MemoryConfig = PrivateAttr()
_storage: StorageBackend = PrivateAttr()
_save_pool: ThreadPoolExecutor = PrivateAttr(         # ★单线程保存池
    default_factory=lambda: ThreadPoolExecutor(
        max_workers=1, thread_name_prefix="memory-save"))
_pending_saves: list[Future[Any]] = PrivateAttr(default_factory=list)  # 挂起的写
_pending_lock: threading.Lock = PrivateAttr(default_factory=threading.Lock)
_reset_lock: Any = PrivateAttr(default_factory=threading.RLock)

Pydantic 模型初始化后自动调 model_post_init（unified_memory.py:206）：

# unified_memory.py:206（节选）
def model_post_init(self, __context: Any) -> None:
    self._config = MemoryConfig(          # ① 把散落的权重打包成一个 config
        recency_weight=self.recency_weight, semantic_weight=self.semantic_weight,
        importance_weight=self.importance_weight, ...)

    self._llm_instance = (                # ② LLM：字符串则延后创建（懒），实例则包装成非流式
        None if isinstance(self.llm, str) else _non_streaming_analysis_llm(self.llm))

    if isinstance(self.storage, str):     # ③ 字符串 storage → 解析成真实后端
        from crewai.memory.storage.factory import resolve_memory_storage
        custom = resolve_memory_storage(self.storage)     # 先问全局工厂
        if custom is not None:
            self._storage = custom
        elif self.storage == "qdrant-edge":
            from crewai.memory.storage.qdrant_edge_storage import QdrantEdgeStorage
            self._storage = QdrantEdgeStorage()
        elif self.storage == "lancedb":
            from crewai.memory.storage.lancedb_storage import LanceDBStorage
            self._storage = LanceDBStorage()
        else:                             # 其他字符串当成路径
            from crewai.memory.storage.lancedb_storage import LanceDBStorage
            self._storage = LanceDBStorage(path=self.storage)
    else:
        self._storage = self.storage      # 已经是实例，直接用

remember 存单条，需要立刻拿到 record（unified_memory.py:430）：

# unified_memory.py:430（节选）
def remember(self, content, scope=None, ..., root_scope=None) -> MemoryRecord | None:
    if self.read_only:
        return None                          # 只读模式：静默不存

    effective_root = root_scope if root_scope is not None else self.root_scope
    try:
        crewai_event_bus.emit(self, MemorySaveStartedEvent(value=content, ...))
        # ★仍走同一个保存池（串行化），但阻塞等它完成
        future = self._submit_save(
            self._encode_batch, [content], scope, categories, metadata,
            importance, source, private, effective_root)
        records = future.result()            # ★阻塞：直到这条存完
        record = records[0] if records else None
        crewai_event_bus.emit(self, MemorySaveCompletedEvent(value=content, ...))
        return record
    except Exception as e:
        crewai_event_bus.emit(self, MemorySaveFailedEvent(value=content, error=str(e), ...))
        raise

批量存不阻塞（unified_memory.py:523）：

# unified_memory.py:523（节选）
def remember_many(self, contents, ..., root_scope=None) -> list[MemoryRecord]:
    if not contents or self.read_only:
        return []
    effective_root = root_scope if root_scope is not None else self.root_scope
    self._submit_save(                       # ★只提交，不 .result()
        self._background_encode_batch,
        contents, scope, categories, metadata, importance,
        source, private, agent_role, effective_root)
    return []                                # ★立刻返回空列表（记录还没生成）

所有写都过这个提交口（unified_memory.py:297）：

# unified_memory.py:297
def _submit_save(self, fn, *args, **kwargs) -> Future[Any]:
    with self._reset_lock:
        ctx = contextvars.copy_context()          # ★复制当前上下文
        try:
            future = self._save_pool.submit(ctx.run, fn, *args, **kwargs)
        except RuntimeError:                       # 池已关闭（close 之后）
            future = Future()                      # ★兜底：同步跑，不丢
            try:
                result = fn(*args, **kwargs)
                future.set_result(result)
            except Exception as exc:
                future.set_exception(exc)
            return future
        with self._pending_lock:
            self._pending_saves.append(future)     # 登记为"挂起"
        future.add_done_callback(self._on_save_done)  # 完成时从挂起表移除
        return future

读屏障本体（unified_memory.py:350）非常短：

# unified_memory.py:350
def drain_writes(self) -> None:
    """Block until all pending background saves have completed."""
    with self._pending_lock:
        pending = list(self._pending_saves)   # 快照当前挂起的 Future
    for future in pending:
        if future.cancelled():
            continue
        future.exception()      # ★阻塞直到完成；不重新抛（失败已由事件报过）

recall 开头第一句就调它（unified_memory.py:711）：

# unified_memory.py:711
def recall(self, query, ..., depth="deep", ...) -> list[MemoryMatch]:
    # Read barrier: wait for any pending background saves to finish
    # so that the search sees all persisted records.
    self.drain_writes()
    ...

排空后，recall 按 depth 分两条路（unified_memory.py:734）：

# unified_memory.py:734（节选）
if depth == "shallow":
    embedding = embed_text(self._embedder, query)     # 直接嵌入查询
    if not embedding:
        results = []
    else:
        raw = self._storage.search(embedding, scope_prefix=effective_scope,
                                   categories=categories, limit=limit, min_score=0.0)
        if not include_private:                        # 隐私过滤
            raw = [(r, s) for r, s in raw if not r.private or r.source == source]
        results = []
        for r, s in raw:
            composite, reasons = compute_composite_score(r, s, self._config)  # 复合评分
            results.append(MemoryMatch(record=r, score=composite, match_reasons=reasons))
        results.sort(key=lambda m: m.score, reverse=True)
else:                                                  # deep：走 RecallFlow（D35）
    from crewai.memory.recall_flow import RecallFlow
    flow = RecallFlow(storage=self._storage, llm=self._llm,
                      embedder=self._embedder, config=self._config)
    flow.kickoff(inputs={"query": query, "scope": effective_scope, ...})
    results = flow.state.final_results