医疗垂直场景RAG技术调优：从知识分段、混合检索到重排溯源的全链路优化实践解析21.7

import re
from typing import List

# 医疗文本语义终止正则规则（适配指南、药品说明、诊断标准）
MEDICAL_SEP_PATTERN = re.compile(
    r"(。|；|：\n|总结|综上|适应症|禁忌症|用法用量|不良反应|诊断标准|治疗原则|随访方案)\s*"
)

def medical_chunk_split(text: str, chunk_size: int = 768, overlap: int = 128) -> List[str]:
    """
    医疗专属自适应分段
    :param text: 原始医疗文本
    :param chunk_size: 单片段最大token长度
    :param overlap: 相邻片段重叠token长度
    :return: 合规医疗知识片段列表
    """
    # 优先按医疗语义断点切割
    raw_sents = MEDICAL_SEP_PATTERN.split(text)
    chunks = []
    current_chunk = ""

    for sent in raw_sents:
        if not sent.strip():
            continue
        # 超出长度则归档当前片段，开启新片段
        if len(current_chunk) + len(sent) > chunk_size:
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = sent
        else:
            current_chunk += sent
    
    # 补充最后一段
    if current_chunk.strip():
        chunks.append(current_chunk.strip())

    # 叠加重叠优化，避免跨知识点缺失
    overlap_chunks = []
    for idx, chunk in enumerate(chunks):
        if idx > 0:
            pre_overlap = chunks[idx-1][-overlap:]
            overlap_chunks.append(pre_overlap + chunk)
        else:
            overlap_chunks.append(chunk)
    return overlap_chunks

# 测试：临床医疗文本分片
if __name__ == "__main__":
    # ---------- 测试1：短文本（不分片） ----------
    print("=" * 60)
    print("  测试1：短文本（chunk_size=768，不触发分片）")
    print("=" * 60)
    short_text = "高血压二级诊断标准为收缩压160-179mmHg，舒张压100-109mmHg。适应症包含原发性高血压患者，可选用氨氯地平、硝苯地平等钙通道阻滞剂。用法用量为成人初始剂量5mg每日一次，最大剂量10mg每日一次。禁忌症为严重低血压、心源性休克患者。"
    res1 = medical_chunk_split(short_text, chunk_size=768)
    print(f"原文长度: {len(short_text)} 字符 → 分片数: {len(res1)}\n")
    for i, c in enumerate(res1):
        print(f"【分片 {i+1}】 {c}")

    # ---------- 测试2：长文本（chunk_size=150，明显分片） ----------
    print("\n\n" + "=" * 60)
    print("  测试2：长文本（chunk_size=150，触发多段分片）")
    print("=" * 60)
    long_text = (
        "高血压是一种常见的慢性疾病，以体循环动脉血压持续升高为主要特征。"
        "高血压一级诊断标准为收缩压140-159mmHg，舒张压90-99mmHg。"
        "高血压二级诊断标准为收缩压160-179mmHg，舒张压100-109mmHg。"
        "高血压三级诊断标准为收缩压≥180mmHg，舒张压≥110mmHg。"
        "适应症包含原发性高血压患者，可选用氨氯地平、硝苯地平等钙通道阻滞剂。"
        "对于合并心力衰竭的患者，推荐使用ACEI类药物如依那普利、贝那普利。"
        "对于合并糖尿病的患者，首选ARB类药物如缬沙坦、厄贝沙坦。"
        "用法用量为成人初始剂量氨氯地平5mg每日一次，可根据血压调整至最大10mg每日一次。"
        "依那普利初始剂量5mg每日一次，维持剂量10-20mg每日一次分两次服用。"
        "不良反应常见头痛、头晕、面部潮红、踝部水肿、心悸和乏力。"
        "少数患者可能出现牙龈增生，停药后可逐渐恢复。"
        "长期使用噻嗪类利尿剂需监测血钾水平，防止低钾血症发生。"
        "禁忌症为严重低血压、心源性休克患者，妊娠期及哺乳期妇女禁用。"
        "双侧肾动脉狭窄患者禁用ACEI和ARB类药物。"
        "治疗原则应个体化给药，从小剂量起始，逐步调整至目标血压。"
        "随访方案建议初始治疗期每2-4周复查一次，血压达标后每3-6个月随访。"
    )
    res2 = medical_chunk_split(long_text, chunk_size=150, overlap=30)
    print(f"原文长度: {len(long_text)} 字符 → 分片数: {len(res2)}\n")
    for i, c in enumerate(res2):
        print(f"【分片 {i+1}】 {c}")
    print(f"\n  [说明] chunk_size=150 强制按语义断点分片，overlap=30 保证相邻分片内容衔接")

