AGENTS.md — 健康画像 ETL 项目 AI 协作指南

项目概述

本项目是一个基于 Python + Neo4j 的医疗健康画像数据 ETL 工程。从 SQL Server 的 ai_patients 表获取患者 ID，调用大数据平台 API 拉取患者健康画像数据，经图谱建模后写入 Neo4j，最终通过 Flask REST API 对外提供查询服务。

代码库结构速查

etl_neo4j/
├── run_api.py                       # ★ API 服务独立启动入口
├── main.py                          # ★ ETL 主入口（增量执行、重试）
├── core/                            # 全局通用组件
│   ├── config.py                    # 全局配置（数据库、API、调度参数）
│   ├── logger.py                    # 日志处理
│   └── db.py                        # Neo4j 单例连接（线程安全）
├── api/                             # REST API 包模块
│   ├── __init__.py                  # Flask App 注册
│   └── routes.py                    # API 路由节点实现
├── etl/                             # 数据提取与转换逻辑
│   ├── extractors/                  # 从外部抓取数据
│   │   ├── health_portrait_api.py   # 大数据平台 HTTP 客户端
│   │   └── sqlserver.py             # SQL Server 客户端
│   ├── core/                        # ★ 图谱核心构建模块
│   │   └── etl_patient.py           # 图谱节点/关系的 Cypher 写入逻辑 
│   └── processors/                  
│       └── health_portrait.py       # 协调 API 调用与事务提交
├── scheduler/                       # 批处理调度
│   ├── job_manager.py               # 批量调度 + 错误队列
│   └── scheduler.py                 # 定时调度器
├── data/                            # 动态产生与运行时数据
│   ├── state/
│   │   └── etl_state.json           # 增量记录（上次成功时间戳）
│   └── samples/                     # 测试用 JSON 样例
├── tests/                           # 单元与集成测试脚本集
├── scripts/                         # 部署/环境验证运维工具
└── docs/                            # 文档说明

数据流与核心调用链

main.py
  └─ SQLServerConnection.load_patient_ids(last_update_time)   # 增量拉取患者 ID
  └─ JobManager.process_batch(empi_list)
       └─ ThreadPoolExecutor → _process_single(empi)
            ├─ HealthPortraitAPI.get_health_portrait(empi)     # 调用大数据平台
            └─ HealthPortraitProcessor.process(patient_data)
                 └─ session.execute_write(import_patient_data_from_json, data)
                      └─ etl_patient.py 中各 import_* 函数（Cypher MERGE 写入）

写入完成后，etl_state.json 记录本次运行时间戳，供下次增量查询使用。

图谱数据模型（节点与关系）

主要节点标签

核心关系类型

API 端点速查

所有端点均以 /api/patients/<patient_id>/ 为前缀：

开发规范与约定

数据写入规范

1. 全部使用 MERGE 而非 CREATE，以保证幂等性。Encounter、Patient、Condition 等核心节点均设有唯一性约束。
2. 批量写入优化：import_diagnoses_from_encounter 采用两阶段写入——先批量 UNWIND 创建节点，再批量创建关系，避免高并发下的锁竞争。
3. 空值过滤：各 import_* 函数内部均对关键字段进行空值检查，缺失时 debug 级别记录日志并跳过，不抛出异常。
4. 家族成员处理：通过 idType + idValue 作为家族成员的唯一标识，支持"认领"机制——若 Neo4j 中已有通过家族关系预建的节点，ETL 会将 patientId 赋给它，实现无缝合并。

日志规范

• 使用 setup_logger(name) 获取 logger，禁止直接 logging.getLogger()。
• 每个模块使用独立的 logger 名称（etl_patient_core、health_portrait、job_manager 等）。
• 正常业务跳过用 logger.debug()，异常用 logger.error()，关键里程碑用 logger.info()。

配置规范

• 所有配置均在 config/settings.py 的 Config 类中集中管理。
• 不得在业务代码中硬编码 URI、密码、表名等敏感信息。
• 修改配置后，先运行 python check_config.py 验证有效性。

错误处理规范

• HealthPortraitProcessor.process() 捕获异常后必须重新 raise，以便 JobManager 将其计入错误队列并触发重试。
• JobManager 的错误重试上限由 Config.RETRY_TIMES 控制，重试间隔由 Config.RETRY_DELAY 控制。
• 若所有重试均失败，etl_state.json 的时间戳不会更新，下次运行会重新处理这些患者。

本地开发快速上手

# 1. 安装依赖
pip install -r requirements.txt

# 2. 检查环境配置
python check_config.py

# 3. 单次运行 ETL（使用本地 JSON 测试数据）
# 在 job_manager.py 中取消注释本地文件读取代码，注释 API 调用
python main.py

# 4. 启动 API 服务
python app.py
# 访问 http://localhost:5000/api/docs

# 5. 启动定时调度
python -c "from scheduler.scheduler import ETLScheduler; ETLScheduler().start(24)"

AI Agent 任务指引

新增实体/关系类型

1. 在 etl/core/etl_patient.py 中新增 import_xxx(tx, patient_id, data_list) 函数。
2. 在 import_patient_data_from_json 的末尾调用新函数。
3. 在 etl/core/__init__.py 的 __all__ 列表中导出。
4. 如需 API 查询，在 app.py 中新增对应端点，参照已有端点的 @neo4j_session 装饰器模式。

修改 Cypher 查询

• 所有 Cypher 均写在各 import_* 函数的局部变量 query 中，直接修改字符串即可。
• 修改后用 files/test_patient.json 作为输入运行单次测试验证。

新增 API 端点

@app.route('/api/patients/<string:patient_id>/new_endpoint', methods=['GET'])
@neo4j_session
def get_new_endpoint(session, patient_id):
    """端点的中文描述（会显示在 /api/docs 页面）"""
    query = "MATCH (p:Patient {patientId: $patientId}) ..."
    results = session.execute_read(lambda tx: list(tx.run(query, patientId=patient_id)))
    return jsonify([serialize_record(r) for r in results])

调试单个患者

# 临时修改 job_manager.py 的 _process_single 方法，
# 用本地 JSON 文件替代 API 调用：
with open("files/test_patient.json", 'r', encoding='utf-8') as f:
    patient_data = json.load(f)
patient_data = patient_data.get('data', patient_data)  # 兼容两种格式

已知限制与注意事项

• MAX_WORKERS 当前建议设为 1（串行），多线程并发写入 Neo4j 在当前图模型下可能引发死锁（MERGE 竞争）。
• archive/ 目录中的文件为历史版本，请勿在新功能中引用。
• app.py 中已注释的部分端点（findings、bodypart 等）依赖图模型中尚未实现的节点类型，暂不可用。
• etl_state.json 存储的时间戳格式为 2026-01-29 16:38:58 (Beijing)，解析时需注意去除后缀。