lambdagent/agentexample/qaagent67lambda/RAG_FUSION_PLAN.md

RAG Fusion 方案设计

Context

在 RAG v2 (BM25 + Embedding + Rerank) 基础上，进一步引入 RAG Fusion 机制：用 LLM 将用户原始 Query 扩展为多个语义等价/互补的子查询，并行检索后通过 RRF 融合排序，显著提升召回覆盖度和鲁棒性。

与 RAG v2 的关系：RAG Fusion 是 RAG v2 的上层增强，不替换 v2，而是在 v2 的混合检索之上叠加 Query Expansion。

架构对比

RAG v1 (当前):
  Query → BM25 → Top-5 → LLM

RAG v2 (混合检索):
  Query → BM25 + Vector → RRF → Rerank → Top-5 → LLM

RAG Fusion (本方案):
  Query → LLM Expansion → N 个子查询
    ├─► Q1 → BM25 + Vector → Top-K1
    ├─► Q2 → BM25 + Vector → Top-K2
    ├─► ...
    └─► QN → BM25 + Vector → Top-KN
         └─► RRF 融合 N 路结果 → Top-20
              └─► Rerank → Top-5 → LLM 生成

核心组件

1. Query Expansion (LLM 查询扩展)

用 LLM 将原始查询扩展为 3-5 个子查询，覆盖不同语义视角。

Prompt 设计:

你是一个搜索查询优化助手。请将以下用户问题改写为 {N} 个不同角度的搜索查询，
用于从海事法规知识库中检索相关文档。每个查询应覆盖不同的关键词和语义角度。
要求：
1. 保持原始问题的核心意图
2. 使用不同的关键词和表达方式
3. 分别侧重不同方面（定义、条款、对比、实例等）
4. 包含原始查询本身作为第一个

用户问题: {query}

请输出 JSON 数组，每个元素是一个查询字符串:

示例:

输入: "海上交通安全法与水污染防治法在船舶管理方面的规定有何异同？"

扩展为:
[
  "海上交通安全法与水污染防治法在船舶管理方面的规定有何异同",
  "海上交通安全法 船舶航行安全 管理规定 具体条款",
  "水污染防治法 船舶排放 防污染 管理要求",
  "船舶管理 安全法 环保法 法律对比",
  "海事法律 船舶监管 安全与环保 异同分析"
]

2. 多路并行检索

每个子查询调用 RAG v2 的 search() (BM25 + Vector 混合检索)，获取 Top-K 结果。

def multi_search(queries, top_k_per_query=10):
    all_results = []
    for q in queries:
        results = hybrid_search(q, top_k=top_k_per_query)  # RAG v2 的 search
        all_results.append(results)
    return all_results

优化: 可以并行化（ThreadPoolExecutor），但需注意：

• BM25 是本地计算，可并行
• Vector 检索是本地 numpy 计算，可并行
• Embedding API 调用，可批量合并（N 个 query 一次 API 调用）

3. RRF 融合 (多路结果合并)

def rrf_fuse_multi(all_results, k=60):
    scores = defaultdict(float)
    chunk_map = {}
    
    for results in all_results:
        for rank, r in enumerate(results):
            doc_id = f"{r['source']}#{r['chunk_id']}"
            scores[doc_id] += 1.0 / (k + rank + 1)
            chunk_map[doc_id] = r  # 保留 chunk 数据
    
    # 按融合分数排序
    sorted_ids = sorted(scores.items(), key=lambda x: -x[1])
    return [chunk_map[did] | {'fusion_score': score} for did, score in sorted_ids]

RRF 的优势:

• "多次出现的文档"得分更高 → 高置信度
• 不依赖原始分数的可比性（BM25 分数 vs 余弦相似度）
• 只用排名，天然归一化

4. Rerank (精排)

与 RAG v2 相同，调 qwen-rerank-8B 对 RRF Top-20 做精排，过滤 threshold=0.5。

5. 动态策略

根据查询复杂度动态调整：

def determine_expansion_count(query):
    """简单问题少扩展，复杂问题多扩展"""
    # 简单启发式：
    # - 包含对比词（异同、比较、区别）→ 5 个
    # - 包含多个实体 → 4 个
    # - 普通事实查询 → 3 个
    compare_words = ['异同', '比较', '区别', '对比', '不同', '相同']
    if any(w in query for w in compare_words):
        return 5
    if len(query) > 30:
        return 4
    return 3

新增/修改文件

新增 1 个文件

文件	用途
qaagent67/scripts/search_engine_fusion.py	RAG Fusion 引擎，包装 v2 的 search

修改 2 个文件

文件	改动
qaagent67/agent-config.yml	新增 rag.fusion 配置段
qademo/app.py	import 改为 from search_engine_fusion import search

