☰
RAG医疗问答系统源码实战:从检索增强到混合检索
2026/10/1 14:02:32 网站建设 项目流程

简介:基于RAG与大模型技术的医疗问答系统源码包,是一套完整的毕业设计项目,面向计算机、人工智能、自动化等专业的学生、老师或从业者,可用于期末课程设计、课程大作业及毕业设计。项目聚焦医疗领域问答,综合运用检索增强生成(RAG)、大模型微调(如ChatGLM)、知识图谱构建与命名实体识别(NER)等技术,覆盖数据预处理、模型训练、推理与前端交互等完整流程,具有较高的学习与借鉴价值。资源包共75个文件,压缩后84.65MB。其中Python源码(py)包含模型微调、推理及图谱构建脚本,Jupyter Notebook(ipynb)提供逐步解析演示,txt/json/csv数据文件为医疗语料及实体关系,yaml为训练配置,png/jpg为界面及流程图,md为文档说明,目录模块划分清晰,便于按需检索。该项目答辩评审分达98分,代码经调试确认可运行。除完整源码与文档外,还提供微调运行示例、数据增强工具、NER结果分析及前端登录界面等,目前已吸引378人学习下载。适合小白系统学习,也可在此基础上修改调整,实现更丰富的功能。

1. 医疗问答系统为什么绕不开 RAG:一份毕设源码的真实边界

把一份药品说明书丢给通用大模型,它通常能答出适应症和常规剂量,但你问“我有高血压,吃了这个药之后头晕,要不要停”,通用模型大概率会给你一段看着像专家共识、实则没法追溯出处的话。医疗问答系统的难点从来不是“模型会不会说话”,而是“它说的每句话能不能找到依据”。基于 RAG 与大模型技术的医疗问答系统源码,解决的问题正是这个:把诊疗指南、药品说明书这类私有文档切成片段、向量化,用户提问时先检索再让大模型基于检索结果作答,答案可溯源、知识可更新。这套源码适合两类人:一类是做毕业设计的学生,想找一个真正能跑通、有文档有评估的完整项目;另一类是刚接触 RAG 的工程师,想看看医疗场景下文档解析、检索、生成这条链路要踩哪些坑。

2. 从裸大模型到 RAG 检索增强:链路选型与关键参数

2.1 为什么医疗问答不能只靠微调

先回答一个绕不开的问题:医疗问答为什么不用微调?我在自己折腾大模型的时候,微调和 RAG 都试过,说下实际感受。微调的本质是把知识灌进权重里,听起来很美,但医疗场景有两个硬伤。第一,知识更新成本高,临床指南、医保目录、药品说明说改就改,每更新一次就要重新训练一轮,时间成本根本扛不住。第二,可解释性差,模型答错了你都不知道它是从哪句话推出来的,医生和患者都不敢用这种“黑匣子”回答。RAG 走的是另一条路:文档在外置知识库里,模型只负责“阅读理解”和“组织语言”。知识更新只需要替换文档、重建索引,答案可以引用片段编号回看原文,这两点恰好命中医疗问答的命门。

我见过不少同学把大量精力花在微调调参上,最后效果还不如一条完整的 RAG 链路。对这份源码来说,设计思路也是典型的新手友好型:不做训练,只做“检索 + 生成”,显卡要求低,部署成本可控。下面是三条路线的对比,能帮你理解为什么选 RAG。

方案知识更新成本答案可溯源部署成本适用场景
裸大模型无更新能力(靠模型内建知识)无低闲聊、通用问答
微调高,每次都要重训无高,需要训练环境固定话术、私有风格
RAG低,换文档重建索引即可有,可回看片段低,普通单卡可跑知识库问答、医疗问答

2.2 源码里的 RAG 链路:向量库、Embedding 与 Top-K 设定

这套系统的主体链路是标准的五段式:文档解析 → 文本切分 → Embedding 向量化 → 向量检索 → 大模型生成。不必把它想得多玄,核心就是“先查资料再回答”。我拆过不少 RAG 项目,凡是效果翻车的,多半不是大模型的问题,而是前面三步没做好。

关键参数通常在源码的配置文件里集中管理。常见做法是放到 config.yaml 或 .env 里,方便切换实验。下面这份参数表基本覆盖了链路里最影响效果的几个旋钮,照着调就行。

