最近在帮几个医学信息工程专业的学弟学妹看毕业设计，发现大家普遍头疼：选题要么太“虚”，做个管理系统；要么技术栈老旧，和现在医院里实际在用的技术脱节。刚好我之前用Python做过一些医疗数据相关的原型，今天就来聊聊，怎么用Python搭建一个既“像样”又能体现专业特色的临床决策支持系统（CDSS）原型。这个项目用到的技术都比较新，而且有真实场景支撑，希望能给正在为毕设发愁的同学一些思路。

1. 医学信息工程毕设的常见“坑”

在动手之前，我们先盘盘大家容易踩的坑，避开这些，项目成功率能高一大截。

• 数据孤岛与模拟数据：真实医疗数据拿不到，自己造的数据又太假，导致系统演示时缺乏说服力。解决方案是使用公开的、符合国际标准的数据模型来模拟，比如FHIR。
• 合规性成为摆设：很多项目只关注功能实现，完全忽略了医疗系统必须的隐私安全要求（如HIPAA、GDPR），这在答辩时会被重点质疑。
• 系统不可演示或“黑盒”：后端逻辑写成一团，没有清晰的API，导致无法提供一个前端界面进行实时交互演示。一个可查询、可触发规则的Web API是演示的利器。
• 技术栈陈旧：还在用传统的Servlet、JSP或者非常基础的Flask写法，没有体现现代Web开发中异步、标准化API设计等思想。

2. 技术选型：轻量、快速、演示友好

针对毕设时间紧、要求可演示的特点，我的选型原则是：轻量、学习曲线平缓、生态活跃。

• Web框架：FastAPI vs Django REST Framework

  • FastAPI：强烈推荐。它异步性能好，自动生成交互式API文档（Swagger UI），这对于毕设演示简直是神器。你不需要额外写接口文档，老师和同学可以直接在浏览器里测试你的API。而且用Python类型提示，开发体验很棒。
  • Django REST Framework：功能大而全，自带Admin后台、用户权限等，如果你需要快速构建一个包含用户管理、内容管理的“管理型”系统，它很合适。但对于专注于数据服务和业务逻辑的CDSS原型，略显笨重。

• 数据库：SQLite vs PostgreSQL

  • SQLite：毕设首选。它是一个文件数据库，无需安装和配置数据库服务，简化了部署。对于原型系统和小规模模拟数据，性能完全足够。数据文件可以直接打包在项目里，方便移植和演示。
  • PostgreSQL：功能更强大，支持更复杂的数据类型和查询。如果你的毕设涉及非常复杂的医疗数据关系分析，可以考虑。但需要额外安装，增加了部署复杂度。

• 规则引擎：本地Python逻辑 vs Drools

  • 本地Python逻辑：对于毕业设计级别的规则（例如：“若患者年龄>65且收缩压>180，触发高血压危象预警”），完全可以用Python的if-else或者简单的策略模式来实现。优点是简单直观，与主程序集成无缝，方便调试和演示。
  • Drools：专业的Java规则引擎，规则与业务代码分离，适合规则非常复杂且频繁变更的企业级场景。但对于Python技术栈的毕设，引入Drools需要搭建Java环境，增加系统复杂度，有点“杀鸡用牛刀”。

小结：我的选择是 FastAPI + SQLite + 本地Python规则逻辑。这个组合能让你快速搭建出可演示、代码清晰、符合现代技术趋势的原型。

3. 核心：用FHIR R4标准建模患者数据

医疗数据交换，标准是关键。FHIR（Fast Healthcare Interoperability Resources）是当前国际主流标准。我们用它的核心资源之一 Patient（患者）和 Observation（观察指标，如血压、血糖）来建模。

为什么用FHIR？

1. 专业性：直接使用国际标准，让项目立意更高，体现医学信息工程的专业性。
2. 结构化：数据定义清晰，避免了自建模型的随意性。
3. 扩展性：FHIR资源设计良好，未来容易扩展其他资源（如Medication、Condition）。

我们简化实现，创建一个本地的FHIR风格数据模型。

4. 完整可运行的代码示例

下面我们来搭建这个系统的核心部分。首先确保安装依赖：pip install fastapi uvicorn sqlalchemy pydantic

项目结构：

cdss_project/
├── main.py          # FastAPI应用主文件
├── models.py        # 数据模型定义 (SQLAlchemy)
├── schemas.py       # Pydantic模型定义 (用于API请求/响应)
├── database.py      # 数据库连接配置
├── rules_engine.py  # 简单的规则引擎
└── test_data.db     # SQLite数据库文件（自动生成）

第一步：定义数据模型和数据库 (models.py, database.py, schemas.py)

# database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 使用SQLite，数据库文件名为`test_data.db`
SQLALCHEMY_DATABASE_URL = "sqlite:///./test_data.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# models.py
from sqlalchemy import Column, Integer, String, Float, Date, Boolean
from database import Base

