Advanced RAG

在 Naive RAG 阶段，我们完成了“加载-索引-检索-生成”的基础闭环。然而，面对大规模、高专业性的技术文档（如学术论文、企业文档）时，Naive RAG 常面临召回噪声多、上下文割裂、专业术语偏差及生成幻觉等性能瓶颈。 本篇文档将带你实施 Advanced RAG 优化策略。通过这套深度调优组合拳，我们在 DeepSeek 核心论文（Conditional Memory via Scalable Lookup: A New Axis of Sparsity for Large Language Models）的测试中，取得了突破性的性能表现：

性能评估对比 (RAGAS)

评估指标	Naive RAG	Advanced RAG	提升幅度 (Delta)
Context Recall (召回率)	0.78	1.00	22%
Faithfulness (忠实度)	0.70	1.00	30%
Answer Relevancy (回答相关度)	0.83	0.89	6%

一、框架概览

• 数据层：
  • 内置示例数据集：平台在内置路径中预置了 “DeepSeek” 论文数据集。用户无需繁琐的数据准备，即可通过调用内置库实现 “一键上手”，快速验证 RAG 流程的闭环性。
  • 私有数据接入：系统具备高度的灵活性，支持用户通过数据通道自行上传 PDF、Markdown 等私有文档。通过简单的路径配置，即可实现从公共知识到行业私有知识的无缝切换。
• 推理层：
  • 核心引擎：采用 vLLM 作为推理后端，利用其 PagedAttention 技术提升并发处理能力。
  • 启动策略：执行显存割让策略（如 60% 显存用于 LLM），为后续的 Embedding（向量化） 和 Rerank（精排） 预留充足的计算余裕，确保单卡环境下多模型的稳定运行。
• 数据存储与检索层：
  • 本地向量库：集成 Milvus Lite。不同于集群版，Lite 版作为嵌入式库运行，仅通过 Python 加载即可操作本地 .db 文件，无需部署额外的数据库服务或维护复杂的容器集群，即可在单机实现百万级检索。
  • 嵌入模型：调用内置库加载 Qwen3-Embedding 系列模型，将 PDF/文本等非结构化数据转化为高维语义向量。
• 逻辑编排层：
  • 工作流管理：基于 LlamaIndex 构建，深度定制文档解析（SimpleDirectoryReader）与查询引擎。
  • Advanced RAG 范式：
    • 分层索引策略：采用层级化切片，确保检索小块（精准度）与提供大块背景（语境感）的平衡。
    • 多路混合检索：结合向量语义检索与 BM25 关键词检索，攻克专业术语缩写识别难题。
    • 重排序 (Rerank)：引入跨编码器模型对初筛结果进行精排，彻底剔除检索噪声，确保喂给 LLM 的知识点“百发百中”。
• 评估与性能量化层：
  • 评估框架：集成 RAGAS ，将 RAG 系统的优化从“感性调优”推向“数据驱动”的科学演进。
  • 核心量化维度：
    • Faithfulness (忠实度)：利用“LLM 裁判”自动化检测回答中的事实断言是否均能从参考资料中找到依据，严控幻觉率。
    • Answer Relevancy (答案相关度)：通过向量空间建模，量化 AI 回答与用户原始意图的契合程度，确保回答精准有力。
    • Context Recall (上下文召回率)：对比“标准答案”与“检索片段”，验证检索系统是否成功捕捉到了解决问题所需的全部关键事实点，精准定位召回漏洞。
  • 持续优化闭环：通过 RAGAS 提供的多维评分报告，实现针对性的参数微调与迭代。

二、优化步骤：构建进阶检索链路

为了彻底解决 Naive RAG 在处理 DeepSeek 论文时出现的知识割裂与召回偏差，我们在代码中实施了以下三大进阶策略：

1：数据层级化处理

传统的 Naive RAG 采用固定尺寸切片，往往会导致“语义断层”：切片太小则丢失语境，切片太大则检索噪声过高。