from rank_bm25 import BM25Okapi
import numpy as np

# 模拟医疗知识库分片（临床、药品、指南数据）
medical_corpus = [
    "氨氯地平成人高血压用法用量：初始5mg/日，最大10mg/日",
    "氨氯地平禁忌症：严重低血压、心源性休克患者禁用",
    "儿童布洛芬混悬液退烧用量：10mg/kg，间隔4-6小时可重复使用",
    "高血压二级诊断标准：收缩压160-179mmHg，舒张压100-109mmHg"
]

# 初始化BM25稀疏检索引擎（中文无空格，用字符级分词）
tokenized_corpus = [list(doc) for doc in medical_corpus]
bm25_model = BM25Okapi(tokenized_corpus)

# 模拟医疗向量相似度计算
def vec_similarity(query_vec, corpus_vec):
    return np.dot(query_vec, corpus_vec) / (np.linalg.norm(query_vec) * np.linalg.norm(corpus_vec))

def medical_hybrid_search(query: str, query_vec, corpus_vec_list, top_k: int = 5):
    """
    医疗混合检索+动态权重融合
    :param query: 用户提问文本
    :param query_vec: 提问向量
    :param corpus_vec_list: 知识库分片向量列表
    :param top_k: 召回数量
    :return: 排序后的优质医疗片段
    """
    # 1. BM25关键词得分（字符级分词，适配中文）
    bm25_scores = bm25_model.get_scores(list(query))
    # 2. 向量语义得分
    vec_scores = [vec_similarity(query_vec, vec) for vec in corpus_vec_list]

    # 3. 场景动态权重分配
    if any(key in query for key in ["用量", "禁忌症", "诊断标准"]):
        bm25_weight, vec_weight = 0.7, 0.3
    elif any(key in query for key in ["怎么办", "如何治疗", "通俗来说"]):
        bm25_weight, vec_weight = 0.3, 0.7
    else:
        bm25_weight, vec_weight = 0.5, 0.5

    # 4. 加权融合得分
    total_scores = [
        bm25_weight * bm_score + vec_weight * vec_score
        for bm_score, vec_score in zip(bm25_scores, vec_scores)
    ]

    # 5. 排序召回
    rank_idx = np.argsort(total_scores)[::-1][:top_k]
    return [medical_corpus[idx] for idx in rank_idx]