参数典型取值作用调试优先级
chunk_size256 / 512单次送入检索单元的文本长度高,太大语义杂,太小上下文碎
chunk_overlap32 / 64相邻片段重叠长度,防止切分切断语义高,与 chunk_size 配合
embedding_modelm3e-base / bge-m3中文语义向量模型高,决定检索质量下限
embedding_dim768 / 1024向量维度,取决于模型中,入库后改要重建索引
top_k5 / 8返回候选片段数量高,医疗场景不建议超过 8
similarity_metricIP / COSINE向量相似度度量方式低,取决于向量模型训练方式

我一般会先把 chunk_size 定在 512、overlap 定在 64,跑一轮看效果再往下调。整套链路在源码里是分离的模块,交换任意一个环节不会牵连其他部分,这也是能作为毕设加分的设计。

2.3 混检索与重排:RAG 效果的分水岭

如果你以为 RAG 就是“向量查一查、拼进 prompt”,那只能说跑通没问题,效果就随缘了。纯向量检索对近义表达很友好,比如“血压偏高”和“高血压”能查到同一篇文档,但它对药品名、剂量这类精确词反而容易翻车:向量表示是语义压缩过的,有时会把“阿莫西林胶囊”和“阿莫西林克拉维酸钾片”混淆。我自己的经验是,医疗文档里专有名词密度高,必须做混合检索。

常见做法是 BM25 关键词检索 + 向量语义检索并行,再把两路结果用 RRF(Reciprocal Rank Fusion)融合。BM25 负责精确匹配药名、剂量、检查项,向量负责语义召回,RRF 把两边的排名合并成一个稳定结果。源码里这层的实现可以单独拎出来用,换到法律、金融文档场景也成立。后面第 3 章我会把这三段的代码拆开讲。

3. 把源码跑起来:医疗文档处理、向量化与检索实现的 Python 细节

3.1 医疗文档清洗与切分:按章节切还是按窗口切

医疗文档的原始形态五花八门:有从 PDF 转出来的文本、有 Word 导出的 Markdown、还有带 OCR 噪声的扫描页。切分之前必须先清洗。我见过最典型的翻车现场是:页眉页脚、“药品名称:XXX”这种重复信息混进片段,检索时十条里有四条是页眉。清洗的两条硬规则是:去掉连续空白和行号,按文档结构切块而不是盲目按字符数切。

import re from typing import List def clean_text(raw: str) -> str: # 去掉行号和页眉页脚常见噪声,保留正文语义 text = re.sub(r'\n\d+\n', '\n', raw) text = re.sub(r'第\s*\d+\s*页', '', text) text = re.sub(r'[ \t]+', ' ', text) text = re.sub(r'\n{3,}', '\n\n', text) return text.strip() def split_document(text: str, chunk_size: int = 512, overlap: int = 64) -> List[str]: # 优先按一级/二级标题切,避免把不同适应症切进同一块 segments = re.split(r'(?=\n【|\n\d+\.|\n[一二三四五六七八九十]+、)', text) chunks = [] for seg in segments: seg = seg.strip() if len(seg) <= chunk_size: chunks.append(seg) continue # 超过阈值的段落再按重叠窗口切 start = 0 while start < len(seg): end = start + chunk_size chunks.append(seg[start:end]) if end >= len(seg): break start = end - overlap return chunks

逻辑说明:clean_text 里的三个正则分别处理行号、页码和多余空白,跑完基本能拿到干净文本。split_document 里re.split的(?=...)是零宽断言,意思是“在标题前面切一刀,但标题不丢”,这样切出来的片段天然带小标题,检索时命中一看标题就知道出自哪一章。最后一段是不得已的兜底:某个小节实在太长,才用固定窗口硬切,overlap 保证跨窗口的语义不中断。

参数说明:chunk_size 不要小于 256,太小会让单条上下文装不下一个完整适应症描述;overlap 一般设成 chunk_size 的 1/8 到 1/4 比较顺。药品说明书我建议优先按“【适应症】【用法用量】【不良反应】”这种结构化标题切,源码里我习惯把标题列表做成可配置项,换科室文档时不用改代码。

3.2 Embedding 模型选型与向量入库

切分完成的片段要转成向量。实际项目中我常看到两个选择:m3e-base 和 bge-m3。m3e-base 是 768 维,模型小、CPU 也能勉强跑,适合毕设和轻量部署;bge-m3 是 1024 维,多语言和长文本能力更强,但对显存要求高一点。这套源码走的是务实路线,默认 m3e-base,后面你要换模型,只需改配置里的模型名和维度,向量库路径会自动分目录,不会污染旧索引。

