简介：面向计算化学与材料科学方向的研究者及研究生，这份资源提供了一套可复用的批量分子动力学模拟脚本与配套示例，用于自动化处理一系列相似的MD任务，减少重复手工配置的工作量。压缩包共427个文件，约14.62MB，以in输入控制文件、ac与ac0等Antechamber中间文件、pdb结构文件、prmtop与inpcrd拓扑及坐标文件、frcmod力场参数、mol2分子文件为主，另含sh批处理脚本、py脚本、log与out运行记录以及一份docx说明文档，覆盖从系统初始化、能量最小化到热平衡与动态演化的完整流程。目前已有694人学习。借助这些脚本与示例，读者可以快速理解批量模拟的参数设置与调用方式，参考预设分子体系与力场文件搭建自己的算例，并借鉴结果处理与日志分析思路，适合作为批量MD工作流的起步模板。
1. 批量分子动力学模拟脚本和示例：从单点跑通到批量产出的落地路径
做材料计算的人大概都有过这种经历：手头一个体系跑通了，能量收敛、轨迹正常，觉得万事大吉；等到要扫参数、换结构、跑几十上百个任务时，才发现真正耗人的不是物理，而是重复劳动——改输入、建目录、提交、盯日志、收数据，一套流程走下来，人比机器还累。这份「批量分子动力学模拟脚本和示例」资源，解决的正是这个断层：它把单次模拟的输入模板、批量调度脚本、结果汇总逻辑打包在一起，让一个能跑通的算例变成一条能批量产出的流水线。适合已经会用主流分子动力学引擎跑单点、但还没建立起批量工作流的研究生和一线计算人员，也适合需要把模拟任务纳入日常数据生产的技术岗。下面按「资源是什么、怎么搭、怎么跑、坑在哪」的顺序拆开讲。
2. 批量模拟的目录结构与输入模板：先把骨架搭对
批量任务翻车，十有八九不是物理参数错，而是目录结构和输入模板没设计好。单点模拟时你怎么放文件都行，一旦上批量，路径、命名、模板变量三者必须解耦，否则脚本跑到一半就会因为某个目录名带空格、某个输入文件缺字段而集体失败。这一章先把骨架讲清楚，后面所有脚本都建立在这个结构上。
2.1 为什么批量任务要先定目录约定
常见做法是「一个任务一个目录」，目录名即任务 ID，目录内固定放输入文件、日志、输出。这样做的直接好处是：调度脚本只需要遍历任务 ID 列表，不需要理解每个任务的物理含义；失败重跑时删掉对应目录即可，不会污染其他任务；结果汇总时按目录名就能反推参数组合。
我一般会把结构定成三层：顶层是项目根目录，第二层是tasks/，第三层是每个任务自己的目录。任务目录内部再固定几个文件名，比如input.in、run.log、output/。文件名固定下来，脚本里就不用写一堆条件判断去猜「这个任务到底叫 in 还是 input」。
# 项目根目录结构（脚本按此约定遍历，不要随意改名） project/ ├── templates/ │ └── input.template # 带占位符的输入模板 ├── tasks/ # 每个子目录是一个独立任务 │ ├── task_0001/ │ │ ├── input.in # 由模板渲染生成 │ │ ├── run.log │ │ └── output/ │ └── task_0002/ ├── scripts/ │ ├── gen_inputs.py # 模板渲染 + 建目录 │ ├── submit_all.sh # 批量提交 │ └── collect.py # 结果汇总 └── params.csv # 参数组合表，一行一个任务这段结构不是随便画的。templates/和tasks/分离，意味着你改模板不会动到已生成的任务；params.csv单独存在，意味着参数组合可以版本化管理，换一批参数只改这一个文件。scripts/集中放脚本，避免脚本散落在任务目录里被误删。
参数说明上，params.csv的列名要和模板占位符一一对应。比如模板里写{temperature}，CSV 里就得有temperature列。列名对不上是渲染阶段最常见的失败原因，脚本里最好加一层校验，缺列直接报错退出，而不是生成一堆残缺输入。
2.2 输入模板的占位符设计与渲染脚本
模板渲染的核心是把「不变的物理设置」和「要扫的参数」分开。不变的写死在模板里，要扫的用占位符。占位符命名建议用有意义的英文单词，别用{0}、{1}这种位置参数，后者在参数列顺序变化时极易错位。
# gen_inputs.py：读参数表，渲染模板，建任务目录 import csv import os import re import sys TEMPLATE_PATH = "templates/input.template" PARAM_PATH = "params.csv" TASK_ROOT = "tasks" def load_params(path): with open(path, newline="", encoding="utf-8") as f: reader = csv.DictReader(f) rows = list(reader) if not rows: sys.exit("参数表为空，检查 params.csv") return rows def render(template_text, row, task_id): # 占位符形如 {temperature}，逐个替换；缺字段直接报错 def repl(match): key = match.group(1) if key not in row: sys.exit(f"任务 {task_id} 缺少参数列: {key}") return str(row[key]) return re.sub(r"\{(\w+)\}", repl, template_text) def main(): with open(TEMPLATE_PATH, encoding="utf-8") as f: template_text = f.read() rows = load_params(PARAM_PATH) for idx, row in enumerate(rows, start=1): task_id = f"task_{idx:04d}" task_dir = os.path.join(TASK_ROOT, task_id) os.makedirs(os.path.join(task_dir, "output"), exist_ok=True) rendered = render(template_text, row, task_id) with open(os.path.join(task_dir, "input.in"), "w", encoding="utf-8") as f: f.write(rendered) print(f"生成 {task_id}") if __name__ == "__main__": main()逻辑说明：load_params用DictReader把 CSV 读成字典列表，列名即键；render用正则匹配{word}形式的占位符，替换时如果发现某个键在行里不存在，直接sys.exit终止，避免生成半成品。任务 ID 用四位补零，保证目录排序和参数表顺序一致，后面汇总时不会乱序。
参数上要注意两点：一是 CSV 里的数值列不要带单位，单位写进模板，否则渲染出来会出现300 K K这种重复；二是如果某个参数是字符串且含空格，CSV 里要用引号包住，DictReader会正确处理。渲染完建议抽查一两个input.in，确认占位符全部被替换，没有残留的{。
3. 批量提交与并发控制：让任务跑满但不炸机
输入生成完，下一步是提交。批量提交最容易出的问题不是脚本写错，而是并发没控住——一次性把上百个任务全推上去，要么把调度队列挤爆，要么把本地机器内存吃光。这一章讲提交脚本怎么写、并发怎么限、失败怎么重试。
3.1 提交脚本的任务发现与状态标记
提交脚本的第一件事是「发现哪些任务还没跑」。做法很简单：遍历tasks/下的目录，检查output/里有没有完成标记文件。完成标记由模拟引擎正常结束后写入，或者由提交脚本在检测到正常退出码后写入。没有标记的才提交，有标记的跳过。这样脚本天然支持断点续跑，中断后重跑不会重复提交已完成任务。
#!/usr/bin/env bash # submit_all.sh：遍历任务目录，跳过已完成，提交未完成 set -u TASK_ROOT="tasks" MAX_JOBS=8 # 同时运行的任务上限，按机器核数和内存调 RUNNING=0 for task_dir in "$TASK_ROOT"/task_*; do [ -d "$task_dir" ] || continue marker="$task_dir/output/.done" if [ -f "$marker" ]; then echo "跳过已完成: $task_dir" continue fi # 后台提交，记录 PID 便于后续等待 ( cd "$task_dir" && ./run_sim.sh > run.log 2>&1 && touch output/.done ) & RUNNING=$((RUNNING + 1)) if [ "$RUNNING" -ge "$MAX_JOBS" ]; then wait -n # 等任意一个结束再继续 RUNNING=$((RUNNING - 1)) fi done wait echo "全部提交完毕"逻辑说明：set -u让未定义变量直接报错，避免路径变量拼错时静默跑错目录。wait -n是并发控制的关键，它等任意一个后台任务结束就返回，从而把同时运行数压在MAX_JOBS以内。完成标记.done只在模拟正常退出（&&链）后才写入，失败任务不会留标记，下次重跑会被重新提交。
参数上，MAX_JOBS要按实际资源定：分子动力学单任务通常吃一个核到几个核，内存占用和体系大小强相关。保守做法是先按「核数 / 单任务核数」算上限，再留一两个余量给系统。如果任务跑在共享集群上，这个值还要和调度器的队列限制对齐，别在本地脚本里限了 8 个、调度器那边又各开多线程，实际并发翻倍。
3.2 并发数与资源配比的确定方法
并发数不是拍脑袋定的。一个可操作的估算方式是：先跑一个代表性任务，用time和内存监控记录它的峰值内存与平均 CPU 占用，再用「总内存 × 0.8 / 单任务峰值内存」和「总核数 / 单任务核数」两个值取小。这样算出来的并发数既不会因为内存不足被系统杀掉，也不会因为核数闲置浪费机时。
| 资源项 | 单任务实测 | 机器总量 | 建议并发 |
|---|---|---|---|
| 内存 | 2.5 GB | 32 GB | 10（取 0.8 系数后约 10） |
| CPU 核 | 2 | 16 | 8 |
| 磁盘 | 500 MB/任务 | 视剩余空间 | 按剩余空间 / 单任务估算 |
上表是示意，实际数值要自己测。注意磁盘这一行常被忽略：批量任务每个都写轨迹文件，几十个任务同时跑，磁盘很快见底，脚本跑到一半报「No space left」是最冤的翻车。建议在提交前先算一遍「任务数 × 单任务预估输出大小」，和剩余空间比一比。
4. 结果汇总与异常排查：从一堆目录里捞出可用数据
任务跑完不等于拿到结果。几十个任务目录里，有的是正常结束，有的是中途报错，有的跑完了但物理量发散。这一章讲怎么把结果汇总成一张表，以及怎么快速定位异常任务。
4.1 汇总脚本的字段抽取与容错
汇总的核心是从每个任务的日志或输出文件里抽几个关键量：最终能量、温度均值、是否收敛、运行时长。抽取脚本要对缺失文件容错，不能因为一个任务缺日志就整个崩掉。
# collect.py：遍历任务目录，抽取关键量，输出汇总 CSV import csv import os import re TASK_ROOT = "tasks" OUT_CSV = "summary.csv" # 按实际日志格式调整正则；这里假设日志里有 "Final energy = -123.45" ENERGY_RE = re.compile(r"Final energy\s*=\s*(-?\d+\.?\d*)") TEMP_RE = re.compile(r"Average temperature\s*=\s*(-?\d+\.?\d*)") def parse_task(task_dir): log_path = os.path.join(task_dir, "run.log") record = {"task": os.path.basename(task_dir), "done": os.path.exists(os.path.join(task_dir, "output", ".done")), "energy": "", "temperature": "", "note": ""} if not os.path.exists(log_path): record["note"] = "缺日志" return record with open(log_path, encoding="utf-8", errors="ignore") as f: text = f.read() m = ENERGY_RE.search(text) if m: record["energy"] = m.group(1) else: record["note"] = "未匹配到能量" t = TEMP_RE.search(text) if t: record["temperature"] = t.group(1) return record def main(): records = [] for name in sorted(os.listdir(TASK_ROOT)): task_dir = os.path.join(TASK_ROOT, name) if os.path.isdir(task_dir) and name.startswith("task_"): records.append(parse_task(task_dir)) with open(OUT_CSV, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=["task", "done", "energy", "temperature", "note"]) writer.writeheader() writer.writerows(records) print(f"汇总 {len(records)} 个任务 -> {OUT_CSV}") if __name__ == "__main__": main()逻辑说明：parse_task先看日志在不在，不在就记「缺日志」并返回，不抛异常；正则匹配不到时记「未匹配到能量」，保留空值。这样汇总表里每一行都有明确状态，note列就是排查入口——按note非空筛一遍，异常任务一目了然。errors="ignore"是为了防止日志里混入非 UTF-8 字符导致读取失败。
参数上，正则要按你实际引擎的日志格式改。不同引擎输出措辞差别很大，硬套别人的正则是常见错误。建议先手动打开一个成功任务的日志，把要抽的那行原样复制出来，再据此写正则，写完拿两三个任务试跑验证。
4.2 异常任务的快速定位路径
汇总表出来后，排查顺序建议是：先看done为 False 的，这些是没跑完的；再看done为 True 但note非空的，这些是跑完了但抽取失败的；最后看数值明显异常的，比如能量为正、温度偏离设定值很远。前两类是流程问题，第三类是物理问题，分开处理效率最高。
流程问题的定位靠日志尾部：tail -n 50 run.log看最后报什么错，常见的是输入文件字段缺失、磁盘满、被系统 OOM 杀掉。物理问题的定位靠轨迹和能量曲线，这时候汇总表里的数值就是线索，顺着异常任务 ID 回去看它的input.in和参数行，往往能发现是某个参数组合本身不合理，而不是脚本的锅。
5. 避坑与常见问题：批量跑模拟最容易踩的五个坑
这一章按「现象 → 原因 → 解决」列五条实际踩过的坑，都是批量场景特有的，单点跑通时根本遇不到。
坑一：任务目录名带空格或中文，脚本遍历时路径断裂。现象是提交脚本报「No such file or directory」，但目录明明存在。原因是 shell 里未加引号的变量展开会把空格当分隔符。解决是目录名只用字母数字下划线，脚本里所有路径变量加双引号，"$task_dir"而不是$task_dir。
坑二：模板渲染后占位符残留，模拟引擎读到非法字段直接退出。现象是任务全部失败，日志里报输入解析错误。原因是 CSV 列名和模板占位符大小写或拼写不一致，render没匹配到就没替换。解决是渲染后加一步校验，grep -n "{" input.in扫一遍，有残留就停下；同时渲染脚本里缺字段直接退出，别静默跳过。
坑三：并发开太大，机器内存耗尽，任务被系统杀掉且不留痕迹。现象是部分任务日志戛然而止，没有报错信息，.done标记也没写。原因是并发数超过内存承载，OOM killer 直接杀进程。解决是按 3.2 的方法实测单任务峰值内存再定并发，并在提交脚本里加内存监控，或者干脆用调度器的资源申请机制代替本地并发控制。
坑四：断点续跑时重复提交已完成任务，结果被覆盖。现象是重跑后某些任务的结果和之前不一致。原因是完成标记没写或写在了错误位置，脚本误判为未完成。解决是完成标记只在模拟正常退出后写入，且写入路径和检查路径严格一致；重跑前先ls tasks/*/output/.done | wc -l核对已完成数量。
坑五：汇总时正则匹配到日志里的中间值而非最终值。现象是汇总表里的能量和手动看日志的最终能量对不上。原因是日志里同一关键词出现多次（比如每步都打印），search只取第一个匹配。解决是用findall取最后一个，或者把正则锚定到日志末尾的汇总段，比如匹配「Final」开头的行而不是任意行。
提示：这五条里，坑二和坑五属于「静默错误」，不会让脚本崩，但会让结果错，危害比直接报错大得多。批量流程里，宁可让脚本在异常时大声失败，也不要让它带着错误数据一路跑完。
6. 进阶：把批量脚本接进参数扫描与结果校验闭环
骨架跑顺之后，真正提升效率的是把「参数生成 → 批量执行 → 结果校验」串成闭环，而不是每次手动改 CSV、手动看汇总。一个实用技巧是给汇总表加一列「校验状态」，用几条简单规则自动标记可疑任务，把人工检查的范围从几十个缩到几个。
校验规则可以很朴素：能量为正标记可疑（多数体系基态能量应为负）；温度偏离设定值超过 10% 标记可疑；运行时长明显短于同批中位数标记可疑（可能提前退出）。这些规则不需要多精确，目的是把明显不对的挑出来。
# validate.py：在 summary.csv 基础上加校验列 import csv import statistics def main(): with open("summary.csv", newline="", encoding="utf-8") as f: rows = list(csv.DictReader(f)) energies = [float(r["energy"]) for r in rows if r["energy"]] if not energies: print("无可校验数据") return median_e = statistics.median(energies) for r in rows: flags = [] if r["energy"]: e = float(r["energy"]) if e > 0: flags.append("能量为正") if abs(e - median_e) > abs(median_e) * 0.5: flags.append("能量偏离中位数") r["flag"] = ";".join(flags) with open("summary_checked.csv", "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter(f, fieldnames=list(rows[0].keys())) writer.writeheader() writer.writerows(rows) print("校验完成 -> summary_checked.csv") if __name__ == "__main__": main()这段脚本的价值不在规则本身，而在「自动缩小人工检查范围」这个思路。跑完批量任务后，先看flag非空的行，逐个回查；flag为空的基本可以放心。参数上，偏离阈值 0.5 是经验值，体系差异大时可以调松或调紧，但别设得太严，否则正常波动也会被标出来，反而增加负担。
从那以后我每次搭批量流程，都会先把「完成标记 + 汇总表 + 校验列」这三件套配齐再开跑，哪怕只有十几个任务。因为批量任务的成本不在跑，而在跑完之后分不清哪些能用、哪些要重来。这套脚本和示例的价值也在这里：它给的不是一个能跑的算例，而是一条从单点到批量、从产出到校验的完整路径，照着搭一遍，后面换体系、换参数都能复用。希望帮到你。
本文还有配套的精品资源，点击获取
