🔍 RAGFlow QA 解析→检索 全流程源码分析

🔌 入口：FACTORY 注册 + Redis Stream 消费

QA 解析器通过 FACTORY 字典注册为 ParserType.QA 的处理器：

# task_executor.py:104-121
FACTORY = {
    ParserType.QA.value: qa,   # ← QA 解析器注册
    ...
}

# task_executor.py:265-316  build_chunks()
chunker = FACTORY[task["parser_id"].lower()]  # 获取 qa 模块
binary = await get_storage_binary(bucket, name)  # 从 MinIO 拉取文件
cks = await thread_pool_exec(
    chunker.chunk,                              # 调用 qa.chunk()
    task["name"], binary=binary,
    from_page=task["from_page"], to_page=task["to_page"],
    lang=task_language, callback=progress_callback,
    kb_id=task["kb_id"], parser_config=parser_config_for_chunk,
    tenant_id=task["tenant_id"],
)

关键点：QA 任务与其他 parser 类型走同一入口，通过 parser_id="qa" 分发到 rag.app.qa.chunk()。

📄 chunk() 按文件扩展名分发到 6 种解析器

🏗️ beAdoc() — Q&A chunk 的核心构建逻辑

# qa.py:291-300
def beAdoc(d, q, a, eng, row_num=-1):
    qprefix = "Question: " if eng else "问题："
    aprefix = "Answer: " if eng else "回答："
    d["content_with_weight"] = "\t".join(
        [qprefix + rmPrefix(q), aprefix + rmPrefix(a)])  # ← Q+A 拼接（向量嵌入 + LLM 上下文）
    d["content_ltks"] = rag_tokenizer.tokenize(q)          # ← 仅 Q 分词（全文检索）
    d["content_sm_ltks"] = rag_tokenizer.fine_grained_tokenize(d["content_ltks"])  # ← 仅 Q 细粒度分词
    if row_num >= 0:
        d["top_int"] = [row_num]
    return d

字段语义表：

🤖 三个 LLM 后处理阶段（可选，由 parser_config 控制）

# task_executor.py:637-648 — embedding 函数中的关键逻辑
async def embedding(docs, mdl, parser_config=None, callback=None):
    for d in docs:
        tts.append(d.get("docnm_kwd", "Title"))
        c = "\n".join(d.get("question_kwd", []))       # ← 优先使用 LLM 生成的问题
        if not c:
            c = d["content_with_weight"]                # ← 降级使用 Q+A 原文
        cnts.append(c)

📐 双向量加权融合

# task_executor.py:637-686
async def embedding(docs, mdl, parser_config=None, callback=None):
    # 1. 准备文本
    for d in docs:
        tts.append(d.get("docnm_kwd", "Title"))         # 标题文本
        c = "\n".join(d.get("question_kwd", []))         # LLM生成的问题（优先）
        if not c:
            c = d["content_with_weight"]                  # 降级：Q+A 原文
        cnts.append(c)

    # 2. 编码标题向量
    vts, c = await thread_pool_exec(mdl.encode, tts[0:1])
    tts = np.tile(vts[0], (len(cnts), 1))               # 所有 chunk 共用标题向量

    # 3. 批量编码内容向量
    for batch in cnts_batches:
        vts, c = await thread_pool_exec(batch_encode, batch)
        cnts_batches.append(vts)

    # 4. 加权融合
    filename_embd_weight = parser_config.get("filename_embd_weight", 0.1)
    title_w = float(filename_embd_weight)                # 默认 0.1
    vects = title_w * tts + (1 - title_w) * cnts        # 标题向量×0.1 + 内容向量×0.9

    # 5. 写入 chunk
    for i, d in enumerate(docs):
        v = vects[i].tolist()
        d["q_%d_vec" % len(v)] = v                       # 例: q_1024_vec

对 QA 的关键影响：

• 如果有 question_kwd（LLM 生成的等价问题），embedding 用它们而非 Q+A 原文
• 标题权重仅 0.1，内容权重 0.9 —— 向量主要反映 Q&A 内容语义
• 向量维度由 embedding 模型决定，动态生成字段名 q_{dim}_vec

💾 批量写入 ES / Infinity / OceanBase

# task_executor.py:1174-1272
async def insert_chunks(task_id, task_tenant_id, task_dataset_id, chunks, progress_callback):
    # 1. 处理母 chunk（mom/mom_with_weight）
    for ck in chunks:
        mom = ck.get("mom") or ck.get("mom_with_weight") or ""
        if mom:
            # 创建 mom chunk（仅保留必要字段，available_int=0 不可被检索）
            mothers.append(mom_ck)

    # 2. 先插入母 chunk
    for b in range(0, len(mothers), settings.DOC_BULK_SIZE):
        await thread_pool_exec(settings.docStoreConn.insert, ...)

    # 3. 批量插入 QA chunks
    for b in range(0, len(chunks), settings.DOC_BULK_SIZE):
        doc_store_result = await thread_pool_exec(
            settings.docStoreConn.insert,
            chunks[b:b + settings.DOC_BULK_SIZE],
            search.index_name(task_tenant_id),
            task_dataset_id,
        )
        # 取消时回滚已插入的 chunk