# 模拟测试
if __name__ == "__main__":
    # 模拟提问向量、知识库向量（随机替代，真实场景替换医疗嵌入模型）
    np.random.seed(42)  # 固定随机种子，保证结果可复现
    test_query_vec = np.random.rand(128)
    test_corpus_vec = [np.random.rand(128) for _ in medical_corpus]

    query = "氨氯地平成人用法用量"
    print(f"用户提问：{query}")
    print(f"知识库文档数：{len(medical_corpus)} 条\n")

    # 1. BM25 关键词检索（字符级分词，适配中文）
    bm25_scores = bm25_model.get_scores(list(query))
    print("--- 第1步：BM25 关键词检索 ---")
    for i, (doc, s) in enumerate(zip(medical_corpus, bm25_scores)):
        print(f"  文档{i} 得分={s:.3f} | {doc[:30]}...")
    print("  [说明] 基于词频和逆文档频率匹配，精确关键词命中得分高")

    # 2. 向量语义检索
    vec_scores = [vec_similarity(test_query_vec, vec) for vec in test_corpus_vec]
    print("\n--- 第2步：向量语义检索 ---")
    for i, (doc, s) in enumerate(zip(medical_corpus, vec_scores)):
        print(f"  文档{i} 得分={s:.3f} | {doc[:30]}...")
    print("  [说明] 基于向量余弦相似度，捕获语义相近但用词不同的内容（本例用随机向量模拟）")

    # 3. 权重分配 & 融合
    if any(key in query for key in ["用量", "禁忌症", "诊断标准"]):
        bm25_weight, vec_weight = 0.7, 0.3
        weight_type = "精确查询"
    elif any(key in query for key in ["怎么办", "如何治疗", "通俗来说"]):
        bm25_weight, vec_weight = 0.3, 0.7
        weight_type = "语义查询"
    else:
        bm25_weight, vec_weight = 0.5, 0.5
        weight_type = "均衡查询"
    print(f"\n--- 第3步：动态权重分配（{weight_type}）---")
    print(f"  关键词权重={bm25_weight}，语义权重={vec_weight}")
    print('  [说明] 查询含"用量/禁忌症"等精确关键词，BM25权重更高')

    total_scores = [
        bm25_weight * bm + vec_weight * sv
        for bm, sv in zip(bm25_scores, vec_scores)
    ]
    rank_idx = np.argsort(total_scores)[::-1]
    print("\n--- 第4步：加权融合 & 排序 ---")
    for rank, idx in enumerate(rank_idx):
        print(f"  Top{rank+1} 文档{idx} | 融合得分={total_scores[idx]:.3f} | {medical_corpus[idx][:30]}...")
    print("  [说明] BM25确保精确命中，向量补足语义泛化，权重按场景动态调整")

    # 最终输出
    res = [medical_corpus[idx] for idx in rank_idx[:2]]
    print(f"\n最终召回结果：{res}")

from typing import List, Tuple

# 定义医疗内容权威权重
AUTHORITY_WEIGHT = {
    "临床指南": 1.0,
    "药典规范": 1.0,
    "核心期刊": 0.8,
    "普通科普": 0.4,
    "小众论文": 0.3
}

# 权威过滤阈值
SCORE_THRESHOLD = 0.6

def medical_rerank(query: str, candidate_list: List[dict]) -> List[dict]:
    """
    医疗内容专属重排、降噪、冲突过滤
    :param query: 用户提问
    :param candidate_list: 初筛候选片段{content, source, time_score, match_score}
    :return: 重排、过滤后的Top5优质内容
    """
    score_results = []
    for item in candidate_list:
        # 四维加权打分：场景匹配、权威度、时效性、语义精度
        total_score = (
            item["match_score"] * 0.4 +
            AUTHORITY_WEIGHT.get(item["source"], 0.2) * 0.3 +
            item["time_score"] * 0.2 +
            item["semantic_score"] * 0.1
        )
        item["total_score"] = total_score
        score_results.append(item)
    
    # 1. 权威阈值过滤
    filter_results = [i for i in score_results if i["total_score"] >= SCORE_THRESHOLD]
    # 2. 分数倒序排序
    sort_results = sorted(filter_results, key=lambda x: x["total_score"], reverse=True)
    # 3. 冲突内容去重，保留最新权威内容
    final_res = []
    content_set = set()
    for res in sort_results:
        if res["content"] not in content_set:
            content_set.add(res["content"])
            final_res.append(res)
    
    return final_res[:5]

