开源大模型医疗问答落地：基于Qwen2.5与RAG构建知识库助手

这次我们来看一个把开源大模型用到医疗知识问答场景的完整落地案例：基于 Qwen2.5-14B-Instruct 构建通义医疗大模型问答助手，先把病理学、诊疗指南这类垂直语料做成向量知识库，再通过 RAG 检索增强生成方式接进大模型，最后用 FastAPI 把能力暴露成 HTTP 接口，方便批量测试和后续集成。

这事的重点不在于“跑通一个模型”，而在于把开源模型变成真正可用的医疗知识问答工具：什么时候该搜知识库，什么时候让模型直接回答，检索结果怎么拼进 Prompt，接口怎么设计才能同时支持单条问答和批量评估。下面直接给一套可复用的技术方案。

1. 核心能力速览

从部署到验证全套流程都在本地闭环，不依赖任何云端推理服务，材料、代码、模型全部可控。

2. 适用场景与合规边界

先说明适合什么场景。医疗大模型问答助手不是给患者“看病”，而是给医学内容生产者、医学生、临床科研人员做知识整理辅助。比如把病理学讲义整理成结构化问答对，把某类疾病诊疗指南检索出来并做摘要，把 WHO 分类标准和 ICD 编码之间的对应关系做成查询接口。这些场景下，大模型加 RAG 的价值是真实存在的。

不适合的场景也讲清楚：不要面向公众提供未审核的诊断建议，不要用模型输出直接替代检验报告解读，不要拿它做药物剂量计算。原因很简单，开源模型存在幻觉，垂直知识库只能降低幻觉概率，不能消除幻觉。做医疗相关系统时，知识库内容必须经过医学专业人员审核，输出必须带引用来源，系统必须明确标识“AI 辅助信息仅供参考”。

版权和隐私边界同样重要。医学教材、诊疗指南、数据库内容往往有版权保护，自建知识库时应优先使用已授权语料、公开的脱敏病例、开源医学数据集。患者数据一律不允许进入本地知识库和模型上下文，如果系统最终要接入医院业务，必须走完整的等保、伦理审批和信息安全评估流程。合规不是文章末尾补一句免责声明就结束，而是在架构设计阶段就要考虑。

3. 环境准备与模型选择

3.1 硬件和系统要求

14B 模型的开源版本推理，最稳妥的是 24GB 显存显卡，加载 4bit 量化后在 16GB 显存附近可跑，但长上下文和并发请求会明显吃紧。CPU 推理可用但速度要低一个数量级，适合只想验证流程的场景。磁盘方面，基础依赖加模型文件建议预留 60GB 以上空间。

系统使用 Linux 服务器最省事，本文命令基于 Ubuntu 22.04 + Conda 给出。macOS 也可以跑，但 Ollama 对 Apple Silicon 支持比较成熟，整体流程差异不大。

3.2 安装依赖

先创建 Python 环境并安装核心依赖。

conda create -n medical_qa python=3.10 -y conda activate medical_qa pip install --upgrade pip pip install fastapi uvicorn langchain langchain-community chromadb sentence-transformers pandas openpyxl requests pydantic

依赖项说明：langchain负责 RAG 流程编排，chromadb是向量存储，sentence-transformers做文本向量化，fastapi提供接口服务。版本建议使用较新的稳定版，避免老版本 API 不兼容问题。

3.3 拉取模型

Ollama 方式最直接，一条命令拉取模型并常驻为本地服务。

ollama pull qwen2.5:14b-instruct-q4_K_M ollama serve

Ollama 默认监听 11434 端口，后续 Python 代码通过http://127.0.0.1:11434访问模型接口。如果要用 Transformers 做更精细的采样控制和 LoRA 微调，则用transformers结合accelerate加载模型，本文以 Ollama 接入为主，但接口调用层留了抽象，替换成本不大。

# 测试 Ollama 服务是否可用 import requests resp = requests.post( "http://127.0.0.1:11434/api/generate", json={"model": "qwen2.5:14b-instruct-q4_K_M", "prompt": "你好", "stream": False} ) print(resp.json().get("response", "")[:200])

