RAG重排序技术是提升RAG（Retrieval-Augmented Generation）系统检索质量的关键。该技术通过在初始检索后增加一个重排序步骤，确保最相关的内容用于生成响应。文章介绍了PDF文本处理、向量生成与存储、LLM重排序、关键词重排序等多个环节，并提供了代码实现和效果展示，旨在帮助读者理解并应用RAG重排序技术。

🔍 **初始检索与文档评分**: RAG系统首先进行初始检索，使用相似度搜索获取初步结果，并评估每个检索到的文档与查询的相关性，为后续重排序做准备。

🔄 **重排序方法**: 提供了多种重排序方法，包括基于LLM的重排序和关键词匹配重排序。LLM重排序利用大语言模型对检索结果进行打分，而关键词匹配重排序则通过关键词匹配对结果进行简单排序。

⚙️ **代码实现与流程**: 详细介绍了PDF文本处理、文本分块、向量生成与存储、LLM重排序、关键词重排序以及回答生成的代码实现，并展示了完整的RAG流程。

💡 **RAG流程**: 完整的RAG流程包括处理PDF文档、提取文本、分块、生成向量、构建向量库、初步检索、重排序和回答生成。流程中，重排序方法可选择LLM或关键词匹配。

RAG重排序

RAG重排序技术以提高RAG系统中的检索质量。重新排序充当初始检索后的第二个过滤步骤，以确保最相关的内容用于响应生成。

重排序的关键概念

1.初始检索：使用基本相似度搜索的第一遍（准确度较低但速度更快）2.文档评分：评估每个检索到的文档与查询的相关性3.重新排序：按相关性分数对文档进行排序4.选择：仅使用最相关的文档生成响应

具体代码实现

PDF文本处理与分块

PDF文件中提取全部文本

def extract_text_from_pdf(pdf_path: str) -> str:    """    从PDF文件中提取全部文本。    """    print(f"[PDF提取] 正在提取: {pdf_path}")    with open(pdf_path, 'rb') as f:        reader = PdfReader(f)        text = ""        for i, page in enumerate(reader.pages):            page_text = page.extract_text()            if page_text:                text += page_text            print(f"  - 已提取第{i+1}页")    print(f"[PDF提取] 完成，总长度: {len(text)} 字符\n")    return text

文本分割为带重叠的块

def chunk_text(text: str, n: int = 1000, overlap: int = 200) -> List[str]:    """    将文本分割为带重叠的块。    """    print(f"[分块] 每块{n}字符，重叠{overlap}字符")    chunks = []    for i in range(0, len(text), n - overlap):        chunks.append(text[i:i + n])    print(f"[分块] 完成，共{len(chunks)}块\n")    return chunks

向量生成与存储

阿里embedding模型批量生成文本向量

def create_embeddings(texts, model: str = EMBEDDING_MODEL) -> List[np.ndarray]:    """    用阿里embedding模型批量生成文本向量。    支持单条或多条文本。    """    if isinstance(texts, str):        texts = [texts]    print(f"[嵌入生成] 正在生成{len(texts)}条文本的向量...")    try:        response = TextEmbedding.call(            model=model,            input=texts,            api_key=ALI_API_KEY        )        if response.status_code == 200:            embeddings = [np.array(item['embedding']) for item in response.output['embeddings']]            print(f"[嵌入生成] 成功，返回{len(embeddings)}条向量\n")            return embeddings if len(embeddings) > 1 else embeddings[0]        else:            print(f"[嵌入生成] 失败: {response.message}")            return [np.zeros(1536)] * len(texts)    except Exception as e:        print(f"[嵌入生成] 异常: {e}")        return [np.zeros(1536)] * len(texts)

简单的向量存储与检索类

