大模型应用：慢病智能筛查与风险预警：XGBoost+规则引擎+混元大模型全解析.106

{
  "rule_id": "hypertension_high_risk",
  "rule_name": "高血压高风险判定规则",
  "condition": "core_abnormal_count >= 2 OR (core_abnormal_count == 1 AND family_history == '是')",
  "action": "risk_level = '高风险'",
  "description": "高风险判定需满足≥2条核心指标异常，或1条核心指标异常+有家族史"
}

{
  "age": 65,
  "gender": "男",
  "systolic_bp": 160,
  "diastolic_bp": 100,
  "family_history": "是",
  "symptom": "头晕"
}

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import LabelEncoder
import xgboost as xgb

# ---------------------- 1. 数据加载与预处理 ----------------------
# 模拟高血压患者数据（实际使用时替换为真实数据）
data = pd.DataFrame({
    'age': [55, 68, 45, 70, 50, 62, 48, 75, 58, 65],
    'gender': ['男', '女', '男', '女', '男', '女', '男', '女', '男', '女'],
    'systolic_bp': [150, 160, 130, 170, 140, 155, 125, 165, 145, 160],  # 收缩压
    'diastolic_bp': [95, 100, 85, 105, 90, 98, 80, 102, 92, 100],  # 舒张压
    'bmi': [28, 29, 24, 30, 26, 28, 25, 31, 27, 29],
    'family_history': ['是', '否', '否', '是', '否', '是', '否', '是', '否', '是'],
    'risk_level': ['中风险', '高风险', '低风险', '高风险', '中风险', '高风险', '低风险', '高风险', '中风险', '高风险']
})

# 特征工程：处理类别特征
# 性别编码：男=1，女=0
le_gender = LabelEncoder()
data['gender_encoded'] = le_gender.fit_transform(data['gender'])
# 家族史编码：是=1，否=0
le_family = LabelEncoder()
data['family_history_encoded'] = le_family.fit_transform(data['family_history'])
# 衍生特征：核心指标异常数
def count_core_abnormal(row):
    count = 0
    # 收缩压≥140mmHg 算异常
    if row['systolic_bp'] >= 140:
        count += 1
    # 舒张压≥90mmHg 算异常
    if row['diastolic_bp'] >= 90:
        count += 1
    # BMI≥28 算异常
    if row['bmi'] >= 28:
        count += 1
    return count

data['core_abnormal_count'] = data.apply(count_core_abnormal, axis=1)

# 风险等级编码：低风险=0，中风险=1，高风险=2
le_risk = LabelEncoder()
data['risk_level_encoded'] = le_risk.fit_transform(data['risk_level'])

# ---------------------- 2. 数据划分 ----------------------
# 特征列：选择核心特征
features = ['age', 'gender_encoded', 'systolic_bp', 'diastolic_bp', 'bmi', 
            'family_history_encoded', 'core_abnormal_count']
X = data[features]
y = data['risk_level_encoded']

# 划分训练集（70%）、验证集（20%）、测试集（10%）
X_train, X_temp, y_train, y_temp = train_test_split(X, y, test_size=0.3, random_state=42)
X_val, X_test, y_val, y_test = train_test_split(X_temp, y_temp, test_size=1/3, random_state=42)

# ---------------------- 3. XGBoost模型训练 ----------------------
# 转换为DMatrix格式（XGBoost专用格式，提升效率）
dtrain = xgb.DMatrix(X_train, label=y_train)
dval = xgb.DMatrix(X_val, label=y_val)
dtest = xgb.DMatrix(X_test, label=y_test)

# 设置模型参数
params = {
    'max_depth': 4,  # 树的深度，医疗数据建议3-5
    'learning_rate': 0.1,  # 学习率
    'n_estimators': 100,  # 树的数量
    'objective': 'multi:softmax',  # 多分类（低/中/高风险）
    'num_class': 3,  # 类别数：3类（0/1/2）
    'eval_metric': 'mlogloss',  # 评估指标：多分类对数损失
    'random_state': 42  # 随机种子，保证结果可复现
}