# 测试示例
if __name__ == "__main__":
    query = "氨氯地平成人用量"
    test_candidate = [
        {"content":"氨氯地平每日最大剂量10mg","source":"临床指南","time_score":0.9,"match_score":0.95,"semantic_score":0.92},
        {"content":"氨氯地平可随意加量服用","source":"普通科普","time_score":0.5,"match_score":0.8,"semantic_score":0.7},
    ]

    print(f"用户提问：{query}")
    print(f"候选文档数：{len(test_candidate)} 条\n")

    # 第1步：四维加权打分
    print("--- 第1步：四维加权打分 ---")
    print(f"  权重分配：match_score(匹配度) 40% + authority(权威度) 30% + time(时效性) 20% + semantic(语义) 10%")
    for item in test_candidate:
        total = (
            item["match_score"] * 0.4 +
            AUTHORITY_WEIGHT.get(item["source"], 0.2) * 0.3 +
            item["time_score"] * 0.2 +
            item["semantic_score"] * 0.1
        )
        print(f"  {item['content'][:25]}... | 来源={item['source']}(权重{AUTHORITY_WEIGHT[item['source']]}) | 综合分={total:.3f}")
    print("  [说明] 临床指南权重高，普通科普权重低，四维加权平衡精度与可信度")

    # 第2步：权威阈值过滤
    print(f"\n--- 第2步：权威阈值过滤（阈值={SCORE_THRESHOLD}）---")
    for item in test_candidate:
        total = (
            item["match_score"] * 0.4 +
            AUTHORITY_WEIGHT.get(item["source"], 0.2) * 0.3 +
            item["time_score"] * 0.2 +
            item["semantic_score"] * 0.1
        )
        status = "保留" if total >= SCORE_THRESHOLD else "淘汰"
        print(f"  {item['content'][:25]}... | 综合分={total:.3f} → {status}")
    print("  [说明] 低于阈值的低质量/低权威内容直接过滤，避免劣质信息混入结果")

    # 最终输出
    res = medical_rerank(query, test_candidate)
    print(f"\n最终结果（Top5）：")
    for i, r in enumerate(res, 1):
        print(f"  {i}. [{r['source']}] {r['content']} | 综合分={r['total_score']:.3f}")

from typing import List, Dict

# 医疗知识库分片+元数据埋点
medical_chunk_source: Dict[str, dict] = {
    "chunk_001": {
        "content": "氨氯地平成人初始剂量5mg每日一次，最大剂量10mg每日一次",
        "source_name": "中国高血压防治指南2023版",
        "authority": "中华医学会心血管病学分会",
        "publish_time": "2023",
        "type": "临床诊疗指南"
    },
    "chunk_002": {
        "content": "氨氯地平禁止用于严重低血压、心源性休克患者",
        "source_name": "中华人民共和国药典2020版",
        "authority": "国家药典委员会",
        "publish_time": "2020",
        "type": "药典规范"
    }
}

def generate_traceable_answer(llm_answer: str, chunk_id_list: List[str]) -> dict:
    """
    生成带溯源的医疗问答结果
    :param llm_answer: 大模型生成答案
    :param chunk_id_list: 生成依赖的分片ID
    :return: 答案+完整溯源清单
    """
    trace_list = []
    for cid in chunk_id_list:
        if cid in medical_chunk_source:
            trace_list.append(medical_chunk_source[cid])
    
    return {
        "answer": llm_answer,
        "trace_source": trace_list,
        "source_count": len(trace_list)
    }

# 落地测试
if __name__ == "__main__":
    # 模拟模型生成答案 & 依赖分片
    ans = "氨氯地平成人常规初始剂量为每日5mg，单日最大剂量不超过10mg，严重低血压患者禁用。"
    used_chunk = ["chunk_001", "chunk_002"]

    print(f"模型生成答案：{ans}\n")

    # 第1步：展示引用分片与原文对照
    print("--- 第1步：引用分片 → 原文溯源 ---")
    for cid in used_chunk:
        if cid in medical_chunk_source:
            meta = medical_chunk_source[cid]
            print(f"  {cid}: {meta['content']}")
            print(f"     来源: {meta['source_name']} | 机构: {meta['authority']} | 年份: {meta['publish_time']} | 类型: {meta['type']}")
    print("  [说明] 大模型生成的每句话，都追溯回原始分片及其出处元数据")

    # 第2步：生成带溯源的最终结果
    res = generate_traceable_answer(ans, used_chunk)
    print(f"\n--- 第2步：组装结构化溯源结果 ---")
    print(f"  答案: {res['answer']}")
    print(f"  引用源数量: {res['source_count']}")
    for i, src in enumerate(res['trace_source'], 1):
        print(f"  源{i}: [{src['type']}] {src['source_name']} ({src['publish_time']}) - {src['authority']}")
    print("  [说明] 答案携带完整出处链，支持逐句回溯至权威文献，确保医疗内容的可信与合规")