class SimpleVectorStore:    """    简单的向量存储与检索类。    """    def __init__(self):        self.vectors = []        self.texts = []        self.metadata = []    def add_item(self, text, embedding, metadata=None):        self.vectors.append(np.array(embedding))        self.texts.append(text)        self.metadata.append(metadata or {})    def similarity_search(self, query_embedding, k=5):        if not self.vectors:            return []        query_vector = np.array(query_embedding)        similarities = []        for i, vector in enumerate(self.vectors):            sim = np.dot(query_vector, vector) / (np.linalg.norm(query_vector) * np.linalg.norm(vector))            similarities.append((i, sim))        similarities.sort(key=lambda x: x[1], reverse=True)        results = []        for i in range(min(k, len(similarities))):            idx, score = similarities[i]            results.append({                "text": self.texts[idx],                "metadata": self.metadata[idx],                "similarity": score            })        return results

文档处理主流程

处理PDF文档，提取文本、分块、生成向量并构建向量库

def process_document(pdf_path: str, chunk_size: int = 1000, chunk_overlap: int = 200) -> SimpleVectorStore:    """    处理PDF文档，提取文本、分块、生成向量并构建向量库。    """    print("[主流程] 开始处理文档...")    extracted_text = extract_text_from_pdf(pdf_path)    text_chunks = chunk_text(extracted_text, chunk_size, chunk_overlap)    print("[主流程] 初始化向量库...")    vector_store = SimpleVectorStore()    print("[主流程] 为每个块生成向量...")    chunk_embeddings = create_embeddings(text_chunks)    for i, (chunk, embedding) in enumerate(zip(text_chunks, chunk_embeddings)):        print(f"[块{i+1}/{len(text_chunks)}] 已生成向量，长度: {len(chunk)} 字符")        vector_store.add_item(chunk, embedding, {"type": "chunk", "index": i})    print("[主流程] 文档处理完毕，向量库构建完成\n")    return vector_store

LLM重排序

LLM对初步检索结果进行重排序

def rerank_with_llm(query: str, results: List[Dict], top_n: int = 3, model: str = LLM_MODEL) -> List[Dict]:    """    使用LLM对初步检索结果进行重排序。    """    print(f"[重排序] 对{len(results)}个文档进行LLM重排序...")    scored_results = []    system_prompt = "你是文档相关性评估专家，请根据用户问题对下列文档相关性打分，0-10分，10分最相关。只返回分数，不要输出其他内容。"    for i, result in enumerate(results):        if i % 5 == 0:            print(f"[重排序] 正在评估第{i+1}/{len(results)}个文档...")        user_prompt = f"""问题: {query}\n\n文档内容:\n{result['text']}\n\n请对该文档与问题的相关性打分（0-10分）："""        try:            response = Generation.call(                model=model,                messages=[                    {"role": "system", "content": system_prompt},                    {"role": "user", "content": user_prompt}                ],                api_key=ALI_API_KEY,                result_format='message'            )            score_text = response.output.choices[0].message.content.strip()            score_match = re.search(r'\b(10|[0-9])\b', score_text)            if score_match:                score = float(score_match.group(1))            else:                print(f"[重排序] 警告: 无法提取分数，使用相似度分数代替: '{score_text}'")                score = result["similarity"] * 10        except Exception as e:            print(f"[重排序] LLM调用异常: {e}，使用相似度分数代替")            score = result["similarity"] * 10        scored_results.append({            "text": result["text"],            "metadata": result["metadata"],            "similarity": result["similarity"],            "relevance_score": score        })    reranked_results = sorted(scored_results, key=lambda x: x["relevance_score"], reverse=True)    print(f"[重排序] 完成，返回Top{top_n}结果\n")    return reranked_results[:top_n]

关键词重排序

关键词匹配对初步检索结果进行简单重排序