search_engine_fusion.py 核心设计

"""
RAG Fusion: Multi-query expansion + parallel hybrid retrieval + RRF fusion + rerank
Wraps search_engine_v2.search() as the base retrieval unit.
"""

# 降级链: fusion → v2 (hybrid) → v1 (BM25)
try:
    from search_engine_v2 import search as hybrid_search
except ImportError:
    from search_engine import search as hybrid_search

def _expand_query(query, n=4):
    """用 LLM 扩展查询为 N 个子查询"""
    prompt = f"""..."""
    # 调 vLLM 本地 (免费) 或 SiliconFlow
    result = llm_call(prompt)
    queries = json.loads(result)
    return [query] + queries[:n-1]  # 原始查询始终保留

def _rrf_fuse_multi(all_results, k=60):
    """RRF 融合多路检索结果"""
    ...

def search(query, top_k=5):
    """RAG Fusion 主入口 (drop-in replacement)"""
    # 1. Query Expansion
    n = determine_expansion_count(query)
    queries = _expand_query(query, n=n)
    
    # 2. 多路检索
    all_results = [hybrid_search(q, top_k=10) for q in queries]
    
    # 3. RRF 融合
    fused = _rrf_fuse_multi(all_results, k=60)
    
    # 4. 去重 + 取 Top-20
    seen = set()
    unique = []
    for r in fused:
        key = f"{r['source']}#{r['chunk_id']}"
        if key not in seen:
            seen.add(key)
            unique.append(r)
    
    # 5. Rerank (复用 v2 的 rerank)
    reranked = _rerank(query, unique[:20])  # 用原始 query rerank
    
    # 6. 过滤 + 返回
    return [r for r in reranked if r['score'] >= threshold][:top_k]

agent-config.yml 新增配置

rag:
  # ... 现有 v2 配置 ...
  
  fusion:
    enabled: true
    defaultExpansions: 4        # 默认扩展查询数
    maxExpansions: 6            # 最大扩展数
    topKPerQuery: 10            # 每路检索的 Top-K
    expansionModel: local       # local (用 vLLM) 或 siliconflow
    # Query Expansion 用本地 vLLM (免费)，不用 SiliconFlow
    expansionPrompt: |
      你是一个搜索查询优化助手。请将以下问题改写为 {n} 个不同角度的搜索查询...

性能与成本分析

延迟分解 (以 4 路查询为例)

阶段	耗时	说明
Query Expansion	~2s	本地 vLLM 生成 4 个子查询
BM25 × 4	<100ms	本地计算，可并行
Embedding × 4	~200ms	批量调 API (4 个 query 一次)
Vector Search × 4	<100ms	numpy 计算
RRF Fusion	<10ms	纯排名合并
Rerank	~200ms	API 调用
总检索	~2.5s
LLM 生成	~10s	vLLM NPU
端到端	~12.5s	vs 当前 ~12s，增加 <1s

成本

项目	每次查询
Query Expansion	免费 (本地 vLLM)
Embedding (4 queries)	~$0.0001
Rerank	~$0.0005
总计	~$0.0006/查询

与各方案对比

方案	检索准确性	延迟	成本/查询	复杂度
v1 BM25	⭐⭐	12s	$0	低
v2 Hybrid	⭐⭐⭐⭐	12.3s	$0.001	中
Fusion	⭐⭐⭐⭐⭐	12.5s	$0.0006	中高

关键发现: Query Expansion 用本地 vLLM (NPU，~2s)，几乎不增加总延迟，但显著提升召回。

针对海事知识库的预期提升

查询类型	v2 Hybrid	Fusion	原因
"海上交通安全法与水污染防治法异同"	检索到法律原文	更全面覆盖两部法律的多个章节	扩展为分别搜索两部法律
"2019年船员注册人数"	精确匹配	同样精确	数据查找不需要多视角
"船员适任证书申请流程"	找到相关条款	覆盖申请条件+流程+所需材料	扩展覆盖不同方面
"琼州海峡定线制"	找到最新版	找到新旧版本+变更说明	扩展到"修订""实施""变化"

实施顺序

1. 先完成 RAG v2 (Embedding + Rerank) 部署
2. 在 v2 基础上叠加 Fusion:
   • 编写 search_engine_fusion.py
   • 更新 config
   • 改 app.py import (1 行)
3. A/B 测试: v1 vs v2 vs Fusion 三方对比

降级策略

Fusion 可用? ─── Yes → 用 Fusion
       │
       No
       │
  v2 可用? ─── Yes → 用 v2 (Hybrid)
       │
       No
       │
  用 v1 (BM25 only)

每一层都是 try/except 保护，任何组件失败不影响基本功能。