Python案例如何用Networkx做贝叶斯网络

本文目录导读：

1. 基础贝叶斯网络构建
2. 高级应用示例
3. 安装必要的库

我来为您详细介绍如何使用Networkx构建贝叶斯网络，虽然Networkx本身不直接支持贝叶斯网络的推理，但可以用它来表示网络结构,并结合其他库进行推理。

基础贝叶斯网络构建

import networkx as nx
import matplotlib.pyplot as plt
import numpy as np
from pgmpy.models import BayesianNetwork
from pgmpy.factors.discrete import TabularCPD
from pgmpy.inference import VariableElimination
# 方法1：使用pgmpy（推荐，支持完整贝叶斯网络功能）
def bayesian_network_example():
    """创建简单贝叶斯网络示例"""
    # 创建贝叶斯网络结构
    model = BayesianNetwork([
        ('Rain', 'WetGrass'),      # 下雨影响草地湿润
        ('Sprinkler', 'WetGrass'), # 洒水器影响草地湿润
        ('Cloudy', 'Rain'),         # 多云影响下雨
        ('Cloudy', 'Sprinkler')     # 多云影响洒水器使用
    ])
    # 定义条件概率表 (CPDs)
    # Cloudy的条件概率
    cpd_cloudy = TabularCPD(
        variable='Cloudy', 
        variable_card=2,
        values=[[0.5], [0.5]],
        state_names={'Cloudy': ['False', 'True']}
    )
    # Sprinkler的条件概率
    cpd_sprinkler = TabularCPD(
        variable='Sprinkler', 
        variable_card=2,
        values=[[0.5, 0.9], [0.5, 0.1]],
        evidence=['Cloudy'],
        evidence_card=[2],
        state_names={'Sprinkler': ['False', 'True'], 'Cloudy': ['False', 'True']}
    )
    # Rain的条件概率
    cpd_rain = TabularCPD(
        variable='Rain', 
        variable_card=2,
        values=[[0.8, 0.2], [0.2, 0.8]],
        evidence=['Cloudy'],
        evidence_card=[2],
        state_names={'Rain': ['False', 'True'], 'Cloudy': ['False', 'True']}
    )
    # WetGrass的条件概率
    cpd_wetgrass = TabularCPD(
        variable='WetGrass', 
        variable_card=2,
        values=[[1.0, 0.1, 0.1, 0.01], [0.0, 0.9, 0.9, 0.99]],
        evidence=['Sprinkler', 'Rain'],
        evidence_card=[2, 2],
        state_names={
            'WetGrass': ['False', 'True'], 
            'Sprinkler': ['False', 'True'], 
            'Rain': ['False', 'True']
        }
    )
    # 将CPDs添加到模型
    model.add_cpds(cpd_cloudy, cpd_sprinkler, cpd_rain, cpd_wetgrass)
    # 检查模型完整性
    print("模型验证:", model.check_model())
    return model
# 方法2：仅使用Networkx可视化结构
def visualize_bayesian_network():
    """可视化贝叶斯网络结构"""
    # 创建有向图
    G = nx.DiGraph()
    # 添加节点
    nodes = ['Cloudy', 'Rain', 'Sprinkler', 'WetGrass']
    G.add_nodes_from(nodes)
    # 添加边
    edges = [
        ('Cloudy', 'Rain'),
        ('Cloudy', 'Sprinkler'),
        ('Rain', 'WetGrass'),
        ('Sprinkler', 'WetGrass')
    ]
    G.add_edges_from(edges)
    # 设置布局
    pos = {
        'Cloudy': (0, 2),
        'Rain': (-1, 1),
        'Sprinkler': (1, 1),
        'WetGrass': (0, 0)
    }
    # 绘制网络
    plt.figure(figsize=(10, 8))
    # 绘制节点
    nx.draw_networkx_nodes(G, pos, node_color='lightblue', 
                          node_size=2000, alpha=0.8)
    # 绘制边
    nx.draw_networkx_edges(G, pos, edge_color='gray', width=2,
                          arrows=True, arrowsize=20)
    # 绘制标签
    nx.draw_networkx_labels(G, pos, font_size=12, font_weight='bold')
    plt.title('贝叶斯网络结构', fontsize=14)
    plt.axis('off')
    plt.show()
    return G