class Patient(Base):
    """患者模型，对应FHIR Patient资源的简化版"""
    __tablename__ = "patients"

    id = Column(Integer, primary_key=True, index=True)
    fhir_id = Column(String, unique=True, index=True)  # 类似FHIR资源的逻辑ID
    name = Column(String)
    birth_date = Column(Date)
    gender = Column(String)  # 'male', 'female', 'other'
    active = Column(Boolean, default=True)

class Observation(Base):
    """观察指标模型，对应FHIR Observation资源的简化版"""
    __tablename__ = "observations"

    id = Column(Integer, primary_key=True, index=True)
    fhir_id = Column(String, unique=True, index=True)
    patient_id = Column(Integer, index=True)  # 关联患者ID
    code = Column(String)  # 指标代码，如 '55284-4' (血压)
    value = Column(Float)
    unit = Column(String)   # 单位，如 'mm[Hg]'
    issued_date = Column(Date)  # 记录日期

# schemas.py
from pydantic import BaseModel
from datetime import date
from typing import Optional

class PatientBase(BaseModel):
    fhir_id: str
    name: str
    birth_date: date
    gender: str

class PatientCreate(PatientBase):
    pass

class Patient(PatientBase):
    id: int
    active: bool

    class Config:
        orm_mode = True  # 允许从ORM对象读取数据

class ObservationBase(BaseModel):
    fhir_id: str
    patient_id: int
    code: str
    value: float
    unit: str
    issued_date: date

class ObservationCreate(ObservationBase):
    pass

class Observation(ObservationBase):
    id: int

    class Config:
        orm_mode = True

第二步：实现简单的规则引擎 (rules_engine.py)

# rules_engine.py
from datetime import date
from typing import Dict, Any, List

def evaluate_patient_rules(patient_data: Dict[str, Any], observations: List[Dict[str, Any]]) -> List[str]:
    """
    评估患者数据和观察指标，触发预警规则。
    返回预警信息列表。
    """
    alerts = []

    # 规则1: 高龄患者风险提示
    if patient_data.get('birth_date'):
        birth_date = patient_data['birth_date']
        if isinstance(birth_date, str):
            # 简单处理，实际应转换日期
            from datetime import datetime
            birth_date = datetime.strptime(birth_date, '%Y-%m-%d').date()
        age = (date.today() - birth_date).days // 365
        if age > 75:
            alerts.append(f"患者年龄({age}岁)大于75岁，属于高龄，需关注综合健康状况。")

    # 规则2: 高血压预警 (基于最近一次血压记录)
    bp_observations = [obs for obs in observations if obs.get('code') == '55284-4']  # 假设55284-4是收缩压代码
    if bp_observations:
        latest_bp = max(bp_observations, key=lambda x: x['issued_date'])
        if latest_bp['value'] > 140:  # 收缩压大于140mmHg
            alerts.append(f"最新血压测量值 {latest_bp['value']} {latest_bp['unit']} 偏高，建议复查。")

    # 规则3: 血糖预警 (示例)
    glucose_obs = [obs for obs in observations if obs.get('code') == '2339-0']  # 空腹血糖代码
    if glucose_obs:
        latest_glucose = max(glucose_obs, key=lambda x: x['issued_date'])
        if latest_glucose['value'] > 7.0:  # 空腹血糖大于7.0 mmol/L
            alerts.append(f"最新空腹血糖 {latest_glucose['value']} {latest_glucose['unit']} 偏高，提示糖尿病风险。")

    return alerts

第三步：构建FastAPI主应用与CRUD API (main.py)

# main.py
from fastapi import FastAPI, Depends, HTTPException, status
from sqlalchemy.orm import Session
from typing import List

import models, schemas
from database import SessionLocal, engine
from rules_engine import evaluate_patient_rules

# 创建数据库表
models.Base.metadata.create_all(bind=engine)

app = FastAPI(title="简易临床决策支持系统(CDSS) API", description="基于FHIR标准的原型系统")

# 依赖项：获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# ---------- 患者 (Patient) CRUD ----------
@app.post("/patients/", response_model=schemas.Patient, status_code=status.HTTP_201_CREATED)
def create_patient(patient: schemas.PatientCreate, db: Session = Depends(get_db)):
    # 检查fhir_id是否重复
    db_patient = db.query(models.Patient).filter(models.Patient.fhir_id == patient.fhir_id).first()
    if db_patient:
        raise HTTPException(status_code=400, detail="FHIR ID already registered")
    db_patient = models.Patient(**patient.dict())
    db.add(db_patient)
    db.commit()
    db.refresh(db_patient)
    return db_patient