• 代码实现逻辑：
  • 多级切分：利用 HierarchicalNodeParser 构建了 [1536, 512, 256] 的三层金字塔结构。
    • 叶子节点：负责极细粒度的语义捕捉，提升关键词命中精度。
    • 父节点：作为叶子节点的“背景库”，在生成阶段提供完整的段落语境。
  • 深度数据清洗：在构建索引前，执行严格的类型检查与编码规范化（isinstance(str) 及 surrogates 过滤），彻底规避了非结构化 PDF 中常见的空块（NaN）导致的 Embedding 崩溃问题。

2：构建多路混合检索

单纯的向量检索依靠“语义相似度”，在面对 DeepSeek 论文中特有的技术术语时容易产生偏差。

• 代码实现逻辑：
  • 语义与词法的融合
    • 向量检索器：负责捕捉语义层面的“意思相近”。
    • BM25 检索器：负责精准匹配论文中的特定算法名称与技术指标。
  • 查询融合与重排序：采用 RRF 倒数排名融合算法，将两路检索结果合并。
  • 动态改写策略：通过 get_retrieval_strategy 函数，根据问题复杂度动态调整查询改写次数（num_queries）。简单不改写，复杂技术问题开启 3 路并行改写，大幅提升召回广度。

3：层级回溯与大模型精排

这是将性能推向 1.0 满分的关键环节，解决了“搜得到但挑不准”的难题。

• 代码实现逻辑：
  • 递归检索回溯：当检索器命中一个 256 或 512 Token 的子节点时，系统会自动“向上回溯”，将其对应的 1536 Token 的父节点提取出来送入大模型。这确保了检索的对象不是碎片，而是完整的知识背景。
  • 大模型重排序
    • 引入 LLM 作为“裁判”，对检索到的 Top 12 片段进行二次打分，仅保留相关度最高的 Top 7 进入生成阶段。
    • 将 choice_batch_size 设为 15，实现了“一次调用、全量评估”，将原本耗时的精排过程缩短了 60% 以上。

三、沐曦 (MetaX) 部署指南

本章节适用于 曦云 C500 等沐曦系列算力卡。

1. 硬件与基础环境

• 算力型号：曦云 C500 (64GB)
• 算力主机：
  • jiajia-mxc：vLLM / vllm：0.11.0 / Python 3.10 / maca 3.3.0.11
  • suanfeng-mxc：vLLM / 0.10.2 / Python 3.10 / maca 3.2.1.3

2. 基础步骤

• 进入算力容器，启动实例后，点击 JupyterLab 进入工作台。

  

3. 实现步骤

3.1 下载 LlamaIndex 与 Milvus Lite 框架

• 创建终端窗口(Terminal)

  

• 输入代码：

  pip install --target /data/llama_libs --no-deps -i https://mirrors.aliyun.com/pypi/simple/ -U \
  "pymilvus==2.6.6" milvus-lite orjson minio pathspec python-dateutil pytz six \
  llama-index-core llama-index-readers-file llama-index-llms-openai llama-index-llms-openai-like \
  llama-index-embeddings-huggingface llama-index-vector-stores-milvus llama-index-postprocessor-sbert-rerank  \
  llama-index-instrumentation llama-index-workflows llama-index-utils-workflow  \
  llama-index-retrievers-bm25 rank-bm25 bm25s  PyStemmer  \
  sentence-transformers pypdf docx2txt nest-asyncio ujson grpcio google-api-core protobuf banks griffe sqlalchemy dataclasses-json marshmallow typing-inspect fsspec filetype deprecated wrapt dirtyjson tenacity jinja2 pyyaml \
  pandas numpy nltk tiktoken requests charset-normalizer urllib3 certifi idna sniffio anyio h11 httpcore httpx mypy_extensions typing_extensions scikit-learn scipy joblib threadpoolctl tqdm pyarrow \
  ragas langchain-core langchain-openai langsmith requests_toolbelt "numpy<2.0" uuid_utils tenacity regex appdirs instructor docstring_parser langchain_community
  