from sentence_transformers import SentenceTransformer import faiss import numpy as np import pickle embedder = SentenceTransformer("m3e-base") def build_index(chunks: List[str], index_path: str = "./data/index"): # 分批编码,避免一次性吃满显存 vectors = [] batch_size = 32 for i in range(0, len(chunks), batch_size): batch = chunks[i : i + batch_size] vec = embedder.encode(batch, normalize_embeddings=True) vectors.append(vec) matrix = np.concatenate(vectors, axis=0) # 内积 + 归一化,等效余弦相似度 index = faiss.IndexFlatIP(matrix.shape[1]) index.add(matrix) # 索引和片段元数据分开存档,防止索引损坏后无法找回原文 faiss.write_index(index, f"{index_path}.faiss") with open(f"{index_path}.meta.pkl", "wb") as f: pickle.dump(chunks, f)

逻辑说明:normalize_embeddings=True这一步很多人会漏,它把向量归一化到单位长度,之后用内积(IndexFlatIP)就能等价于余弦相似度,检索分数更稳定。batch_size 设 32 是为了在普通显卡上控制峰值显存,你的卡大可以调大,卡小就调小。索引文件和片段元数据分开存,因为 faiss 索引里只存向量,不存原文,如果只存索引,检索结果没法映射回文本。

参数说明:m3e-base 的向量维度是 768,如果你后面换 bge-m3,维度变成 1024,旧索引不能复用,必须重建。faiss 的 IndexFlatIP 是暴力精确检索,数据量在十万级以下性能完全没问题,不必上 IVF 或 HNSW 这种索引,省得给自己加复杂度。

3.3 检索接口实现:从 query 到候选上下文

检索层是这套源码里最值得单独抽出来复用的部分。用户的 query 进来,先做一遍同类清洗,然后走双路检索:一路用 BM25 做关键词精确匹配,一路用向量做语义召回,最后用 RRF 合并。这里给一个不带重排的简化实现,重排器和生成层分开,方便你单独测检索质量。

import jieba from rank_bm25 import BM25Okapi from typing import List class HybridRetriever: def __init__(self, chunks: List[str], faiss_index, embedder): self.chunks = chunks self.index = faiss_index self.embedder = embedder # BM25 对中文需要切词,这里直接喂 jieba 分词的 token 列表 tokenized = [list(jieba.cut(c)) for c in chunks] self.bm25 = BM25Okapi(tokenized) def search(self, query: str, top_k: int = 5) -> List[dict]: q_vec = self.embedder.encode([query], normalize_embeddings=True) scores, ids = self.index.search(q_vec, top_k) dense_hits = [(idx, 3.0) for idx in ids[0]] bm25_scores = self.bm25.get_scores(list(jieba.cut(query))) bm25_hits = sorted( enumerate(bm25_scores), key=lambda x: x[1], reverse=True )[:top_k] bm25_hits = [(idx, 2.0) for idx, _ in bm25_hits] # RRF 融合,k=60 是经验值 fused = {} for idx, rank_score in dense_hits + bm25_hits: fused[idx] = fused.get(idx, 0.0) + 1.0 / (60 + rank_score) ranked = sorted(fused.items(), key=lambda x: x[1], reverse=True) return [ {"chunk": self.chunks[idx], "score": score, "idx": idx} for idx, score in ranked[:top_k] ]

逻辑说明:dense_hits 的两个分数不是相似度,而是在 RRF 里表示“这一路结果的先验权重”,向量路给 3.0、BM25 路给 2.0,意思是向量语义召回优先级略高。1.0 / (60 + rank_score)是 RRF 的核心公式,k 值越大,两路结果越“平均”;k 越小,rank 靠前的结果优势越明显。医疗场景我一般把 k 留在 60,因为药品名精确匹配和语义召回同样重要,不能一边倒。

参数说明:top_k 是最终返回给生成层的片段数。取 5 是权衡后的结果,太大 prompt 会被无关片段淹没,让大模型抓到噪声;太小又可能漏掉关键依据。如果后面加了重排器,这里可以放宽到 10,重排器会二次筛掉无关项。

4. 生成回答与效果评估:Prompt 约束、Hit Rate 与可复现实验

4.1 Prompt 模板:让大模型只答知识库里的内容

检索做得再好,生成层用裸 prompt 还是会一本正经地胡说八道。医疗场景的 prompt 必须做三件事:限定知识来源、要求引用编号、允许说“不知道”。模板里给一个{context}占位符,检索结果按[片段 1] [片段 2]的格式填进去,模型在输出里直接引用对应编号,方便人工核对。

