RAGFlow 源码阅读与实践：如何针对“医学检验报告”做 Chunk 深度优化？

Moon指针.

384人浏览 · 2026-05-07 15:49:49

Moon指针. · 2026-05-07 15:49:49 发布

摘要

在医疗垂类大模型落地中，RAG（检索增强生成）是不可或缺的一环。然而，标准的文本分块（Chunking）策略在面对高度结构化的医学检验报告时，往往会导致“指标名称”与“数值/参考区间”的割裂。本文基于开源框架 RAGFlow 的源码阅读，深入剖析其文档处理流程，并提供一套针对医学检验报告的定制化 Chunk 优化方案与代码实现。

1. 痛点分析：为什么标准 Chunker 搞不定检验报告？

一份标准的医学血常规报告通常包含两部分：

全局上下文：患者姓名、性别、年龄、检验日期。
表格数据：项目名称（如白细胞计数）、结果（如 12.5）、提示（↑）、参考区间（3.5-9.5）。

如果使用传统的关键字符或者按 Token 长度强行切分：

丢失上下文：切出来的 Chunk 里只有“白细胞计数 12.5”，大模型不知道这是谁、哪天的报告。
数据割裂：表格的一行被切成两半，“指标”在 Chunk A，“参考区间”被切到了 Chunk B，导致 RAG 回答幻觉。

2. RAGFlow 源码探秘：它的解析强在哪里？

阅读 RAGFlow 的核心模块可以发现，它依托强大的视觉模型OCR + 布局分析，能够精准识别出 Document 中的 Table（表格） 和 Text（文本）。

同时RAGFlow适配多种不同的信息源格式，例如MD，PDF，txt，word各种。

默认策略下，RAGFlow 会尽量保持表格的完整性转化为 Markdown 格式，这种无法解决医学检验报告RAG的痛点，比如表格太长，仍然会被截断。所以我们需要在它解析出表格后，进行“行级（Row-level） + 上下文注入（Context Injection）+动态语义映射”的二次优化。

3. 医学报告 Chunk 优化策略 (Medical-Chunking)

为了彻底解决大模型在医学报告上的“检索幻觉”和“语义鸿沟”（例如用户问“肝功能”，但报告里只有“ALT”），我设计了一套 “行级切分 + 动态语义映射” 的混合优化逻辑：

Header 上下文提取：提取 Patient_Info（姓名、日期等）。
行级切割 (Row-based Split)：以 Markdown 表格的行边界为基准，确保单行指标不被腰斩。
向量语义匹配与增强 (Vector Semantic Enrichment)：
- 预先定义一个医学检验大类知识库（如：肝功能、肾功能、血常规、凝血等）。
- 在切分每一行时，利用轻量级 Embedding 模型（本文采用的是BGE）将当前“检验项目名称”与知识库进行向量相似度计算。
- 如果相似度超过阈值（本文涉及的是0.6），则将匹配到的“语义分类”动态注入到该 Chunk 中。
组装终极 Chunk：最终每个 Chunk 不仅包含完整数据和患者上下文，还带上了高维语义标签。

4. 核心改造代码实现

import re
import warnings
from typing import List
from sentence_transformers import SentenceTransformer, util

warnings.filterwarnings("ignore")