@app.get("/patients/", response_model=List[schemas.Patient])
def read_patients(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    patients = db.query(models.Patient).offset(skip).limit(limit).all()
    return patients

@app.get("/patients/{patient_id}", response_model=schemas.Patient)
def read_patient(patient_id: int, db: Session = Depends(get_db)):
    db_patient = db.query(models.Patient).filter(models.Patient.id == patient_id).first()
    if db_patient is None:
        raise HTTPException(status_code=404, detail="Patient not found")
    return db_patient

# ---------- 观察指标 (Observation) CRUD ----------
@app.post("/observations/", response_model=schemas.Observation)
def create_observation(observation: schemas.ObservationCreate, db: Session = Depends(get_db)):
    # 可选：验证patient_id是否存在
    db_observation = models.Observation(**observation.dict())
    db.add(db_observation)
    db.commit()
    db.refresh(db_observation)
    return db_observation

@app.get("/patients/{patient_id}/observations", response_model=List[schemas.Observation])
def read_patient_observations(patient_id: int, db: Session = Depends(get_db)):
    observations = db.query(models.Observation).filter(models.Observation.patient_id == patient_id).all()
    return observations

# ---------- 核心功能：触发临床决策规则 ----------
@app.get("/patients/{patient_id}/alerts")
def get_patient_alerts(patient_id: int, db: Session = Depends(get_db)):
    """
    获取针对特定患者的临床预警信息。
    1. 获取患者基本信息。
    2. 获取该患者的所有观察指标。
    3. 送入规则引擎进行评估。
    4. 返回预警列表。
    """
    # 1. 获取患者
    patient = db.query(models.Patient).filter(models.Patient.id == patient_id).first()
    if not patient:
        raise HTTPException(status_code=404, detail="Patient not found")

    # 2. 获取观察指标
    observations = db.query(models.Observation).filter(models.Observation.patient_id == patient_id).all()

    # 3. 准备数据并评估规则
    patient_dict = schemas.Patient.from_orm(patient).dict()
    obs_list = [schemas.Observation.from_orm(obs).dict() for obs in observations]

    alerts = evaluate_patient_rules(patient_dict, obs_list)

    # 4. 返回结果
    return {
        "patient_id": patient_id,
        "patient_name": patient.name,
        "alerts": alerts
    }


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

5. 性能与安全考量（毕设答辩加分项）

在答辩时，如果能提到以下几点，会显得你对生产环境有思考。

• 基础合规性 (HIPAA/GDPR)：
  • 数据加密：SQLite数据库文件本身不加密。在生产环境中，应对磁盘上的数据库文件进行加密，或使用支持加密的数据库。在演示时，可以提及此点。
  • 访问控制：我们的API目前是开放的。实际应添加身份认证（如OAuth2、JWT）和角色权限控制（RBAC），确保只有授权人员能访问患者数据。
• 敏感字段脱敏：在查询日志或返回非必要数据时，应对患者姓名、身份证号等字段进行脱敏处理（如显示为“张*”、“110101****1234”）。
• 请求限流：使用FastAPI中间件或像slowapi这样的库，对API接口进行限流，防止恶意请求或意外过载。

6. 生产环境避坑指南

• 绝对不要硬编码密钥：将数据库连接字符串、API密钥等敏感信息放在环境变量或配置文件中（如.env），使用python-dotenv读取。千万不要直接写在代码里并上传到GitHub！
• 谨慎记录日志：避免在应用程序日志中直接记录完整的患者个人信息（如logger.info(f”Processing patient {patient.name} with ID {patient.fhir_id}”)）。记录脱敏后的信息或仅记录操作类型和资源ID。
• 输入验证：FastAPI通过Pydantic已经做了基础类型验证。但对于医疗数据，还需要业务逻辑验证，例如性别字段是否只接受‘male’、‘female’、‘other’，日期是否合理等。
• 依赖管理：使用requirements.txt或Pipenv或Poetry精确管理项目依赖库及其版本，确保项目在任何地方都能以相同的环境运行。

结尾与展望

按照上面的步骤，你应该能跑起来一个拥有患者管理、指标录入和预警触发功能的CDSS原型了。访问 http://localhost:8000/docs 就能看到自动生成的API文档，进行测试，这足够作为你毕设的核心演示部分。

这个系统现在是一个独立的“数据孤岛”。一个很自然的延伸思考是：如何让它支持多家医院或机构在数据不出本地的前提下进行协作分析？ 这正是当前医疗AI的热点——联邦学习。

你可以将这个系统扩展为联邦学习的前端节点：

1. 本地模型训练：每个医院的CDSS系统在本地使用自己的患者数据训练一个机器学习模型（例如预测住院风险）。
2. 参数交换：只将训练好的模型参数（而非原始数据）加密后上传到一个协调服务器。
3. 全局模型聚合：协调服务器聚合所有节点的模型参数，生成一个更强大的全局模型。
4. 模型更新：将改进后的全局模型参数下发给各个节点。

这样，你的毕设就从“一个系统”的设计，上升到了“一种协作模式”的探讨，格局一下子就打开了。希望这篇笔记能帮你把抽象的毕业设计，变成一个看得见、摸得着、能演示、有深度的实战项目。祝你答辩顺利！