• 完成下载后，新建一个新的终端:

3.2 启动 vLLM 推理

• 在新的终端内输入代码：

  python -m vllm.entrypoints.openai.api_server \
      --model /mnt/moark-models/Qwen3-8B \
      --gpu-memory-utilization 0.6 \
      --port 8000

• 当终端提示INFO： Application startup compete，则完成vLLM启动步骤。

3.3 创建并运行 Python 脚本

• 点击 Python File：

  

• 输入代码：

  import sys, os, asyncio, nest_asyncio, torch
  
  #  环境初始化
  PRIVATE_LIB = "/data/llama_libs"
  if PRIVATE_LIB not in sys.path:
      sys.path.insert(0, PRIVATE_LIB)
  
  nest_asyncio.apply()
  
  from datasets import Dataset
  from ragas import evaluate, RunConfig
  from ragas.metrics import Faithfulness, AnswerRelevancy, ContextRecall
  from ragas.llms import LlamaIndexLLMWrapper
  from ragas.embeddings import LlamaIndexEmbeddingsWrapper
  
  from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, StorageContext, Settings, PromptTemplate
  from llama_index.embeddings.huggingface import HuggingFaceEmbedding
  from llama_index.llms.openai_like import OpenAILike
  from llama_index.vector_stores.milvus import MilvusVectorStore
  from llama_index.core.node_parser import HierarchicalNodeParser, get_leaf_nodes
  from llama_index.core.retrievers import RecursiveRetriever, QueryFusionRetriever
  from llama_index.core.postprocessor import LLMRerank
  from llama_index.core.query_engine import RetrieverQueryEngine
  from llama_index.retrievers.bm25 import BM25Retriever
  
  # --- 工具函数 ---
  
  def clean_think_tag(text):
      text = str(text).strip()
      if not text:
          return "（粒术正在组织语言，请再试一次）"
      if "</think>" in text:
          res = text.split("</think>")[-1].strip()
          return res if res else "（粒术思考了很久，但没能生成有效回答）"
      return text
  
  
  def get_retrieval_strategy(user_input):
      """根据问题复杂度动态决定查询改写次数"""
      vague_keywords = ["总结", "概括", "讲了什么", "主要内容", "核心思想", "这本书", "介绍", "你好", "是谁"]
      if len(user_input) < 10 or any(k in user_input for k in vague_keywords):
          return 1  # 简单问候或极其模糊的问题不浪费 Token 改写
      return 3      # 复杂技术问题开启 3 路改写
  
  # 配置路径
  DATA_DIR = "/mnt/moark-models/deepseek_papers" 
  EMBED_PATH = "/mnt/moark-models/Qwen3-Embedding-8B"
  LLM_MODEL = "/mnt/moark-models/Qwen3-8B"
  
  #  主程序
  async def main():
      # --- Step 1: 模型初始化 ---
      print(">>> 正在初始化 8B 规模双模型...")
      Settings.embed_model = HuggingFaceEmbedding(
          model_name=EMBED_PATH, device="cuda", trust_remote_code=True,
          model_kwargs={"torch_dtype": torch.float16} 
      )
      Settings.llm = OpenAILike(
          model=LLM_MODEL, api_base="http://localhost:8000/v1", 
          api_key="fake", is_chat_model=True, timeout=120.0 
      )
  
      # --- Step 2: 层级索引构建---
      print(">>> 正在解析文档并进行深度清理...")
      reader = SimpleDirectoryReader(input_dir=DATA_DIR, recursive=True)
      documents = reader.load_data()
      
      # 1. 初始文档过滤
      documents = [doc for doc in documents if isinstance(doc.text, str) and len(doc.text.strip()) > 0]
  
      node_parser = HierarchicalNodeParser.from_defaults(chunk_sizes=[1536, 512,256])
      all_nodes = node_parser.get_nodes_from_documents(documents)
      raw_leaf_nodes = get_leaf_nodes(all_nodes) 
  
      # 2. 【核心修复】：深度类型检查与清理
      import math
      leaf_nodes = []
      for n in raw_leaf_nodes:
          text_content = n.get_content() # 获取节点完整内容
          
          # 严格检查：必须是字符串、不能为空、不能是 NaN
          if text_content is not None and isinstance(text_content, str):
              clean_text = text_content.strip()
              if clean_text:
                  # 额外保护：确保文本中没有会导致 Tokenizer 崩溃的非法字符
                  n.text = clean_text.encode('utf-8', 'ignore').decode('utf-8')
                  leaf_nodes.append(n)
          else:
              # 打印出问题节点的信息，方便排查
              print(f" 跳过非法节点: 类型={type(text_content)}, 预览={str(text_content)[:20]}")
  
      print(f">>> 深度过滤完成，有效叶子节点: {len(leaf_nodes)} / 原始: {len(raw_leaf_nodes)}")
  
      # 3. 索引构建
      vector_store = MilvusVectorStore(uri="./advanced_rag_final.db", dim=4096, overwrite=True)
      storage_context = StorageContext.from_defaults(vector_store=vector_store)
      storage_context.docstore.add_documents(all_nodes) 
      
      # 确保传入的是过滤后的 leaf_nodes
      index = VectorStoreIndex(leaf_nodes, storage_context=storage_context, show_progress=True)
  
      # --- Step 3: 检索器配置 ---
      vector_retriever = index.as_retriever(similarity_top_k=12)
      bm25_retriever = BM25Retriever.from_defaults(nodes=leaf_nodes, similarity_top_k=12)
      
      fusion_retriever = QueryFusionRetriever(
          [vector_retriever, bm25_retriever],
          similarity_top_k=12, 
          num_queries=1, # 初始值，后续动态改
          mode="reciprocal_rerank",
          use_async=False # Milvus 异步不稳定时建议设为 False
      )
  
      recursive_retriever = RecursiveRetriever(
          "vector",
          retriever_dict={"vector": fusion_retriever},
          node_dict={node.node_id: node for node in all_nodes}
      )
  
      # --- Step 4: 显存优化版精排 ---
      # choice_batch_size=15 大幅减少请求次数
      reranker = LLMRerank(llm=Settings.llm, choice_batch_size=15, top_n=7)
  
      # 有温度、有性格的 Prompt
      qa_prompt_tmpl = PromptTemplate(
      "【身份设定】你是DeepSeek RAG 知识库 AI 助理，名字叫做‘粒术’。请优先判断用户的意图\n\n"
      "【场景 A：社交寒暄】\n"
      "如果用户在打招呼（如：你好、早上好）或询问你是谁，请忽略下方的参考资料，用温暖、幽默、有礼貌的口吻直接回答。\n\n"
      "【场景 B：技术咨询】\n"
      "如果涉及 DeepSeek 论文或具体技术问题，请严格查阅资料：\n"
      "参考资料：\n---------------------\n{context_str}\n---------------------\n"
      "准则：1. 严禁发挥，必须引用资料数据。2. 资料未提及请幽默回复‘大脑里没存这段信息’。3. 保持专业严谨，带一点点幽默。\n\n"
      "用户提问：{query_str}\n"
      )
  
      query_engine = RetrieverQueryEngine.from_args(
          retriever=recursive_retriever,
          node_postprocessors=[reranker],
          text_qa_template=qa_prompt_tmpl
      )
  
  # --- Step 5: 测试与交互 ---
      print("\n" + "="*50 + "\nDeepSeek RAG 论文知识库已就绪！")
      
      while True:
          try:
              # 1. 获取输入
              raw_input = input("\n用户 >> ").strip()
              if raw_input.lower() in ['exit', 'quit', '退出']:
                  print("再见！祝你今天有个好心情。")
                  break
              if not raw_input: continue
  
              # 2. 编码清洗
              user_input = raw_input.encode('utf-8', 'ignore').decode('utf-8')
  
              # 3. 意图识别：判断是否为寒暄
              greetings = ["你好", "嗨", "hello", "你是谁", "早安", "午安", "在吗", "粒术"]
              # 如果输入包含寒暄词，或者输入非常短（通常是无意义的社交）
              is_greeting = any(g in user_input.lower() for g in greetings) or len(user_input) < 5
  
              if is_greeting:
                  # 【关键修改】：通过特定指令限定开场白的范围
                  greeting_prompt = (
                      "你是 DeepSeek 论文研究专家‘粒术’。请用温暖、专业且富有逻辑的口吻回应用户的问候。"
                      "在回复中明确告知用户：你已经深度研读了 DeepSeek 论文，专门负责解答论文中的技术架构、"
                      "实验结论及算法细节。不要提及任何生活技巧或电脑维修等无关内容。"
                  )
                  fallback_res = Settings.llm.complete(f"{greeting_prompt}\n用户输入：{user_input}")
                  print(f"\n粒术 >> {clean_think_tag(fallback_res.text)}")
              else:
                  # 【技术查询场景】：严格执行 RAG
                  n_queries = get_retrieval_strategy(user_input)
                  fusion_retriever.num_queries = n_queries
                  
                  # 执行检索与生成
                  response = query_engine.query(user_input)
                  final_answer = clean_think_tag(response.response)
  
                  # 4. 严格响应判定
                  # 如果 Reranker 过滤后没有节点，或者模型返回了空/默认错误词
                  if not final_answer or "Empty Response" in final_answer or len(final_answer) < 5:
                      print("\n粒术 >> 抱歉，由于知识库中没有直接相关的细节，我暂时无法给出精准的技术回答。您可以尝试换个问法。")
                  else:
                      print(f"\n粒术 >> {final_answer}")
  
          except UnicodeEncodeError:
              print("输入包含非法字符，请尝试手动输入。")
              continue
          except Exception as e:
              # 这里的打印是为了调试方便，如果不需要看到底层报错，可以改成自定义提示
              # print(f"DEBUG ERROR: {e}") 
              print("\n粒术 >> 抱歉，处理该请求时遇到了一些技术困难，请尝试换个问法或稍后再试。")
              continue
  
  if __name__ == "__main__":
      try:
          asyncio.run(main())
      except KeyboardInterrupt:
          pass

