所有检索方式钉在 RetrievalMethod（core/rag/retrieval/retrieval_methods.py:4）：

# core/rag/retrieval/retrieval_methods.py:4
class RetrievalMethod(StrEnum):
    SEMANTIC_SEARCH = "semantic_search"      # 语义：向量最近邻（懂意思）
    FULL_TEXT_SEARCH = "full_text_search"    # 全文：倒排索引（抠字眼）
    HYBRID_SEARCH = "hybrid_search"          # 混合：语义 + 全文，再 rerank
    KEYWORD_SEARCH = "keyword_search"        # 关键词：经济模式的字面匹配

    @staticmethod
    def is_support_semantic_search(retrieval_method: str) -> bool:
        return retrieval_method in {RetrievalMethod.SEMANTIC_SEARCH, RetrievalMethod.HYBRID_SEARCH}

    @staticmethod
    def is_support_fulltext_search(retrieval_method: str) -> bool:
        return retrieval_method in {RetrievalMethod.FULL_TEXT_SEARCH, RetrievalMethod.HYBRID_SEARCH}

对外入口是 RetrievalService.retrieve（core/rag/datasource/retrieval_service.py:114）。它本身也用线程池，把"文本查询"和"多个附件查询"并发发出去：

# core/rag/datasource/retrieval_service.py:114
def retrieve(cls, retrieval_method, dataset_id, query, top_k=4, score_threshold=0.0,
             reranking_model=None, reranking_mode="reranking_model", weights=None,
             document_ids_filter=None, attachment_ids=None):
    if not query and not attachment_ids:
        return []                                     # ① 空查询直接返回
    dataset = cls._get_dataset(dataset_id)
    if not dataset:
        return []
    all_documents: list[Document] = []
    exceptions: list[str] = []
    with ThreadPoolExecutor(max_workers=dify_config.RETRIEVAL_SERVICE_EXECUTORS) as executor:  # ② 线程池
        futures = []
        retrieval_service = RetrievalService()
        if query:
            futures.append(executor.submit(_propagate_otel_context(retrieval_service._retrieve),
                flask_app=current_app._get_current_object(), retrieval_method=retrieval_method,
                dataset=dataset, query=query, top_k=top_k, ..., all_documents=all_documents, exceptions=exceptions))
        if attachment_ids:
            for attachment_id in attachment_ids:      # 多个附件各起一个任务
                futures.append(executor.submit(..., attachment_id=attachment_id, ...))
        if futures:
            for _ in concurrent.futures.as_completed(futures, timeout=3600):
                if exceptions:                        # ③ 任何子任务出错就取消其余、尽早失败
                    for f in futures:
                        f.cancel()
                    break
    if exceptions:
        raise ValueError(";\n".join(exceptions))
    return all_documents

真正干活的 _retrieve（core/rag/datasource/retrieval_service.py:781）里，又开一层线程池，按检索方式把"关键词/语义/全文"三路并发提交：

# core/rag/datasource/retrieval_service.py:781
def _retrieve(self, flask_app, retrieval_method, dataset, all_documents, exceptions,
              query=None, top_k=4, score_threshold=0.0, reranking_model=None, ...):
    with flask_app.app_context():
        all_documents_item: list[Document] = []
        with ThreadPoolExecutor(max_workers=dify_config.RETRIEVAL_SERVICE_EXECUTORS) as executor:
            futures = []
            if retrieval_method == RetrievalMethod.KEYWORD_SEARCH and query:
                futures.append(executor.submit(self.keyword_search, ...))          # 关键词路
            if RetrievalMethod.is_support_semantic_search(retrieval_method):        # ★ 语义/混合 → 开
                if query:
                    futures.append(executor.submit(self.embedding_search, ..., query_type=QueryType.TEXT_QUERY))
                if attachment_id:
                    futures.append(executor.submit(self.embedding_search, ..., query_type=QueryType.IMAGE_QUERY))
            if RetrievalMethod.is_support_fulltext_search(retrieval_method) and query:   # ★ 全文/混合 → 开
                futures.append(executor.submit(self.full_text_index_search, ...))
            if futures:
                for future in concurrent.futures.as_completed(futures, timeout=300):
                    if future.exception():
                        for f in futures: f.cancel()
                        break

三路召回汇齐后，混合检索要做"去重 + 重排"这一步（core/rag/datasource/retrieval_service.py:881）：

