RAG 查询转换六种策略：Demo、融合链路与 MemoryOS 选型

1
2
3
4
5

请针对用户的原始问题，从3个不同的语义角度生成3条独立的检索子查询，要求子查询覆盖问题的不同侧面，便于更全面地召回知识库相关文档。
仅输出3条查询，每行一条，不要序号和额外解释。

原始问题：{user_question}
子查询列表：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 初始化LLM与向量检索器
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_db = FAISS.load_local("./knowledge_base", embeddings, allow_dangerous_deserialization=True)
base_retriever = vector_db.as_retriever(search_kwargs={"k": 4})

# 多查询生成提示词
multi_query_prompt = ChatPromptTemplate.from_template("""
请针对用户的原始问题，从3个不同的语义角度生成3条独立的检索子查询，要求子查询覆盖问题的不同侧面，便于更全面地召回知识库相关文档。
仅输出3条查询，每行一条，不要序号和额外解释。

原始问题：{user_question}
子查询列表：
""")

def generate_multi_queries(query: str) -> list[str]:
    """生成多路子查询"""
    chain = multi_query_prompt | llm
    result = chain.invoke({"user_question": query}).content
    return [q.strip() for q in result.split("\n") if q.strip()]

def multi_query_retrieve(query: str) -> list:
    """多查询检索 + 简单去重合并"""
    queries = generate_multi_queries(query)
    all_docs = []
    for q in queries:
        docs = base_retriever.get_relevant_documents(q)
        all_docs.extend(docs)
    # 按文档内容去重
    seen = set()
    unique_docs = []
    for doc in all_docs:
        if doc.page_content not in seen:
            seen.add(doc.page_content)
            unique_docs.append(doc)
    return unique_docs

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

def rrf_fuse(doc_lists: list[list], k: int = 60, weights: list[float] = None) -> list:
    """
    RRF倒数排名融合
    :param doc_lists: 多路检索结果列表，每个元素是一路文档
    :param k: RRF常数，默认60
    :param weights: 每路检索的权重，默认全1
    :return: 融合排序后的文档列表
    """
    if weights is None:
        weights = [1.0] * len(doc_lists)
    
    fused_scores = {}
    doc_map = {}
    
    # enumerate遍历每一路检索，匹配对应权重
    for i, docs in enumerate(doc_lists):
        weight = weights[i]
        for rank, doc in enumerate(docs):
            doc_key = doc.page_content
            doc_map[doc_key] = doc
            # 计算单路RRF得分并加权
            score = (1 / (rank + k)) * weight
            fused_scores[doc_key] = fused_scores.get(doc_key, 0) + score
    
    # 按总分降序排序
    sorted_items = sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)
    return [doc_map[key] for key, _ in sorted_items]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

from langchain_community.retrievers import BM25Retriever
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain.retrievers import EnsembleRetriever

# 准备知识库文档列表
docs = [...]  # 你的Document对象列表

# 1. 初始化BM25关键词检索器
bm25_retriever = BM25Retriever.from_documents(docs)
bm25_retriever.k = 4

# 2. 初始化FAISS向量检索器
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
faiss_db = FAISS.from_documents(docs, embeddings)
faiss_retriever = faiss_db.as_retriever(search_kwargs={"k": 4})

# 3. 集成检索器（底层内置RRF融合，支持自定义权重）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, faiss_retriever],
    weights=[0.5, 0.5]
)

# 直接调用检索
result_docs = ensemble_retriever.get_relevant_documents("耳机进水保修多久？")

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

示例1：
用户问题：耳机进水保修多久？
回退后问题：电子产品售后保修政策

示例2：
用户问题：月度报表里研发费用抵扣标准？
回退后问题：企业研发费用财务抵扣管理规定

现在请按上面例子，改写问题：{user_question}
仅输出改写后的回退问题，不要额外解释、不要序号。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# Step-Back 提示词模板
step_back_prompt = ChatPromptTemplate.from_template("""
示例1：
用户问题：耳机进水保修多久？
回退后问题：电子产品售后保修政策

示例2：
用户问题：月度报表里研发费用抵扣标准？
回退后问题：企业研发费用财务抵扣管理规定

现在请按上面例子，改写问题：{user_question}
仅输出改写后的回退问题，不要额外解释、不要序号。
""")

def generate_step_back_query(query: str) -> str:
    """生成回退宏观问题"""
    chain = step_back_prompt | llm
    return chain.invoke({"user_question": query}).content

def step_back_retrieve(query: str, retriever) -> list:
    """Step-Back双路检索 + RRF融合"""
    # 1. 生成两路查询
    original_query = query
    step_back_query = generate_step_back_query(query)
    # 2. 双路独立检索
    original_docs = retriever.get_relevant_documents(original_query)
    step_back_docs = retriever.get_relevant_documents(step_back_query)
    # 3. RRF融合排序
    fused_docs = rrf_fuse([original_docs, step_back_docs], k=60, weights=[0.6, 0.4])
    return fused_docs

# 使用示例
final_docs = step_back_retrieve("耳机进水保修多久？", ensemble_retriever)

1
2
3
4
5