• 按Ctrl + S保存文件，并完成文件命名test。新建一个终端，输入python test.py，即可进入 RAG 系统。

  

四、燧原 (Enflame) 部署指南

本章节适用于 燧原 S60 等燧原系列算力卡。

1. 硬件与基础环境

• 算力型号：燧原 S60 (48GB)
• 算力主机：enflame-node：Ubuntu / 22.04 / Python 3.13 / ef 1.5.0.604

2. 基础步骤

• 进入算力容器，启动实例后，点击 JupyterLab 进入工作台。

  

3. 实现步骤

3.1 下载 LlamaIndex 与 Milvus Lite 框架

• 创建终端窗口(Terminal)

  

• 输入代码：

  pip install --target /data/llama_libs --no-deps -i https://mirrors.aliyun.com/pypi/simple/ -U \
  "pymilvus==2.6.6" milvus-lite orjson minio pathspec python-dateutil pytz six \
  llama-index-core llama-index-readers-file llama-index-llms-openai llama-index-llms-openai-like \
  llama-index-embeddings-huggingface llama-index-vector-stores-milvus llama-index-postprocessor-sbert-rerank \
  llama-index-instrumentation llama-index-workflows llama-index-utils-workflow \
  llama-index-retrievers-bm25 rank-bm25 bm25s  PyStemmer \
  sentence-transformers pypdf docx2txt nest-asyncio ujson grpcio google-api-core protobuf banks griffe sqlalchemy dataclasses-json marshmallow typing-inspect fsspec filetype deprecated wrapt dirtyjson tenacity jinja2 pyyaml \
  pandas numpy nltk tiktoken requests charset-normalizer urllib3 certifi idna sniffio anyio h11 httpcore httpx mypy_extensions typing_extensions scikit-learn scipy joblib threadpoolctl tqdm pyarrow \
  ragas langchain-core langchain-openai langsmith requests_toolbelt "numpy<2.0" uuid_utils tenacity regex appdirs instructor docstring_parser langchain_community jsonpatch colorama