# core/rag/datasource/retrieval_service.py:881
if retrieval_method == RetrievalMethod.HYBRID_SEARCH:
    all_documents_item = self._deduplicate_documents(all_documents_item)     # ① 去重（同一片可能被多路命中）
    data_post_processor = DataPostProcessor(
        str(dataset.tenant_id), reranking_mode, reranking_model, weights, False)  # ② 造重排器
    if query:
        rerank_query, query_type = query, QueryType.TEXT_QUERY
    elif attachment_id:
        rerank_query, query_type = attachment_id, QueryType.IMAGE_QUERY
    else:
        return
    all_documents_item = data_post_processor.invoke(                         # ③ ★重排：统一打分排序取 top_k
        query=rerank_query, documents=all_documents_item,
        score_threshold=score_threshold, top_n=top_k, query_type=query_type)
    if not data_post_processor.rerank_runner and score_threshold:            # ④ 没重排器才在这兜底过滤阈值
        all_documents_item = self._filter_documents_by_vector_score_threshold(
            all_documents_item, score_threshold)
all_documents.extend(all_documents_item)                                     # 并回总结果

DataPostProcessor 在构造时就按 reranking_mode 选好了哪种重排器（core/rag/data_post_processor/data_post_processor.py:65）：

# core/rag/data_post_processor/data_post_processor.py:65
def _get_rerank_runner(self, reranking_mode, tenant_id, reranking_model=None, weights=None):
    if reranking_mode == RerankMode.WEIGHTED_SCORE and weights:        # ① 权重重排
        return RerankRunnerFactory.create_rerank_runner(
            runner_type=reranking_mode, tenant_id=tenant_id,
            weights=Weights(vector_setting=VectorSetting(...), keyword_setting=KeywordSetting(...)))
    elif reranking_mode == RerankMode.RERANKING_MODEL:                 # ② 模型重排
        rerank_model_instance = self._get_rerank_model_instance(tenant_id, reranking_model)
        if rerank_model_instance is None:
            return None
        return RerankRunnerFactory.create_rerank_runner(
            runner_type=reranking_mode, rerank_model_instance=rerank_model_instance)
    return None

权重重排不调模型，纯靠"关键词分 × 权重 + 向量分 × 权重"算加权分。看 WeightRerankRunner.run（core/rag/rerank/weight_rerank.py:59）的核心公式：

# core/rag/rerank/weight_rerank.py:59
query_scores = self._calculate_keyword_score(query, documents)            # 关键词相似度(TF-IDF)
query_vector_scores = self._calculate_cosine(self.tenant_id, query, documents, self.weights.vector_setting)  # 向量余弦相似度
rerank_documents = []
for document, query_score, query_vector_score in zip(documents, query_scores, query_vector_scores):
    score = (self.weights.vector_setting.vector_weight * query_vector_score      # ★ 加权融合
             + self.weights.keyword_setting.keyword_weight * query_score)
    if score_threshold and score < score_threshold:
        continue
    document.metadata["score"] = score
    rerank_documents.append(document)

重排器统一入口 DataPostProcessor.invoke（core/rag/data_post_processor/data_post_processor.py:49）——重排完还能接一个 reorder（把最相关的放两端，缓解"中间遗忘"）：

# core/rag/data_post_processor/data_post_processor.py:49
def invoke(self, query, documents, score_threshold=None, top_n=None, query_type=QueryType.TEXT_QUERY):
    if self.rerank_runner:
        documents = self.rerank_runner.run(query, documents, score_threshold, top_n, query_type)   # 重排
    if self.reorder_runner:
        documents = self.reorder_runner.run(documents)     # 可选：重新摆放（头尾放最相关的）
    return documents

cd /Users/bitmart/work/codes/github/AI_WORK/dify/api

# 1. 四种方式
cat core/rag/retrieval/retrieval_methods.py

# 2. 并发召回
sed -n '114,190p' core/rag/datasource/retrieval_service.py   # retrieve 入口
sed -n '781,875p' core/rag/datasource/retrieval_service.py   # _retrieve 三路并发
sed -n '881,909p' core/rag/datasource/retrieval_service.py   # 混合：去重 + rerank

# 3. 两种重排
sed -n '49,97p' core/rag/data_post_processor/data_post_processor.py  # invoke + 选 runner
sed -n '59,73p' core/rag/rerank/weight_rerank.py                     # 加权融合公式