SYSTEM_PROMPT = """你是一名基于知识库回答问题的医疗问答助手。 规则: 1. 只依据 <context> 中提供的片段回答,禁止使用你自己的知识。 2 回答必须标注引用的片段编号,例如 [1][2]。 3. 如果 <context> 中没有足够信息,回答:未在现有资料中检索到相关内容,建议线下就医。 4. 不要输出诊断、不要给出用药剂量建议,只做资料性整理。 <context> {context} </context> """ def build_prompt(query: str, hits: List[dict]) -> str: context = "\n".join( f"[{i + 1}] {hit['chunk']}" for i, hit in enumerate(hits) ) return SYSTEM_PROMPT.format(context=context) + f"\n用户问题:{query}"

逻辑说明:规则 1 是防幻觉的第一道闸,规则 2 把回答变成可核验的,规则 3 给了兜底话术,规则 4 是医疗合规边界。这套系统定位是资料整理助手,不是在线医生,所以强制限定“不做诊断、不给剂量”,这一点在毕设答辩里反而是加分项,说明你有安全意识。

补充说明:很多人会把“不要输出诊断”写进 prompt 就完事,实际测试里模型还是会偶尔越界。可靠的做法是在输出端再加一道正则检查,命中“建议服用”“剂量”等敏感词时直接切换到兜底回答,代码量很少,但能挡住绝大多数风险。

4.2 评估集构造与 hit rate / recall 计算

RAG 系统最怕的事情是“感觉还行,但说不清哪里不行”。要量化效果,就得有评估集。做法是手工构造 20 个问句,每个问句标注一个或多个“标准答案所在的片段编号”,然后跑检索,看这 20 个问句的 top-K 结果里有没有包含标准片段。hit rate 就是“命中的问句数 / 总问句数”,这个指标专门衡量检索质量,和大模型无关,定位问题非常有用。

