产生式系统的完整示例——医疗诊断系统（附python代码）
产生式系统在现实社会中依然有很重要的应用价值，为了更好的理解它的构建、推理以及完整的实现过程，本文一医疗诊断系统为例，进行详细说明。以下是一个包含10条规则的医疗诊断产生式系统示例，包含完整的规则定义、推理流程图、Python代码实现及推理过程可视化。
有关知识表示方法之二：产生式表示法（Production System）的定义、原理等内容可以先看我的CSDN文章：知识表示方法之二：产生式表示法（Production System）-CSDN博客
一、医疗诊断系统规则定义
1.疾病与症状假设
传统的产生式系统中，前件和后件都是有专家定义的。
（1）目标疾病：流感（Flu）、麻疹（Measles）、肺炎（Pneumonia）、链球菌性喉炎（StrepThroat）、普通感冒（Cold）。
（2）症状与检测指标：发烧（Fever）、咳嗽（Cough）、皮疹（Rash）、喉咙痛（SoreThroat）、头痛（Headache）、流涕（RunnyNose）、呼吸困难（ShortnessOfBreath）、白细胞计数高（HighWBC）、接触流感患者（ContactFlu）、接触麻疹患者（ContactMeasles）。
产生式系统的目的是根据症状与检测指标（前件）推理目标疾病（后件）。
2.规则库
将规则分为初步怀疑规则和确诊规则，并说明每条规则的医学依据：
| 规则类型 | 规则编号 | 条件（前件） | 结论（后件） | 优先级 | 医学解释 |
| 初步怀疑 | R1 | Fever ∧ Cough ∧ Headache | SuspectFlu | 1 | 流感典型症状为发热、咳嗽、头痛。 |
| R2 | Fever ∧ Rash ∧ ContactMeasles | SuspectMeasles | 1 | 麻疹需发热、皮疹，且有接触史。 | |
| R3 | Fever ∧ SoreThroat ∧ HighWBC | SuspectStrepThroat | 2 | 链球菌性喉炎伴随高白细胞计数。 | |
| R4 | Cough ∧ ShortnessOfBreath | SuspectPneumonia | 3 | 肺炎表现为咳嗽且呼吸困难。 | |
| R5 | RunnyNose ∧ SoreThroat | SuspectCold | 1 | 普通感冒以流涕和喉咙痛为主。 | |
| 确诊 | R6 | SuspectFlu ∧ HighWBC | ConfirmFlu | 2 | 流感确诊需实验室检测白细胞升高。 |
| R7 | SuspectMeasles ∧ Rash | ConfirmMeasles | 2 | 麻疹确诊需皮疹持续扩散。 | |
| R8 | SuspectStrepThroat ∧ SoreThroat | ConfirmStrepThroat | 2 | 链球菌性喉炎确诊需咽拭子检测阳性。 | |
| R9 | SuspectPneumonia ∧ ShortnessOfBreath | ConfirmPneumonia | 3 | 肺炎确诊需胸部X光或CT显示肺部阴影。 | |
| R10 | SuspectCold ∧ ¬Fever | ConfirmCold | 1 | 普通感冒通常不伴随高热。 |
规则冲突与优先级策略：
（1）优先级权重：确诊规则（R6-R10）优先级 > 初步怀疑规则（R1-R5）。
（2）特异性原则：条件更多的规则优先级更高（如R3条件数=3 > R1条件数=3，但因高白细胞权重更高）。
（3）动态调整：根据实时症状更新优先级（如呼吸困难为紧急症状，触发R9优先级提升）。
二、推理流程图
1.产生式系统求解问题的一般步骤
（1）初始化综合数据库，把问题的初始已知事实送入综合数据库（事实库）中。
（2）若规则库中存在尚未使用过的规则，而且它的前提（前件）可与综合数据库中的已知事实匹配，则转第（3）步；若不存在这样的事实，则转第（5）步。
（3）执行当前选中的规则，并对该规则做上标记，把该规则执行后得到的结论送入综合数据库中。如果该规则的结论部分指出的是某些操作，则执行这些操作。
（4）检查综合数据库中是否已包含了问题的解，若已包含，则终止问题的求解过程；否则转第（2）步。
（5）要求用户提供进一步的关于问题的已知事实，若能提供，则转第（2）步；否则终止问题的求解过程。
（6）若规则库中不再有未使用过的规则，则终止问题的求解过程。
在上述第（4）步中，为了检查综合数据库中是否包含问题的解，可采用如下两种简单的处理方法：
（1）把问题的全部最终结论，如医疗诊断系统中的流感、麻疹、肺炎、链球菌性喉炎、普通感冒等五种疾病的名称全部列于一张表中，每当执行一条规则得到一个结论时，就检查该结论是否包含在表中，若包含在表中，说明它就是最终结论，求得了问题的解。
（2）对每条结论部分是最终结论的产生式规则，如医疗诊断系统中的规则R7至R9，分别做一标记，当执行到上述一般步骤中的第（3）步时，首先检查该选中的规则是否带有这个标记，若带有，则由该规则推出的结论就是最终结论，即求得了问题的解。
最后，需要特别说明的是，问题的求解过程与推理的控制策略有关，上述的一般步骤只是针对正向推理而言的，而且它只是粗略地描述了产生式系统求解问题的大致步骤，许多细节均未考虑，如冲突消解、不确定性的处理等，这些问题都将在后面的内容中分别讨论。
2.医疗诊断系统的流程图
正向推理引擎层级结构如下：
│
├─ 初始化事实库（输入症状）
│
├─ 循环：
│ ├─ 遍历规则库：
│ │ ├─ 检查规则前件是否匹配事实库
│ │ └─ 若匹配且未被触发 → 触发规则，添加后件到事实库
│ │
│ └─ 若无新事实触发 → 退出循环
│
└─ 输出最终诊断结论
正向推理引擎流程图如下：
推理步骤详解：
（1）输入阶段：
用户输入症状集合（如 {"Fever", "Cough", "HighWBC"}）。
系统自动补全否定条件（如未输入 Rash 则隐含 ¬Rash）。
（2）规则匹配阶段：
遍历规则库，检查每条规则的前件是否被当前事实库包含。
否定条件处理：若规则含 ¬P，需确保 P 不在事实库中。
（3）冲突消解阶段：
按优先级排序触发规则，执行最高优先级规则的后件。
记录触发顺序（用于回溯诊断逻辑）。
（4）终止条件：
无新事实触发，或达到预设最大循环次数（防止死循环）。
三、Python代码实现
1.规则与事实库定义
class Rule:
def __init__(self, antecedent, consequent, priority=1):
self.antecedent = set(antecedent) # 前件（集合）
self.consequent = consequent # 后件（字符串）
self.priority = priority # 优先级
# 定义规则库（10条规则）
rules = [
Rule(["Fever", "Cough", "Headache"], "SuspectFlu", priority=1),
Rule(["Fever", "Rash", "ContactMeasles"], "SuspectMeasles", priority=1),
Rule(["Fever", "SoreThroat", "HighWBC"], "SuspectStrepThroat", priority=2),
Rule(["Cough", "ShortnessOfBreath"], "SuspectPneumonia", priority=3),
Rule(["RunnyNose", "SoreThroat"], "SuspectCold", priority=1),
Rule(["SuspectFlu", "HighWBC"], "ConfirmFlu", priority=2),
Rule(["SuspectMeasles", "Rash"], "ConfirmMeasles", priority=2),
Rule(["SuspectStrepThroat", "SoreThroat"], "ConfirmStrepThroat", priority=2),
Rule(["SuspectPneumonia", "ShortnessOfBreath"], "ConfirmPneumonia", priority=3),
Rule(["SuspectCold", "¬Fever"], "ConfirmCold", priority=1),
]
# 初始化事实库（患者症状）
facts = {"Fever", "Cough", "SoreThroat", "HighWBC"}2.正向推理算法
def forward_chaining(rules, facts):
triggered_rules = []
new_facts = set()
while True:
triggered = False
# 按优先级排序规则（高优先级优先）
sorted_rules = sorted(rules, key=lambda x: x.priority, reverse=True)
for rule in sorted_rules:
# 检查前件是否全部满足
antecedent_met = True
for condition in rule.antecedent:
# 处理否定条件（如 ¬Fever）
if condition.startswith("¬"):
if condition[1:] in facts:
antecedent_met = False
break
else:
if condition not in facts:
antecedent_met = False
break
if antecedent_met and rule.consequent not in facts:
facts.add(rule.consequent)
triggered_rules.append(rule.consequent)
new_facts.add(rule.consequent)
triggered = True
if not triggered:
break
return facts, triggered_rules, new_facts
# 执行推理
final_facts, triggered_rules, new_facts = forward_chaining(rules, facts)3.推理过程可视化
def visualize_inference(triggered_rules, final_facts):
print("=== 推理过程 ===")
for i, rule in enumerate(triggered_rules, 1):
print(f"步骤 {i}: 触发规则 → {rule}")
print("\n=== 最终诊断结论 ===")
confirmed = [fact for fact in final_facts if fact.startswith("Confirm")]
if confirmed:
print("确诊疾病:", ", ".join(confirmed))
else:
print("无法确诊，需进一步检查。")
# 输出结果
visualize_inference(triggered_rules, final_facts)4.示例运行结果
（1）输入症状：
facts = {"Fever", "Cough", "SoreThroat", "HighWBC"}根据输入症状，依靠产生式系统，推理出得的是什么疾病。
（2）输出：
=== 推理过程 ===
步骤 1: 触发规则 → SuspectStrepThroat
步骤 2: 触发规则 → ConfirmStrepThroat
=== 最终诊断结论 ===
确诊疾病: ConfirmStrepThroat代码解释与关键点
（1）规则优先级：优先级高的规则先触发（如肺炎规则优先级3 > 流感优先级1）。
（2）否定条件处理：支持 ¬Fever 表示“无发烧”。
（3）动态更新事实库：每次循环检查所有规则，直到无新事实触发。
（4）冲突消解策略：按优先级排序，确保高优先级规则优先触发。
5.扩展：生成推理流程图（Graphviz）
（1）安装Graphviz并将其添加到系统环境变量PATH中
下载地址：https://graphviz.org/download/
安装时勾选 Add Graphviz to the system PATH。
如果没有添加系统变量，或可以在代码中指定Graphviz路径：
import os
os.environ["PATH"] += os.pathsep + 'C:/Program Files/Graphviz/bin/' # 替换为你的安装路径以上是默认安装路径，如果自己安装在了其他地方，要更改成自己安装的路径。之后将代码放在文件前面位置即可。
（2）python中安装Graphviz插件
先安装Graphviz，再可以通过在CMD中使用pip安装python插件，命令是：
pip install graphviz -i http://pypi.tuna.tsinghua.edu.cn/simple之后在上文的内容之后，添加如下内容，并可生成流程图：
from graphviz import Digraph
def generate_flowchart(triggered_rules: list):
"""根据实际触发的规则动态生成流程图"""
dot = Digraph(comment='Medical Diagnosis Flow')
# 创建节点
dot.node('Start', '开始')
dot.node('Input', '输入症状')
dot.node('Process', '规则引擎')
dot.node('End', '输出结论')
# 主流程连接
dot.edge('Start', 'Input')
dot.edge('Input', 'Process')
dot.edge('Process', 'End')
# 动态添加规则触发节点
for i, rule_name in enumerate(triggered_rules):
node_id = f'R{i}'
dot.node(node_id, f'触发规则\n{rule_name}')
dot.edge('Process', node_id)
dot.edge(node_id, 'Process')
dot.render('diagnosis_flow.gv', view=True)
# 调用时传入实际触发的规则列表
generate_flowchart(triggered_rules)（3）生成效果：
6.完整代码（含注释与异常处理）
以下代码在上面代码的基础上，有所删减。
# 导入必要的库和模块
from typing import Set, List # 用于类型注解
import os # 用于操作系统路径配置
from graphviz import Digraph # 用于生成流程图
# 配置Graphviz的可执行文件路径（根据实际安装路径修改）
os.environ["PATH"] += os.pathsep + 'C:/Program Files/Graphviz/bin/'
# 定义规则类，表示产生式系统中的单条规则
class Rule:
def __init__(self, antecedent: Set[str], consequent: str, priority: int = 1):
"""
初始化规则
:param antecedent: 前件集合，表示触发规则需要满足的条件（支持否定条件，如"¬Fever"）
:param consequent: 后件，表示规则触发后得到的结论
:param priority: 规则优先级，数值越大越优先触发
"""
self.antecedent = antecedent # 规则前件（条件集合）
self.consequent = consequent # 规则后件（结论）
self.priority = priority # 规则优先级
self.triggered = False # 标记规则是否已被触发
def __repr__(self):
"""返回规则的字符串表示，便于调试"""
return f"Rule({self.antecedent} → {self.consequent})"
# 定义产生式系统核心类
class ProductionSystem:
def __init__(self, rules: List[Rule]):
"""
初始化产生式系统
:param rules: 规则列表，包含所有预先定义的规则
"""
self.rules = rules # 系统规则库
self.facts: Set[str] = set() # 综合数据库（当前已知事实）
self.triggered_rules: List[str] = [] # 记录已触发的规则顺序
def add_fact(self, fact: str):
"""
向事实库中添加新事实
:param fact: 要添加的事实（不能是否定条件）
"""
if fact.startswith("¬"):
raise ValueError("否定条件需通过规则推导，不可直接添加")
self.facts.add(fact)
def reset(self):
"""重置系统状态，清空事实库和触发记录"""
self.facts.clear()
self.triggered_rules.clear()
for rule in self.rules:
rule.triggered = False
def visualize_step(self, step: int):
"""
可视化当前推理步骤的状态
:param step: 当前步骤编号
"""
print(f"\n=== 步骤 {step} ===")
print("当前事实:", self.facts)
print("可触发规则:", [rule for rule in self.rules if not rule.triggered])
def forward_chaining(self, max_iterations: int = 100) -> Set[str]:
"""
正向推理引擎（数据驱动推理）
:param max_iterations: 最大迭代次数，防止无限循环
:return: 最终事实库
"""
iteration = 0
while iteration < max_iterations:
triggered = False # 标记本轮是否触发过规则
# 对未触发的规则进行排序（优先级降序 > 条件数量降序）
sorted_rules = sorted(
[r for r in self.rules if not r.triggered],
key=lambda x: (-x.priority, -len(x.antecedent)))
# 遍历所有可触发规则
for rule in sorted_rules:
antecedent_met = True # 标记前件是否满足
# 检查每个条件是否满足
for cond in rule.antecedent:
# 处理否定条件（例如"¬Fever"表示没有发烧）
if cond.startswith("¬"):
if cond[1:] in self.facts: # 如果事实库中存在相反条件
antecedent_met = False
break
else:
if cond not in self.facts: # 如果条件不在事实库中
antecedent_met = False
break
# 如果满足所有前件条件
if antecedent_met:
# 更新事实库和触发记录
self.facts.add(rule.consequent)
self.triggered_rules.append(rule.consequent)
rule.triggered = True
triggered = True
break # 每次只触发最高优先级的规则
# 显示当前步骤状态
self.visualize_step(iteration + 1)
# 如果没有触发新规则则终止推理
if not triggered:
break
iteration += 1
return self.facts
# ---------------------- 规则定义 ----------------------
# 定义医疗诊断规则库（10条规则）
rules = [
# 怀疑类规则（优先级1-3）
Rule({"Fever", "Cough"}, "SuspectFlu", 1), # R1: 发烧+咳嗽 → 怀疑流感
Rule({"Fever", "Rash", "ContactMeasles"}, "SuspectMeasles", 1), # R2: 发烧+皮疹+接触史 → 怀疑麻疹
Rule({"Fever", "SoreThroat", "HighWBC"}, "SuspectStrepThroat", 2), # R3: 发烧+喉咙痛+高白细胞 → 怀疑链球菌喉炎
Rule({"Cough", "ShortnessOfBreath"}, "SuspectPneumonia", 3), # R4: 咳嗽+呼吸困难 → 怀疑肺炎（高优先级）
Rule({"RunnyNose", "SoreThroat"}, "SuspectCold", 1), # R5: 流涕+喉咙痛 → 怀疑感冒
# 确诊类规则（优先级2-3）
Rule({"SuspectFlu", "HighWBC"}, "ConfirmFlu", 2), # R6: 怀疑流感+高白细胞 → 确诊流感
Rule({"SuspectMeasles", "Rash"}, "ConfirmMeasles", 2), # R7: 怀疑麻疹+皮疹 → 确诊麻疹
Rule({"SuspectStrepThroat", "SoreThroat"}, "ConfirmStrepThroat", 2), # R8: 怀疑链球菌喉炎+喉咙痛 → 确诊
Rule({"SuspectPneumonia", "ShortnessOfBreath"}, "ConfirmPneumonia", 3), # R9: 怀疑肺炎+呼吸困难 → 确诊（高优先级）
Rule({"SuspectCold", "¬Fever"}, "ConfirmCold", 1) # R10: 怀疑感冒+无发烧 → 确诊感冒
]
# ---------------------- 系统初始化 ----------------------
# 创建产生式系统实例
system = ProductionSystem(rules)
# 添加初始症状（用户输入）
system.add_fact("Fever") # 发烧
system.add_fact("Cough") # 咳嗽
system.add_fact("HighWBC") # 高白细胞计数
# ---------------------- 执行推理 ----------------------
diagnosis = system.forward_chaining()
# ---------------------- 结果输出 ----------------------
print("\n=== 最终诊断 ===")
# 过滤出所有确诊结论（以Confirm开头的事实）
confirmed = [d for d in diagnosis if d.startswith("Confirm")]
print("确诊疾病:", confirmed if confirmed else "无法确诊")
# ---------------------- 可视化流程图生成 ----------------------
def generate_flow_chart(triggered_rules: List[str]):
"""生成推理流程图"""
dot = Digraph(comment='医疗诊断流程图') # 创建有向图
# 定义主要节点
dot.node('Start', '开始', shape='ellipse')
dot.node('Input', '输入症状', shape='parallelogram')
dot.node('Process', '规则引擎', shape='rectangle')
dot.node('End', '输出结果', shape='ellipse')
# 连接主流程节点
dot.edge('Start', 'Input')
dot.edge('Input', 'Process')
dot.edge('Process', 'End')
# 添加触发的规则节点
for i, rule in enumerate(triggered_rules):
# 每个规则节点显示触发顺序和规则内容
dot.node(f'R{i}', f'规则 {i+1}\n{rule}', shape='diamond')
# 连接规则引擎与规则节点（双向箭头表示反复匹配）
dot.edge('Process', f'R{i}', label='匹配')
dot.edge(f'R{i}', 'Process', label='继续匹配')
# 生成图片文件并自动打开
dot.render('diagnosis_flow', format='png', view=True)
# 生成流程图（需安装Graphviz）
generate_flow_chart(system.triggered_rules)代码说明：
（1）Graphviz路径配置
通过os.environ添加Graphviz的安装路径，确保程序能找到dot可执行文件
（2）Rule类
antecedent: 使用集合存储前件条件，便于快速查找
triggered标记: 防止规则被重复触发
（3）冲突消解策略
规则排序逻辑：sorted(rules, key=lambda x: (-x.priority, -len(x.antecedent)))
优先级优先：优先级数值大的先执行（降序）
条件数量优先：条件更多的规则更具体，优先触发
（4）否定条件处理
通过检查cond.startswith("¬")处理否定逻辑，例如¬Fever表示事实库中没有Fever。
（5）可视化增强
visualize_step()方法展示每轮推理状态
流程图使用不同形状表示节点类型：
椭圆：开始/结束节点
平行四边形：输入输出节点
矩形：处理过程
菱形：规则判断节点
（6）诊断规则设计
将规则分为"怀疑"和"确诊"两个阶段
肺炎相关规则（R4、R9）优先级较高，符合医疗紧急情况处理原则
代码关键点解释：
（1）规则类（Rule）：封装前件、后件、优先级及触发状态。
（2）综合数据库（Production System）：通过集合存储事实，自动处理重复项。
（3）否定条件处理：通过检查条件字符串是否以 ¬ 开头，动态验证事实库。
（4）冲突消解策略：按优先级和条件数量双重排序，确保更具体的规则优先触发。
输出示例：
=== 步骤 1 ===
当前事实: {'SuspectFlu', 'HighWBC', 'Cough', 'Fever'}
可触发规则: [Rule({'ContactMeasles', 'Rash', 'Fever'} → SuspectMeasles), Rule({'HighWBC', 'SoreThroat', 'Fever'} → SuspectStrepThroat), Rule({'Cough', 'ShortnessOfBreath'} → SuspectPneumonia), Rule({'SoreThroat', 'RunnyNose'} → SuspectCold), Rule({'SuspectFlu', 'HighWBC'} → ConfirmFlu), Rule({'Rash', 'SuspectMeasles'} → ConfirmMeasles), Rule({'SoreThroat', 'SuspectStrepThroat'} → ConfirmStrepThroat), Rule({'SuspectPneumonia', 'ShortnessOfBreath'} → ConfirmPneumonia), Rule({'SuspectCold', '¬Fever'} → ConfirmCold)]
=== 步骤 2 ===
当前事实: {'HighWBC', 'Cough', 'ConfirmFlu', 'Fever', 'SuspectFlu'}
可触发规则: [Rule({'ContactMeasles', 'Rash', 'Fever'} → SuspectMeasles), Rule({'HighWBC', 'SoreThroat', 'Fever'} → SuspectStrepThroat), Rule({'Cough', 'ShortnessOfBreath'} → SuspectPneumonia), Rule({'SoreThroat', 'RunnyNose'} → SuspectCold), Rule({'Rash', 'SuspectMeasles'} → ConfirmMeasles), Rule({'SoreThroat', 'SuspectStrepThroat'} → ConfirmStrepThroat), Rule({'SuspectPneumonia', 'ShortnessOfBreath'} → ConfirmPneumonia), Rule({'SuspectCold', '¬Fever'} → ConfirmCold)]
=== 步骤 3 ===
当前事实: {'HighWBC', 'Cough', 'ConfirmFlu', 'Fever', 'SuspectFlu'}
可触发规则: [Rule({'ContactMeasles', 'Rash', 'Fever'} → SuspectMeasles), Rule({'HighWBC', 'SoreThroat', 'Fever'} → SuspectStrepThroat), Rule({'Cough', 'ShortnessOfBreath'} → SuspectPneumonia), Rule({'SoreThroat', 'RunnyNose'} → SuspectCold), Rule({'Rash', 'SuspectMeasles'} → ConfirmMeasles), Rule({'SoreThroat', 'SuspectStrepThroat'} → ConfirmStrepThroat), Rule({'SuspectPneumonia', 'ShortnessOfBreath'} → ConfirmPneumonia), Rule({'SuspectCold', '¬Fever'} → ConfirmCold)]生成流程图的效果图如下：
四、扩展性与实际应用分析
1.扩展功能建议
（1）不确定性推理：为规则和事实添加置信度（如 ConfirmFlu: 0.8）。
（2）动态规则加载：从JSON或数据库读取规则，支持在线更新。
（3）多疾病协同诊断：处理并发疾病（如同时患流感和感冒）。
2.局限性及改进方向
（1）缺点：无法处理模糊症状（如“轻微咳嗽”）。规则数量爆炸时效率下降。
（2）改进方案：结合模糊逻辑（Fuzzy Logic）量化症状强度。使用Rete算法优化规则匹配速度。
3.实际应用场景
（1）急诊分诊：快速识别危重病患（如优先触发肺炎规则）。
（2）电子健康档案（EHR）：自动生成诊断建议供医生参考。
（3）公共卫生监控：通过症状数据实时预测流行病趋势。
通过细化规则语义、优化冲突消解策略、增强代码健壮性及可视化交互，产生式系统在医疗诊断中的实用性显著提升。结合扩展功能后，可逐步过渡到真实临床辅助决策场景，但仍需与医生经验结合以避免误诊。