引言

上一篇博客，简单讲述了如何构建一个简单的 RAG 系统，但存在检索信息不精确、召回率低的问题，在这篇博客中，我们将引入 reranker 技术，实现 RAG 检索信息的重排序，提升检索信息的质量。

系统架构概述

我们的 RAG 系统包含以下核心组件：

1. 文档加载与处理：读取和解析文本文件

2. 文本分割：将长文档切分为适合处理的片段

3. 向量化与存储：将文本转换为向量并存入向量数据库

4. 检索与重排序：查找相关文档并优化排序

5. 答案生成：基于检索到的内容生成准确回答

核心实现详解

1. 文档加载与文本分割

def load_text_file(file_path: str) -> List[Document]:
    text_loader = TextLoader(file_path, encoding="utf-8")
    documents = text_loader.load()
    return documents

def split_documents(documents: List[Document],
                    spit_way: str = "\n\n",
                    chunk_size: int = 400,
                    chunk_overlap: int = 100) -> List[str]:
    text_splitter = RecursiveCharacterTextSplitter(
        separators=[spit_way],
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap
    )
    documents = text_splitter.split_documents(documents)
    doc_strs = []
    for doc in documents:
        doc_strs.append(doc.page_content.replace(spit_way, ""))
    return doc_strs

文本分割是 RAG 系统的基础，合适的块大小和重叠策略能平衡信息的完整性和检索的精准度。我们使用 LangChain 的 RecursiveCharacterTextSplitter，并支持自定义分隔符。

2. 向量化与存储

emb = OpenAIEmbeddings(
    model="Qwen/Qwen3-Embedding-8B",
    openai_api_base="https://api.siliconflow.cn/v1",
    openai_api_key="您的API密钥"
)

# 向量化存储
vector_store = FAISS.from_texts(split_docs, emb)

我们使用硅基流动平台的 Qwen 嵌入模型将文本转换为高维向量表示，并使用 FAISS 作为向量数据库进行高效相似性搜索。

3. 检索与重排序

传统的 RAG 系统直接使用向量检索结果，但这可能存在"语义相似但不相关"的问题。我们引入了重排序步骤来解决这一痛点：

def reranker(query: str, search_docs: List[str], top_k: int = 3) -> List[str]:
    reranker_result = call_siliconflow_rerank_api(query=query, documents=search_docs, top_k=top_k)
    if reranker_result['error'] is not None:
        print(reranker_result['error'])
    else:
        print("*" * 200)
        print("Reranker返回结果：")
        print(reranker_result['raw_response'])

    return reranker_result['reranked_text']

def process_retriever_output(input_dict):
    # 获取检索结果
    documents = input_dict["rag_search_documents"]
    question = input_dict["question"]

    # 合并检索结果
    docs_list = merge_rag_search_result(documents, is_show_log=True)

    # 重新排序 - 关键步骤！
    reranked_docs = reranker(question, docs_list, top_k=3)

    # 合并为上下文
    context = "\n".join(reranked_docs)

    return {
        "context": context,
        "question": question
    }

重排序使用专门的模型对初步检索结果进行精细打分和排序，筛选出真正与问题最相关的文档片段，大幅减少噪声并提高答案准确性。

4. 提示工程与答案生成

prompt = PromptTemplate.from_template(
    "- Role: 人事客服助理\n"
    "- Background: 用户在寻求与人事相关的问题解答...\n"
    # 详细的提示词设计
)

llm = ChatOpenAI(
    model="deepseek-ai/DeepSeek-R1-Distill-Qwen-7B",
    openai_api_base="https://api.siliconflow.cn",
    openai_api_key="您的API密钥",
    temperature=0.1,  # 低温度确保答案确定性
    max_tokens=2000
)

我们设计了详细的提示词，明确设定了 AI 的角色、背景、技能和约束条件，确保生成答案的准确性和专业性。

5. 完整流水线构建

# 创建检索器
retriever = vector_store.as_retriever(search_kwargs={'k': 10})

