简介:一套基于Python的医疗知识图谱知识问答系统课程设计资源包,适合知识图谱与问答方向的初学者,帮助理解医疗实体关系建模、规则式对话流程及问句查询解析。压缩包共58个文件,约20MB,核心内容涵盖设计报告、源代码、医疗知识图谱数据、实体词典与规则文本、项目运行截图,目录按基础图谱、医疗图谱、问答机器人三个模块组织,便于按阶段学习。词典覆盖疾病、药品、食物、检查、科室等实体,支持问句分类与答案抽取,能有效展示知识图谱问答的完整链路。整体来看,图谱构建与问答查询全程无需训练、响应迅速,可作为课程设计与项目演示的完整方案。已有2570人学习,配套的设计报告、说明文档与可视化截图能帮助快速上手,既可直接运行体验,也可在此基础上进一步探索深度学习对问答灵活性的提升。
1. 医疗知识图谱问答系统:先想清楚这个 zip 里装的到底是什么
打开这类压缩包之前,我的第一个提醒是:基于 Python 实现的医疗知识图谱知识问答系统,绝大多数能分享出来的,都不是大模型对话机器人,而是「干净数据 + 图数据库 + 规则模板」的组合。它先把疾病、症状、药品、检查、科室之间的关系抽成一张图,再把用户的自然语言问句翻译成图查询,最终返回一段结构化答案。这种方案的问题定位很明确:常见病症咨询、科室导诊、用药常识这类封闭域问答;它能给你一个可解释、可离线、单机能跑的完整链路。适合正在做医疗信息化方向、入门 NLP 想走通全流程、以及需要课程设计落地的开发者。你真正要投入精力的,不是最后那个问答函数,而是建图谱之前的数据清洗和跑通之后的一整套排错思路。
2. 医疗知识图谱怎么建:本体设计、数据清洗与 Neo4j 入库
2.1 先定本体:为什么从五类实体、五种关系起步
拿到这类项目压缩包,我建议先别急着跑代码,第一件事是看数据长什么样。常见的数据格式是一张或几张二维表,列名类似“疾病名称、症状、常用药、就诊科室、推荐检查”,偶尔还混着几百条非结构化的文本主诉。如果数据本身是乱的,后面所有链路都会跟着乱。
本体设计我一般控制在“五实体五关系”这个量级,具体如下表。
| 实体类型 | 类标签 | 典型来源 |
|---|---|---|
| 疾病 | Disease | 病历诊断、教材目录、挂号主诉 |
| 症状 | Symptom | 主诉、体征描述、健康科普文本 |
| 药品 | Medicine | 处方记录、药品说明书 |
| 检查 | Check | 检验项目、影像检查项目 |
| 科室 | Department | 门诊科室表、导诊规则 |
关系我固定为五种:疾病到症状用 HAS_SYMPTOM,疾病到药品用 RECOMMEND_DRUG,疾病到检查用 SUGGEST_CHECK,疾病到科室用 VISIT_DEPARTMENT,药品到禁忌人群用 CONTRAINDICATION。这套设计的取舍在于:问答层是模板驱动的,每一种关系都要对应两到三个问法模板,关系一旦超过十种,模板维护成本会直接失控。五实体五关系刚好覆盖最常见的四类提问:有什么症状、吃什么药、挂什么科、做什么检查,剩下的需求等真出现了再一个关系一个关系地扩展。
存储选 Neo4j 而不是关系型数据库,理由很直接:多跳查询。用户问“高血压患者头疼应该挂什么科”,潜在路径是 疾病到症状再回疾病再到科室,这种查询用 SQL 要写多个 JOIN,而在图数据库里就是一条路径遍历的事。几十万三元组的数据规模,单机 Neo4j 完全扛得住,没必要一上来就上分布式。
2.2 数据清洗:从 GBK 乱码到标准三元组
医疗数据的编码问题是我见过最普遍的坑。很多脱敏导出的 Excel 或 CSV 是 GBK 编码,直接用 pandas 读取会得到一堆乱码。我有一次导入后查询,Neo4j 里全是“???”,排查了半小时才发现源头文件是 GBK。所以清洗的第一步永远是统一编码。
import pandas as pd # 医疗导出数据很多是 GBK 编码,read_csv 时先指定编码 df = pd.read_csv("raw_medical.csv", encoding="gbk") # 去掉疾病为空的行,以及完全重复的三元组 df = df.dropna(subset=["disease"]) df = df.drop_duplicates(subset=["disease", "symptom", "drug", "check", "department"]) # 去掉字段首尾空格,中文数据里全角半角空格都可能有 df["disease"] = df["disease"].str.strip() df["symptom"] = df["symptom"].str.strip() # 白名单式的别名收敛,不要用正则把“病”字全局删掉 df.loc[df["disease"] == "高血压病", "disease"] = "高血压" df.loc[df["disease"] == "上呼吸道感染", "disease"] = "感冒" # 统一写成 UTF-8,带 BOM 是为了照顾后续要读这个文件的 Windows 工具 df.to_csv("clean_medical.csv", index=False, encoding="utf-8-sig")这段代码里有三个细节值得单独说明。第一,encoding="gbk"只是常见值,如果报错可以依次试gb18030和utf-8,有些文件是混合编码,那就得用二进制模式逐行嗅探。第二,drop_duplicates的subset不是只留疾病那一列,而是几个关键字段一起判断,避免把同一疾病在不同症状上的有效记录删掉。第三,疾病名收敛建议用白名单映射,像“高血压病”到“高血压”这种是安全的,但“糖尿病”如果被正则砍掉“病”字会变成“糖尿”,反而是事故。
清洗完成后,我会顺手统计一下三元组总数和空值率。这一步值得做,因为后面图谱建好后发现某类关系明显偏少,回来查数据时,你还能想起来当时哪些行是被丢弃的。
2.3 LOAD CSV 批量写入:MERGE、索引、事务三个关键点
数据清洗成标准 CSV 后,常见的做法是用 py2neo 把 Cypher 发给 Neo4j 执行。文件要放在 Neo4j 的 import 目录里,路径是服务器视角,不是你 Python 脚本所在的目录,这个边界很多人第一次都会搞反。
from py2neo import Graph graph = Graph("bolt://localhost:7687", auth=("neo4j", "your_password")) # 先建唯一性约束,相当于关系型数据库里的唯一索引 graph.run("CREATE CONSTRAINT disease_name IF NOT EXISTS " "FOR (d:Disease) REQUIRE d.name IS UNIQUE") graph.run("CREATE CONSTRAINT symptom_name IF NOT EXISTS " "FOR (s:Symptom) REQUIRE s.name IS UNIQUE") # 用 LOAD CSV 批量导入,一次 Cypher 完成建节点和建关系 graph.run(""" LOAD CSV WITH HEADERS FROM 'file:///clean_medical.csv' AS row MERGE (d:Disease {name: row.disease}) MERGE (s:Symptom {name: row.symptom}) MERGE (d)-[:HAS_SYMPTOM]->(s) """)这里必须用 MERGE 而不是 CREATE,原因很直接:同一份 CSV 里,一个疾病会对应多个症状,如果后面继续导入药品、科室关系,你希望节点被复用而不是每次重建一条。CREATE 会在第二次导入时生成重复节点,MERGE 则会先查再写,保证同名的疾病始终是同一个节点。代价是 MERGE 比 CREATE 慢,但数据量在几十万这个量级时,差别可以忽略。
约束的作用是给 MERGE 的查询兜底。Disease 的 name 属性有了唯一约束后,Neo4j 会自动建索引,后续问答里大量MATCH (d:Disease {name: $name})查询就会走索引而不是全库扫描。等查询慢到不能忍再回来补索引,等于吃后悔药,不如建库时一步到位。
如果是自己写 Python 循环逐行插入而不是 LOAD CSV,那就必须处理事务分批。一次性把所有数据包在一个事务里,内存压力会很大;每行单独提交,速度又慢得让人怀疑人生。我一般按 1000 条一批提交:
tx = graph.begin() for idx, row in df.iterrows(): tx.run("MERGE (d:Disease {name: $name})", name=row["disease"]) if idx % 1000 == 0: tx.commit() tx = graph.begin() tx.commit()这个idx % 1000就是事务边界参数,它决定了单次事务里塞多少条写入。太小则提交频繁、总耗时变长;太大则单事务内存占用高、失败回滚代价大。1000 是我常用的起点,如果你一次性导入的是几十万行,可以压到 500,给旧机器留点余量。
入库之后别急着写问答,先用几条聚合查询验证数据质量。比如查每种疾病下挂了多少个症状,排序看看有没有明显偏少的:
MATCH (d:Disease)-[r:HAS_SYMPTOM]->(s:Symptom) RETURN d.name AS disease, count(r) AS symptom_count ORDER BY symptom_count DESC LIMIT 20;如果发现某类疾病只有一条症状,大概率是清洗阶段丢数据了。图谱里的关系数量不对,问答做得再花哨也是白搭,因为答案本质上就是把图里的关系翻译成人话。我会在这个阶段多花半小时把数据补全,而不是等到问答返空再回头找原因。
3. 问答链路拆开看:从实体识别、意图分类到答案生成的完整流程
3.1 先建两张词典表:实体词典、同义词映射
问答链路的第一环是实体识别,目标是问句里的“高血压”对上图谱里的疾病节点。这一步我不用命名实体识别模型,而是用自定义词典。理由很实际:医疗实体是封闭集合,图谱里有哪些病、哪些药,词典里就列哪些词;模型在几百个实体这种规模上,效果并不会比词典好,反而更难解释为什么抽错。
jieba 自定义词典的格式是“词语 词频 词性”三列,词频越高,分词时越容易被当作一个整体词。我通常在医疗问法里给到 100 以上,避免长词被切散。
import jieba # 每行格式:词语 词频 词性 jieba.load_userdict("data/medical_dict.txt")词典文件内容大致如下:
高血压 100 n 高血压病 100 n 上呼吸道感染 100 n 布洛芬 100 n 偏头痛 100 n词频这个参数是 jieba 分词结果的关键。给太低,词典词在和其他词竞争时会被切开;给太高,则会过度拟合,把“高血压患者”整个切成一个词,虽然不影响实体匹配,但会影响后面的意图关键词判断。100 是我试下来比较稳的起点。
实体词典解决“这是不是医疗实体”的问题,同义词映射解决“用户说的和图里存的不是同一个词”的问题。图谱里存的是标准名,用户嘴里说的是口语名,这个映射表就是两者之间的翻译层。
ALIAS_MAP = { "高血压": "高血压", "血压高": "高血压", "血糖高": "糖尿病", "着凉": "感冒", "拉肚子": "腹泻", }注意映射表的键要多收录口语词,值一律指向图谱里的标准名。这个表不需要一次做全,是随着问答测试持续补充的。每发现一个查不到答案的问法,先看实体识别有没有抽出来,再查图谱里有没有这个标准名,最后决定是加词典还是加同义词。这个过程看起来原始,但它是整个问答系统里投入产出比最高的环节。
3.2 意图分类不需要模型:四类规则模板怎么落地
实体识别之后,下一步是判断用户想问这个实体的哪方面信息。意图分类在这个场景下不值得上模型,原因有三:意图类别只有症状、用药、科室、检查四类;问法高度模板化;规则漏了可以直接补关键词,模型漏了要重新标注数据训练。我用一张关键词规则表就能覆盖绝大多数问法。
| 意图 | 触发关键词 | 对应图谱关系 |
|---|---|---|
| symptom | 什么症状、表现、症 | HAS_SYMPTOM |
| drug | 吃什么药、用什么药、用药 | RECOMMEND_DRUG |
| department | 挂什么科、哪个科、科室 | VISIT_DEPARTMENT |
| check | 做什么检查、检查、确诊要查 | SUGGEST_CHECK |
规则分类的代码很简单,本质就是看问句里有没有命中这些关键词:
INTENT_RULES = { "symptom": ["什么症状", "有哪些症状", "表现", "症"], "drug": ["吃什么药", "用什么药", "用药", "药"], "department": ["挂什么科", "哪个科", "科室", "看什么科"], "check": ["做什么检查", "检查", "确诊", "查什么"], } def classify_intent(question: str) -> str: # 规则命中的优先级和列表顺序有关,前面的先匹配 for intent, keywords in INTENT_RULES.items(): for kw in keywords: if kw in question: return intent # 没命中就兜底到症状,这是医疗问答里最常见的问法 return "symptom"这段代码有两个地方值得说。第一,关键词的优先级由字典顺序决定。比如“吃什么药需要做什么检查”同时命中 drug 和 check,规则会先走 drug;从产品体验看,“吃什么药”是更具体的诉求,优先给用药没问题。第二,默认兜底是 symptom,因为“XX 有什么症状”是这一类系统里最高频的问法;如果以后统计数据发现另一类占比更高,把默认值改掉就行。
有人会把规则关键词写复杂,比如用正则匹配“有什么.{0,3}症状”来容忍插入语气词。我的经验是,直接多列几个关键词比写花哨的正则更容易维护,因为每一种新说法你都看得见、改得快。正则一旦复杂,后期排查时还得现推语义,反而慢。
3.3 把意图翻译成 Cypher:参数化查询与三种常用模板
意图确定后,就要拼 Cypher。这一步最容易掉进去的坑是字符串拼接,把用户输入的实体名直接 f-string 拼到查询里,遇到引号或特殊字符就会让整个查询语法报错。我习惯统一用参数化查询,py2neo 的graph.run()直接支持关键字参数绑定。
def build_cypher(intent: str, entity: str) -> str: # 每种意图对应一条查询模板,返回的是带 $name 占位符的 Cypher if intent == "symptom": return ("MATCH (d:Disease {name: $name})-[:HAS_SYMPTOM]->(s:Symptom) " "RETURN s.name AS name LIMIT 5") if intent == "drug": return ("MATCH (d:Disease {name: $name})-[:RECOMMEND_DRUG]->(m:Medicine) " "RETURN m.name AS name LIMIT 5") if intent == "department": return ("MATCH (d:Disease {name: $name})-[:VISIT_DEPARTMENT]->(dep:Department) " "RETURN dep.name AS name LIMIT 3") if intent == "check": return ("MATCH (d:Disease {name: $name})-[:SUGGEST_CHECK]->(c:Check) " "RETURN c.name AS name LIMIT 5") return "" def run_query(graph, cypher: str, entity: str): # 实体名作为参数传入,不是拼接进字符串 return graph.run(cypher, name=entity).data()LIMIT 这个参数值得根据场景调一下。症状和检查的项目通常比较多,返回 5 条比较合理;科室一般就一个到两个,LIMIT 3 足够。有些项目喜欢把 LIMIT 写 10 或 20,答案拼出来一大段,用户阅读体验反而差。问答不是查全量,是给一个够用的答案。
参数化还有一层好处是 Neo4j 会缓存查询计划。同样结构的 Cypher 用参数化写法,服务端可以将 parse 和 plan 的结果复用,虽然小数据量下感知不明显,但并发上来以后能省一部分 CPU。
3.4 空了怎么办:同义词兜底与答案格式化
查询结果为空的情况,在医疗问答里非常常见。我会按“直查 → 同义词替换 → 放弃”的顺序做兜底,而不是直接告诉用户没找到。
def answer_question(graph, question: str) -> str: entity = find_entity(question) if entity is None: # 实体词典都没命中,说明问法超出覆盖范围 return "暂时没听懂,换个方式描述问题,比如换成疾病名称加提问" intent = classify_intent(question) cypher = build_cypher(intent, entity) rows = run_query(graph, cypher, entity) # 第一轮查询为空,尝试同义词映射后再查一次 if not rows and entity in ALIAS_MAP: target = ALIAS_MAP[entity] rows = run_query(graph, cypher, target) return format_answer(intent, entity, rows)这里有个细节:format_answer里展示实体名时,要保留用户原始输入的那个词,而不是同义词替换后的标准名。用户问“血压高怎么调理”,如果回答里说“高血压”的用药,用户会觉得自己被纠正了,体验不好。用原词组织答案就行,反正嵌套在自然语言里不影响理解。
答案格式化的模板我通常直接写在单独的映射里,方便统一修改生成规则:
def format_answer(intent: str, entity: str, rows: list) -> str: if not rows: return f"在现有知识库里,暂时没找到「{entity}」相关的{intent}信息" names = "、".join([r["name"] for r in rows]) if intent == "symptom": return f"「{entity}」的常见症状包括:{names}" if intent == "drug": return f"「{entity}」常用的药物有:{names}" if intent == "department": return f"「{entity}」建议就诊的科室是:{names}" if intent == "check": return f"「{entity}」通常需要做的检查有:{names}" return "查无结果"整个问答链路到这里就闭环了。你会发现它没有任何一个环节用了机器学习模型,但回答的确定性和可解释性都很好,这恰恰是这个方案在医疗场景的价值:答案错了,你能顺着实体识别、意图分类、知识图谱三层一路查下去,知道是哪一环出了问题。
4. 用 Flask 封装成 Web 服务:最小接口与三个必调参数
4.1 最小问答接口:POST /qa 与统一返回格式
问答链路在 Python 函数里跑通后,下一步是用 Flask 把它变成一个可以调用的 Web 服务。我不建议在这个阶段引重型框架,Flask 足够。接口设计上,我坚持用 POST + JSON,不用 GET,因为问题文本可能很长,而且后续要加用户标识、多轮上下文时,JSON 格式扩展起来方便。
from flask import Flask, request, jsonify from functools import lru_cache app = Flask(__name__) # graph 在模块加载时初始化一次,后续所有请求复用同一个连接 graph = Graph("bolt://localhost:7687", auth=("neo4j", "your_password")) @lru_cache(maxsize=512) def cached_answer(question: str) -> str: # 相同的问题直接走缓存,不再打 Neo4j return answer_question(graph, question) @app.route("/qa", methods=["POST"]) def qa(): payload = request.get_json(force=True) question = (payload.get("question") or "").strip() if not question: return jsonify({"code": 400, "message": "question 不能为空"}), 400 answer = cached_answer(question) return jsonify({"code": 0, "question": question, "answer": answer}) if __name__ == "__main__": # 开发阶段用内置服务,生产环境不要直接这样跑 app.run(host="0.0.0.0", port=5000, debug=False)接口写好后,用 curl 验证是最快的路径:
curl -X POST http://localhost:5000/qa \ -H "Content-Type: application/json" \ -d '{"question": "高血压有什么症状"}'request.get_json(force=True)这个参数是我刻意加的。它允许客户端即使没有正确设置 Content-Type 也能解析 JSON,代价是宽容了一点请求格式。内部工具这么做省去很多联调摩擦,但如果对外提供服务,建议还是严格要求 Content-Type。
lru_cache(maxsize=512)是进程内缓存,参数值根据单机内存和问题量来调。512 个常见问题的答案,每条平均 200 字,内存占用可以忽略。要注意缓存的是传入的原始 question 字符串,所以接口层要先做 strip 再进缓存,否则“高血压”和“高血压 ”是两个 key。
4.2 三个必调参数:单例 Graph、缓存、gunicorn 并发
把这套服务从“本地能跑”变成“别人也能用”,我一般会先检查三个参数,少了任何一个都会在正式使用阶段翻车。
第一个参数是 Graph 对象的作用域。Graph 对象底层维护的是到 Neo4j 的连接池,如果在每次请求里都 new 一个新的 Graph,等于每次请求都新建物理连接,连接池里的旧连接又得不到复用,文件描述符耗尽只是时间问题。把它定义成模块级全局单例,所有请求共享,这是性能上最便宜的一笔优化。
第二个参数是上面的lru_cache(maxsize=512)。医疗问答的重复提问率极高,同一批常见病在一天内会被问几十遍。没有缓存时,每遍都打一次 Neo4j,虽然单次只要几十毫秒,但并发一上来,Neo4j 的事务并发能力就成了瓶颈。缓存命中后,接口耗时能压到 5 毫秒以内,用户体验差异非常大。
第三个参数是部署时的 worker 数量。Flask 内置的app.run()是单进程单线程开发服务器,三个并发请求就能把它卡住。生产环境我一般用 gunicorn 起多 worker:
gunicorn -w 4 -b 0.0.0.0:5000 app:app --timeout 60-w 4是 worker 进程数,小项目从 4 起步就行;--timeout 60表示单个请求超过 60 秒会被强制结束,这个值要看你的 Cypher 查询耗时来定,如果图谱查询偶尔有慢查询,60 秒是一个比较安全的余量。这里有个副作用值得提前知道:lru_cache是进程内的,4 个 worker 的缓存互不相通,等于同一条问题最多会在 4 个进程里各查一次。数据量小的时候无所谓,量大时再把缓存换成 Redis 统一管理。
5. 避坑:医疗问答系统跑不通时的 6 个高频排查点
这篇文章写到这,我整理一下自己调试这类系统时的血泪经验。以下 6 个问题按出现频率排序,前两个几乎每个项目都会遇到。
5.1 py2neo 与 Neo4j 版本不一致:最常见的“装好连不上”
现象:代码里from py2neo import Graph完全正常,但执行graph.run()时直接报连接错误,或抛出与 Bolt 协议相关的异常;Neo4j 浏览器却可以正常打开。
原因:py2neo 的最后稳定版停留在 2021.2.x,其底层驱动对较新的 Neo4j 服务端协议兼容性不够,尤其是 Neo4j 5.x 之后,默认启用的协议版本和旧驱动对不上。这不是代码写错,是版本矩阵没对上。
解决:最简单的办法是把 Neo4j 固定在 4.4 版本;想继续用新版本的话,直接换 Neo4j 官方驱动:
from neo4j import GraphDatabase driver = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "your_password")) with driver.session() as session: result = session.run( "MATCH (d:Disease {name: $name})-[:HAS_SYMPTOM]->(s:Symptom) RETURN s.name", name="高血压" ) rows = [record["s.name"] for record in result]两种方案我都用过。图省事就锁 Neo4j 4.4,代码改动最小;走官方驱动则更接近生产环境的生态,但 py2neo 的graph.run(...).data()写法要改成官方 session 的写法,改造量几百行。取决于你的项目里 py2neo 封装得有多深。
5.2 导入中文乱码:LOAD CSV 编码与路径两个坑
现象:LOAD CSV 导入后,查询出来的中文全是“???”,或者直接报编码错误;有时候同一份文件在 Windows 能导入,在 Linux 上就失败。
原因:Neo4j 的 LOAD CSV 要求文件是 UTF-8 编码,很多医疗导出数据是 GBK。另一个坑是路径,LOAD CSV 里的file:///指向的是 Neo4j 服务端的 import 目录,不是 Python 脚本旁边。
解决:数据清洗阶段统一用encoding="utf-8-sig"写文件,UTF-8 带 BOM 对中文兼容性更好;文件放进 Neo4j 的 import 目录后再执行 LOAD CSV。我自己踩过一次之后,就再也没跨目录导过文件,所有 CSV 一律先落到 import 目录再做导入。
5.3 jieba 把“高血压”切成“高血/压”:词典与长词优先
现象:jieba.lcut("高血压有什么症状")返回的结果是["高血", "压", "有什么", "症状"],实体识别匹配不到图谱里的任何疾病节点。
原因:自定义词典没加载,或者加载了但词频太低,分词器把长词让位给了更细粒度的切分。这是 jieba 默认词库在通用场景下的常见行为。
解决:加载自定义词典时把词频写到 100 以上,并且把常见的口语变体也加进去,比如“血压高”。更稳妥的方案是实体匹配不完全依赖 jieba 分词,直接遍历实体词典,看哪个词出现在问句里;如果实体词有长有短,先匹配长度更长的那个,避免“高血压”还没匹配上,“血压”先被抽出来。
5.4 并发一上来就 Transaction timeout:索引与跳数两个方向
现象:单条测试查询一切正常,并发到十几路请求后,Neo4j 报事务超时,错误信息里带 timeout 字样。
原因:两个叠加因素。一是实体名上没有索引,查询全库扫描;二是查询路径超过两跳,Neo4j 的规划器在小数据量下不会主动优化深路径。
解决:建库时给所有节点的 name 属性建唯一约束,这个前面已经说过;再就是检查问答模板,把 Cypher 的路径控制在两跳以内。疾病查症状是一跳,疾病查科室是一跳,疾病查症状再查药品是两跳,尽量别让问答模板出现三跳以上的路径。真出现三跳需求,我会考虑把冗余关系也建模成一跳,比如直接建“症状到药品”的直达边,用空间换查询时间。
5.5 缓存只在单个 worker 内生效:多进程部署后缓存失效
现象:本地 Flask 单进程测试,重复问同一个问题特别快;一上 gunicorn 多 worker,同样的问题每隔一段时间就变慢一次。
原因:lru_cache是进程内缓存,gunicorn 的 4 个 worker 各自持有一份,A worker 缓存过的问题,轮到 B worker 处理时又是冷缓存。
解决:如果请求量不大,workers=1就能彻底避免这个认知偏差;请求量起来后,把缓存换成 Redis,question做 key,答案做 value,所有 worker 共享同一份缓存。Redis 的引入会多一个运维组件,但这个代价在缓存命中率高的场景里是值得的。
5.6 用 app.run() 直接扛生产:开发服务器并发一多就卡
现象:服务虽然后台能跑,但一旦有前端页面或脚本并发调用,请求排队明显,卡顿到超时。
原因:Flask 内置的 Werkzeug 开发服务器默认单进程单线程,它的定位是给开发调试用的,不是给生产流量用的。
解决:换 gunicorn 启动,这是改动最小收益最明显的一步。命令只需要一行:
gunicorn -w 4 -b 0.0.0.0:5000 app:app --timeout 60如果服务器 Windows 环境用不了 gunicorn,可以用 waitress 替代,接口代码完全不用动,只是启动命令从python app.py变成waitress-serve --port=5000 app:app。这个问题不解决,后面任何关于性能和并发的讨论都没有意义。
6. 进阶:用向量检索兜底,覆盖用户说不全实体名的问法
规则模板最大的痛点是覆盖不全。用户问“血压高该注意点什么”,实体识别能抽出“血压高”,同义词表里如果有映射就能转到“高血压”;但如果用户问“老是头晕该挂哪个科”,实体识别抽出来的是“头晕”而不是“高血压”,问题就断在实体链接这一步了。我常用的做法,是给问答链路加一层向量检索兜底,专门处理这类不按模板说话的情况。
思路是预先生成一批问法模板,让它们只作为“问题范式”而不是具体答案。比如:
templates = [ "{disease}有什么症状", "{disease}吃什么药", "{disease}挂什么科", "{disease}需要做什么检查", ]然后用 SentenceTransformer 里的一个多语言 MiniLM 模型,把这些模板的占位符替换成“疾病”后统一编码;用户问题进来后也做一次编码,和模板库算余弦相似度,取 top1 模板,再从用户问句里抽取实体词做槽位填充。
from sentence_transformers import SentenceTransformer import numpy as np model = SentenceTransformer("paraphrase-multilingual-MiniLM-L12-v2") # 模板统一编码,占位符先填上常见词,避免模板本身太短导致编码不稳定 template_embs = model.encode([t.replace("{disease}", "疾病") for t in templates]) def match_template(question: str) -> str: q_emb = model.encode([question])[0] scores = np.dot(template_embs, q_emb) # 简化版,实际建议先归一化 return templates[int(np.argmax(scores))]向量召回只解决模板选择的问题,实体抽取还是靠前面的词典。它的价值在于:用户没说出“药”“症状”“挂科”这些关键词时,规则分类器会无脑兜底到 symptom,而向量模型能识别出“该注意点什么”更接近用药或生活建议类意图。模板命中后,再回到原来的意图分类或直接映射到对应 Cypher 模板。
验证这个方法值不值得接入,我的习惯是准备 100 条真实问法,跑一遍旧规则,再跑一遍规则加向量兜底,对比两类问题的答案命中率。如果提升不到 10 个百分点,我会认为是同义词表没补够而不是向量方案的问题。实际经验里,向量兜底对“口语化表述”的提升最明显,但对“实体词本身说错”的情况无效,后者只能靠图谱端把别名收敛做好。
如果重做一次,我会在一开始就把同义词表和实体词典补到尽可能全,把向量兜底直接设计进主流程,而不是等规则漏了一轮再去查缺口。这两个东西一个管实体,一个管意图,互补之后这个医疗问答系统的可用度才会真正上去。希望帮到你。
本文还有配套的精品资源,点击获取