能看到正常回复，说明模型服务没问题，可以进入知识库构建阶段。

4. 本地医学知识库构建

4.1 语料整理

知识库质量决定问答质量，模型再强也救不了混乱的语料。我把参考语料分为三类：结构化权威内容，比如诊疗指南、药品说明书、病理学分类标准；教材章节，比如病理学各论；问答对，比如从题库整理出的“问题-标准答案”。

语料整理成统一格式，每一条记录包含content和source两个字段，source是来源标注，后续答案会带着来源一起返回。

documents = [ { "content": "WHO中枢神经系统肿瘤分类将脑膜瘤分为WHO 1级、2级和3级，其中大多数脑膜瘤为WHO 1级，属于良性肿瘤。", "source": "WHO CNS Tumor Classification (2021)" }, { "content": "肺腺癌的常用免疫组化标志物包括TTF-1、Napsin A、CK7，其中TTF-1在肺腺癌中阳性率较高。", "source": "病理学教材第12章" } ]

4.2 文本切分

长文本必须切分。切分策略影响检索效果：太短导致语义不完整，太长导致向量表示被无关内容稀释。推荐先按段落切，再用滑窗补充上下文。

from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=100, separators=["\n\n", "\n", "。", "；"] ) all_chunks = [] for doc in documents: chunks = splitter.split_text(doc["content"]) for chunk in chunks: all_chunks.append({"text": chunk, "source": doc["source"]}) print(f"切分后文本块数量: {len(all_chunks)}")

医学文本里句子比较长，分隔符要放。和；，只按换行切会把大段内容切碎。

4.3 向量化并写入 ChromaDB

向量化模型选择很关键。通用向量模型对医学专业术语支持有限，优先选择在中文语料上训练且对领域词汇友好的模型，比如BAAI/bge-m3。如果确实没有条件下载 embedding 模型，也可以用text2vec-large-chinese这类通用中文模型做兜底，但要意识到医学实体召回会有损失。

from chromadb import PersistentClient from chromadb.utils import embedding_functions client = PersistentClient(path="./medical_kb_chroma") embedding_fn = embedding_functions.SentenceTransformerEmbeddingFunction( model_name="BAAI/bge-m3" ) collection = client.get_or_create_collection( name="medical_kb", embedding_function=embedding_fn ) ids = [f"chunk_{i}" for i in range(len(all_chunks))] documents = [c["text"] for c in all_chunks] metadatas = [{"source": c["source"]} for c in all_chunks] # 如果集合已有数据，先清空再写入，避免重复 collection.delete(where={}) collection.add( ids=ids, documents=documents, metadatas=metadatas ) print(f"向量库已写入 {len(ids)} 条记录")

写入完成后，可以跑一次检索验证召回效果。

query = "肺腺癌的免疫组化标志物有哪些？" results = collection.query(query_texts=[query], n_results=3) for idx, doc in enumerate(results["documents"][0]): meta = results["metadatas"][0][idx] print(f"--- Top {idx+1} ---") print(f"来源: {meta['source']}") print(doc[:150]) print()

如果检索出的内容和问题明显不相关，优先检查切分粒度、embedding 模型和语料质量，而不是先调 Prompt。

5. RAG 问答流程实现

5.1 基础 Prompt 设计

医疗场景的 Prompt 要同时约束三个点：只根据检索内容回答、不编造、引用来源。基础模板如下。

SYSTEM_PROMPT = """ 你是一名严谨的医学知识助手。你的任务是基于提供的医学参考资料回答用户问题。 规则： 1. 优先使用参考资料中的信息作答；如果资料不完整，必须说明信息不足。 2. 禁止编造诊断、药物剂量、治疗建议。 3. 输出结尾需要列出引用来源。 4. 如果问题不在你的知识范围内，明确回答“暂无法基于现有知识库回答”。 参考资料： {context} 用户问题： {question} """

5.2 检索增强生成

问答函数先检索向量库，再把结果拼进 Prompt，最后调用 Ollama。