def rerank_with_keywords(query: str, results: List[Dict], top_n: int = 3) -> List[Dict]:    """    使用关键词匹配对初步检索结果进行简单重排序。    """    print(f"[关键词重排序] 对{len(results)}个文档进行关键词重排序...")    keywords = [word.lower() for word in query.split() if len(word) > 3]    scored_results = []    for result in results:        document_text = result["text"].lower()        base_score = result["similarity"] * 0.5        keyword_score = 0        for keyword in keywords:            if keyword in document_text:                keyword_score += 0.1                first_position = document_text.find(keyword)                if first_position < len(document_text) / 4:                    keyword_score += 0.1                frequency = document_text.count(keyword)                keyword_score += min(0.05 * frequency, 0.2)        final_score = base_score + keyword_score        scored_results.append({            "text": result["text"],            "metadata": result["metadata"],            "similarity": result["similarity"],            "relevance_score": final_score        })    reranked_results = sorted(scored_results, key=lambda x: x["relevance_score"], reverse=True)    print(f"[关键词重排序] 完成，返回Top{top_n}结果\n")    return reranked_results[:top_n]

回答生成

def generate_response(query: str, context: str, model: str = LLM_MODEL) -> str:    """    用大模型基于上下文生成回答。    """    print("[生成] 正在调用大模型生成回答...")    system_prompt = "你是一个AI助手，只能基于给定上下文回答问题。如果上下文无法直接回答，请回复：'信息不足，无法回答。'"    user_prompt = f"""上下文：\n{context}\n\n问题：{query}\n\n请只基于上述上下文简明准确作答。"""    try:        response = Generation.call(            model=model,            messages=[                {"role": "system", "content": system_prompt},                {"role": "user", "content": user_prompt}            ],            api_key=ALI_API_KEY,            result_format='message'        )        answer = response.output.choices[0].message.content.strip()        print(f"[生成] 回答生成成功: {answer}\n")        return answer    except Exception as e:        print(f"[生成] 回答生成异常: {e}")        return ""

RAG主流程

def rag_with_reranking(query: str, vector_store: SimpleVectorStore, reranking_method: str = "llm", top_n: int = 3, model: str = LLM_MODEL) -> Dict:    """    完整RAG流程，包含初步检索、重排序和回答生成。    """    print(f"[RAG] 开始RAG流程，检索+重排序方法: {reranking_method}")    query_embedding = create_embeddings(query)    initial_results = vector_store.similarity_search(query_embedding, k=10)    if reranking_method == "llm":        reranked_results = rerank_with_llm(query, initial_results, top_n=top_n)    elif reranking_method == "keywords":        reranked_results = rerank_with_keywords(query, initial_results, top_n=top_n)    else:        reranked_results = initial_results[:top_n]    context = "\n\n===\n\n".join([result["text"] for result in reranked_results])    response = generate_response(query, context, model)    print(f"[RAG] 流程完成\n")    return {        "query": query,        "reranking_method": reranking_method,        "initial_results": initial_results[:top_n],        "reranked_results": reranked_results,        "context": context,        "response": response    }

附录

效果展示