# 训练模型
watchlist = [(dtrain, 'train'), (dval, 'val')]  # 监控训练集和验证集效果
model = xgb.train(params, dtrain, num_boost_round=100, evals=watchlist, early_stopping_rounds=10)

# ---------------------- 4. 模型评估 ----------------------
# 测试集预测
y_pred = model.predict(dtest)
# 计算准确率
accuracy = np.mean(y_pred == y_test)
print(f"测试集准确率：{accuracy:.2f}")

# 特征重要性（查看哪些特征对风险判定最关键）
importance = model.get_score(importance_type='weight')
print("\n特征重要性：")
for feat, score in sorted(importance.items(), key=lambda x: x[1], reverse=True):
    print(f"{feat}: {score}")

# ---------------------- 5. 模型保存 ----------------------
model.save_model('hypertension_risk_model.model')
print("\n模型已保存为：hypertension_risk_model.model")

# ---------------------- 1. 定义事实（患者数据） ----------------------
class PatientFact:
    """患者事实类：存储患者的核心指标和模型预测结果"""
    def __init__(self, age, core_abnormal_count, family_history, model_risk_level):
        self.age = age  # 年龄
        self.core_abnormal_count = core_abnormal_count  # 核心指标异常数
        self.family_history = family_history  # 家族史：True=有，False=无
        self.model_risk_level = model_risk_level  # 模型预测的风险等级：低/中/高


# ---------------------- 2. 规则引擎定义 ----------------------
class HypertensionRuleEngine:
    """高血压风险规则引擎：基于国家规范校验模型结果"""
    
    @staticmethod
    def evaluate(patient):
        """评估患者风险等级"""
        age = patient.age
        core_abnormal_count = patient.core_abnormal_count
        family_history = patient.family_history
        model_risk = patient.model_risk_level
        
        # 规则1：低风险判定规则
        if (core_abnormal_count == 0 and 
            not family_history and 
            age < 65):
            final_risk = "低风险"
            reason = "无核心指标异常，无家族史，年龄<65岁"
        
        # 规则2：中风险判定规则
        elif (core_abnormal_count == 1 or 
              (age >= 65 and core_abnormal_count == 0)):
            final_risk = "中风险"
            reason = "1条核心指标异常，或年龄≥65岁且无核心指标异常"
        
        # 规则3：高风险判定规则
        elif (core_abnormal_count >= 2 or 
              (core_abnormal_count == 1 and family_history)):
            final_risk = "高风险"
            reason = "≥2条核心指标异常，或1条核心指标异常+有家族史"
        
        # 规则4：模型结果修正规则（模型结果与规则不符时）
        is_corrected = False
        corrected_reason = None
        if model_risk != final_risk:
            is_corrected = True
            corrected_reason = f"模型预测风险等级为{model_risk}，但不符合国家规范，修正为{final_risk}"
        
        return {
            'final_risk_level': final_risk,
            'reason': reason,
            'is_corrected': is_corrected,
            'corrected_reason': corrected_reason
        }


# ---------------------- 3. 规则引擎执行 ----------------------
def run_rule_engine(patient_data):
    """
    执行规则引擎校验
    :param patient_data: 患者数据字典，包含age, core_abnormal_count, family_history, model_risk_level
    :return: 最终风险等级、判定依据、是否修正
    """
    # 创建患者事实对象
    patient = PatientFact(
        age=patient_data['age'],
        core_abnormal_count=patient_data['core_abnormal_count'],
        family_history=patient_data['family_history'],
        model_risk_level=patient_data['model_risk_level']
    )
    
    # 执行规则评估
    return HypertensionRuleEngine.evaluate(patient)


