2026实测！多模态RAG幻觉率仍超40%，5段代码构建可信检索增强生成管道

import fitz  # PyMuPDF
from PIL import Image
import io

class MultimodalDocumentParser:
    def __init__(self, vlm_client):
        self.vlm_client = vlm_client # 接入多模态大模型API进行图像理解

    def parse_pdf(self, pdf_path: str) -> List[dict]:
        doc = fitz.open(pdf_path)
        parsed_chunks = []
        
        for page_num, page in enumerate(doc):
            text_blocks = page.get_text("dict")["blocks"]
            page_text = ""
            image_descriptions = []
            
            for block in text_blocks:
                if "lines" in block:
                    for line in block["lines"]:
                        page_text += "".join([span["text"] for span in line["spans"]]) + " "
                        
                elif block["type"] == 1: # 图像块
                    img_bytes = block["image"]
                    img = Image.open(io.BytesIO(img_bytes))
                    # 调用VLM生成图像的Markdown结构化描述
                    desc = self.vlm_client.describe_image(img, prompt="将图表数据转化为Markdown表格或结构化文本。")
                    image_descriptions.append(desc)
                    
            # 将文本与同页的图像描述强绑定为一个Chunk
            chunk_content = page_text.strip()
            if image_descriptions:
                chunk_content += "\n\n[Attached Charts]\n" + "\n".join(image_descriptions)
                
            parsed_chunks.append({
                "page": page_num + 1,
                "content": chunk_content,
                "metadata": {"source": pdf_path, "page": page_num + 1}
            })
        return parsed_chunks

from sentence_transformers import CrossEncoder

class CrossEncoderReranker:
    def __init__(self, model_name: str = "BAAI/bge-reranker-v2-m3"):
        # 加载支持多语言与长文本的交叉编码器
        self.model = CrossEncoder(model_name, max_length=8192)

    def rerank(self, query: str, retrieved_chunks: List[dict], top_k: int = 3) -> List[dict]:
        """对初步召回的Chunks进行深度语义重排"""
        if not retrieved_chunks:
            return []
            
        # 构建 [Query, Document] 对
        pairs =  for chunk in retrieved_chunks]
        
        # 交叉编码器同时输入Query和Doc，计算精确的相关性得分
        scores = self.model.predict(pairs)
        
        # 将得分附加到Chunk上并排序
        for chunk, score in zip(retrieved_chunks, scores):
            chunk["rerank_score"] = float(score)
            
        retrieved_chunks.sort(key=lambda x: x["rerank_score"], reverse=True)
        
        # 过滤掉得分过低的噪声数据
        filtered_chunks = [c for c in retrieved_chunks[:top_k] if c["rerank_score"] > 0.1]
        return filtered_chunks

import re

class CitationGenerator:
    def __init__(self):
        self.reference_map = {}

    def build_prompt_with_anchors(self, chunks: List[dict]) -> str:
        """构建带有显式引用锚点的系统提示"""
        context_str = ""
        for i, chunk in enumerate(chunks):
            anchor_id = f"[DOC-{i+1}-P{chunk['metadata']['page']}]"
            self.reference_map[anchor_id] = chunk["metadata"]
            context_str += f"{anchor_id}\n{chunk['content']}\n\n"
            
        prompt = f"""基于以下参考资料回答问题。
要求：
1. 必须在陈述事实或数据后，使用对应的锚点标记（如[DOC-1-P2]）进行引用。
2. 如果资料中没有相关信息，请直接回答“资料中未提及”，严禁编造。

参考资料：
{context_str}"""
        return prompt

    def validate_citations(self, generated_text: str) -> dict:
        """后处理：校验大模型生成的引用标记是否合法且存在"""
        found_citations = re.findall(r'\[DOC-\d+-P\d+\]', generated_text)
        valid_citations = []
        hallucinated_citations = []
        
        for cite in found_citations:
            if cite in self.reference_map:
                valid_citations.append(cite)
            else:
                hallucinated_citations.append(cite)
                
        return {
            "is_trustworthy": len(hallucinated_citations) == 0,
            "valid": valid_citations,
            "hallucinated": hallucinated_citations,
            "sources": [self.reference_map[c] for c in valid_citations]
        }

class ConfidenceFilter:
    def __init__(self, min_score_threshold: float = 0.65, score_gap_threshold: float = 0.15):
        self.min_score = min_score_threshold
        self.gap_threshold = score_gap_threshold

    def evaluate_retrieval_quality(self, reranked_chunks: List[dict]) -> bool:
        """评估召回结果的整体置信度，决定是否允许大模型生成"""
        if not reranked_chunks:
            return False
            
        top_score = reranked_chunks[0]["rerank_score"]
        
        # 条件1：最高分必须超过绝对阈值
        if top_score < self.min_score:
            return False
            
        # 条件2：检查分数断层（Score Gap）
        # 如果第一名分数很高，但第二名断崖式下跌，说明只有第一条相关，属于弱召回
        if len(reranked_chunks) > 1:
            second_score = reranked_chunks[1]["rerank_score"]
            if (top_score - second_score) > self.gap_threshold and top_score < 0.8:
                return False
                
        return True

    def safe_generate(self, query: str, reranker: CrossEncoderReranker, llm_client) -> str:
        retrieved = reranker.rerank(query, initial_search(query))
        
        if not self.evaluate_retrieval_quality(retrieved):
            return "抱歉，当前知识库中未找到与您问题高度相关的信息，无法提供准确回答。"
            
        prompt = CitationGenerator().build_prompt_with_anchors(retrieved)
        return llm_client.generate 31251.t.kuaisou.com

import json

class RAGHallucinationEvaluator:
    def __init__(self, judge_llm_client):
        self.judge = judge_llm_client

    def evaluate_faithfulness(self, query: str, context: str, generated_answer: str) -> dict:
        """使用LLM-as-a-Judge评估生成答案的事实一致性"""
        eval_prompt = f"""你是一个严格的RAG系统评估专家。
请判断【生成答案】中的每一个事实陈述，是否都能在【参考资料】中找到明确依据。

参考资料：
{context}

生成答案：
{generated_answer}

请输出JSON格式：
{{
  "faithfulness_score": 0.0到1.0之间的浮点数,
  "unsupported_claims": ["列出答案中缺乏依据的幻觉陈述"],
  "reasoning": "简要说明评分理由"
}}"""
        response = self.judge.generate(eval_prompt, response_format={"type": "json_object"})
        return json.loads(response)

    def run_batch_evaluation(self, test_dataset: List[dict]) -> float:
        """批量运行评估集，计算系统整体幻觉率"""
        total_score = 0
        for item in test_dataset:
            result = self.evaluate_faithfulness(item["query"], item["context"], item["answer"])
            total_score += result["faithfulness_score"]
            
        avg_faithfulness = total_score / len(31252.t.kuaisou.com)
        hallucination_rate = 1.0 - avg_faithfulness
        print(f"Batch Evaluation Complete. Hallucination Rate: {hallucination_rate:.2%}")
        return hallucination_rate