请基于用户的问题，生成一段完整、专业的假设性回答文档，要求内容符合企业知识库的陈述风格，包含对应领域专业术语，逻辑通顺，长度控制在150-200字。
仅输出假设文档内容，不要额外解释、不要标题。

用户问题：{user_question}
假设回答文档：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

hyde_prompt = ChatPromptTemplate.from_template("""
请基于用户的问题，生成一段完整、专业的假设性回答文档，要求内容符合企业知识库的陈述风格，包含对应领域专业术语，逻辑通顺，长度控制在150-200字。
仅输出假设文档内容，不要额外解释、不要标题。

用户问题：{user_question}
假设回答文档：
""")

def generate_hyde_doc(query: str) -> str:
    """生成假设伪文档"""
    chain = hyde_prompt | llm
    return chain.invoke({"user_question": query}).content

def hyde_retrieve(query: str, retriever) -> list:
    """HyDE混合检索 + RRF融合"""
    # 原始问句检索
    original_docs = retriever.get_relevant_documents(query)
    # HyDE伪文档检索
    hyde_doc = generate_hyde_doc(query)
    hyde_docs = retriever.get_relevant_documents(hyde_doc)
    # 两路RRF融合，伪文档权重更高
    fused_docs = rrf_fuse([original_docs, hyde_docs], k=60, weights=[0.4, 0.6])
    return fused_docs

1
2
3
4
5

请将用户的复杂综合问题拆解为3-4个互相独立的子问题，每个子问题对应一个独立的知识维度，确保所有子问题覆盖原始问题的全部核心需求。
仅输出子问题列表，每行一个，不要序号和额外解释。

原始问题：{user_question}
子问题列表：

1
2
3
4
5
6
7
8

请结合以下所有子问题的检索文档与子答案，针对用户的原始问题生成一份完整、逻辑连贯的最终回答。
要求：1. 覆盖所有子维度信息，结构清晰；2. 所有结论均有文档依据；3. 语言专业严谨。

原始问题：{original_question}
子问题与对应参考资料：
{sub_results}

最终回答：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

map_prompt = ChatPromptTemplate.from_template("""
请将用户的复杂综合问题拆解为3-4个互相独立的子问题，每个子问题对应一个独立的知识维度，确保所有子问题覆盖原始问题的全部核心需求。
仅输出子问题列表，每行一个，不要序号和额外解释。

原始问题：{user_question}
子问题列表：
""")

reduce_prompt = ChatPromptTemplate.from_template("""
请结合以下所有子问题的检索文档与子答案，针对用户的原始问题生成一份完整、逻辑连贯的最终回答。
要求：1. 覆盖所有子维度信息，结构清晰；2. 所有结论均有文档依据；3. 语言专业严谨。

原始问题：{original_question}
子问题与对应参考资料：
{sub_results}

最终回答：
""")

def map_reduce_rag(query: str, retriever) -> str:
    # 1. Map阶段：拆分子问题
    map_chain = map_prompt | llm
    sub_queries = [q.strip() for q in map_chain.invoke({"user_question": query}).content.split("\n") if q.strip()]
    
    # 2. 每个子问题独立执行Step-Back检索 + 生成子答案（生产环境可异步并发）
    sub_results = []
    all_fused_docs = []
    for sq in sub_queries:
        # 子问题嵌套Step-Back双路检索
        docs = step_back_retrieve(sq, retriever)
        all_fused_docs.extend(docs)
        # 生成子答案
        context = "\n".join([f"参考文档{i+1}：{d.page_content}" for i, d in enumerate(docs[:3])])
        sub_answer = llm.invoke(f"参考资料：\n{context}\n请回答问题：{sq}").content
        sub_results.append(f"【子问题】{sq}\n【子答案】{sub_answer}")
    
    # 3. 全局文档去重重排
    final_docs = rrf_fuse([all_fused_docs], k=60)
    
    # 4. Reduce阶段：汇总生成最终答案
    reduce_chain = reduce_prompt | llm
    final_answer = reduce_chain.invoke({
        "original_question": query,
        "sub_results": "\n---\n".join(sub_results)
    }).content
    return final_answer

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

用户输入Query
  ↓
【前置智能路由层】规则+LLM分类，判定问题复杂度等级
  ├─ 等级1：极简单点FAQ → 直接Ensemble混合检索 → LLM生成答案 → 返回
  ├─ 等级2：细节专业问题 → Step-Back+HyDE双改写 → 多路Ensemble检索 → RRF融合 → LLM生成答案 → 返回
  └─ 等级3：复杂综合问题 → 开启Map-Reduce全链路
        ↓ Map阶段：LLM拆分为多个独立子问题Q1/Q2/Q3
        ├─ 每个子问题单独执行 Step-Back + HyDE 改写：原问句 + 回退宏观问句 + 伪文档三检索
        ├─ 每路子问题产出3路文档列表，每路均为BM25+向量混合检索结果
        ↓ 全部多路文档汇总 → RRF统一融合重排 → 去重 → MMR多样性过滤
        ↓ Reduce阶段：把融合后的全部上下文交给LLM合并输出最终答案 → 返回