def evaluate(retriever, eval_set, top_k: int = 5): hit = 0 mrr = 0.0 for query, golden_ids in eval_set: results = retriever.search(query, top_k=top_k) retrieved_ids = [r["idx"] for r in results] if any(g in retrieved_ids for g in golden_ids): hit += 1 for rank, r in enumerate(results, start=1): if r["idx"] in golden_ids: mrr += 1.0 / rank break n = len(eval_set) return {"hit_rate": hit / n, "mrr": mrr / n} eval_set = [ ("阿莫西林胶囊的用法用量是什么", [12, 13]), ("高血压患者能否同时服用布洛芬", [204]), # ... 实际评估集建议准备 20 组以上 ] print(evaluate(retriever, eval_set))

逻辑说明:hit_rate 是召回质量的及格线,mrr 是“标准片段排在第几位”的加权指标,排名越靠前,生成效果越好。跑评估时我会把 top_k 固定为 5,这样 hit_rate 高说明检索链路没毛病;如果 hit_rate 低于 0.8,先别调生成层,回头查切分和 Embedding。

建议:每个问句的标准片段编号要从切分后的 chunks 列表里挑,而不是从原文档里标注。因为检索返回的是切分后的片段,标注错了编号,评估脚本就会误判。这里建议在评估集里同时保存“原文档段落”和“chunk 编号”,避免标注错位。

4.3 流式输出与超时控制

后端接口这一层,源码用的是 FastAPI。实测下来有两个细节值得照着做:第一,大模型生成必须用流式,否则一个长回答可能要等十几秒,前端体验很糟糕;第二,检索和生成要设置各自独立的超时,检索超过 3 秒直接返回兜底,不能让用户永远转圈。

from fastapi import FastAPI from fastapi.responses import StreamingResponse app = FastAPI() def generate_stream(prompt: str): # 实际项目中这里调用大模型接口,此处伪代码示意 for token in chat_model.stream(prompt): yield token @app.post("/chat") async def chat(query: str, top_k: int = 5): hits = retriever.search(query, top_k=top_k) if not hits: return {"answer": "未在现有资料中检索到相关内容,建议线下就医。"} prompt = build_prompt(query, hits[:top_k]) return StreamingResponse(generate_stream(prompt), media_type="text/plain")

逻辑说明:StreamingResponse把生成器的输出逐个发给前端,用户看到的是“打字机”效果。检索为空时直接返回固定话术,不再调用大模型,省一次耗时和成本。这套接口设计在你做压力测试时也省心,每条请求的耗时都在生成阶段,检索阶段相对稳定,调优方向清晰。

5. 避坑排查:医疗问答系统最常见的五个翻车现场

5.1 症状:检索召回全是无关段落

现象:无论问什么,返回的片段都是文档里的通用内容,比如“请仔细阅读说明书并在医师指导下使用”这种句子。 原因:这类句子几乎出现在每份文档里,向量分布和所有问句的匹配度都高,把真正相关的片段挤出了 top_k。 解决:清洗阶段做一遍“高频模板句过滤”,正则把“请仔细阅读”“本品为”这类通用套句直接剔除;如果不想写死在代码里,就把这些模板句放进一个黑名单文本文件,清洗时逐行匹配删除。

5.2 症状:模型答非所问,引用与答案对不上

现象:答案内容看着合理,但标注的 [1][2] 对应片段根本不包含答案依据。 原因:prompt 里没强制“引用必须来自片段原文”,模型按自己的知识作答,随口编了编号。 解决:生成后加一道校验:把引用编号对应的 chunk 原文和答案一起送去算相似度,低于阈值就拒绝该引用。我这个校验脚本是后置的,不改造生成层,部署阶段也敢用。

5.3 症状:启动即 OOM 或者向量库加载慢

现象:服务一起,显存或内存直接爆掉,或者加载索引就要好几分钟。 原因:向量索引全量加载进内存,加之大模型权重也在显存里,两边抢资源。 解决:faiss 索引加载改用faiss.read_index后再index.reconstruct的按需方式,或者换IndexIVF这类压缩索引,代价是召回率略微下降;大模型推理用量化版本。经验值是 10 万条 chunk 以内,IndexFlatIP完全够用,不用过度设计。

5.4 症状:文档更新后检索结果没变化

现象:替换了知识库文档,但同一个问题返回的片段还是旧的。 原因:索引没有重建,或者服务进程缓存了旧的 Embedding 结果。 解决:把“重建索引”做成单独的脚本,文档变更后先跑清洗与切分,再跑向量化入库。这个步骤不要和启动服务绑在一起,否则每次启动都重建,时间全浪费掉了。

5.5 症状:中文医疗术语被切碎

现象:检索“阿莫西林克拉维酸钾片”,BM25 路召回的是“阿莫西林”和“克拉维酸”的碎片文档,反而不如纯向量。 原因:jieba 默认词典对专业药品名不识别的结果。 解决:给 jieba 加载一个医疗词典,词典行就是把常见药名、检查名、疾病名写进去,用jieba.load_userdict加载。效果立竿见影:BM25 的 token 不再碎片化,精确匹配能力一下子上来了。

6. 进阶一步:把单轮 RAG 升级为带追踪的 Agentic RAG

6.1 从单轮到多跳:源码扩展点在哪

这套源码跑通之后,如果你还有余力,最值得升级的方向是 Agentic RAG。单轮 RAG 是“问一次查一次”,碰上“高血压患者感冒了能吃哪种感冒药”这种要跨两份文档推理的问题就会卡住。常见做法是给检索加一个 query 改写器:首轮检索不满意就自动拆解成子问题,逐个子问题检索,再合并进 prompt。这个扩展点建议落在源码的检索层,而不是生成层,因为生成层改动会影响你之前调好的效果。

def rewrite_query(query: str, hit_scores: List[float], threshold: float = 0.4) -> str: # 首轮检索分数普遍偏低时,尝试改写为更聚焦的问法 avg_score = sum(hit_scores) / len(hit_scores) if avg_score >= threshold: return query # 实际项目中这里调用大模型改写,伪代码示意 return f"针对{query},列出与诊断、用药、禁忌相关的要点"

逻辑说明:threshold是触发改写的分数阈值,低于它就说明首轮检索没找到强相关片段,把问句转成“要点式提问”再查一次,往往能命中指南里的核心章节。这套做法在 agentic rag 项目里被反复验证过,原理是让检索目标从“用户原话”变成“文档更可能写的表达方式”。

6.2 验证方法:用 20 个病例把系统压一遍

升级完别急着开心,把评估集扩成 20 个真实病例类问句,跑一遍第 4.2 节的评估脚本,重点看两个数字:hit_rate 是否维持在 0.8 以上,引用覆盖率(答案里标注了有效编号的比例)是否超过 90%。有这两个数兜底,系统拿去答辩或演示才有底气。我自己的教训是:曾经跳过了这步,直接演示,结果现场一个问题召回翻车,场面一度尴尬。从那以后,每次交出去的问答系统我都强制走一遍 20 问回归,先看检索指标再看引用完整率,这两个数字不过关,绝不进演示环节。这套源码已经帮你把文档处理、检索、评估和部署的骨架搭好了,照着跑一遍,再按你自己的场景调一遍参,比从零开始省太多时间。希望帮到你。

本文还有配套的精品资源,点击获取

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询