class MedicalSemanticChunkerBGE:

    def __init__(self, model_name: str = "BAAI/bge-small-zh-v1.5", similarity_threshold: float = 0.65):
        """
        初始化基于 BGE 的医学语义分块器
        :param model_name: BGE 模型名称，默认使用 bge-small-zh-v1.5，生产环境推荐 bge-large-zh-v1.5 或 bge-m3
        :param similarity_threshold: 语义匹配相似度阈值
        """
        self.similarity_threshold = similarity_threshold
        
        print(f"正在加载 BGE 向量模型 [{model_name}]")
        self.encoder = SentenceTransformer(model_name)
        print("模型加载完成！")
        
        # 模拟知识库
        # 预定义医学大类知识库（可根据实际医院科室和检验单类型扩充）
        self.medical_categories = [
            "肝功能", "肾功能", "血常规", "血脂", 
            "凝血功能", "甲状腺功能", "心肌酶", "肿瘤标志物", "电解质"
        ]
        
        # 预先计算医学大类的 Embedding 向量 (只需计算一次)
        # BGE 官方建议：对于非对称检索任务，query 需要加 prompt，但我们这里是短文本对称语义匹配，直接 encode 即可
        self.category_embeddings = self.encoder.encode(self.medical_categories, convert_to_tensor=True)

    def extract_global_context(self, text: str) -> str:
        """
        从报告原文中提取全局上下文信息（如患者基本信息）
        """

        # 使用正则提取患者属性
        context_pattern = r"(姓名[:：].*?|年龄[:：].*?|性别[:：].*?|检验日期[:：].*?(?=\n|\s|$))"
        matches = re.findall(context_pattern, text)
        
        if matches:
            return "  ".join([m.strip() for m in matches])
        return "未知患者上下文"

    def _get_semantic_category(self, item_name: str) -> str:

        """
        核心亮点：使用 BGE 模型计算检验项目与医学大类的余弦相似度
        """

        if not item_name:
            return "综合指标"
            

        # 计算当前项目的 Embedding
        item_embedding = self.encoder.encode(item_name, convert_to_tensor=True)
        

        # 计算当前项目与所有大类的余弦相似度
        similarities = util.cos_sim(item_embedding, self.category_embeddings)[0]
        
        # 获取最大相似度及其对应的索引
        max_sim_value = similarities.max().item()
        max_sim_idx = similarities.argmax().item()
        
        # 如果最高相似度超过设定的阈值，则认定属于该分类
        if max_sim_value >= self.similarity_threshold:
            return self.medical_categories[max_sim_idx]
        
        return "其他/综合指标"

    def parse_and_enrich_table(self, global_context: str, table_text: str) -> List[str]:
        """
        按行解析 Markdown 表格，并进行向量语义增强注入
        """
        lines = table_text.strip().split('\n')
        if len(lines) < 3: 
            return [] # 不是标准的 Markdown 表格
        
        headers = lines[0].strip()
        chunks = []
        
        for line in lines[2:]:
            line = line.strip()
            if not line or set(line.replace('|', '').replace(' ', '')) == {'-'}: 
                continue # 过滤空行或异常分隔线
            
            # 尝试提取检验项目名称
            columns = [col.strip() for col in line.split('|') if col.strip()]
            item_name = columns[0] if columns else ""
            
            # 通过 BGE 获取高维语义标签
            semantic_tag = self._get_semantic_category(item_name)
            
            # Chunk
            enriched_chunk = (
                f"【报告上下文】{global_context}\n"
                f"【语义分类】{semantic_tag}\n"
                f"{headers}\n"
                f"{line}"
            )
            chunks.append(enriched_chunk)
            
        return chunks


# ==========================================
# 测试模块：模拟 RAGFlow deepdoc 解析后的输出
# ==========================================
if __name__ == "__main__":




    # 模拟一份原始检验报告文本
    sample_doc = """
【检验报告单】
姓名：王建国  性别：男  年龄：62岁  检验日期：2023-12-15

| 检验项目 | 结果 | 提示 | 参考区间 | 单位 |
| --- | --- | --- | --- | --- |
| 白细胞计数(WBC) | 11.2 | ↑ | 3.5-9.5 | 10^9/L |
| 谷丙转氨酶(ALT) | 75 | ↑ | 9-50 | U/L |
| 尿酸(UA) | 450 | ↑ | 210-430 | μmol/L |
| 总胆固醇(TC) | 6.2 | ↑ | 3.0-5.7 | mmol/L |
| 促甲状腺激素(TSH) | 2.5 | 正常 | 0.3-4.2 | mIU/L |
    """




    # 初始化
    chunker = MedicalSemanticChunkerBGE()
    
    # 提取全局上下文
    global_ctx = chunker.extract_global_context(sample_doc)
    
    table_text = sample_doc.split("检验日期：2023-12-15\n\n")[1]

    result_chunks = chunker.parse_and_enrich_table(global_ctx, table_text)
    
    # 打印最终生成的 Chunk 列表
    for i, chunk_text in enumerate(result_chunks):
        print(f"\n{"-"*15} 最终切片 (Chunk {i+1}) {"-"*15}")
        print(chunk_text)