产生式系统的完整示例——医疗诊断系统（附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 = Truefor condition in rule.antecedent:# 处理否定条件（如 ¬Fever）if condition.startswith("¬"):if condition[1:] in facts:antecedent_met = Falsebreakelse:if condition not in facts:antecedent_met = Falsebreakif antecedent_met and rule.consequent not in facts:facts.add(rule.consequent)triggered_rules.append(rule.consequent)new_facts.add(rule.consequent)triggered = Trueif not triggered:breakreturn 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 osos.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 Digraphdef 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 = Falsedef 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 = 0while 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 = Falsebreakelse:if cond not in self.facts:  # 如果条件不在事实库中antecedent_met = Falsebreak# 如果满足所有前件条件if antecedent_met:# 更新事实库和触发记录self.facts.add(rule.consequent)self.triggered_rules.append(rule.consequent)rule.triggered = Truetriggered = Truebreak  # 每次只触发最高优先级的规则# 显示当前步骤状态self.visualize_step(iteration + 1)# 如果没有触发新规则则终止推理if not triggered:breakiteration += 1return 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)]

生成流程图的效果图如下：