===== RAG重排序增强示例 =====[主流程] 开始处理文档...[PDF提取] 正在提取: data/2888年Java程序员找工作最新场景题.pdf  - 已提取第1页  - 已提取第2页  - 已提取第3页  - 已提取第4页  - 已提取第5页  - 已提取第6页  - 已提取第7页  - 已提取第8页  - 已提取第9页  - 已提取第10页[PDF提取] 完成，总长度: 6984 字符[分块] 每块1000字符，重叠200字符[分块] 完成，共9块[主流程] 初始化向量库...[主流程] 为每个块生成向量...[嵌入生成] 正在生成9条文本的向量...[嵌入生成] 成功，返回9条向量[块1/9] 已生成向量，长度: 1000 字符[块2/9] 已生成向量，长度: 1000 字符[块3/9] 已生成向量，长度: 1000 字符[块4/9] 已生成向量，长度: 1000 字符[块5/9] 已生成向量，长度: 1000 字符[块6/9] 已生成向量，长度: 1000 字符[块7/9] 已生成向量，长度: 1000 字符[块8/9] 已生成向量，长度: 1000 字符[块9/9] 已生成向量，长度: 584 字符[主流程] 文档处理完毕，向量库构建完成[主流程] 向量库已构建，开始不同重排序方法对比...=== 标准检索（无重排序） ===[RAG] 开始RAG流程，检索+重排序方法: none[嵌入生成] 正在生成1条文本的向量...[嵌入生成] 成功，返回1条向量[生成] 正在调用大模型生成回答...[生成] 回答生成成功: 根据上下文，Java程序员面试中常见的技术问题包括但不限于：- 分布式架构相关知识：如CAP原则、微服务架构、弹性设计以及Spring Cloud、CloudNative等相关技术框架的理解。- 网络编程技能：理解TCP/IP协议的三次握手与四次挥手过程，掌握Socket编程基础，了解select、poll、epoll等I/O多路复用技术。- 项目经验分享：面试官会询问候选人关于他们参与过的项目细节，特别是那些最自豪或最具挑战性的项目经历，以及如何解决遇到的技术难题。此外，还可能要求讲述曾经犯下的最大技术失误及从中学习到的经验教训。这些问题旨在考察求职者的实际工作经验、解决问题的能力和技术深度。[RAG] 流程完成Query: Java程序员面试中常见的技术问题有哪些？Response:根据上下文，Java程序员面试中常见的技术问题包括但不限于：- 分布式架构相关知识：如CAP原则、微服务架构、弹性设计以及Spring Cloud、CloudNative等相关技术框架的理解。- 网络编程技能：理解TCP/IP协议的三次握手与四次挥手过程，掌握Socket编程基础，了解select、poll、epoll等I/O多路复用技术。- 项目经验分享：面试官会询问候选人关于他们参与过的项目细节，特别是那些最自豪或最具挑战性的项目经历，以及如何解决遇到的技术难题。此外，还可能要求讲述曾经犯下的最大技术失误及从中学习到的经验教训。这些问题旨在考察求职者的实际工作经验、解决问题的能力和技术深度。=== LLM重排序 ===[RAG] 开始RAG流程，检索+重排序方法: llm[嵌入生成] 正在生成1条文本的向量...[嵌入生成] 成功，返回1条向量[重排序] 对9个文档进行LLM重排序...[重排序] 正在评估第1/9个文档...[重排序] 正在评估第6/9个文档...[重排序] 完成，返回Top3结果[生成] 正在调用大模型生成回答...[生成] 回答生成成功: 根据上下文，Java程序员面试中常见的技术问题涉及以下几个方面：- 并发编程、NIO、JVM等进阶知识。- Spring、Netty等流行框架的基本原理。- 对于分布式架构的理解，包括CAP原则、微服务架构（如Spring Cloud）、弹性设计及云原生(Cloud Native)技术。- 网络编程相关知识，比如TCP/IP协议的三次握手和四次挥手过程，Socket编程基础，以及I/O多路复用技术(select, poll, epoll)。这些问题旨在评估应聘者对于所列技术的实际掌握程度和技术深度。[RAG] 流程完成Query: Java程序员面试中常见的技术问题有哪些？Response:根据上下文，Java程序员面试中常见的技术问题涉及以下几个方面：- 并发编程、NIO、JVM等进阶知识。- Spring、Netty等流行框架的基本原理。- 对于分布式架构的理解，包括CAP原则、微服务架构（如Spring Cloud）、弹性设计及云原生(Cloud Native)技术。- 网络编程相关知识，比如TCP/IP协议的三次握手和四次挥手过程，Socket编程基础，以及I/O多路复用技术(select, poll, epoll)。这些问题旨在评估应聘者对于所列技术的实际掌握程度和技术深度。=== 关键词重排序 ===[RAG] 开始RAG流程，检索+重排序方法: keywords[嵌入生成] 正在生成1条文本的向量...[嵌入生成] 成功，返回1条向量[关键词重排序] 对9个文档进行关键词重排序...[关键词重排序] 完成，返回Top3结果[生成] 正在调用大模型生成回答...[生成] 回答生成成功: 根据上下文，Java程序员面试中常见的技术问题包括但不限于：- 分布式架构相关：CAP原则的理解、微服务架构的应用、弹性设计的概念以及Spring Cloud和Cloud Native等技术框架的使用。- 网络编程基础：TCP/IP协议的工作机制（如三次握手、四次挥手），Socket编程的基础知识，还有I/O多路复用技术（select, poll, epoll）的应用。此外，还强调了对项目经验和技术深度思考的考察，比如要求候选人分享他们最自豪或最近完成的项目、解决过的复杂难题、经历过的挑战性项目以及曾经犯下的最大技术失误等。这些问题旨在评估候选人的实际操作能力、解决问题的能力及面对错误的态度。[RAG] 流程完成Query: Java程序员面试中常见的技术问题有哪些？Response:根据上下文，Java程序员面试中常见的技术问题包括但不限于：- 分布式架构相关：CAP原则的理解、微服务架构的应用、弹性设计的概念以及Spring Cloud和Cloud Native等技术框架的使用。- 网络编程基础：TCP/IP协议的工作机制（如三次握手、四次挥手），Socket编程的基础知识，还有I/O多路复用技术（select, poll, epoll）的应用。此外，还强调了对项目经验和技术深度思考的考察，比如要求候选人分享他们最自豪或最近完成的项目、解决过的复杂难题、经历过的挑战性项目以及曾经犯下的最大技术失误等。这些问题旨在评估候选人的实际操作能力、解决问题的能力及面对错误的态度。进程已结束，退出代码为 0