chain = (
        {
            "rag_search_documents": RunnableLambda(debug_retriever_input) | retriever,
            "question": RunnablePassthrough()
        }
        | RunnableLambda(process_retriever_output)
        | prompt
        | llm
        | StrOutputParser()
)

这个链式调用实现了完整的 RAG 流程：

1. 检索相关文档片段（初始检索）

2. 对结果进行重排序（精排）

3. 组合检索结果和用户问题

4. 使用精心设计的提示词

5. 调用语言模型生成答案

重排序的重要性

重排序是提升 RAG 系统性能的关键技术，它的价值体现在：

1. 解决语义相似性问题：向量检索可能返回语义相似但实际不相关的文档，重排序通过更精细的交互计算筛选真正相关内容

2. 提升答案准确性：通过提供更精准的上下文，大幅提高生成答案的准确度

3. 减少幻觉：减少无关上下文对 LLM 的干扰，降低模型编造信息的可能性

完整代码

访问硅基流动 reranker API

  from typing import List, Dict, Any

import requests


def call_siliconflow_rerank_api(query: str, documents: List[str], top_k: int = 3) -> Dict[str, Any]:
    """调用重排序API核心接口

    Args:
        query: 查询语句
        documents: 待重排序的文档列表
        top_k: 返回前K个最相关的文档

    Returns:
        包含重排序结果和原始响应的字典
    """
    if not query or not documents:
        return {"reranked_text": documents[:top_k], "raw_response": None, "error": "Invalid input"}

    url = "https://api.siliconflow.cn/v1/rerank"
    payload = {
        "query": query,
        "documents": documents,
        "return_documents": False,
        "max_chunks_per_doc": 1024,
        "overlap_tokens": 80,
        "model": "BAAI/bge-reranker-v2-m3"
    }
    headers = {
        "Authorization": "Bearer YOUR-API-KEY",
        "Content-Type": "application/json"
    }

    try:
        response = requests.post(url, json=payload, headers=headers, timeout=30)
        response.raise_for_status()

        res_json = response.json()
        reranker_result = res_json.get("results", [])
        # f返回文档会按照分数进行从大到小排序，相关性越高，排名靠前，可按照此顺序，利用index下标从原始数据中提取top_k个最相关的文档
        # 按照相关性得分排序并提取top_k文档
        reranked_documents = []
        for i, item in enumerate(reranker_result):
            if i >= top_k:
                break
            index = item.get('index', -1)
            if 0 <= index < len(documents):
                reranked_documents.append(documents[index])

        return {
            "reranked_text": reranked_documents,
            "raw_response": res_json,
            "error": None
        }

    except requests.exceptions.Timeout:
        return {
            "reranked_text": documents,
            "raw_response": None,
            "error": "Request timeout"
        }
    except requests.exceptions.RequestException as e:
        return {
            "reranked_text": documents,
            "raw_response": None,
            "error": f"API call failed: {str(e)}"
        }
    except (KeyError, ValueError, IndexError) as e:
        return {
            "reranked_text": documents,
            "raw_response": None,
            "error": f"Response parsing failed: {str(e)}"
        }

langchain+rag+reranker 代码

from typing import List

from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_text_splitters import RecursiveCharacterTextSplitter

emb = OpenAIEmbeddings(
    model="Qwen/Qwen3-Embedding-8B",
    openai_api_base="https://api.siliconflow.cn/v1",
    openai_api_key="YOUR-API-KEY",
)

llm = ChatOpenAI(
    model="deepseek-ai/DeepSeek-R1-Distill-Qwen-7B",
    openai_api_base="https://api.siliconflow.cn",
    openai_api_key="YOUR-API-KEY",
    temperature=0.1,
    max_tokens=2000
)


def load_text_file(file_path: str) -> List[Document]:
    text_loader = TextLoader(file_path, encoding="utf-8")
    documents = text_loader.load()
    return documents


