簡介基于Python的醫(yī)療知識圖譜自動問答系統(tǒng)源碼包適合對自然語言處理、知識圖譜和智慧醫(yī)療感興趣的開發(fā)者、研究員及畢業(yè)設(shè)計選題學(xué)生。項目從數(shù)據(jù)預(yù)處理入手覆蓋醫(yī)學(xué)詞匯、病癥、藥物、治療方法等信息的清洗、去重與標準化進而借助Neo4j等圖數(shù)據(jù)庫將實體和關(guān)系建模構(gòu)建結(jié)構(gòu)化的醫(yī)療知識圖譜同時使用NLTK、spaCy等工具完成問句的語義分析、實體識別與關(guān)系抽取配合Cypher查詢語句實現(xiàn)問題與圖譜節(jié)點的智能匹配最終通過Flask、Django搭建直觀的問答交互界面形成從數(shù)據(jù)到回答的完整閉環(huán)。壓縮包共26個文件以8個py源碼腳本、8個txt詞典與知識數(shù)據(jù)、5個xml工程配置、2個json規(guī)則文件為主大小15.54MB目錄拆分清晰包含數(shù)據(jù)準備、圖譜構(gòu)建、問題分類、答案搜索等模塊并附帶演示圖片方便對照運行結(jié)果。已有184人學(xué)習(xí)下載可作為從零搭建醫(yī)療問答系統(tǒng)的完整參考幫助掌握實體識別、意圖分類、圖譜查詢、回答生成等關(guān)鍵技術(shù)也為擴展其他垂直領(lǐng)域問答提供了可復(fù)用的代碼框架。1. 醫(yī)療知識圖譜自動問答系統(tǒng)先搞清楚它解決什么問題前臺護士每天要回答幾十次“頭疼掛什么科”一套基于 Python 的醫(yī)療知識圖譜自動問答系統(tǒng)會先在知識圖譜上查“頭痛→神經(jīng)內(nèi)科”這條邊再返回科室和下一步建議。它不是大模型不靠猜答案來自圖中真實事實可解釋、可控、可離線運行。這類系統(tǒng)由三塊組成醫(yī)療知識圖譜構(gòu)建、問句解析與意圖識別、查詢與應(yīng)答。標題里的“源碼”就是這條鏈路的完整工程形態(tài)。適合正在入門 python、想用知識圖譜做項目的開發(fā)者也適合醫(yī)療信息化團隊拿來快速搭一套可演示的問答原型。2. 圖譜構(gòu)建從表格到Neo4j決定問答上限的一步2.1 一份最小可用的醫(yī)療數(shù)據(jù)與關(guān)系設(shè)計先解決“數(shù)據(jù)從哪來”。常見做法是收集三類結(jié)構(gòu)化表格疾病-癥狀、疾病-科室、藥品-適應(yīng)癥。很多公開醫(yī)療網(wǎng)站的“疾病百科”“就醫(yī)指南”欄目就是這類知識抓下來整理成 CSV。這里有個誤區(qū)覺得“數(shù)據(jù)量越大系統(tǒng)越聰明”。實際上圖庫問答的鏈路不是靠規(guī)模撐起來的100 個高頻疾病、300 個癥狀、30 個科室、50 種常用藥就能讓系統(tǒng)有內(nèi)容可答關(guān)鍵是覆蓋高頻問法和高頻詞。真正要花心思的是關(guān)系設(shè)計。我第一版會只保留四條關(guān)系邊關(guān)系類型起點實體終點實體示例HAS_SYMPTOM疾病癥狀高血壓 - 頭暈DEPARTMENT疾病科室高血壓 - 心血管內(nèi)科TREATS藥品疾病硝苯地平 - 高血壓CHECKS疾病檢查高血壓 - 血壓監(jiān)測這張表就是知識圖譜的骨架。實體類型也不多Disease、Symptom、Department、Drug、Check后面所有問答都圍著這五種節(jié)點轉(zhuǎn)。關(guān)系一旦定下來別急著加“并發(fā)癥”“相關(guān)疾病”這類邊先把主干跑通。2.2 數(shù)據(jù)清洗與三元組生成pandas先把臟數(shù)據(jù)擋在門外拿到 CSV 之后第一步不是入庫是清洗。網(wǎng)頁復(fù)制出來的表格經(jīng)常帶首尾空格、全角空格、換行符不清洗就會在 Neo4j 里形成“高血壓”和“高血壓 ”兩個節(jié)點問答匹配時時好時壞非常折磨人。import pandas as pd def clean_entity_name(value): 統(tǒng)一實體名去掉首尾空格、全角空格、換行符避免匹配時翻車 if pd.isna(value): return None return str(value).strip().replace( , ).replace(\u3000, ).replace(\n, ) def build_triples(disease_symptom_file, disease_dept_file): ds pd.read_csv(disease_symptom_file, encodingutf-8-sig) dd pd.read_csv(disease_dept_file, encodingutf-8-sig) triples [] for _, row in ds.iterrows(): disease clean_entity_name(row[疾病]) symptom clean_entity_name(row[癥狀]) if disease and symptom: triples.append((Disease, disease, HAS_SYMPTOM, Symptom, symptom)) for _, row in dd.iterrows(): disease clean_entity_name(row[疾病]) dept clean_entity_name(row[科室]) if disease and dept: triples.append((Disease, disease, DEPARTMENT, Department, dept)) # 跨表去重保持首次出現(xiàn)順序 seen set() unique_triples [] for t in triples: if t not in seen: seen.add(t) unique_triples.append(t) return unique_triples編碼用 utf-8-sig 而不是 utf-8是因為很多從 Excel 另存出來的 CSV 帶 BOM用 utf-8 讀第一列會多一個不可見字符實體名匹配直接失敗。clean_entity_name 是我做圖譜項目必寫的第一個函數(shù)它同時處理了末尾換行和全角空格。參數(shù)說明兩個入?yún)⒎謩e是疾病-癥狀表、疾病-科室表的路徑返回三元組列表元素結(jié)構(gòu)是起點類型起點名關(guān)系類型終點類型終點名。去重放在最后做因為同一個“疾病-癥狀”對可能出現(xiàn)在多張來源表里跨表去重必須統(tǒng)一處理。2.3 用py2neo批量寫入Neo4j先建約束再按關(guān)系分組MERGE入庫順序比入庫本身更重要。我一般先建唯一約束再用 UNWIND 批量寫入。約束保證實體名唯一它同時也是查詢索引問答階段的匹配速度靠它兜底。from py2neo import Graph from collections import defaultdict graph Graph(bolt://localhost:7687, auth(neo4j, your_password)) # 先建唯一約束舊版Neo4j用ASSERT語法4.4用REQUIRE graph.run(CREATE CONSTRAINT disease_name IF NOT EXISTS FOR (n:Disease) REQUIRE n.name IS UNIQUE) graph.run(CREATE CONSTRAINT symptom_name IF NOT EXISTS FOR (n:Symptom) REQUIRE n.name IS UNIQUE) graph.run(CREATE CONSTRAINT dept_name IF NOT EXISTS FOR (n:Department) REQUIRE n.name IS UNIQUE) graph.run(CREATE CONSTRAINT drug_name IF NOT EXISTS FOR (n:Drug) REQUIRE n.name IS UNIQUE) graph.run(CREATE CONSTRAINT check_name IF NOT EXISTS FOR (n:Check) REQUIRE n.name IS UNIQUE) def import_triples(triples, batch_size500): # 按關(guān)系類型分組避免一個事務(wù)里混合多種關(guān)系模板 grouped defaultdict(list) for start_type, start_name, rel_type, end_type, end_name in triples: grouped[(start_type, rel_type, end_type)].append({ start: start_name, end: end_name }) for (start_type, rel_type, end_type), group in grouped.items(): for i in range(0, len(group), batch_size): rows group[i:i batch_size] cypher f UNWIND $rows AS row MERGE (s:{start_type} {{name: row.start}}) MERGE (e:{end_type} {{name: row.end}}) MERGE (s)-[:{rel_type}]-(e) graph.run(cypher, rowsrows) print(f關(guān)系 {rel_type} 已寫入 {min(i batch_size, len(group))} / {len(group)} 條)三個 MERGE 不等于三個 CREATE。MERGE 在節(jié)點已存在時只做匹配不重復(fù)創(chuàng)建配合唯一約束能避免二次導(dǎo)入時報錯。按關(guān)系類型分組是為了讓每個事務(wù)里的 Cypher 模板一致UNWIND 一批 500 條既快又容易定位失敗批次。類型名和關(guān)系名我直接拼進了 Cypher因為這些值來自代碼里的固定常量不是用戶輸入。如果哪天數(shù)據(jù)表里混入了外部傳入的關(guān)系名必須先做白名單校驗再拼不要直接格式化進查詢。batch_size 也是可調(diào)參數(shù)本地 Neo4j 可以調(diào)到 1000遠程實例建議 200防止單事務(wù)過大把內(nèi)存打滿。2.4 本體建模的邊界先少后多別讓關(guān)系成為負擔(dān)這一步其實就是知識圖譜構(gòu)建里最常說的本體建模定義實體類型和關(guān)系類型。醫(yī)療場景有個反直覺的經(jīng)驗——關(guān)系寧少勿多。常見做法是先覆蓋“癥狀-疾病-科室-藥品”四條主干邊等問答跑通、確實回答不了一些問題時再按需補“并發(fā)癥”“檢查”邊。我在早期項目里一口氣加了七種關(guān)系結(jié)果每種問法都要重新設(shè)計 Cypher 模板維護成本成倍上漲。還有一個容易被忽略的查詢邊界盡量別用可變長度路徑比如“-[:HAS_SYMPTOM*1..2]-”。在概念演示里它很漂亮但真實問答時它會把多條路徑的答案混在一起系統(tǒng)很難向用戶解釋“這個答案是怎么推出來的”。醫(yī)療場景要的是可解釋路徑越短越清晰。3. 問答引擎意圖識別、實體對齊與Cypher生成讓機器聽懂人話3.1 五類問題的意圖識別先用規(guī)則模板跑通再談模型自動問答的“自動”第一步是判斷用戶在問什么。常見問題類型有五類問癥狀、問科室、問藥物、問病因、問預(yù)防。很多人上來就想用 BERT 分類模型但手里訓(xùn)練樣本可能只有幾百條效果反而不如規(guī)則模板穩(wěn)定。import re INTENT_MAP { query_symptom: [什么癥狀, 有哪些癥狀, 癥狀表現(xiàn), 會有什么表現(xiàn)], query_department: [掛什么科, 什么科室, 哪個科室, 應(yīng)該看什么科], query_drug: [吃什么藥, 用什么藥, 藥物, 怎么用藥], query_cause: [病因, 什么原因, 怎么會得, 誘發(fā)因素], query_prevent: [怎么預(yù)防, 如何預(yù)防, 預(yù)防措施, 注意事項], } def classify_intent(question: str) - str: for intent, keywords in INTENT_MAP.items(): if any(kw in question for kw in keywords): return intent return query_general沒有用正則表達式只做子串匹配原因是醫(yī)療問句的變體實在太多正則寫起來又長又脆。每類意圖維護一個關(guān)鍵詞數(shù)組新增問法就加一行字符串代價最低。query_general 是兜底意圖表示系統(tǒng)沒聽明白后面會走回退邏輯。這個函數(shù)是純字符串匹配中文分詞不參與跑得極快。3.2 實體抽取與別名對齊jieba加自定義詞典命中“頭疼頭痛”問句進來之后要找出里面的疾病名、癥狀名。最常用的做法是 jieba 分詞加自定義詞典。醫(yī)療詞有兩個特點詞長不穩(wěn)定“神經(jīng)內(nèi)科”在默認詞典里會被切成“神經(jīng)/內(nèi)科”經(jīng)常會切錯別名多“頭疼”和“頭痛”在知識圖譜里可能是兩個寫法。import jieba # 自定義詞典提高醫(yī)療實體分詞穩(wěn)定度 for word in [神經(jīng)內(nèi)科, 心血管內(nèi)科, 高血壓, 糖尿病, 腦梗死, 冠心病]: jieba.add_word(word) # 別名表口語/別名 - 圖譜標準名 ALIAS_MAP { 頭疼: 頭痛, 血壓高: 高血壓, 血糖高: 糖尿病, 腦袋疼: 頭痛, 胸口悶: 胸悶, } def extract_entities(question: str) - list: tokens jieba.lcut(question) hits [] for token in tokens: token ALIAS_MAP.get(token, token) # 先做別名歸一 for entity_type, entity_names in ENTITY_NAMES.items(): if token in entity_names: hits.append((entity_type, token)) break return hitsENTITY_NAMES 是“實體類型 - 標準名集合”可以直接從圖譜里讀出來保持和庫內(nèi)完全一致。這里的關(guān)鍵是把別名映射放在分詞之后、匹配標準名之前。順序錯了比如先匹配再歸一結(jié)果就是“頭疼”在標準名集合里查不到直接被丟棄。參數(shù)說明jieba.add_word 會提升該詞在后續(xù)切分中的權(quán)重但它不是百分百強制極端情況下還是可能切錯。更保險的做法是把自定義詞典寫進 jieba 的 user_dict 文件并把這個文件放進項目倉庫里換環(huán)境時不用重新寫一遍。3.3 相似度兜底分詞沒命中系統(tǒng)還能不能救回來分詞畢竟有邊界。用戶說“血壓偏高”圖譜里只有“高血壓”jieb 切出來是“血壓”“偏高”實體抽取直接撲空。這種時候要有一個模糊匹配兜底常見做法是 difflib 的 SequenceMatcher。from difflib import SequenceMatcher def fuzzy_entity_lookup(name: str, entity_list: list, min_ratio: float 0.6): 在實體名列表里找最相似的候選低于min_ratio則不返回 best_match, best_ratio None, 0.0 for entity in entity_list: ratio SequenceMatcher(None, name, entity).ratio() if ratio best_ratio: best_match, best_ratio entity, ratio return best_match if best_ratio min_ratio else Nonemin_ratio 默認 0.6這個值是血淚經(jīng)驗調(diào)出來的。設(shè)太低比如 0.4“小兒肺炎”會匹配到“肺炎”給出的答案不準設(shè)太高比如 0.85別名基本救不回來。0.6 是在實體長度普遍為 2 到 6 個漢字時的折中值。要注意“血壓偏高”這句由兩個 token 組成fuzzy_entity_lookup 只能處理單個 token所以我會先把“血壓偏高”整體拼回去做一次查詢再考慮 token 級匹配。順序是完整句子先試一次分詞結(jié)果再試一次最后才是單 token 模糊。這一層兜底的意義是讓系統(tǒng)回答“不確定”時有個合理的近似答案而不是直接報錯。醫(yī)療場景寧可給近似條目并提示“為您找到相近的是”也不該空手而歸。3.4 生成Cypher查詢模板拼接與參數(shù)化一個都不能少意圖和實體都齊了下一步是把它們翻譯成圖查詢。這里的模板要按意圖和實體類型同時匹配比如“高血壓有什么癥狀”和“頭痛掛什么科”兩個問題的起點實體類型、關(guān)系邊都不一樣。def build_cypher(intent: str, entity_type: str, entity_name: str): if intent query_symptom: cypher ( fMATCH (d:{entity_type} {{name: $name}}) f-[:HAS_SYMPTOM]-(s:Symptom) RETURN s.name AS answer ) elif intent query_department: cypher ( fMATCH (d:{entity_type} {{name: $name}}) f-[:DEPARTMENT]-(dept:Department) RETURN dept.name AS answer ) elif intent query_drug: cypher ( fMATCH (d:{entity_type} {{name: $name}}) f-[:TREATS]-(drug:Drug) RETURN drug.name AS answer ) else: return None, None return cypher, {name: entity_name} def query_answer(cypher: str, params: dict): if not cypher: return [抱歉這個問題我暫時回答不了] records graph.run(cypher, parametersparams).data() return [record[answer] for record in records] or [知識圖譜里暫未查到相關(guān)記錄]實體名通過 $name 參數(shù)傳入而不是直接拼進 Cypher。參數(shù)化在這里有兩個作用一是避免實體名里帶單引號、雙引號時把查詢語法破壞掉二是防止注入雖然問答場景風(fēng)險有限但工程上保留這個習(xí)慣沒有壞處。entity_type 依然走白名單拼裝來源是 2.2 節(jié)三元組里的固定類型不受用戶輸入影響。到這里一個完整問答請求已經(jīng)能夠閉環(huán)問句進來識別意圖抽取實體拼 Cypher查庫返回答案列表。完整系統(tǒng)通常會把 query_answer 包成一個 Flask 接口前端 POST 一句 json 就能拿到答案。接口本身是殼問答核心全在前面幾個函數(shù)里。4. 五個高頻踩坑記錄現(xiàn)象、原因、解決4.1 “頭疼”查得到“頭痛”查不到現(xiàn)象用戶輸入“頭疼”返回“知識圖譜里暫未查到相關(guān)記錄”輸入“頭痛”卻有完整答案。原因圖譜標準名只收了“頭痛”問句里的“頭疼”沒進別名表實體抽取階段就丟了。解決先建別名表把口語說法、方言、縮寫統(tǒng)一映射到標準名映射發(fā)生在實體匹配之前。別只做兩三個別名實際收集時用搜索詞的聯(lián)想詞去補表比如“血壓高”“血壓偏高”“HBP”都要收斂到“高血壓”。4.2 圖譜越大越慢約束沒建、索引沒有現(xiàn)象幾百條數(shù)據(jù)時秒回到幾千條后查詢明顯變慢有的 Cypher 要跑幾秒鐘。原因?qū)霑r沒建唯一約束也沒建屬性索引Neo4j 只能按 label 掃描所有節(jié)點做 name 匹配。解決導(dǎo)入前執(zhí)行 2.3 節(jié)那幾條 CREATE CONSTRAINT 語句它同時創(chuàng)建索引。索引要建在查詢的匹配字段上也就是 name 屬性。這個問題在數(shù)據(jù)量過萬后會更明顯項目初期就要建別等卡了再補。4.3 換臺電腦跑不起來py2neo與Neo4j版本不匹配現(xiàn)象代碼在自己機器上跑得好好的換臺電腦裝好 Neo4j啟動 Python 腳本直接報認證失敗或者提示 Graph.run 方法不存在。原因py2neo 4.x 和 5.x 的部分 API 不兼容Neo4j 5.x 對舊版驅(qū)動支持也有限兩個版本對不上就會出這類玄學(xué)問題。解決項目根目錄放 requirements.txt鎖死 py2neo 與 neo4j 驅(qū)動版本同時把 Neo4j 版本寫進 README。我一般也會順手寫一個連接測試腳本啟動后先跑一句 RETURN 1確認環(huán)境通了再跑導(dǎo)入和問答。4.4 問“病因”返回“科室”意圖模板太粗現(xiàn)象用戶問“高血壓是什么原因造成的”系統(tǒng)返回“心血管內(nèi)科”。原因INTENT_MAP 里 query_cause 只有“病因”“什么原因”這句恰好沒觸發(fā)整體落進了 query_department 的關(guān)鍵詞“什么”。解決不要只擴充關(guān)鍵詞要建一個意圖自檢集每個意圖至少準備 15 個變體問句跑一遍自動測試看哪些問句被分錯了類。規(guī)則模板的優(yōu)勢就是修起來快但也要有測試集兜住回歸否則改一個詞可能帶崩另一類問題。4.5 導(dǎo)入時報ConstraintValidationFailed重復(fù)實體沒有先合并現(xiàn)象第二次運行導(dǎo)入腳本圖庫報“Node already exists with label Disease and property name”。原因唯一約束建立后CREATE 語句試圖新建重名節(jié)點會直接拋錯。解決所有節(jié)點寫入統(tǒng)一用 MERGE它存在就匹配、不存在才創(chuàng)建。另外 2.2 節(jié)的三元組去重也不是可有可無同一對“疾病-癥狀”可能在多張來源表里重復(fù)出現(xiàn)導(dǎo)入前去重能減少一半這類報錯。5. 進階從能問答到答得可靠先過評估、歸一化、鏈指三關(guān)5.1 建一份測試問句集算準召評價問答系統(tǒng)不能靠拍腦袋。常見做法是準備 100 條測試問句每條標注標準實體、標準答案然后跑一個簡單評估腳本def evaluate(test_set, answer_func): total_recall total_precision 0.0 n 0 for case in test_set: answers set(answer_func(case[question])) gold set(case[gold_answers]) if not gold: continue total_recall len(answers gold) / len(gold) total_precision (len(answers gold) / len(answers)) if answers else 0 n 1 return total_recall / n, total_precision / n召回率衡量“該給的答案漏沒漏”精確率衡量“給的答案準不準”。醫(yī)療問答里我一般更看重召回答不全比答錯好補救但精確率低于 0.7 時就要檢查實體對齊和意圖模板了。5.2 術(shù)語歸一化把“血壓高”“HBP”都收斂到“高血壓”別名表做到后期會成為一張標準術(shù)語表這是醫(yī)療問答從“能用”到“好用”的分水嶺。做法是給每個標準實體維護一個別名數(shù)組在問句預(yù)處理階段統(tǒng)一替換。這一步值得提前投入因為后續(xù)做實體鏈指、接口拓展都依賴一套穩(wěn)定的標準名。我見過團隊在這里偷懶結(jié)果換個數(shù)據(jù)源別名表推倒重來代價遠大于一開始就建詞表。5.3 實體鏈指處理“它”“這個病”的指代連續(xù)對話里用戶會說“它”“這個病”實體抽取抓不到。簡單做法是保存上一輪命中的實體當(dāng)前問句出現(xiàn)代詞時替換回去def resolve_coreference(question, last_entity): if not last_entity: return question return question.replace(它, last_entity).replace(這個病, last_entity)只適用于單輪指代。多輪對話的指代消解要上模型醫(yī)療場景里結(jié)論必須保守指代不確定時寧可多問一句“您說的是哪個病”也不要自作聰明。我最早搭這個系統(tǒng)時數(shù)據(jù)表里有一行疾病名后面帶了個換行符結(jié)果實體匹配時好時壞查了一晚上才發(fā)現(xiàn)是清洗環(huán)節(jié)漏了換行符。后來我把所有實體名統(tǒng)一走 clean_entity_name順序也固定為先洗數(shù)據(jù)、再建約束、最后寫問答邏輯。這個順序值得你記下來。希望幫到你。本文還有配套的精品資源點擊獲取