完整代码示例

# -*- coding: utf-8 -*-"""RAG重排序增强脚本（基于阿里云通义千问Qwen）========================================本脚本实现了RAG检索增强中的重排序（rerank）流程，所有大模型调用均基于阿里云通义千问Qwen，密钥从api_keys.py读取。每一步均有详细控制台输出，便于观察效果。"""import osimport numpy as npimport jsonimport refrom typing import List, Dictfrom PyPDF2 import PdfReaderfrom dashscope import Generation, TextEmbedding# ========== 密钥配置 ==========# try:#     from test.api_keys import ALI_API_KEY# except ImportError:#     raise RuntimeError("未找到test/api_keys.py或未定义ALI_API_KEY，请配置API密钥！")ALI_API_KEY="sk-148deabc0bcf4fdeaa70a78eaa829c7e"# =============================LLM_MODEL = "qwen-max"  # 通义千问主力模型EMBEDDING_MODEL = "text-embedding-v2"  # 阿里云嵌入模型# ========== PDF文本处理与分块 ==========def extract_text_from_pdf(pdf_path: str) -> str:    """    从PDF文件中提取全部文本。    """    print(f"[PDF提取] 正在提取: {pdf_path}")    with open(pdf_path, 'rb') as f:        reader = PdfReader(f)        text = ""        for i, page in enumerate(reader.pages):            page_text = page.extract_text()            if page_text:                text += page_text            print(f"  - 已提取第{i+1}页")    print(f"[PDF提取] 完成，总长度: {len(text)} 字符\n")    return textdef chunk_text(text: str, n: int = 1000, overlap: int = 200) -> List[str]:    """    将文本分割为带重叠的块。    """    print(f"[分块] 每块{n}字符，重叠{overlap}字符")    chunks = []    for i in range(0, len(text), n - overlap):        chunks.append(text[i:i + n])    print(f"[分块] 完成，共{len(chunks)}块\n")    return chunks# ========== 向量生成与存储 ==========def create_embeddings(texts, model: str = EMBEDDING_MODEL) -> List[np.ndarray]:    """    用阿里embedding模型批量生成文本向量。    支持单条或多条文本。    """    if isinstance(texts, str):        texts = [texts]    print(f"[嵌入生成] 正在生成{len(texts)}条文本的向量...")    try:        response = TextEmbedding.call(            model=model,            input=texts,            api_key=ALI_API_KEY        )        if response.status_code == 200:            embeddings = [np.array(item['embedding']) for item in response.output['embeddings']]            print(f"[嵌入生成] 成功，返回{len(embeddings)}条向量\n")            return embeddings if len(embeddings) > 1 else embeddings[0]        else:            print(f"[嵌入生成] 失败: {response.message}")            return [np.zeros(1536)] * len(texts)    except Exception as e:        print(f"[嵌入生成] 异常: {e}")        return [np.zeros(1536)] * len(texts)class SimpleVectorStore:    """    简单的向量存储与检索类。    """    def __init__(self):        self.vectors = []        self.texts = []        self.metadata = []    def add_item(self, text, embedding, metadata=None):        self.vectors.append(np.array(embedding))        self.texts.append(text)        self.metadata.append(metadata or {})    def similarity_search(self, query_embedding, k=5):        if not self.vectors:            return []        query_vector = np.array(query_embedding)        similarities = []        for i, vector in enumerate(self.vectors):            sim = np.dot(query_vector, vector) / (np.linalg.norm(query_vector) * np.linalg.norm(vector))            similarities.append((i, sim))        similarities.sort(key=lambda x: x[1], reverse=True)        results = []        for i in range(min(k, len(similarities))):            idx, score = similarities[i]            results.append({                "text": self.texts[idx],                "metadata": self.metadata[idx],                "similarity": score            })        return results# ========== 文档处理主流程 ==========def process_document(pdf_path: str, chunk_size: int = 1000, chunk_overlap: int = 200) -> SimpleVectorStore:    """    处理PDF文档，提取文本、分块、生成向量并构建向量库。    """    print("[主流程] 开始处理文档...")    extracted_text = extract_text_from_pdf(pdf_path)    text_chunks = chunk_text(extracted_text, chunk_size, chunk_overlap)    print("[主流程] 初始化向量库...")    vector_store = SimpleVectorStore()    print("[主流程] 为每个块生成向量...")    chunk_embeddings = create_embeddings(text_chunks)    for i, (chunk, embedding) in enumerate(zip(text_chunks, chunk_embeddings)):        print(f"[块{i+1}/{len(text_chunks)}] 已生成向量，长度: {len(chunk)} 字符")        vector_store.add_item(chunk, embedding, {"type": "chunk", "index": i})    print("[主流程] 文档处理完毕，向量库构建完成\n")    return vector_store# ========== LLM重排序 ==========def rerank_with_llm(query: str, results: List[Dict], top_n: int = 3, model: str = LLM_MODEL) -> List[Dict]:    """    使用LLM对初步检索结果进行重排序。    """    print(f"[重排序] 对{len(results)}个文档进行LLM重排序...")    scored_results = []    system_prompt = "你是文档相关性评估专家，请根据用户问题对下列文档相关性打分，0-10分，10分最相关。只返回分数，不要输出其他内容。"    for i, result in enumerate(results):        if i % 5 == 0:            print(f"[重排序] 正在评估第{i+1}/{len(results)}个文档...")        user_prompt = f"""问题: {query}\n\n文档内容:\n{result['text']}\n\n请对该文档与问题的相关性打分（0-10分）："""        try:            response = Generation.call(                model=model,                messages=[                    {"role": "system", "content": system_prompt},                    {"role": "user", "content": user_prompt}                ],                api_key=ALI_API_KEY,                result_format='message'            )            score_text = response.output.choices[0].message.content.strip()            score_match = re.search(r'\b(10|[0-9])\b', score_text)            if score_match:                score = float(score_match.group(1))            else:                print(f"[重排序] 警告: 无法提取分数，使用相似度分数代替: '{score_text}'")                score = result["similarity"] * 10        except Exception as e:            print(f"[重排序] LLM调用异常: {e}，使用相似度分数代替")            score = result["similarity"] * 10        scored_results.append({            "text": result["text"],            "metadata": result["metadata"],            "similarity": result["similarity"],            "relevance_score": score        })    reranked_results = sorted(scored_results, key=lambda x: x["relevance_score"], reverse=True)    print(f"[重排序] 完成，返回Top{top_n}结果\n")    return reranked_results[:top_n]# ========== 关键词重排序 ==========def rerank_with_keywords(query: str, results: List[Dict], top_n: int = 3) -> List[Dict]:    """    使用关键词匹配对初步检索结果进行简单重排序。    """    print(f"[关键词重排序] 对{len(results)}个文档进行关键词重排序...")    keywords = [word.lower() for word in query.split() if len(word) > 3]    scored_results = []    for result in results:        document_text = result["text"].lower()        base_score = result["similarity"] * 0.5        keyword_score = 0        for keyword in keywords:            if keyword in document_text:                keyword_score += 0.1                first_position = document_text.find(keyword)                if first_position < len(document_text) / 4:                    keyword_score += 0.1                frequency = document_text.count(keyword)                keyword_score += min(0.05 * frequency, 0.2)        final_score = base_score + keyword_score        scored_results.append({            "text": result["text"],            "metadata": result["metadata"],            "similarity": result["similarity"],            "relevance_score": final_score        })    reranked_results = sorted(scored_results, key=lambda x: x["relevance_score"], reverse=True)    print(f"[关键词重排序] 完成，返回Top{top_n}结果\n")    return reranked_results[:top_n]# ========== 回答生成 ==========def generate_response(query: str, context: str, model: str = LLM_MODEL) -> str:    """    用大模型基于上下文生成回答。    """    print("[生成] 正在调用大模型生成回答...")    system_prompt = "你是一个AI助手，只能基于给定上下文回答问题。如果上下文无法直接回答，请回复：'信息不足，无法回答。'"    user_prompt = f"""上下文：\n{context}\n\n问题：{query}\n\n请只基于上述上下文简明准确作答。"""    try:        response = Generation.call(            model=model,            messages=[                {"role": "system", "content": system_prompt},                {"role": "user", "content": user_prompt}            ],            api_key=ALI_API_KEY,            result_format='message'        )        answer = response.output.choices[0].message.content.strip()        print(f"[生成] 回答生成成功: {answer}\n")        return answer    except Exception as e:        print(f"[生成] 回答生成异常: {e}")        return ""# ========== RAG主流程 ==========def rag_with_reranking(query: str, vector_store: SimpleVectorStore, reranking_method: str = "llm", top_n: int = 3, model: str = LLM_MODEL) -> Dict:    """    完整RAG流程，包含初步检索、重排序和回答生成。    """    print(f"[RAG] 开始RAG流程，检索+重排序方法: {reranking_method}")    query_embedding = create_embeddings(query)    initial_results = vector_store.similarity_search(query_embedding, k=10)    if reranking_method == "llm":        reranked_results = rerank_with_llm(query, initial_results, top_n=top_n)    elif reranking_method == "keywords":        reranked_results = rerank_with_keywords(query, initial_results, top_n=top_n)    else:        reranked_results = initial_results[:top_n]    context = "\n\n===\n\n".join([result["text"] for result in reranked_results])    response = generate_response(query, context, model)    print(f"[RAG] 流程完成\n")    return {        "query": query,        "reranking_method": reranking_method,        "initial_results": initial_results[:top_n],        "reranked_results": reranked_results,        "context": context,        "response": response    }# ========== main方法示例 ==========if __name__ == "__main__":    # 示例PDF路径（请替换为实际文件）    pdf_path = "data/2888年Java程序员找工作最新场景题.pdf"    # 示例问题    query = "Java程序员面试中常见的技术问题有哪些？"    print("\n===== RAG重排序增强示例 =====\n")    vector_store = process_document(pdf_path)    print("[主流程] 向量库已构建，开始不同重排序方法对比...\n")    # 1. 标准检索（无重排序）    print("\n=== 标准检索（无重排序） ===")    standard_results = rag_with_reranking(query, vector_store, reranking_method="none")    print(f"\nQuery: {query}")    print(f"\nResponse:\n{standard_results['response']}")    # 2. LLM重排序    print("\n=== LLM重排序 ===")    llm_results = rag_with_reranking(query, vector_store, reranking_method="llm")    print(f"\nQuery: {query}")    print(f"\nResponse:\n{llm_results['response']}")    # 3. 关键词重排序    print("\n=== 关键词重排序 ===")    keyword_results = rag_with_reranking(query, vector_store, reranking_method="keywords")    print(f"\nQuery: {query}")    print(f"\nResponse:\n{keyword_results['response']}")