def split_documents(documents: List[Document],
                    spit_way: str = "\n\n",
                    chunk_size: int = 400,
                    chunk_overlap: int = 100) -> List[str]:
    text_splitter = RecursiveCharacterTextSplitter(
        separators=[spit_way],
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap
    )
    documents = text_splitter.split_documents(documents)
    doc_strs = []
    for doc in documents:
        doc_strs.append(doc.page_content.replace(spit_way, ""))
    return doc_strs


def merge_rag_search_result(documents: List[Document], is_show_log: bool = True) -> List[str]:
    context = []
    if is_show_log:
        print("*" * 200)
        print("RAG检索结果：")
    for i, doc in enumerate(documents):  # 修复：添加 enumerate
        if is_show_log:
            print(f"第{i}个文档，{doc.page_content}")
            print("-" * 100)
        context.append(doc.page_content)
    return context


def reranker(query: str, search_docs: List[str], top_k: int = 3) -> List[str]:
    reranker_result = call_siliconflow_rerank_api(query=query, documents=search_docs, top_k=top_k)
    if reranker_result['error'] is not None:
        print(reranker_result['error'])
    else:
        print("*" * 200)
        print("Reranker返回结果：")
        print(reranker_result['raw_response'])

    return reranker_result['reranked_text']


def process_retriever_output(input_dict):
    # 获取检索结果
    documents = input_dict["rag_search_documents"]
    question = input_dict["question"]

    # 合并检索结果
    docs_list = merge_rag_search_result(documents, is_show_log=True)

    # 重新排序
    reranked_docs = reranker(question, docs_list, top_k=3)

    # 合并为上下文
    context = "\n".join(reranked_docs)

    return {
        "context": context,
        "question": question
    }


def debug_retriever_input(input_data):
    """
    调试RAG检索器输入的辅助函数
    """
    # 打印分隔线和查询问题信息
    print("*" * 200)
    print(f"传递给RAG检索器的查询问题: {input_data}")
    return input_data


if __name__ == '__main__':
    # 加载文本文件
    docs = load_text_file("../公司人事管理流程章程-V1.txt")
    split_way = "---#*split*#---"
    # 文本分割
    split_docs = split_documents(docs, split_way, 300, 50)
    # 向量化存储
    vector_store = FAISS.from_texts(split_docs, emb)
    # 定义提示词
    prompt = PromptTemplate.from_template(
        "- Role: 人事客服助理\n"
        "- Background: 用户在寻求与人事相关的问题解答，需要基于提供的参考信息（{context}）获取准确答案。用户可能正在处理与人力资源相关的事务，如招聘、员工关系、薪酬福利等，需要明确、准确的信息来解决问题。\n"
        "- Profile: 你是一位经验丰富的人事客服助理，对人力资源管理的各个模块有着扎实的了解，熟悉招聘流程、员工关系维护、薪酬福利政策等。你擅长从提供的文件或信息中快速提取关键内容，并以简洁明了的方式回答用户的问题。\n"
        "- Skills: 你具备快速阅读和理解文件的能力，能够准确识别与问题相关的关键信息。你擅长逻辑分析，能够将复杂的人事政策或流程简化为易于理解的答案。同时，你具备良好的沟通能力，能够以礼貌、专业的态度回应用户。\n"
        "- Goals: 基于提供的参考信息（{context}），准确回答用户的问题（{question}），确保回答内容严格遵循参考信息，不使用任何其他信息。如果无法从参考信息中找到答案，明确告知用户\"不知道\"，不编造答案。\n"
        "- Constrains: 你只能使用提供的参考信息（{context}）作为回答的依据，不能使用任何外部资源或编造答案。如果参考信息中没有相关内容，必须明确告知用户\"不知道\"。\n"
        "- OutputFormat: 以简洁明了的语言回答用户的问题，确保回答内容直接、准确。\n"
        "- Workflow:\n"
        " 1. 仔细阅读用户提供的参考信息（{context}），提取与问题（{question}）相关的关键内容。\n"
        " 2. 分析问题（{question}），确定其核心要点。\n"
        " 3. 根据提取的关键内容，结合问题的核心要点，给出准确的回答。如果参考信息中没有相关内容，明确告知用户\"不知道\"。\n"
        "- Examples:\n"
        " - 例子1：\n"
        "   - 参考信息：公司规定，员工每月可享受3天带薪病假。\n"
        "   - 问题：员工每月可以享受多少天带薪病假？\n"
        "   - 回答：根据公司规定，员工每月可享受3天带薪病假。\n"
        " - 例子2：\n"
        "   - 参考信息：公司目前没有关于远程工作的政策。\n"
        "   - 问题：公司是否有远程工作的政策？\n"
        "   - 回答：不知道。"
    )

    # 创建检索器
    retriever = vector_store.as_retriever(search_kwargs={'k': 10})

    chain = (
            {
                "rag_search_documents": RunnableLambda(debug_retriever_input) | retriever,
                "question": RunnablePassthrough()
            }
            | RunnableLambda(process_retriever_output)
            | prompt
            | llm
            | StrOutputParser()
    )
    
   
    result = chain.invoke("员工入职多久内必须签订劳务合同？")
    print("*" * 200)
    print("大模型返回结果：")
    print(result)