# ---------------------- 4. 测试规则引擎 ----------------------
if __name__ == "__main__":
    # 测试案例1：模型预测高风险，规则引擎验证合规
    patient1 = {
        'age': 68,
        'core_abnormal_count': 2,
        'family_history': False,
        'model_risk_level': '高风险'
    }
    result1 = run_rule_engine(patient1)
    print("测试案例1结果：")
    print(f"最终风险等级：{result1['final_risk_level']}")
    print(f"判定依据：{result1['reason']}")
    print(f"是否修正：{result1['is_corrected']}")
    
    # 测试案例2：模型预测高风险，规则引擎修正为中风险
    patient2 = {
        'age': 55,
        'core_abnormal_count': 1,
        'family_history': False,
        'model_risk_level': '高风险'
    }
    result2 = run_rule_engine(patient2)
    print("\n测试案例2结果：")
    print(f"最终风险等级：{result2['final_risk_level']}")
    print(f"判定依据：{result2['reason']}")
    print(f"是否修正：{result2['is_corrected']}")
    print(f"修正原因：{result2['corrected_reason']}")

from openai import OpenAI
import json

# ---------------------- 1. 配置大模型API ----------------------
# 腾讯混元API配置
api_key = 'sk-bWlJ*************8Ze'

client = OpenAI(
    api_key=api_key,
    base_url="https://api.hunyuan.cloud.tencent.com/v1",
)

# ---------------------- 2. 大模型解读函数 ----------------------
def generate_risk_interpretation(risk_result, patient_info):
    """
    生成风险结果的自然语言解读
    :param risk_result: 规则引擎输出的风险结果字典
    :param patient_info: 患者基本信息字典
    :return: 自然语言解读文本
    """
    # 构建提示词（Prompt）：结合医疗规则和患者信息
    prompt = f"""
    你是一名专业的高血压慢病管理医生，需要根据以下信息为患者生成易懂的风险解读：
    1. 患者基本信息：
       - 年龄：{patient_info['age']}岁
       - 性别：{patient_info['gender']}
       - 收缩压：{patient_info['systolic_bp']}mmHg
       - 舒张压：{patient_info['diastolic_bp']}mmHg
       - BMI：{patient_info['bmi']}
       - 家族史：{'有' if patient_info['family_history'] else '无'}
    2. 风险判定结果：
       - 最终风险等级：{risk_result['final_risk_level']}
       - 判定依据：{risk_result['reason']}
       - 是否修正：{'是' if risk_result['is_corrected'] else '否'}
       - 修正原因：{risk_result['corrected_reason'] if risk_result['is_corrected'] else '无'}
    
    要求：
    1. 语言通俗易懂，避免专业术语；
    2. 先告知风险等级，再解释判定依据，最后给出针对性的干预建议；
    3. 若有修正，需简要说明修正原因；
    4. 建议具体可行（比如饮食、运动、监测频率）；
    5. 语气亲切，符合医疗沟通规范。
    """
    
    # 调用腾讯混元API生成解读
    response = client.chat.completions.create(
        model="hunyuan-lite",  # 混元轻量级模型
        messages=[
            {"role": "system", "content": "你是专业的慢病管理医生，擅长用通俗易懂的语言解释风险等级和干预建议。"},
            {"role": "user", "content": prompt}
        ],
        temperature=0.3,  # 温度越低，结果越稳定
        max_tokens=500  # 最大生成字数
    )
    
    # 提取生成的文本
    interpretation = response.choices[0].message.content.strip()
    return interpretation

# ---------------------- 3. 测试大模型解读 ----------------------
if __name__ == "__main__":
    # 患者信息
    patient_info = {
        'age': 68,
        'gender': '女',
        'systolic_bp': 160,
        'diastolic_bp': 100,
        'bmi': 29,
        'family_history': False
    }
    
    # 规则引擎输出结果
    risk_result = {
        'final_risk_level': '高风险',
        'reason': '≥2条核心指标异常，或1条核心指标异常+有家族史',
        'is_corrected': False,
        'corrected_reason': None
    }
    
    # 生成解读
    interpretation = generate_risk_interpretation(risk_result, patient_info)
    print("患者风险解读：")
    print(interpretation)