两个 Flow 都关了自身事件、且标记"别对我自己开记忆"（recall_flow.py:66 / encoding_flow.py:85）：

# recall_flow.py:58
class RecallFlow(Flow[RecallState]):
    _skip_auto_memory: bool = True          # ★别给记忆流水线自身再开记忆（防递归）
    initial_state: type[RecallState] = RecallState
    def __init__(self, storage, llm, embedder, config=None):
        super().__init__(suppress_flow_events=True)   # 内部流水线，不往事件总线刷屏
        self._storage = storage; self._llm = llm
        self._embedder = embedder; self._config = config or MemoryConfig()

RecallFlow 的起点（recall_flow.py:178）：

# recall_flow.py:178（节选）
@start()
def analyze_query_step(self) -> QueryAnalysis:
    self.state.exploration_budget = self._config.exploration_budget
    query_len = len(self.state.query)
    skip_llm = query_len < self._config.query_analysis_threshold    # ★短查询跳过 LLM

    if skip_llm:
        analysis = QueryAnalysis(keywords=[], suggested_scopes=[],
            complexity="simple", recall_queries=[self.state.query])
        self.state.query_analysis = analysis
    else:
        available = self._storage.list_scopes(self.state.scope or "/") or ["/"]
        scope_info = (self._storage.get_scope_info(self.state.scope or "/")
                      if self.state.scope else None)
        analysis = analyze_query(self.state.query, available, scope_info, self._llm)
        self.state.query_analysis = analysis
        if analysis.time_filter:                        # LLM 抽出的时间过滤
            try:
                self.state.time_cutoff = datetime.fromisoformat(analysis.time_filter)
            except ValueError:
                pass
    queries = (analysis.recall_queries or [self.state.query])[:3]   # 最多 3 个子查询
    embeddings = embed_texts(self._embedder, queries)               # ★一次批量嵌入
    pairs = [(q, emb) for q, emb in zip(queries, embeddings) if emb]
    ...
    self.state.query_embeddings = pairs
    return analysis

filter_and_chunk 选定候选 scope 后，search_chunks 调 _do_search（recall_flow.py:87）：

# recall_flow.py:87（节选）
def _do_search(self) -> list[dict[str, Any]]:
    def _search_one(embedding, scope):
        raw = self._storage.search(embedding, scope_prefix=scope,
            categories=search_categories,
            limit=self.state.limit * _RECALL_OVERSAMPLE_FACTOR,   # ★多取 2 倍候选
            min_score=0.0)
        if self.state.time_cutoff and raw:                        # 时间过滤
            raw = [(r, s) for r, s in raw if r.created_at >= self.state.time_cutoff]
        if not self.state.include_private and raw:                # 隐私过滤
            raw = [(r, s) for r, s in raw if not r.private or r.source == self.state.source]
        return scope, raw

    tasks = [(emb, scope)                             # ★笛卡尔积：每个子查询 × 每个 scope
             for _q, emb in self.state.query_embeddings
             for scope in self.state.candidate_scopes]

    findings = []
    if len(tasks) <= 1:
        ... # 单个直接跑
    else:
        with ThreadPoolExecutor(max_workers=min(len(tasks), 4)) as pool:  # 最多 4 并发
            futures = {pool.submit(contextvars.copy_context().run, _search_one, emb, sc): (emb, sc)
                       for emb, sc in tasks}
            for future in as_completed(futures):
                scope, results = future.result()
                if results:
                    top_composite, _ = compute_composite_score(results[0][0], results[0][1], self._config)
                    findings.append({"scope": scope, "results": results, "top_score": top_composite})
    self.state.chunk_findings = findings
    self.state.confidence = max((f["top_score"] for f in findings), default=0.0)  # ★置信度
    return findings

核心的自适应逻辑在 @router（recall_flow.py:271）：

# recall_flow.py:271
@router(search_chunks)
def decide_depth(self) -> str:
    analysis = self.state.query_analysis
    if (analysis and analysis.complexity == "complex"
            and self.state.confidence < self._config.complex_query_threshold):
        if self.state.exploration_budget > 0:
            return "explore_deeper"          # 复杂查询 + 置信不足 + 有预算 → 深挖
    if self.state.confidence >= self._config.confidence_threshold_high:
        return "synthesize"                  # 置信很高（≥0.8）→ 直接收工
    if (self.state.exploration_budget > 0
            and self.state.confidence < self._config.confidence_threshold_low):
        return "explore_deeper"              # 置信很低（<0.5）+ 有预算 → 深挖
    return "synthesize"                      # 其余 → 收工

深挖分支（recall_flow.py:291）先扣预算，再让 LLM 从当前结果里提炼：