运行结果

********************************************************************************************************************************************************************************************************
传递给RAG检索器的查询问题: 员工入职多久内必须签订劳务合同？
********************************************************************************************************************************************************************************************************
RAG检索结果：
第0个文档，
第二十一条 人事信息系统
1. 公司建立并使用人事信息管理系统（HRIS），实现员工信息、组织架构、招聘、考勤、薪酬、绩效、培训等模块的数字化管理。
2. 确保系统数据准确、及时更新。设置严格的访问权限，保障员工个人信息安全。

第七章 附则
----------------------------------------------------------------------------------------------------
第1个文档，
第二十四条 生效与执行
本章程经[公司最高权力机构，如：董事会/总经理办公会]审议通过，自[XXXX年XX月XX日]起正式生效执行。原相关人事管理规定同时废止。
----------------------------------------------------------------------------------------------------
第2个文档，
第六条 招聘渠道与方式
1. 人力资源部根据岗位特点，选择合适渠道（如：公司官网/招聘页、主流招聘网站、校园招聘、猎头推荐、内部推荐、人才市场等）发布招聘信息。
2. 招聘方式包括但不限于：笔试、面试（初试、复试、终试）、测评（性格、能力、专业技能等）、背景调查。关键岗位或管理岗位需增加高管面试环节。
----------------------------------------------------------------------------------------------------
第3个文档，
 第五章 离职管理
----------------------------------------------------------------------------------------------------
第4个文档，
 第九条 劳动合同签订
1. 人力资源部应在员工入职之日起一个月内，与其依法签订书面《劳动合同》。合同文本由公司根据国家规定统一制定。
2. 合同内容应包含：用人单位和劳动者基本信息、合同期限、工作内容和工作地点、工作时间和休息休假、劳动报酬、社会保险、劳动保护、劳动条件和职业危害防护、法律规定的其他事项等。
3. 劳动合同一式两份，公司和员工各执一份。员工需亲自签署，公司加盖公章。
----------------------------------------------------------------------------------------------------
第5个文档，
 第二章 招聘与录用

 第五条 招聘需求与计划
1. 各部门根据业务发展、人员流动及编制情况，于每年末或项目启动前向人力资源部提交年度或专项《人员需求计划表》，说明需求岗位、人数、任职资格、到岗时间等。
2. 人力资源部汇总、审核各部门需求，结合公司战略、预算及编制，拟定公司年度/季度招聘计划，报公司管理层审批。
----------------------------------------------------------------------------------------------------
第6个文档，
第三章 入职管理

