简介:基于Rasa框架的智能医疗机器人项目,面向毕业设计、课程设计与实际项目开发场景,提供从医药问答、疾病诊断到语音对话的完整功能链。资源共141个文件,以Python源码、Rasa配置文件、Neo4j数据库备份及开发文档为主,另有语音与测试数据,整体约100.72MB,能支撑环境搭建到功能扩展的全流程。已有41人学习下载。项目集成知识图谱、语音识别/合成和开放天气API,配有技术架构说明与数据库脚本,参考价值高,可直接理解业务逻辑、对话流程设计及图谱查询方式,适合需要快速上手Rasa或构建医疗领域智能助手的开发者。
1. 为什么医疗问答机器人要基于 Rasa 自建,而不是直接套对话 API
一位患者在小程序里输入"这两天一直咳嗽,嗓子疼,该吃什么药"。如果后端把这句直接丢给通用大模型接口,返回内容既无法保证剂量来自本院药品目录,也没法给每条回答挂上免责和溯源字段。Rasa 的做法是把"理解"和"查询"拆开:NLU 从口语里抽意图和实体,自定义 Action 去查药品库、疾病库,再由对话策略决定是直答、追问还是转人工。
模型在本地 CPU 就能推理,患者主诉和查询日志不出内网,这是医疗场景的硬要求,也是 Rasa 相对云端对话平台的核心优势。标题里的医药问答、智能问药、疾病诊断、病症查询、症状查询和语音对话,落地时对应意图识别、槽位抽取、规则化对话流程、知识库查询、STT/TTS 接入五块。下面按这条链路展开,代码基于 Rasa 3.x 和 Python 3.8 以上环境,先跑通骨架,再逐步加厚。
2. 搭一个能跑通的 Python/Rasa 医疗机器人骨架:意图、实体与领域规则
2.1 先划领域边界:医疗机器人至少要定义 6 个意图
写 nlu.yml 之前,先列意图清单。意图不是越细越好,医疗场景里"症状查询"和"疾病诊断"边界最容易混:用户说"头疼怎么办"是症状查询,"持续低烧三天是不是肺炎"才是诊断意图。我给这类项目定的初始意图表如下。
| intent | 典型例句 | 要抽取的实体 | 回答策略 |
|---|---|---|---|
| greet / goodbye | 你好 / 再见 | 无 | 固定话术 |
| ask_medicine | 感冒了吃什么药 | medicine_disease, symptom | 查药品库 |
| ask_symptom | 头疼挂哪个科 | symptom | 查症状-科室映射 |
| ask_disease | 糖尿病有什么症状 | disease_name | 查疾病库 |
| diagnose_intent | 反复咳嗽发烧是肺炎吗 | symptom + duration | 多轮追问后给建议 |
| ask_drug_usage | 布洛芬怎么吃 | medicine_name | 查药品说明书字段 |
表格里的"回答策略"直接决定后续 Action 怎么写。ask_disease 是一次查询,diagnose_intent 必须走多轮:先确认症状、再问持续时间、最后给"疑似范围 + 就诊建议",不能在首轮就下结论。这六类之外的话术,统一走 FallbackClassifier 的兜底分支。
2.2 nlu.yml 与 domain.yml:症状、药品、疾病三类实体的标注写法
环境准备好之后,先建项目骨架:conda create -n rasa python=3.8,然后pip install rasa rasa-sdk,再rasa init --no-prompt生成标准目录。nlu.yml 里实体要标注够,也要留改写空间。药品名和疾病名适合用 lookup 表兜底,症状名适合直接写进 example。下面是一个能直接放进 data/nlu.yml 的最小集合。
version: "3.1" nlu: - intent: ask_medicine examples: | - 感冒了吃什么[药](medicine_disease) - [发烧](symptom)应该吃什么[药](medicine_disease) - 有没有治[咳嗽](symptom)的[药](medicine_disease) - 帮我查一下[布洛芬](medicine_name) - intent: ask_disease examples: | - [糖尿病](disease_name)有什么症状 - [高血压](disease_name)平时要注意什么 - intent: diagnose_intent examples: | - 持续[低烧](symptom)[三天](duration)是不是[肺炎](disease_name) - 我[干咳](symptom)还[胸闷](symptom),会不会是[支气管炎](disease_name) - lookup: medicine_name examples: | - data/lookup/medicine.txt实体名用了中性的槽位:medicine_disease 表示"症状对应的病症说法",medicine_name 才是具体药品。这样设计的好处是,用户说"感冒了吃什么药",NLU 能把"感冒"识别成症状实体而不是药品名,Action 再拿 symptom 值去查疾病-药品关联表。lookup 文件每行一个药品名,覆盖院内药品目录即可,不需要上模型去认。
domain.yml 里要把 intent、entity、slot、action 全部声明一遍。Rasa 3.x 的规则化能力取决于 domain 里的 action 列表,漏声明会直接报错。
version: "3.1" intents: - greet - ask_medicine - ask_disease - ask_symptom - diagnose_intent entities: - symptom - disease_name - medicine_name - duration slots: symptom: type: list influence_conversation: true disease_name: type: text influence_conversation: true actions: - action_query_medicine - action_query_disease - action_diagnose注意 symptom 用了 list 类型的槽位,因为一个句子里可能出现"干咳 + 胸闷"两个症状,text 类型只能存最后一个。duration 这类补充信息不进槽位也行,诊断 Action 里直接从 latest_message 的实体里取,减少槽位互相覆盖带来的误判。
2.3 rules.yml 控制诊断边界:症状没齐就不给结论
诊断流程必须用 rules 而不是 stories 来锁。stories 适合学出来的自由对话,诊断是医疗安全边界,规则要硬编码。diagnose 的三步规则如下。
rules: - rule: 开始诊断先问症状 steps: - intent: diagnose_intent - action: action_diagnose - active_loop: nullaction_diagnose 内部维护一个状态:第一次进来只追问"具体是哪里不舒服,持续多久",第二次带着之前对话的实体再查知识库。为了不让规则可复现性受模型训练影响,不要把追问逻辑拆成多条 rule,而是在一个自定义 Action 里根据 tracker 拿历史实体。这样训练集里只要一条规则,追问节奏全部由代码控制,后续改追问话术也不动模型。
提示:rules.yml 里同一条 rule 不要既匹配 diagnose_intent 又匹配后续的确认意图,否则 RulePolicy 会和其他 policy 抢优先级。追问循环收进 Action 内部状态,是最省心的做法。
3. 医药问答与疾病诊断的检索管线:自定义 Action 才是核心
3.1 为什么查询逻辑不进 Story:Action 与规则的分工
Rasa 新手最容易犯的错,是把"查药品"当成一个故事写进 stories.yml,然后在 NLU 里堆几千条例句,指望模型学会回答。实际跑起来会发现:药品目录一变、科室改名、同义词新增,全要重新训练。正确的分工是:NLU 只负责把意图和实体从口语里抠出来,任何涉及知识库的查询、条件判断、拼装回复,全部放进 actions.py 的自定义 Action。Action 跑的是普通 Python 代码,更新药品库只需要重跑一次 SQL,不用碰模型。
这种分工还带来测试上的好处:Action 可以脱离 Rasa 单独用 pytest 打。把 tracker 的实体传进去,断言返回值里包含某个药品名,回归成本远低于端到端对话测试。
3.2 用 SQLite 当知识库:智能问药 Action 的完整代码
知识库选 SQLite 而不是 JSON 文件,是因为药品数据天然有结构化关联:药品、成分、适应症、禁忌、科室。下面这张表是智能问药模块的最小形态。
CREATE TABLE medicine ( id INTEGER PRIMARY KEY, name TEXT NOT NULL UNIQUE, -- 药品通用名 brand TEXT, -- 商品名 indications TEXT, -- 适应症,逗号分隔 dosage TEXT, -- 用法用量 contraindications TEXT, -- 禁忌 department TEXT, -- 对应科室 source TEXT -- 数据来源,用于溯源 );查询 Action 的核心逻辑是"先精确匹配,再模糊匹配"。用户说"布洛芬怎么吃",NLU 抽出 medicine_name=布洛芬,直接按 name 精确查;用户说"感冒吃什么药",拿到的实体是 symptom=感冒,就得先查症状-药品关联,再把关联结果按匹配度排序返回。
import sqlite3 from typing import Any, Dict, List, Text from rasa_sdk import Action, Tracker from rasa_sdk.executor import CollectingDispatcher DB_PATH = "data/medical.db" class ActionQueryMedicine(Action): def name(self) -> Text: return "action_query_medicine" def run(self, dispatcher: CollectingDispatcher, tracker: Tracker, domain: Dict[Text, Any]) -> List[Dict[Text, Any]]: conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row entities = list(tracker.latest_message.get("entities", [])) med_name = next((e["value"] for e in entities if e["entity"] == "medicine_name"), None) if med_name: rows = conn.execute( "SELECT * FROM medicine WHERE name = ? OR brand LIKE ?", (med_name, f"%{med_name}%")).fetchall() else: symptom = next((e["value"] for e in entities if e["entity"] == "symptom"), None) rows = conn.execute( "SELECT * FROM medicine WHERE indications LIKE ?", (f"%{symptom}%",)).fetchall() if not rows: dispatcher.utter_message(text="药品库里没有查到相关条目,建议先到门诊确认。") return [] for r in rows[:3]: dispatcher.utter_message( text=f"{r['name']}:{r['dosage']}。禁忌:{r['contraindications']}(来源:{r['source']})") conn.close() return []代码里的模糊查询用 LIKE 而不是全文检索,是因为 symptom 大概率是短词,"%感冒%"能覆盖大多数情况。rows[:3] 限制条数,避免一次抛出一长串药品让患者难以选择。每条回复都带 source 字段,这是后面做溯源和免责的地基。生产环境里可以把 dispatcher.utter_message 换成模板加结构化返回,前端拿到 JSON 再渲染成卡片。
3.3 症状匹配的权重算法:从 Jaccard 到同义词映射
疾病诊断查库时,患者原话和库里的症状词很少逐字一致。"胸口疼"要匹配"胸痛","拉肚子"要匹配"腹泻"。纯 SQL LIKE 在这里会大量漏检,需要在 Action 里做一次症状归一化。常用做法是两层匹配:第一层维护一个同义词映射表,把所有口语说法映射到标准症状词;第二层用 Jaccard 相似度对未命中文本做兜底。
SYNONYMS = { "胸口疼": "胸痛", "胸闷": "胸闷", "拉肚子": "腹泻", "头疼": "头痛", "脑袋晕": "头晕", "嗓子疼": "咽痛", } def normalize_symptom(text: str) -> str: return SYNONYMS.get(text.strip(), text.strip()) def jaccard(a: set, b: set) -> float: if not a or not b: return 0.0 return len(a & b) / len(a | b) def match_disease(symptom_text: str, disease_rows: list) -> list: norm = normalize_symptom(symptom_text) scored = [] for row in disease_rows: std_symptom = set(row["symptom_keywords"].split(",")) score = jaccard({norm}, std_symptom) if norm in std_symptom: score = 1.0 scored.append((score, row)) scored.sort(key=lambda x: x[0], reverse=True) return [row for score, row in scored if score > 0.3]normalize_symptom 必须放在 jaccard 之前,因为"胸口疼"和"胸痛"字符级相似度是 0,不归一化就永远匹配不上。0.3 这个阈值是经验值:太低会把不相关科室的疾病列出来,太高会漏掉只有一个症状词重合的情况。实际项目里建议用一批真实门诊主诉做离线评测,把准确率和召回率打出来再定阈值,不要拍脑袋。
4. 语音对话接入:本地 STT、TTS 与 Rasa 的异步组合
4.1 语音链路先选型:离线识别还是云端识别
标题里的语音对话,在医疗场景几乎必须选离线方案。患者主诉是敏感数据,走云端识别意味着音频出内网,合规上很难解释。延迟反而是次要矛盾。常见做法是两种:一是服务端常驻本地 STT 模型,接收客户端上传的音频文件,识别完转文本再 POST 给 Rasa 的 REST 通道;二是客户端直接装轻量识别库,本地出文本,只把文本发给后端。医院导诊机里跑的是前者,小程序里跑的是后者,两者代码结构几乎一样。
| 方案 | 延迟 | 离线 | 适用位置 |
|---|---|---|---|
| Vosk 本地模型 | 约 1-2 秒 | 完全离线 | 服务端集中识别 |
| Sherpa-ONNX 端侧 | 约 0.5-1 秒 | 完全离线 | 小程序/H5 端 |
Vosk 的中文小模型约 40MB,CPU 上识别一句 5 秒语音在 1 秒左右,够用。Sherpa-ONNX 的优势是端侧推理,手机端不用上传音频,但模型裁剪和前端工作量更大。先按服务端 Vosk 打通,后面再考虑端侧。
4.2 Vosk 识别加 REST 通道:音频转文本再交给 Rasa
服务端接法分三步:Vosk 初始化一次,接收音频流分块喂给识别器,识别结果整句 POST 到 Rasa。代码用 asyncio 包起来,避免音频上传阻塞 Rasa 的消息循环。环境里需要pip install vosk aiohttp。
import asyncio import json import aiohttp from vosk import Model, KaldiRecognizer, SetLogLevel SetLogLevel(-1) model = Model("models/vosk-model-small-cn-0.22") RASA_URL = "http://127.0.0.1:5005/webhooks/rest/webhook" async def process_voice(sender: str, audio_path: str) -> str: rec = KaldiRecognizer(model, 16000) rec.SetWords(True) with open(audio_path, "rb") as f: while True: data = f.read(4000) if not data: break rec.AcceptWaveform(data) result = json.loads(rec.FinalResult()) text = result.get("text", "") if not text: return "" async with aiohttp.ClientSession() as session: payload = {"sender": sender, "message": text} async with session.post(RASA_URL, json=payload) as resp: replies = await resp.json() return repliesVosk 的 AcceptWaveform 必须按固定采样率喂数据,音频文件如果不是 16000Hz 单声道,要先在客户端或服务端用 ffmpeg 转码,否则识别率会断崖式下降。REST 通道的返回是 JSON 数组,每个元素对应 Rasa 端 dispatcher 报出去的一条消息,前端拿到后逐条播放或渲染。sender 字段建议用会话 ID,不能用用户真实姓名,日志脱敏从这一层就要做。
4.3 TTS 回读与槽位确认的联动
语音对话和纯文本的差别在于,文本回复可以一次给三条,语音一次只能读一条,而且读药品剂量信息时用户大概率记不住。常见做法是:TTS 只读结论句和确认句,详细说明以卡片形式同步推给前端。
def tts_reply(text: str, out_path: str) -> None: import pyttsx3 engine = pyttsx3.init() engine.setProperty("rate", 160) engine.save_to_file(text, out_path) engine.runAndWait()pyttsx3 是纯本地 TTS,医院内网里最省事,音色机械但清楚。对音质有要求就用 edge-tts 的离线缓存方案,把常用话术预生成音频文件,动态内容才实时合成。槽位确认尤其适合预生成:诊断 Action 追问"具体是哪里不舒服,持续多久"这句高频话术,完全可以提前合成,减少每次请求的合成耗时。
提示:语音交互里要区分"识别失败"和"没匹配到意图"。Vosk 返回空文本时直接回"没有听清,请再说一遍",不要再进 Rasa,否则空字符串会被 NLU 当成未知意图,触发兜底话术,体验会很怪。
5. 医疗机器人的兜底、日志回流与答案溯源
5.1 三档置信度阈值:确定回答、澄清、转人工
config.yml 里给 DIETClassifier 加一个 FallbackClassifier,设 confidence_threshold。线上建议按三档处理:置信度高于 0.85 直接回答;0.5 到 0.85 触发单轮澄清,把识别到的意图和实体回显给用户确认;低于 0.5 转人工或给出挂号引导,不要让模型硬答。医疗场景宁可多问一句,也不要答错一句。
pipeline: - name: WhitespaceTokenizer - name: CountVectorsFeaturizer - name: DIETClassifier epochs: 100 - name: FallbackClassifier threshold: 0.85 ambiguity_threshold: 0.1澄清话术要放在 rules.yml 里固定住,写成"您是想查药品、查症状,还是做疾病咨询?",不要用模型生成,免得澄清本身也跑偏。
5.2 把"没答上来的话"回流成训练数据
医疗机器人上线三个月后会积累大量兜底回答,这些兜底话术对应的用户原话,是最值钱的训练数据。做法是给兜底 Action 单独开一张日志表,每天跑一次离线脚本,把日志里置信度低于阈值但意图明确的句子抽出来,人工标注后追加进 nlu.yml。一个可用的抽取脚本大概长这样。
sqlite3 data/logs.db "SELECT sender, user_text, intent, confidence FROM nlu_feedback WHERE confidence BETWEEN 0.3 AND 0.85 ORDER BY confidence ASC LIMIT 200;"拿到这批句子后按意图归组,每组挑 20 到 50 条有代表性的补进 nlu.yml,再rasa train一次。这个回流循环跑上两轮,兜底率通常会明显下降,而且新增的例句都是真实用户说法,比人工编的训练数据泛化好得多。
5.3 每条回答都必须带溯源字段
医疗回答和闲聊最大的区别是责任边界。action_query_medicine 的回复里带 source 字段,前端渲染成"来源:院内药品目录 2025-01 版";疾病诊断回复统一追加一句"以上为症状分析,不能替代线下诊断,请及时就医"。这句话不放在 NLU 训练数据里,而是放在每个诊断类 Action 的回复模板末尾,保证任何路径触达诊断结论都带同样的免责声明,不会因为话术训练不完整而漏掉。
最后补一个验证细节:用 pytest 对 Action 写断言,检查返回文本里是否包含 source 和免责句,缺一个就算回归失败。把安全边界变成可执行的测试,比任何代码审查都可靠。返回兜底的阈值、同义词映射、日志回流脚本,这三样东西在项目里是绑在一起迭代的,缺一个,另外两个的效果都会打折扣。
本文还有配套的精品资源,点击获取