# recall_flow.py:291（节选）
@listen("explore_deeper")
def recursive_exploration(self) -> list[Any]:
    self.state.exploration_budget -= 1          # ★先扣预算，保证循环终止
    enhanced = []
    for finding in self.state.chunk_findings:
        content_parts = [r[0].content for r in finding["results"][:5]]
        chunk_text = "\n---\n".join(content_parts)
        prompt = (f"Query: {self.state.query}\n\nRelevant memory excerpts:\n{chunk_text}\n\n"
                  "Extract the most relevant information ... If something is missing, "
                  "say what's missing in one short line.")
        response = self._llm.call([{"role": "user", "content": prompt}])
        if isinstance(response, str) and "missing" in response.lower():
            self.state.evidence_gaps.append(response[:200])   # ★记下"缺什么"
        enhanced.append({"scope": finding["scope"], "extraction": response, "results": finding["results"]})
    self.state.chunk_findings = enhanced
    return enhanced

@listen(recursive_exploration)
def re_search(self) -> list[Any]:
    return self._do_search()                    # 重搜，更新 confidence 给路由再判

@router(re_search)
def re_decide_depth(self) -> str:
    return self.decide_depth()                  # 复用同一套路由逻辑

最终合成（recall_flow.py:343）：去重 + 复合评分 + 排序 + 挂证据缺口：

# recall_flow.py:343（节选）
@listen("synthesize")
def synthesize_results(self) -> list[MemoryMatch]:
    seen_ids, matches = set(), []
    for finding in self.state.chunk_findings:
        for item in finding.get("results", []):
            record, score = item[0], item[1]
            if isinstance(record, MemoryRecord) and record.id not in seen_ids:  # 去重
                seen_ids.add(record.id)
                composite, reasons = compute_composite_score(record, float(score), self._config)
                matches.append(MemoryMatch(record=record, score=composite, match_reasons=reasons))
    matches.sort(key=lambda m: m.score, reverse=True)     # 复合分降序
    final_results = matches[: self.state.limit]           # 截断到 limit
    self.state.final_results = final_results
    if self.state.evidence_gaps and self.state.final_results:
        self.state.final_results[0].evidence_gaps = list(self.state.evidence_gaps)  # 挂"缺什么"
    return final_results

换到编码侧。EncodingFlow 是 5 步顺序 Flow（encoding_flow.py:75），头两步：

# encoding_flow.py:110
@start()
def batch_embed(self) -> None:
    """Embed all items in a single embedder call."""
    items = list(self.state.items)
    texts = [item.content for item in items]
    embeddings = embed_texts(self._embedder, texts)      # ★一次调用嵌入全部
    for item, emb in zip(items, embeddings):
        item.embedding = emb

@listen(batch_embed)
def intra_batch_dedup(self) -> None:
    """Drop near-exact duplicates within the batch."""
    items = list(self.state.items)
    if len(items) <= 1:
        return
    threshold = self._config.batch_dedup_threshold      # 默认 0.98
    n = len(items)
    for j in range(1, n):
        if items[j].dropped or not items[j].embedding: continue
        for i in range(j):
            if items[i].dropped or not items[i].embedding: continue
            sim = self._cosine_similarity(items[i].embedding, items[j].embedding)
            if sim >= threshold:
                items[j].dropped = True                 # ★近乎重复→标记丢弃
                self.state.items_dropped_dedup += 1
                break

第三步并行找库里已有的相似记录（encoding_flow.py:152），为下一步"要不要合并"做准备——同样用 ThreadPoolExecutor(max_workers=min(len(active), 8)) 并发搜、copy_context().run 传上下文。

第四步是最精妙的：按"字段全不全 × 有没有相似记录"把每条分成四组（encoding_flow.py:221）：

# encoding_flow.py:264（节选）
fields_provided = (item.scope is not None and item.categories is not None
                   and item.importance is not None)
has_similar = item.top_similarity >= threshold          # ≥0.85

if fields_provided and not has_similar:                 # A 组：0 次 LLM
    self._apply_defaults(item)
    item.plan = ConsolidationPlan(actions=[], insert_new=True)   # 直接插入
elif fields_provided and has_similar:                   # B 组：1 次（只判合并）
    self._apply_defaults(item)
    consol_futures[i] = pool.submit(..., analyze_for_consolidation, ...)
elif not fields_provided and not has_similar:           # C 组：1 次（只补字段）
    save_futures[i] = pool.submit(..., analyze_for_save, ...)
else:                                                   # D 组：2 次并发（补字段 + 判合并）
    save_futures[i] = pool.submit(..., analyze_for_save, ...)
    consol_futures[i] = pool.submit(..., analyze_for_consolidation, ...)

第五步 execute_plans（encoding_flow.py:369）把所有计划落库，关键是跨 item 去重动作：

# encoding_flow.py:390（节选）
# 多个 item 的 similar_records 可能重叠 → 对同一 record_id 只保留第一个动作，防 LanceDB 提交冲突
for i, item in enumerate(items):
    for action in item.plan.actions:
        rid = action.record_id
        if action.action == "delete" and rid not in dedup_deletes and rid not in dedup_updates:
            dedup_deletes.add(rid)
        elif action.action == "update" and action.new_content and rid not in ...:
            dedup_updates[rid] = (i, action.new_content)
# 之后：批量 re-embed 更新内容 → delete → update → 一次 bulk save(records)