第八条 入职手续办理
1. 新员工按《录用通知书》要求，携带个人身份证、学历学位证书、离职证明（或应届生派遣证/就业协议）、体检报告、银行卡、社保公积金转移资料、照片等原件及复印件，于指定日期到人力资源部报到。
2. 人力资源部负责：
 审核资料真实性、完整性。
 指导员工填写《员工入职登记表》、《员工手册签收确认单》、《劳动合同》等文件。
 介绍公司基本情况、组织架构、主要规章制度（尤其是考勤、薪酬、保密、安全等）。
 安排工位、办公用品、门禁卡、邮箱账号等。
 引导至用人部门报到。
----------------------------------------------------------------------------------------------------
第7个文档，
第二条 适用范围
本章程适用于与公司建立劳动关系的所有员工（含试用期员工）。公司高级管理人员及其他特殊岗位人员，如劳动合同或聘用协议另有约定，从其约定；无特别约定者，适用本章程。
----------------------------------------------------------------------------------------------------
第8个文档，公司人事管理流程章程

 第一章 总则

 第一条 目的与宗旨
为建立科学、规范、高效、公正的人事管理体系，优化人力资源配置，保障公司与员工的合法权益，明确人力资源管理各环节的权责与流程，营造和谐稳定的劳动关系，提升组织效能与核心竞争力，依据《中华人民共和国劳动法》、《中华人民共和国劳动合同法》及相关法律法规，结合本公司实际情况，特制定本章程。本章程旨在确保人事管理的透明度、一致性与合规性，促进员工与公司共同发展。
----------------------------------------------------------------------------------------------------
第9个文档，
第六章 人事档案与信息系统

第二十条 人事档案管理
1. 人力资源部负责建立、保管和维护员工人事档案。
2. 档案内容：包括但不限于：应聘登记表、身份证件复印件、学历证明、前单位离职证明、劳动合同、录用/转正/调岗/晋升/奖惩通知、绩效考核表、培训记录、薪资调整记录、重要协议（保密、竞业限制等）、离职文件等。
3. 管理原则：确保档案的完整性、准确性、保密性和安全性。员工个人档案信息属保密范畴，查阅需经授权审批并登记。
----------------------------------------------------------------------------------------------------
********************************************************************************************************************************************************************************************************
Reranker返回结果：
{'id': '0198e58a20417cf8bda8279a5515607b', 'results': [{'index': 4, 'relevance_score': 0.98403233}, {'index': 6, 'relevance_score': 0.02784845}, {'index': 7, 'relevance_score': 0.0037654135}, {'index': 9, 'relevance_score': 0.0021912407}, {'index': 5, 'relevance_score': 0.0014437558}, {'index': 1, 'relevance_score': 0.00094361923}, {'index': 8, 'relevance_score': 0.000493605}, {'index': 2, 'relevance_score': 0.00037850367}, {'index': 0, 'relevance_score': 0.00021995338}, {'index': 3, 'relevance_score': 0.00020662983}], 'meta': {'billed_units': {'input_tokens': 1087, 'output_tokens': 0, 'search_units': 0, 'classifications': 0}, 'tokens': {'input_tokens': 1087, 'output_tokens': 0}}}
********************************************************************************************************************************************************************************************************
大模型返回结果：

根据参考信息，员工入职后一个月内必须签订书面《劳动合同》。

Process finished with exit code 0
 

痛点

细心的朋友可能已经发现，此处rag检索时，top_k设置为10，为何要设置这么大？聪明的你果真发现了问题！

我们曾以为引入ReRanker是解决问题的银弹，但事实证明它更像是一个“成本优化工具”，它筛选并压缩了信息，却未能改变“召回池中缺乏正确答案”这一根本事实。真正的挑战在于：即便设置top_k=10，正确的答案也可能湮没在后几位；一旦调小k值，它便彻底消失。这并非ReRanker的失败，而是整个检索链路的第一棒就出现了偏差。

接下来，让我们开启一场优化之旅，直击召回源头，精准捕获那些被遗漏的答案。

敬请期待~