QA chunk 在 ES/Infinity 中的最终字段：

🔍 用户 query → 结构化全文查询表达式

# query.py:28-41 — 字段权重配置
class FulltextQueryer(QueryBase):
    def __init__(self):
        self.query_fields = [
            "title_tks^10",          # 标题分词 boost=10
            "title_sm_tks^5",        # 标题细粒度 boost=5
            "important_kwd^30",      # LLM 关键词 boost=30 ← 最高权重！
            "important_tks^20",      # LLM 关键词分词 boost=20
            "question_tks^20",       # LLM 生成问题分词 boost=20
            "content_ltks^2",        # Q-only 内容分词 boost=2
            "content_sm_ltks",       # Q-only 细粒度 boost=1（默认）
        ]

    def question(self, txt, tbl="qa", min_match: float = 0.6):
        # 1. 文本清洗：繁简转换、全半角转换、特殊字符过滤
        txt = rag_tokenizer.tradi2simp(rag_tokenizer.strQ2B(txt.lower()))
        
        # 2. 英文路径：词权重 + 同义词扩展 + bigram
        if not self.is_chinese(txt):
            tks_w = self.tw.weights(tks, preprocess=False)  # 词权重
            syns = [self.syn.lookup(tk) for tk, w in tks_w]  # 同义词
            # 构建 "(tk^weight synonym_expansion)" 查询表达式
        
        # 3. 中文路径：细粒度分词 + 同义词 + 子词 n-gram
        # 构建匹配表达式
        
        # 4. 返回 MatchTextExpr → 用于 ES/Infinity 全文检索

🎯 retrieval() — 6 阶段检索引擎

sequenceDiagram
    participant U as 用户查询
    participant D as Dealer.retrieval()
    participant Q as FulltextQueryer
    participant DS as DocStore (ES/Infinity)
    participant E as EmbeddingModel
    participant R as RerankModel

    U->>D: retrieval(question, embd_mdl, ...)
    D->>Q: question(txt) → MatchTextExpr
    D->>E: encode_queries(txt) → MatchDenseExpr
    D->>DS: search(text_expr, dense_expr, fusion)
    DS-->>D: 初检结果 (top_k × 2)
    D->>D: _prune_deleted_chunks()
    alt rerank_by_model
        D->>R: similarity(query, chunks)
        R-->>D: 精排分数
    else 本地重排
        D->>D: rerank() / rerank_with_knn()
    end
    D->>D: 按 similarity 阈值过滤
    D->>D: 分页 + group_docs 聚合
    D-->>U: SearchResult (ids, field, highlight, group_docs)
    

QA 检索的混合匹配逻辑：

1. 全文检索：用户 query 分词 → 匹配 content_ltks（Q-only 分词）、important_tks、question_tks 等字段
2. 向量检索：用户 query → embedding → 与 q_X_vec（Q+A 向量）做 cosine 相似度
3. 融合：全文分数 + 向量分数按权重合并
4. 精排：可选 Cross-Encoder 模型对 Top-N 做精确重排序
5. 过滤：相似度低于 threshold 的 chunks 被丢弃
6. 聚合：同一文档的 chunks 合并为 group_docs

📎 rerank_by_model() + insert_citations()

重排序（可选）：

# search.py — 使用 Cross-Encoder 模型精排
def rerank_by_model(self, rerank_mdl, sres, query, tkweight, vtweight, cfield, rank_feature):
    # 对初检结果的每个 chunk 文本，用 Cross-Encoder 模型重新打分
    # 结合 token 相似度和向量相似度
    ...
    return sres  # 按新分数重排

引用标注：

# search.py:242-332  insert_citations()
# 1. 将 LLM 回答按句子切分
# 2. 对每个句子编码为向量
# 3. 计算句子与检索到的 chunks 的混合相似度
# 4. 对相似度 > 0.63 的句子标注 [ID:n] 引用
# 5. 阈值逐步降低（0.63 → 0.50 → 0.40...）直到找到引用
for i, a in enumerate(pieces_):
    sim, tksim, vtsim = self.qryr.hybrid_similarity(
        ans_v[i], chunk_v, a_tks, chunks_tks, tkweight, vtweight)
    if mx >= thr:
        cites[idx[i]] = [str(ii) for ii in range(len(chunk_v)) if sim[ii] > mx]
# 输出: "退款请登录账户...[ID:3] 进入订单页面...[ID:7]"

📝 检索结果 → LLM Prompt 的格式化

# prompts/generator.py  kb_prompt()
def kb_prompt(kbinfos, max_tokens, hash_id):
    # 将检索到的 chunks 格式化为结构化 Prompt：
    # [ID]: 1
    # [Title]: 常见问题.xlsx
    # [Content]: 问题：如何退款？\t回答：请登录账户，进入订单页面...
    # [Metadata]: {"tag": "退款"}
    #
    # [ID]: 2
    # ...
    # Token 预算控制：超过 max_tokens 时截断

LLM 收到的上下文包含完整的 Q+A，可以直接基于标准答案回复，而非重新生成。