import requests OLLAMA_URL = "http://127.0.0.1:11434/api/generate" MODEL_NAME = "qwen2.5:14b-instruct-q4_K_M" def build_context(query: str, top_k: int = 3): results = collection.query(query_texts=[query], n_results=top_k) chunks = [] sources = [] for doc, meta in zip(results["documents"][0], results["metadatas"][0]): chunks.append(doc) if meta.get("source") not in sources: sources.append(meta["source"]) context = "\n\n".join([f"[{i+1}] {c}" for i, c in enumerate(chunks)]) return context, sources def ask_medical_llm(question: str, top_k: int = 3): context, sources = build_context(question, top_k) prompt = SYSTEM_PROMPT.format(context=context, question=question) resp = requests.post( OLLAMA_URL, json={ "model": MODEL_NAME, "prompt": prompt, "stream": False, "options": {"temperature": 0.2, "max_tokens": 1024} }, timeout=180 ) answer = resp.json().get("response", "").strip() return { "question": question, "answer": answer, "sources": sources }

温度设置为 0.2 是为了压低随机性。医疗问答和创意写作不同，输出稳定性优先。

5.3 无知识库兜底

本地知识库不可能覆盖所有问题。对知识库检索不到相关内容的查询，直接把问题交给模型，但模型要用更保守的系统提示词。

FALLBACK_SYSTEM_PROMPT = """ 你是一名医学信息整理助手。你可以基于常识回答通识性医学问题，但必须注明“此为AI生成信息，仅供参考”。 涉及具体诊断、用药、治疗方案时，必须建议用户咨询执业医师。 """ def ask_fallback_llm(question: str): prompt = f"{FALLBACK_SYSTEM_PROMPT}\n\n用户问题：{question}" resp = requests.post( OLLAMA_URL, json={"model": MODEL_NAME, "prompt": prompt, "stream": False}, timeout=180 ) return resp.json().get("response", "").strip()

判断是否走兜底逻辑，可以在检索阶段设定相似度阈值，比如 Top-1 分数低于 0.45 就视为无相关材料。阈值以实际语料表现为准，没有统一的绝对标准。

6. 部署为 FastAPI 服务

6.1 接口设计

服务端提供三个接口：健康检查、单条问答、批量问答。批量接口直接读取 JSONL 文件，避免一次请求塞太多文本导致超时。

from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn app = FastAPI(title="Medical QA API") class QueryRequest(BaseModel): question: str top_k: int = 3 use_rag: bool = True @app.get("/health") def health(): return {"status": "ok", "model": MODEL_NAME} @app.post("/ask") def ask(request: QueryRequest): if not request.question.strip(): raise HTTPException(status_code=400, detail="question 不能为空") if request.use_rag: result = ask_medical_llm(request.question, request.top_k) else: answer = ask_fallback_llm(request.question) result = {"question": request.question, "answer": answer, "sources": []} return result if __name__ == "__main__": uvicorn.run(app, host="127.0.0.1", port=8000)

6.2 启动服务

conda activate medical_qa python api_server.py

服务启动后，通过http://127.0.0.1:8000/docs可以直接打开 Swagger 调试页面，这个很适合第一次验证接口参数格式。

6.3 调用示例

用 curl 或 Python requests 都能测试。

curl -X POST http://127.0.0.1:8000/ask \ -H "Content-Type: application/json" \ -d '{"question": "脑膜瘤WHO分级有哪几级？", "top_k": 3}'

import requests resp = requests.post( "http://127.0.0.1:8000/ask", json={"question": "肺腺癌常用免疫组化标志物有哪些？", "top_k": 3}, timeout=300 ) data = resp.json() print("答案:", data["answer"]) print("来源:", data["sources"])

返回值里带sources字段，这是医疗问答重要的可信度来源。没有引用来源的答案无法审计，不符合医疗辅助场景要求。

7. 批量评测与效果优化

7.1 批量评测流程

医疗问答系统上线前必须做批量评测。准备一个 JSONL 格式的测试集，每行包含question和reference_answer。