• 完成下载后，新建一个新的终端:

3.2 启动 vLLM 推理

• 在新的终端内输入代码：

  vllm serve /mnt/moark-models/Qwen3-8B  --gpu-memory-utilization 0.6 --port 8000

• 当终端提示INFO： Application startup compete，则完成vLLM启动步骤。

3.3 创建并运行 Python 脚本

• 点击 Python File：

  

• 输入代码：

  import sys, os, asyncio, nest_asyncio, torch
  import torch_gcu
  from torch_gcu import transfer_to_gcu
  
  #  环境初始化
  PRIVATE_LIB = "/data/llama_libs"
  if PRIVATE_LIB not in sys.path:
      sys.path.insert(0, PRIVATE_LIB)
  
  nest_asyncio.apply()
  
  from datasets import Dataset
  from ragas import evaluate, RunConfig
  from ragas.metrics import Faithfulness, AnswerRelevancy, ContextRecall
  from ragas.llms import LlamaIndexLLMWrapper
  from ragas.embeddings import LlamaIndexEmbeddingsWrapper
  
  from llama_index.core import SimpleDirectoryReader, VectorStoreIndex, StorageContext, Settings, PromptTemplate
  from llama_index.embeddings.huggingface import HuggingFaceEmbedding
  from llama_index.llms.openai_like import OpenAILike
  from llama_index.vector_stores.milvus import MilvusVectorStore
  from llama_index.core.node_parser import HierarchicalNodeParser, get_leaf_nodes
  from llama_index.core.retrievers import RecursiveRetriever, QueryFusionRetriever
  from llama_index.core.postprocessor import LLMRerank
  from llama_index.core.query_engine import RetrieverQueryEngine
  from llama_index.retrievers.bm25 import BM25Retriever
  
  # --- 工具函数 ---
  
  def clean_think_tag(text):
      text = str(text).strip()
      if not text:
          return "（粒术正在组织语言，请再试一次）"
      if "</think>" in text:
          res = text.split("</think>")[-1].strip()
          return res if res else "（粒术思考了很久，但没能生成有效回答）"
      return text
  
  
  def get_retrieval_strategy(user_input):
      """根据问题复杂度动态决定查询改写次数"""
      vague_keywords = ["总结", "概括", "讲了什么", "主要内容", "核心思想", "这本书", "介绍", "你好", "是谁"]
      if len(user_input) < 10 or any(k in user_input for k in vague_keywords):
          return 1  # 简单问候或极其模糊的问题不浪费 Token 改写
      return 3      # 复杂技术问题开启 3 路改写
  
  # 配置路径
  DATA_DIR = "/mnt/moark-models/deepseek_papers" 
  EMBED_PATH = "/mnt/moark-models/Qwen3-Embedding-8B"
  LLM_MODEL = "/mnt/moark-models/Qwen3-8B"
  
  # 主程序
  async def main():
      # --- Step 1: 模型初始化 ---
      print(">>> 正在初始化 8B 规模双模型...")
      Settings.embed_model = HuggingFaceEmbedding(
          model_name=EMBED_PATH, device="cuda", trust_remote_code=True,
          model_kwargs={"torch_dtype": torch.float16} 
      )
      Settings.llm = OpenAILike(
          model=LLM_MODEL, api_base="http://localhost:8000/v1", 
          api_key="fake", is_chat_model=True, timeout=120.0 
      )
  
      # --- Step 2: 层级索引构建 ---
      print(">>> 正在解析文档并进行深度清理...")
      reader = SimpleDirectoryReader(input_dir=DATA_DIR, recursive=True)
      documents = reader.load_data()
      
      # 1. 初始文档过滤
      documents = [doc for doc in documents if isinstance(doc.text, str) and len(doc.text.strip()) > 0]
  
      node_parser = HierarchicalNodeParser.from_defaults(chunk_sizes=[1536, 512,256])
      all_nodes = node_parser.get_nodes_from_documents(documents)
      raw_leaf_nodes = get_leaf_nodes(all_nodes) 
  
      # 2. 【核心修复】：深度类型检查与清理
      import math
      leaf_nodes = []
      for n in raw_leaf_nodes:
          text_content = n.get_content() # 获取节点完整内容
          
          # 严格检查：必须是字符串、不能为空、不能是 NaN
          if text_content is not None and isinstance(text_content, str):
              clean_text = text_content.strip()
              if clean_text:
                  # 额外保护：确保文本中没有会导致 Tokenizer 崩溃的非法字符
                  n.text = clean_text.encode('utf-8', 'ignore').decode('utf-8')
                  leaf_nodes.append(n)
          else:
              # 打印出问题节点的信息，方便排查
              print(f" 跳过非法节点: 类型={type(text_content)}, 预览={str(text_content)[:20]}")
  
      print(f">>> 深度过滤完成，有效叶子节点: {len(leaf_nodes)} / 原始: {len(raw_leaf_nodes)}")
  
      # 3. 索引构建
      vector_store = MilvusVectorStore(uri="./advanced_rag_final.db", dim=4096, overwrite=True)
      storage_context = StorageContext.from_defaults(vector_store=vector_store)
      storage_context.docstore.add_documents(all_nodes) 
      
      # 确保传入的是过滤后的 leaf_nodes
      index = VectorStoreIndex(leaf_nodes, storage_context=storage_context, show_progress=True)
  
      # --- Step 3: 检索器配置 ---
      vector_retriever = index.as_retriever(similarity_top_k=12)
      bm25_retriever = BM25Retriever.from_defaults(nodes=leaf_nodes, similarity_top_k=12)
      
      fusion_retriever = QueryFusionRetriever(
          [vector_retriever, bm25_retriever],
          similarity_top_k=12, 
          num_queries=1, # 初始值，后续动态改
          mode="reciprocal_rerank",
          use_async=False # Milvus 异步不稳定时建议设为 False
      )
  
      recursive_retriever = RecursiveRetriever(
          "vector",
          retriever_dict={"vector": fusion_retriever},
          node_dict={node.node_id: node for node in all_nodes}
      )
  
      # --- Step 4: 显存优化版精排 ---
      # choice_batch_size=15 大幅减少请求次数
      reranker = LLMRerank(llm=Settings.llm, choice_batch_size=15, top_n=7)
  
      # 有温度、有性格的 Prompt
      qa_prompt_tmpl = PromptTemplate(
      "【身份设定】你是DeepSeek RAG 知识库 AI 助理，名字叫做‘粒术’。请优先判断用户的意图。\n\n"
      "【场景 A：社交寒暄】\n"
      "如果用户在打招呼（如：你好、早上好）或询问你是谁，请忽略下方的参考资料，用温暖、幽默、有礼貌的口吻直接回答。\n\n"
      "【场景 B：技术咨询】\n"
      "如果涉及 DeepSeek 论文或具体技术问题，请严格查阅资料：\n"
      "参考资料：\n---------------------\n{context_str}\n---------------------\n"
      "准则：1. 严禁发挥，必须引用资料数据。2. 资料未提及请幽默回复‘大脑里没存这段信息’。3. 保持专业严谨，带一点点幽默。\n\n"
      "用户提问：{query_str}\n"
      )
  
      query_engine = RetrieverQueryEngine.from_args(
          retriever=recursive_retriever,
          node_postprocessors=[reranker],
          text_qa_template=qa_prompt_tmpl
      )
  
  # --- Step 5: 测试与交互 ---
      print("\n" + "="*50 + "\nDeepSeek RAG 论文知识库已就绪！")
      
      while True:
          try:
              # 1. 获取输入
              raw_input = input("\n用户 >> ").strip()
              if raw_input.lower() in ['exit', 'quit', '退出']:
                  print("再见！祝你今天有个好心情。")
                  break
              if not raw_input: continue
  
              # 2. 编码清洗
              user_input = raw_input.encode('utf-8', 'ignore').decode('utf-8')
  
              # 3. 意图识别：判断是否为寒暄
              greetings = ["你好", "嗨", "hello", "你是谁", "早安", "午安", "在吗", "粒术"]
              # 如果输入包含寒暄词，或者输入非常短（通常是无意义的社交）
              is_greeting = any(g in user_input.lower() for g in greetings) or len(user_input) < 5
  
              if is_greeting:
                  # 【关键修改】：通过特定指令限定开场白的范围
                  greeting_prompt = (
                      "你是 DeepSeek 论文研究专家‘粒术’。请用温暖、专业且富有逻辑的口吻回应用户的问候。"
                      "在回复中明确告知用户：你已经深度研读了 DeepSeek 论文，专门负责解答论文中的技术架构、"
                      "实验结论及算法细节。不要提及任何生活技巧或电脑维修等无关内容。"
                  )
                  fallback_res = Settings.llm.complete(f"{greeting_prompt}\n用户输入：{user_input}")
                  print(f"\n粒术 >> {clean_think_tag(fallback_res.text)}")
              else:
                  # 【技术查询场景】：严格执行 RAG
                  n_queries = get_retrieval_strategy(user_input)
                  fusion_retriever.num_queries = n_queries
                  
                  # 执行检索与生成
                  response = query_engine.query(user_input)
                  final_answer = clean_think_tag(response.response)
  
                  # 4. 严格响应判定
                  # 如果 Reranker 过滤后没有节点，或者模型返回了空/默认错误词
                  if not final_answer or "Empty Response" in final_answer or len(final_answer) < 5:
                      print("\n粒术 >> 抱歉，由于知识库中没有直接相关的细节，我暂时无法给出精准的技术回答。您可以尝试换个问法。")
                  else:
                      print(f"\n粒术 >> {final_answer}")
  
          except UnicodeEncodeError:
              print("输入包含非法字符，请尝试手动输入。")
              continue
          except Exception as e:
              # 这里的打印是为了调试方便，如果不需要看到底层报错，可以改成自定义提示
              # print(f"DEBUG ERROR: {e}") 
              print("\n粒术 >> 抱歉，处理该请求时遇到了一些技术困难，请尝试换个问法或稍后再试。")
              continue
  
  if __name__ == "__main__":
      try:
          asyncio.run(main())
      except KeyboardInterrupt:
          pass

• 按Ctrl + S保存文件，并完成文件命名test。新建一个终端，输入python test.py，即可进入 RAG 系统。

  

五、附录：评估原始数据

为了确保测试结果的可追溯性，以下附上两套系统在终端输出的 RAGAS 评估原始记录。

• 1. Advance RAG：

表现评价：各项指标表现极其优异，成功解决了 Naive 架构下的知识割裂问题。

• 
• 2. Naive RAG：

表现评价：在处理高专业性的 DeepSeek 论文时，召回率与忠实度均存在明显短板。

•