# 方法3：使用Networkx创建自定义贝叶斯网络类
class CustomBayesianNetwork:
    """自定义贝叶斯网络类"""
    def __init__(self):
        self.graph = nx.DiGraph()
        self.cpds = {}  # 条件概率表存储
    def add_node(self, node, states=None):
        """添加节点"""
        if states is None:
            states = ['True', 'False']
        self.graph.add_node(node, states=states)
    def add_edge(self, parent, child, cpd=None):
        """添加边"""
        self.graph.add_edge(parent, child)
        if cpd:
            self.cpds[(parent, child)] = cpd
    def get_parents(self, node):
        """获取节点的父节点"""
        return list(self.graph.predecessors(node))
    def get_children(self, node):
        """获取节点的子节点"""
        return list(self.graph.successors(node))
    def get_topological_order(self):
        """获取拓扑排序"""
        try:
            return list(nx.topological_sort(self.graph))
        except nx.NetworkXUnfeasible:
            print("图中存在环")
            return None
    def display_structure(self):
        """显示网络结构"""
        print("网络结构:")
        print("节点:", list(self.graph.nodes()))
        print("边:", list(self.graph.edges()))
        print("\n依赖关系:")
        for node in self.graph.nodes():
            parents = self.get_parents(node)
            if parents:
                print(f"  {node} 依赖于 {parents}")
            else:
                print(f"  {node} 是根节点")
    def visualize(self):
        """可视化网络"""
        pos = nx.spring_layout(self.graph)
        plt.figure(figsize=(10, 8))
        # 绘制节点
        nx.draw_networkx_nodes(self.graph, pos, node_color='lightgreen',
                              node_size=1500, alpha=0.9)
        # 绘制边
        nx.draw_networkx_edges(self.graph, pos, edge_color='gray',
                              width=2, arrows=True, arrowsize=20)
        # 添加节点标签
        labels = {node: node for node in self.graph.nodes()}
        nx.draw_networkx_labels(self.graph, pos, labels, font_size=10)
        plt.title('自定义贝叶斯网络')
        plt.axis('off')
        plt.show()
# 使用示例
if __name__ == "__main__":
    print("=== 方法1: 使用pgmpy ===")
    model = bayesian_network_example()
    # 进行推理
    from pgmpy.inference import VariableElimination
    inference = VariableElimination(model)
    # 查询：给定草地湿润，计算下雨的概率
    query_result = inference.query(
        variables=['Rain'],
        evidence={'WetGrass': 'True'}
    )
    print("\n给定草地湿润时下雨的概率:")
    print(query_result)
    print("\n=== 方法2: 使用Networkx可视化 ===")
    G = visualize_bayesian_network()
    print("\n=== 方法3: 自定义贝叶斯网络类 ===")
    custom_bn = CustomBayesianNetwork()
    # 添加节点
    custom_bn.add_node('Cloudy', ['True', 'False'])
    custom_bn.add_node('Rain', ['True', 'False'])
    custom_bn.add_node('Sprinkler', ['True', 'False'])
    custom_bn.add_node('WetGrass', ['True', 'False'])
    # 添加边
    custom_bn.add_edge('Cloudy', 'Rain')
    custom_bn.add_edge('Cloudy', 'Sprinkler')
    custom_bn.add_edge('Rain', 'WetGrass')
    custom_bn.add_edge('Sprinkler', 'WetGrass')
    # 显示网络信息
    custom_bn.display_structure()
    print(f"\n拓扑排序: {custom_bn.get_topological_order()}")
    # 可视化
    custom_bn.visualize()

高级应用示例

import networkx as nx
import matplotlib.pyplot as plt
from pgmpy.models import BayesianNetwork
from pgmpy.factors.discrete import TabularCPD
from pgmpy.inference import VariableElimination
import pandas as pd
# 医疗诊断贝叶斯网络
def medical_diagnosis_example():
    """医疗诊断贝叶斯网络示例"""
    # 定义网络结构
    model = BayesianNetwork([
        ('Smoking', 'LungCancer'),
        ('Smoking', 'Bronchitis'),
        ('LungCancer', 'ChestXray'),
        ('LungCancer', 'Dyspnea'),
        ('Bronchitis', 'Dyspnea')
    ])
    # 定义CPDs
    cpd_smoking = TabularCPD(
        variable='Smoking', 
        variable_card=2,
        values=[[0.7], [0.3]],  # 非吸烟者 vs 吸烟者
        state_names={'Smoking': ['No', 'Yes']}
    )
    cpd_lungcancer = TabularCPD(
        variable='LungCancer', 
        variable_card=2,
        values=[[0.95, 0.9], [0.05, 0.1]],
        evidence=['Smoking'],
        evidence_card=[2],
        state_names={'LungCancer': ['No', 'Yes'], 'Smoking': ['No', 'Yes']}
    )
    cpd_bronchitis = TabularCPD(
        variable='Bronchitis', 
        variable_card=2,
        values=[[0.9, 0.7], [0.1, 0.3]],
        evidence=['Smoking'],
        evidence_card=[2],
        state_names={'Bronchitis': ['No', 'Yes'], 'Smoking': ['No', 'Yes']}
    )
    cpd_chestxray = TabularCPD(
        variable='ChestXray', 
        variable_card=2,
        values=[[0.99, 0.05], [0.01, 0.95]],
        evidence=['LungCancer'],
        evidence_card=[2],
        state_names={'ChestXray': ['Normal', 'Abnormal'], 
                    'LungCancer': ['No', 'Yes']}
    )
    cpd_dyspnea = TabularCPD(
        variable='Dyspnea', 
        variable_card=2,
        values=[[0.9, 0.3, 0.2, 0.1], [0.1, 0.7, 0.8, 0.9]],
        evidence=['LungCancer', 'Bronchitis'],
        evidence_card=[2, 2],
        state_names={
            'Dyspnea': ['No', 'Yes'], 
            'LungCancer': ['No', 'Yes'], 
            'Bronchitis': ['No', 'Yes']
        }
    )
    # 添加CPDs到模型
    model.add_cpds(cpd_smoking, cpd_lungcancer, cpd_bronchitis, 
                   cpd_chestxray, cpd_dyspnea)
    return model