{"question": "肾透明细胞癌最常累及的基因是什么？", "reference_answer": "VHL基因"} {"question": "胃镜活检病理提示印戒细胞癌，最常见好发部位是？", "reference_answer": "胃窦"}

批量评测代码逐个读取题目，调用本地服务，输出结果保存为 CSV。

import json import csv import requests API_URL = "http://127.0.0.1:8000/ask" def read_test_set(path: str): items = [] with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: items.append(json.loads(line)) return items def run_batch(input_path: str, output_path: str): items = read_test_set(input_path) results = [] for idx, item in enumerate(items): print(f"处理进度: {idx+1}/{len(items)}") try: resp = requests.post(API_URL, json={"question": item["question"]}, timeout=300) data = resp.json() results.append({ "question": item["question"], "reference_answer": item["reference_answer"], "model_answer": data["answer"].replace("\n", " "), "sources": "; ".join(data["sources"]) }) except Exception as e: print(f"第 {idx+1} 条失败: {e}") results.append({ "question": item["question"], "reference_answer": item["reference_answer"], "model_answer": f"ERROR: {e}", "sources": "" }) with open(output_path, "w", newline="", encoding="utf-8-sig") as f: writer = csv.DictWriter(f, fieldnames=["question", "reference_answer", "model_answer", "sources"]) writer.writeheader() writer.writerows(results) print(f"批量评测完成，结果保存至 {output_path}") if __name__ == "__main__": run_batch("./test_set.jsonl", "./eval_result.csv")

输出 CSV 用utf-8-sig编码，是为了直接 Office 打开不乱码。

7.2 评估和优化方向

客观题可以做关键词匹配或语义相似度打分，主观题建议人工抽检。重点看三类错误：知识库有对应内容但模型没检索到，说明分块或 embedding 有问题；检索到了但答案偏离，说明 Prompt 约束不足；知识库没有但模型硬答，需要加强拒答逻辑。

优化方向按性价比排序：

1. 增加高质量语料，覆盖高频问题。
2. 调整chunk_size和chunk_overlap，让切块更贴合语义单元。
3. 换更强的 embedding 模型，提升专业术语召回。
4. 增加对比示例 Few-shot，让模型学会标准回答格式。
5. 如果预算充足，用评测集在 Qwen2.5-14B 上做 LoRA 微调。

RAG 系统先保证检索对，再优化生成。检索不到正确答案时，换 Prompt 换多少次都没用。

8. 常见问题与排查方法

单条请求超时后，建议在客户端做指数退避重试，不要频繁重发。批量任务里如果前几条已经超时，多半不是单条问题，而是模型服务整体不可用，此时应停掉整个批量任务排查服务状态。

9. 最佳实践与后续扩展

先用最小用例验证三条链路：Ollama 模型调用通不通、ChromaDB 检索到不到内容、FastAPI 接口返回格式对不对。链路通了再扩大语料和生产化改造。

工程化时，有几点值得留意：

1. 模型和知识库分开目录管理，模型文件与知识库都做版本号标记，知识库更新后要重建向量集合，不能原地增量写入。
2. 日志要记录question、retrieved_chunks、model_answer和response_time，后续做质量分析才有数据。
3. 接口服务必须加访问限制，医疗数据接口不要直接暴露到公网，用内网部署加 API Key 认证。
4. 知识库内容变更后要跑一遍回归测试集，确认新增语料没有让旧问题的回答质量下降。
5. 评估结果要人审，尤其是主观题和涉及具体治疗建议的问题。

后续可以做几个方向的扩展：接语音识别做成医学语音问诊记录助手；用 LoRA 对模型做医学指令微调；把单机服务改成多卡部署并发推理。但每一步扩展都要重新评估合规边界，特别是涉及真实患者信息时，必须重新设计数据流和访问控制。

这套架构的价值在于，它能用开源模型和开源工具在可控预算内搭建一个领域知识问答系统。先跑通，再评估，最后再谈优化，这是最务实的路径。建议收藏备用，直接按文中流程在自己机器上验证一遍。