# 可视化医疗诊断网络
def visualize_medical_network():
    """可视化医疗诊断网络"""
    # 创建Networkx图
    G = nx.DiGraph()
    # 添加节点和边
    nodes = ['吸烟', '肺癌', '支气管炎', '胸部X光', '呼吸困难']
    edges = [
        ('吸烟', '肺癌'),
        ('吸烟', '支气管炎'),
        ('肺癌', '胸部X光'),
        ('肺癌', '呼吸困难'),
        ('支气管炎', '呼吸困难')
    ]
    G.add_nodes_from(nodes)
    G.add_edges_from(edges)
    # 绘制网络
    plt.figure(figsize=(12, 8))
    # 设置节点颜色
    node_colors = ['lightcoral' if n == '吸烟' else 
                   'lightblue' for n in nodes]
    pos = nx.spring_layout(G, seed=42)
    # 绘制网络
    nx.draw(G, pos, with_labels=True, node_color=node_colors,
            node_size=3000, font_size=12, font_weight='bold',
            edge_color='gray', width=2, arrowsize=20)
    plt.title('医疗诊断贝叶斯网络', fontsize=14)
    plt.show()
    return G
# 运行推理
def run_inference_examples():
    """运行推理示例"""
    model = medical_diagnosis_example()
    inference = VariableElimination(model)
    print("\n=== 医疗诊断推理示例 ===")
    # 示例1: 给定胸部X光异常，计算肺癌概率
    result1 = inference.query(
        variables=['LungCancer'],
        evidence={'ChestXray': 'Abnormal', 'Smoking': 'Yes'}
    )
    print(f"\n吸烟者胸部X光异常时患肺癌的概率:")
    print(result1)
    # 示例2: 给定呼吸困难，计算原因
    result2 = inference.query(
        variables=['LungCancer', 'Bronchitis'],
        evidence={'Dyspnea': 'Yes'}
    )
    print(f"\n出现呼吸困难时各种病因的概率:")
    print(result2)
    # 示例3: 反向推理
    result3 = inference.query(
        variables=['Smoking'],
        evidence={'ChestXray': 'Abnormal', 'Dyspnea': 'Yes'}
    )
    print(f"\n胸部X光异常且呼吸困难时吸烟的概率:")
    print(result3)
# 使用示例
if __name__ == "__main__":
    print("=== 医疗诊断贝叶斯网络 ===")
    # 可视化网络结构
    G = visualize_medical_network()
    # 运行推理
    run_inference_examples()
    # 网络分析
    print("\n=== 网络分析 ===")
    print(f"节点数: {G.number_of_nodes()}")
    print(f"边数: {G.number_of_edges()}")
    # 查找有向路径
    for source in G.nodes():
        for target in G.nodes():
            if source != target:
                if nx.has_path(G, source, target):
                    paths = list(nx.all_simple_paths(G, source, target))
                    print(f"从 {source} 到 {target} 的路径: {paths}")

安装必要的库

# 安装必要的库
pip install networkx
pip install pgmpy
pip install matplotlib
pip install numpy

这个示例展示了：

1. 如何使用pgmpy创建完整的贝叶斯网络
2. 如何使用Networkx进行网络结构可视化和分析
3. 如何进行概率推理
4. 如何构建自定义贝叶斯网络类
5. 实际医疗诊断应用

通过这些示例,您可以学习如何在Python中使用贝叶斯网络进行概率推理和决策支持。