AI Agent驱动计算材料结构优化:ASE EMT最小闭环与生产扩展
做计算化学和材料计算的朋友大多有过这样的体验:一个结构优化的任务,从建模、选泛函、设收敛标准、跑任务、看轨迹、判断是否收敛,到最后一堆能量文件需要手工整理成表。整个链路里真正需要人做判断的环节其实只有几个,剩下的全是重复劳动和格式转换。而偏偏这几个判断环节又强依赖经验,新人很难独立完成。
这篇文章想聊的就是:能不能用 AI Agent 把这套链路串起来,让模型负责"决定下一步做什么",让程序负责"真正去做"。文章会从概念讲起,然后手写一个不依赖任何黑盒框架的最小 Agent,用 ASE 内置的 EMT 计算器跑通"建模—优化—校验—落盘"的完整闭环,最后再讨论怎么把这个玩具级实现扩展到真实的第一性原理计算和机器学习势场景。适合有 Python 基础、做过一些原子尺度模拟、想了解 Agent 工程化的读者。
1. 为什么要把 AI Agent 引入计算化学与材料计算
1.1 Agent 在这个语境下到底指什么
先把这个词说清楚,因为它在不同圈子里含义差别很大。在这里,AI Agent 指的是一个由大语言模型驱动的决策循环:模型读取当前的任务状态,从一组预先定义好的工具里挑选一个调用,拿到工具返回的结构化结果,再决定下一步,如此往复直到任务结束或者触发终止条件。
它和"用 ChatGPT 问一句然后复制粘贴结果"有本质区别。后者模型只是一次性地产生文本,所有执行动作由人完成;前者模型处在一个闭环里,它的每一次输出都会被程序解析成具体的函数调用,执行结果又会回到模型的上下文里。换句话说,Agent 的价值不在于模型懂多少化学,而在于它能把"意图"翻译成"操作序列",并且根据中间结果动态调整。
在计算化学和材料学里,这个特性的用处非常明显。因为这类任务的本质是"搜索"——搜索构型空间、搜索参数组合、搜索收敛路径。人的经验本质上是一套启发式规则,而 Agent 恰好可以把这套规则形式化成工具调用序列,同时保留在遇到异常时改弦更张的能力。
需要强调的是,Agent 不是万能的。凡是需要高精度数值判断的地方,比如某个泛函对某类体系的系统性误差,模型基本给不出可靠结论。它的定位应该是"实验记录员 + 操作员",而不是"理论家"。
1.2 与传统脚本、工作流引擎的区别
很多人会问:我写一个 Python 脚本,把 ASE 的步骤串起来不就完了?或者用工作流引擎编排一下?这三者的差别值得掰开讲。
传统脚本是确定性的,路径写死。好处是可复现、易调试;坏处是遇到边界情况就崩,比如某个元素 EMT 不支持、某个晶格常数导致优化发散,脚本没法自己绕过去,只能人来改代码或者改参数。
工作流引擎(比如各种 DAG 编排工具)解决了任务依赖和失败重试的问题,但它的分支逻辑依然是人预先写死的。你可以配"失败了重试三次",但没法配"如果能量异常升高,就换个初始晶格常数再试"。这类判断需要语义理解,工作流引擎做不到。
Agent 补的正是这一块:把分支决策交给模型。代价是引入了不确定性——同样的输入,模型两次可能给出不同的操作序列。所以在工程上,一个靠谱的 Agent 系统必须有确定性的校验层兜底,这一点后面会反复提到。
一个实用的判断标准是:如果你的任务流程完全固定、输入格式完全可控,那就老老实实写脚本,别上 Agent。如果你的任务需要"看情况",比如结构来源五花八门、收敛行为不可预测、需要根据中间结果选下一步,那 Agent 才有发挥空间。
1.3 计算材料学里适合 Agent 化的任务
不是所有任务都值得 Agent 化。根据实际使用情况,下面这几类收益比较明显。
第一类是结构准备与预处理。从 CIF、POSCAR、SMILES 等各种来源读入结构,判断晶胞是否合理、是否需要加真空层、是否需要扩胞、表面模型该切哪个晶面。这些判断规则性强但情况琐碎,很适合交给 Agent 调度。
第二类是计算流程编排。给定一个体系和一个目标(比如"算弹性常数"或"算形成能"),Agent 需要拆解成一系列计算任务,决定顺序,检查每一步的输出是否正常,异常时调整参数重跑。
第三类是结果解析与汇总。从大量输出文件里提取能量、力、带隙、晶格常数等量,做单位换算,整理成表格,标注异常值。这类工作纯体力活,Agent 做起来出错率反而比人低。
第四类是文献与数据库的信息抽取辅助。比如从材料数据库批量拉取结构并做一致性检查。注意这里说的是"辅助",涉及判断的部分仍然需要人复核。
反过来,纯粹的精度评估、方法选择的理论论证、需要长期经验的误差分析,这些不建议交给 Agent,至少不能让它做最终决策。
2. 环境准备与项目骨架
2.1 运行环境与依赖说明
本文示例以 Python 环境为例。Python 版本建议 3.9 以上,因为代码里用了一些较新的类型标注写法(from __future__ import annotations可以兼容低版本,但为了少踩坑,建议直接用较新的小版本)。
核心依赖只有两个方向:计算侧和编排侧。计算侧本文用 ASE(Atomic Simulation Environment),它自带的 EMT 计算器可以对部分金属做快速近似计算,正好适合做演示,不需要安装任何第一性原理软件。编排侧如果要接真实大模型,需要对应的 SDK;如果只想先把流程跑通,可以完全不用,本文提供了一个离线的脚本化 Planner。
需要说明的是,具体依赖版本请以你本地环境的实际情况为准,本文重点演示的是架构和配置思路,不绑定某个特定版本号。安装命令大致如下:
# 建议在虚拟环境中操作 python -m venv venv source venv/bin/activate # Windows 下用 venv\Scripts\activate # 计算侧依赖 pip install ase numpy # 可选:如果需要接大模型,再安装对应 SDK(版本以官方文档为准) # pip install openaiASE 的 EMT 计算器支持的元素有限,主要覆盖 Al、Cu、Ag、Au、Ni、Pd、Pt、H、C、N、O 等常见元素的一部分。如果你的体系包含它不支持的元素,跑起来会直接报错。这一点在演示阶段要特别注意,后面常见问题一节会专门讲这个报错怎么排查。
2.2 项目目录结构
为了让代码可维护,建议一开始就把结构分清楚。工具层负责真正干活,Agent 层负责调度,校验层负责兜底,三者不要混在一个文件里。推荐的结构如下:
chem-agent/ ├── agent/ │ ├── __init__.py │ ├── session.py # 计算会话,持有当前结构等状态 │ ├── tools.py # 工具函数实现 │ ├── schema.py # 工具描述(给模型看的) │ ├── validators.py # 结果校验 │ └── loop.py # Agent 主循环 ├── planners/ │ ├── __init__.py │ ├── scripted.py # 离线规则 Planner,用于测试 │ └── llm.py # 大模型 Planner(可选) ├── runs/ # 运行产物,自动创建 └── main.py # 入口这个划分的好处是:计算逻辑(tools.py)完全不感知模型的存在,可以被普通脚本直接调用,也可以在 Agent 里被调用。这一点非常重要,因为它意味着你的计算代码永远可以独立测试,不会因为模型抽风而不可用。
2.3 单位约定
在动手写代码之前,必须先把单位约定固定下来,否则后面一定乱。ASE 内部统一使用 eV 作为能量单位、Å 作为长度单位、eV/Å 作为力的单位。工具函数的入参和返回值都遵循这套约定,不要在中途混入 kcal/mol 或者 Hartree。
如果最终要和其他软件对接,单位换算放在结果输出环节统一做,不要在工具内部做隐式转换。这个原则看似啰嗦,但在实际项目里能省掉大量对不上数的排查时间。
3. 核心原理拆解:一个可落地的 Agent 由什么组成
3.1 四要素模型
抛开各种框架的包装,一个能干活的计算 Agent 由四部分组成:状态(State)、工具(Tools)、决策器(Planner)、循环控制器(Loop Controller)。
状态是当前任务的全部上下文,包括已构建的原子结构、已执行的操作历史、中间计算结果。在计算化学场景里,状态管理是最容易被低估的部分。因为原子结构是个复杂对象,没法直接塞进模型的上下文,只能序列化成摘要信息(元素组成、原子数、晶格参数、当前能量等)传给模型。
工具是一组有明确输入输出的函数。每个工具应该只做一件事,并且可以被独立测试。工具设计得好不好,直接决定了 Agent 的上限——模型再聪明,如果工具粒度太粗(比如一个工具直接跑完整套流程),模型就没有调度空间;如果太细(比如一个工具只移动一个原子),模型会被淹没在细节里。
决策器负责根据当前状态选择下一步动作。它可以是规则脚本,也可以是大模型。把它抽成接口的好处是:开发阶段用规则脚本跑通流程,上线前再换成模型,两者可以随时切换对比。
循环控制器负责防止 Agent 跑飞。必须有最大步数限制、重复动作检测、异常中断机制。一个没有终止条件的 Agent 是很危险的,尤其是当工具会消耗真实计算资源的时候。
3.2 工具契约设计的三条原则
写工具函数的时候,有三条原则值得坚持。
第一条:参数用基本类型。模型输出的参数是 JSON,只能表达字符串、数字、布尔、数组、对象。所以工具参数不要设计成Atoms对象这种复杂类型,而应该用symbol="Cu"、a=3.6这样的基本类型,在函数内部再去构造对象。
第二条:返回值必须可序列化且有明确状态标识。建议统一返回一个 dict,里面至少包含status字段(ok/error)和必要的数值结果。出错时不要抛异常穿透到循环层,而是在工具内部捕获后返回{"status": "error", "message": "..."}。这样模型能看到错误信息并自行调整,而不是整个程序崩掉。这点是 Agent 和普通脚本在设计理念上最大的不同之一。
第三条:单位写进描述里。工具的 description 字段里要明确写出参数和返回值的单位,比如"能量,单位 eV"、"力收敛判据,单位 eV/Å"。模型对单位非常敏感,描述里不写清楚,它很容易传错量级。
3.3 循环控制与终止条件
一个最基本的 Agent 循环如下:
- 把当前状态和历史整理成消息列表。
- 交给决策器,得到下一个动作。
- 如果是终止动作,结束循环并输出结果。
- 否则解析动作名和参数,调用对应工具。
- 记录工具返回结果,回到第 1 步。
这里面有几个必须设的护栏。最大步数限制(比如 20 步)防止无限循环;重复动作检测防止模型卡在同一个调用上;工具执行超时防止某个计算卡死;异常动作(工具名不存在、参数缺失)要有明确的兜底返回,而不是直接抛异常。
在实际项目中,还有一个容易被忽略的护栏:状态一致性检查。比如模型调用了"优化结构"之后又调用了"读取能量",如果优化失败导致结构状态不合法,读取能量就应该返回明确的错误而不是给出无意义的数值。
3.4 校验层为什么不能省
这一点值得单独强调。模型给出的操作序列在语义上可能是合理的,但在数值上可能完全错误。比如它可能把晶格常数 3.6 写成 36,或者把力的收敛判据 0.05 写成 5。这类错误如果直接传给计算程序,轻则白跑一次,重则产生一堆垃圾数据污染结果库。
校验层的作用就是在工具执行前后做数值合理性检查:晶格常数是否在合理区间(比如 2 到 10 Å)、原子间距是否过近(小于 0.5 Å 通常意味建模出错)、能量是否为有限值、优化后的最大力是否真的低于判据。
这些检查写起来不复杂,但能让整个系统的可靠性提升一个量级。后面实战部分会把它们具体实现出来。
4. 完整实战:搭建一个结构优化 Agent
4.1 初始化项目与依赖
先创建目录结构并写一个空的__init__.py,然后在agent/session.py里定义会话对象。这个对象是整段代码的状态中心。
# 文件路径:agent/session.py from __future__ import annotations import os from typing import Any, Dict, List, Optional from ase import Atoms class ChemSession: """计算会话:持有当前结构、历史记录和运行目录。""" def __init__(self, workdir: str = "runs"): self.workdir = workdir os.makedirs(self.workdir, exist_ok=True) self.atoms: Optional[Atoms] = None self.energy: Optional[float] = None self.max_force: Optional[float] = None self.history: List[Dict[str, Any]] = [] def require_atoms(self) -> Atoms: """获取当前结构,没有则抛出可被工具层捕获的错误。""" if self.atoms is None: raise RuntimeError("当前会话还没有结构,请先调用 build_bulk") return self.atoms def summary(self) -> Dict[str, Any]: """生成给模型看的结构摘要,不要塞完整坐标。""" if self.atoms is None: return {"has_structure": False} return { "has_structure": True, "formula": self.atoms.get_chemical_formula(), "natoms": len(self.atoms), "cell_lengths": [round(x, 4) for x in self.atoms.cell.lengths()], "energy_eV": None if self.energy is None else round(self.energy, 6), "max_force_eV_per_A": ( None if self.max_force is None else round(self.max_force, 6) ), } def log(self, step: int, action: str, payload: Dict[str, Any]) -> None: self.history.append({"step": step, "action": action, "payload": payload})注意summary()里只返回摘要而不是完整坐标,这是刻意设计的。一个 100 原子的结构如果完整序列化进上下文,会占用大量 token,而且模型也看不懂坐标数组。摘要里给元素组成、原子数、晶格参数、当前能量,已经足够支撑决策了。
4.2 编写核心工具函数
接下来是工具层。每个函数都遵循前面说的契约:基本类型参数、dict 返回值、错误内部捕获。
# 文件路径:agent/tools.py from __future__ import annotations import json import os from typing import Any, Dict from ase.build import bulk from ase.calculators.emt import EMT from ase.io import write as ase_write from ase.optimize import BFGS from agent.session import ChemSession def build_bulk( session: ChemSession, symbol: str, crystalstructure: str = "fcc", a: float = 3.6, ) -> Dict[str, Any]: """按指定晶格常数构建体相结构。a 的单位是 Å。""" try: if not (2.0 <= float(a) <= 10.0): return { "status": "error", "message": f"晶格常数 {a} 不在合理区间 [2.0, 10.0] Å,请检查是否单位写错", } atoms = bulk(symbol, crystalstructure, a=float(a), cubic=True) session.atoms = atoms session.energy = None session.max_force = None return { "status": "ok", "formula": atoms.get_chemical_formula(), "natoms": len(atoms), "cell_lengths_A": [round(x, 4) for x in atoms.cell.lengths()], } except Exception as exc: # noqa: BLE001 return {"status": "error", "message": f"构建结构失败: {exc}"} def optimize( session: ChemSession, fmax: float = 0.05, steps: int = 200, ) -> Dict[str, Any]: """使用 BFGS 做结构优化。fmax 单位 eV/Å。""" try: atoms = session.require_atoms() if not (1e-4 <= float(fmax) <= 1.0): return { "status": "error", "message": f"收敛判据 {fmax} eV/Å 不合理,常见取值 0.01 ~ 0.1", } if not (1 <= int(steps) <= 2000): return {"status": "error", "message": f"步数 {steps} 超出允许范围"} atoms.calc = EMT() optimizer = BFGS(atoms, logfile=os.devnull) converged = optimizer.run(fmax=float(fmax), steps=int(steps)) forces = atoms.get_forces() max_force = float((forces ** 2).sum(axis=1).max() ** 0.5) energy = float(atoms.get_potential_energy()) session.energy = energy session.max_force = max_force return { "status": "ok", "converged": bool(converged), "energy_eV": round(energy, 6), "max_force_eV_per_A": round(max_force, 6), "cell_lengths_A": [round(x, 4) for x in atoms.cell.lengths()], } except Exception as exc: # noqa: BLE001 return {"status": "error", "message": f"优化失败: {exc}"} def get_state(session: ChemSession) -> Dict[str, Any]: """读取当前会话状态摘要。""" return {"status": "ok", **session.summary()} def save_structure(session: ChemSession, filename: str = "final.xyz") -> Dict[str, Any]: """把当前结构写入 runs 目录,返回绝对路径。""" try: atoms = session.require_atoms() if os.path.basename(filename) != filename: return {"status": "error", "message": "filename 不能包含路径分隔符"} path = os.path.join(session.workdir, filename) ase_write(path, atoms) return {"status": "ok", "path": os.path.abspath(path), "natoms": len(atoms)} except Exception as exc: # noqa: BLE001 return {"status": "error", "message": f"写入结构失败: {exc}"}这里有几点值得解释。
BFGS的run()方法返回布尔值表示是否收敛,这个返回值要透传给模型,因为模型需要据此判断是继续调大步数还是调整初始结构。logfile=os.devnull是为了避免优化过程刷屏,实际项目中建议改成写文件,方便事后复盘。
最大力的计算用了(forces ** 2).sum(axis=1).max() ** 0.5,这是对每行三个分量求平方和再开方,得到每个原子的力矢量模长,取最大值。注意不要写成forces.max(),那是对分量取最大,物理意义完全不同。
save_structure里对filename做了路径分隔符检查,这是一个简单但必要的安全边界。如果不检查,模型完全可能传入../../etc/passwd这种值。虽然这只是个演示项目,但这种习惯要养成。
4.3 注册工具 Schema
工具写好了,还要告诉模型有哪些工具、参数是什么。这里用标准的 function calling 格式组织。
# 文件路径:agent/schema.py TOOL_SCHEMA = [ { "type": "function", "function": { "name": "build_bulk", "description": ( "按给定晶格常数构建体相晶体结构。" "参数 a 的单位是埃(Å),常见金属约在 3.0 到 4.5 之间。" ), "parameters": { "type": "object", "properties": { "symbol": {"type": "string", "description": "元素符号,如 Cu"}, "crystalstructure": { "type": "string", "description": "晶体结构类型,如 fcc、bcc、hcp", }, "a": {"type": "number", "description": "晶格常数,单位 Å"}, }, "required": ["symbol", "a"], }, }, }, { "type": "function", "function": { "name": "optimize", "description": "对当前结构做 BFGS 结构优化,返回最终能量和最大力。", "parameters": { "type": "object", "properties": { "fmax": { "type": "number", "description": "力的收敛判据,单位 eV/Å,常用 0.05", }, "steps": {"type": "integer", "description": "最大优化步数"}, }, "required": [], }, }, }, { "type": "function", "function": { "name": "get_state", "description": "读取当前结构摘要,包括组成、原子数、晶格常数、能量。", "parameters": {"type": "object", "properties": {}, "required": []}, }, }, { "type": "function", "function": { "name": "save_structure", "description": "把当前结构保存为 xyz 文件到运行目录。", "parameters": { "type": "object", "properties": { "filename": {"type": "string", "description": "文件名,不要带路径"} }, "required": ["filename"], }, }, }, ]Schema 里的 description 是给模型看的,写得好不好直接影响调用准确率。经验是:把单位、常见取值、参数间的依赖关系都写进去,模型出错的概率会明显下降。比如"常见金属约在 3.0 到 4.5 之间"这句话,能有效避免模型传 36 这种量级错误。
4.4 实现 Agent 主循环
主循环负责把 Planner、工具、会话串起来。这里刻意不依赖任何 Agent 框架,就是为了让每一步都看得见。
# 文件路径:agent/loop.py from __future__ import annotations import json from typing import Any, Callable, Dict, List from agent.session import ChemSession from agent.tools import build_bulk, get_state, optimize, save_structure MAX_STEPS = 20 class AgentLoop: def __init__(self, session: ChemSession, planner, max_steps: int = MAX_STEPS): self.session = session self.planner = planner self.max_steps = max_steps self.tool_registry: Dict[str, Callable[..., Dict[str, Any]]] = { "build_bulk": lambda **kw: build_bulk(session, **kw), "optimize": lambda **kw: optimize(session, **kw), "get_state": lambda **kw: get_state(session), "save_structure": lambda **kw: save_structure(session, **kw), } self.recent_actions: List[str] = [] self.trace: List[Dict[str, Any]] = [] def _is_repeat(self, action_key: str) -> bool: return len(self.recent_actions) >= 2 and self.recent_actions[-2:] == [action_key, action_key] def run(self, task: str) -> Dict[str, Any]: for step in range(1, self.max_steps + 1): decision = self.planner.next_action(task=task, state=self.session.summary()) if decision.get("name") == "finish": self.trace.append({"step": step, "action": "finish", "payload": decision}) return {"status": "ok", "steps": step, "answer": decision.get("arguments", {}), "trace": self.trace} name = decision.get("name", "") args = decision.get("arguments", {}) or {} action_key = f"{name}:{json.dumps(args, sort_keys=True)}" if name not in self.tool_registry: result = {"status": "error", "message": f"未知工具: {name}"} elif self._is_repeat(action_key): result = {"status": "error", "message": "检测到连续重复调用,请换一个动作"} else: try: result = self.tool_registry[name](**args) except TypeError as exc: result = {"status": "error", "message": f"参数不匹配: {exc}"} except Exception as exc: # noqa: BLE001 result = {"status": "error", "message": f"工具执行异常: {exc}"} self.recent_actions.append(action_key) self.session.log(step, name, {"args": args, "result": result}) self.trace.append({"step": step, "action": name, "args": args, "result": result}) return {"status": "timeout", "message": f"超过最大步数 {self.max_steps}", "trace": self.trace}几个细节说明。tool_registry里用 lambda 把 session 闭包进去,这样 Planner 只需要关心业务参数,不需要处理 session。_is_repeat检测连续两次完全相同的调用,这是最常见的卡死场景——模型看不到进展,就一直重复同一个动作。TypeError单独捕获是因为参数不匹配在 Agent 场景里极其常见,需要给模型明确的错误提示让它自我修正。
另外注意返回值里带了完整trace,这个在调试时非常有用,可以看到模型每一步的原始决策。生产环境里应该把它写进日志文件而不是直接返回给用户。
4.5 两个 Planner:离线脚本与大模型
先写一个离线的脚本化 Planner,用于在不接大模型的情况下验证整条链路。
# 文件路径:planners/scripted.py from __future__ import annotations from typing import Any, Dict, List class ScriptedPlanner: """按预设脚本顺序返回动作,用于离线联调和回归测试。""" def __init__(self, script: List[Dict[str, Any]]): self.script = list(script) self.cursor = 0 def next_action(self, task: str, state: Dict[str, Any]) -> Dict[str, Any]: if self.cursor >= len(self.script): return {"name": "finish", "arguments": {"summary": "脚本执行完毕"}} action = self.script[self.cursor] self.cursor += 1 return action对应的入口脚本:
# 文件路径:main.py from agent.loop import AgentLoop from agent.session import ChemSession from planners.scripted import ScriptedPlanner SCRIPT = [ {"name": "build_bulk", "arguments": {"symbol": "Cu", "crystalstructure": "fcc", "a": 3.6}}, {"name": "get_state", "arguments": {}}, {"name": "optimize", "arguments": {"fmax": 0.05, "steps": 200}}, {"name": "get_state", "arguments": {}}, {"name": "save_structure", "arguments": {"filename": "cu_final.xyz"}}, {"name": "finish", "arguments": {"summary": "Cu 结构优化完成"}}, ] def main() -> None: session = ChemSession(workdir="runs") loop = AgentLoop(session, ScriptedPlanner(SCRIPT)) result = loop.run(task="优化 fcc Cu 的晶格常数") for item in result.get("trace", []): print(item) if __name__ == "__main__": main()运行之后预期能看到这样一串输出:先构建出含 4 个原子的 fcc 铜晶胞(cubic=True会生成常规立方胞),然后优化几步,能量大约在 -0.02 eV/atom 量级(EMT 的绝对值本身没有物理意义,只看相对变化),最后在当前目录下生成runs/cu_final.xyz。
如果要在真实场景里换成大模型,Planner 需要实现同样的next_action接口。核心逻辑是把工具 schema 一起传给模型,然后解析返回的 tool_calls:
# 文件路径:planners/llm.py from __future__ import annotations import json from typing import Any, Dict, List from agent.schema import TOOL_SCHEMA # 说明:以下为调用思路示例,SDK 版本与方法名请以官方文档为准 class LLMPlanner: def __init__(self, client, model: str, system_prompt: str = ""): self.client = client self.model = model self.messages: List[Dict[str, Any]] = [] self.system_prompt = system_prompt or ( "你是一个计算材料学助手。你只能通过提供的工具操作结构," "不要凭空给出数值结果。每次只调用一个工具。" ) def next_action(self, task: str, state: Dict[str, Any]) -> Dict[str, Any]: if not self.messages: self.messages.append({"role": "system", "content": self.system_prompt}) self.messages.append( { "role": "user", "content": f"任务:{task}\n当前状态:{json.dumps(state, ensure_ascii=False)}", } ) response = self.client.chat.completions.create( model=self.model, messages=self.messages, tools=TOOL_SCHEMA, tool_choice="auto", ) message = response.choices[0].message self.messages.append(message.model_dump() if hasattr(message, "model_dump") else message) if not getattr(message, "tool_calls", None): return {"name": "finish", "arguments": {"summary": message.content}} call = message.tool_calls[0] try: arguments = json.loads(call.function.arguments or "{}") except json.JSONDecodeError: arguments = {} return {"name": call.function.name, "arguments": arguments} def observe(self, result: Dict[str, Any]) -> None: """把工具结果回填到消息历史,供下一轮决策使用。""" self.messages.append( {"role": "tool", "content": json.dumps(result, ensure_ascii=False)} )这里要提醒一句:不同版本 SDK 的工具调用回填格式不完全一样,tool消息通常需要带tool_call_id。实际接入时请对照官方文档调整,不要直接照抄。这也是本文故意把 Planner 抽成接口的原因——换个模型或者换个 SDK,只需要改这一个文件。
4.6 加入校验层
前面说的校验层,现在补上。它的作用是拦截那些"看起来能跑但结果没意义"的情况。
# 文件路径:agent/validators.py from __future__ import annotations import math from typing import Any, Dict from ase import Atoms def check_structure(atoms: Atoms, min_distance: float = 0.5) -> Dict[str, Any]: """检查结构的基本合理性,返回校验结论。""" issues = [] if len(atoms) == 0: issues.append("结构中没有原子") cell_lengths = atoms.cell.lengths() if any(x <= 0 for x in cell_lengths): issues.append(f"晶胞长度异常: {cell_lengths}") distances = atoms.get_all_distances(mic=True) for i in range(len(atoms)): for j in range(i + 1, len(atoms)): if distances[i][j] < min_distance: issues.append( f"原子 {i} 与 {j} 距离 {distances[i][j]:.3f} Å 过近,疑似建模错误" ) return {"ok": not issues, "issues": issues} def check_energy(energy: float) -> Dict[str, Any]: """检查能量是否为有限值。""" if energy is None: return {"ok": False, "issues": ["能量为空"]} if not math.isfinite(energy): return {"ok": False, "issues": [f"能量不是有限值: {energy}"]} return {"ok": True, "issues": []}get_all_distances(mic=True)里的mic表示使用最小镜像约定,对周期性体系这是必要的,否则会误判跨边界的原子距离。这个检查在扩胞、切表面之后特别有用,能抓出很多建模错误。
把校验接进优化流程,可以在optimize工具里返回值加一个validation字段,也可以在循环层统一做。实践上更推荐在工具内部做,因为工具知道自己的语义。
5. 从演示走向生产:扩展方向
5.1 接入第一性原理计算与机器学习势
演示用的是 EMT,速度极快但没有精度。真实场景里通常要换成第一性原理计算器或机器学习势。ASE 的好处是接口统一,只要计算器实现了get_potential_energy和get_forces,上层代码基本不用改。
下面是一个示意写法,展示如何把计算器做成可配置的:
# 文件路径:agent/calculators.py from __future__ import annotations from ase.calculators.emt import EMT def make_calculator(name: str = "emt", **kwargs): """根据名称返回计算器实例。真实项目里在这里分支到 VASP、CP2K 或机器学习势。""" name = (name or "emt").lower() if name == "emt": return EMT() # 其他计算器请按实际安装情况补充,例如: # if name == "mace": # from mace.calculators import mace_mp # return mace_mp(model=kwargs.get("model", "medium")) raise ValueError(f"不支持的计算器: {name}")这里的关键设计是把"计算器选择"也变成一个参数,让 Agent 可以在成本和质量之间做权衡。比如粗筛阶段用机器学习势,最后精修用第一性原理。这种分阶段策略在真实项目里非常常见,因为它能把计算成本降下来一大截。
需要提醒的是,真正的第一性原理计算动辄几十分钟到几小时,Agent 的循环必须支持异步——不能让整个进程阻塞等待。实践做法是把任务提交到作业调度系统,工具返回一个作业 ID,然后另设一个查询工具轮询状态。这样 Agent 的循环节奏和计算任务的生命周期就解耦了。
5.2 批量任务与结果汇总
单个结构的优化只是起点。真实场景往往是几十上百个结构,这时候 Agent 的角色就变成了"调度员":决定先跑哪些、哪些失败需要重试、哪些结果异常需要人工复核。
批量场景下最重要的工程点是幂等性和缓存。每个结构用内容哈希做 key,算过的直接读缓存,不要重复算。这一步能省掉大量算力,也能让整个流程可以断点续跑。
结果汇总建议输出结构化格式(JSON 或 CSV),字段固定,不要用自然语言描述结果。自然语言适合给模型读,不适合给人做后续分析。
5.3 让 Agent 输出可复现的记录
一个容易被忽略的要求:Agent 的每一次运行都应该产出一份可复现的记录,包括任务描述、模型版本、工具调用序列、每个工具的输入输出、最终结果。这份记录的价值在于,当结果出现争议时,你可以一层层回溯到底是哪一步出了问题。
实践上可以把前面循环里的trace直接序列化成 JSON 落盘,文件名带上时间戳。如果接了真实模型,建议把模型的原始响应也存下来,方便排查是模型的问题还是解析的问题。
6. 常见问题与排查清单
6.1 高频问题对照表
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| 报错说某元素没有对应的 EMT 参数 | EMT 计算器只支持部分元素 | 换用支持该元素的计算器,或在工具层做元素白名单校验并提前返回明确错误 |
优化跑完converged为 False | 初始结构离平衡位置太远、步数不够、判据过严 | 先放宽 fmax 到 0.1 粗优化,再收紧到 0.02 精优化;或调整初始晶格常数 |
能量出现nan或极大正值 | 原子重叠、晶胞退化、计算器参数缺失 | 用check_structure检查最小原子间距,确认晶胞长度为正 |
| 模型反复调用同一个工具 | 工具返回值没有提供新信息,模型无法判断进展 | 检查工具返回是否包含关键数值;在主循环加重复动作检测并返回提示 |
| 工具调用报参数不匹配 | 模型传了多余参数或类型不对 | 在循环里捕获TypeError,把错误信息回填给模型;同时精简 schema 的 required 字段 |
| 优化后晶格常数几乎没变 | 用了固定晶胞的自由度设置,只优化了原子位置 | 明确是否需要优化晶胞;ASE 中 BFGS 默认会优化所有自由度,但需确认Atoms是否带正确的 pbc |
| 结果每次跑都不完全一样 | 初始结构不同、优化器随机性、并行归约顺序 | 固定初始结构来源、固定优化器参数、记录完整输入 |
| 保存文件时报路径错误 | 文件名包含路径分隔符或非法字符 | 在工具层做文件名白名单校验,只允许字母数字下划线和点 |
6.2 建议的排查顺序
遇到 Agent 跑飞的情况,建议按这个顺序查:先看 trace 里最后一个成功的工具调用是什么,确认状态是否合法;再看模型给出的参数是否在合理范围;然后单独把那个工具函数拎出来,脱离 Agent 手动调用一次,确认计算本身没问题;最后才怀疑模型。这个顺序能帮你快速区分"计算问题"和"调度问题",避免在错误的方向上浪费时间。
一个很实用的技巧是准备一组固定的回归用例:几个典型结构、几组典型参数,每次改动 Agent 逻辑后跑一遍,确认结果没变。Agent 系统的不确定性很强,没有回归测试的话,改一行代码可能带来完全意想不到的行为变化。
7. 最佳实践与工程建议
7.1 工具粒度的把握
工具粒度是设计 Agent 时最需要反复调整的地方。太粗,模型没有调度空间,Agent 退化成一个固定脚本;太细,模型要在琐碎操作上消耗大量 token 和步数。
一个经验法则是:一个工具对应一个"物理上有意义的操作"。构建结构、优化结构、计算单点能、保存结构,这些都是有明确物理含义的原子操作。而"把第 3 个原子的 x 坐标加 0.1"这种就不适合做成工具,因为它不对应任何有意义的研究动作。
另一个法则是:如果一个操作几乎总是跟着另一个操作出现,考虑把它们合并。比如"优化并返回能量"就比"优化"和"读能量"两个工具更实用,因为分开调用只会增加步数。
7.2 安全边界与资源控制
涉及真实计算资源的场景,安全边界必须提前设计。至少要有这几层:文件写入限制在指定目录内;计算任务有超时和资源上限;单次会话的总计算量有配额;危险操作(比如删除文件、提交大规模作业)需要显式确认。
不要指望模型自己遵守这些规则。模型的输出是不可信的输入,所有约束必须在代码层强制实施。这和 Web 开发里"永远不要信任前端传参"是同一个道理。
7.3 可复现性与版本记录
计算科学最讲究可复现。引入 Agent 之后,可复现性面临新的挑战:同一个任务,模型可能给出不同的操作序列。应对办法有两个方向:一是把所有输入(任务描述、模型版本、温度参数、工具 schema 版本)都记录下来;二是关键流程提供"固化"能力,也就是把一次成功的 Agent 运行固化成脚本,后续直接跑脚本。
第二个方向在实际项目中价值很高。Agent 负责探索,一旦找到可行路径就固化成确定性流程,既保留了灵活性,又保证了稳定性。
7.4 日志与可观测性
建议给每次会话分配一个唯一 ID,所有日志带上这个 ID。日志内容至少包括:每步的动作名、参数、返回结果摘要、耗时。有了这些,出问题时可以像查分布式系统调用链一样回溯。
另外建议对工具调用做耗时统计。如果某个工具平均耗时特别长,说明它可能被调用得过于频繁,或者内部有优化空间。这类数据驱动的优化比凭感觉调整有效得多。
7.5 成本控制
如果接的是商用大模型 API,token 成本需要关注。控制手段主要有三个:精简 schema(只保留必要工具和参数)、精简状态摘要(不要传完整结构)、限制历史长度(只保留最近若干轮对话)。
一个容易忽略的点是工具返回值的长度。如果工具返回了完整的坐标数组或者几千行的输出文件,token 会瞬间爆掉。所有工具返回值都应该做摘要化处理,只保留决策需要的数值。
8. 学习路线与后续方向
如果想把这条路走下去,建议分三个阶段推进。
第一阶段是把本文的演示跑通并且改造它。换几个不同的元素和晶体结构,故意传入错误参数看系统怎么反应,故意让优化不收敛看 Agent 能否自行调整。这个阶段的目的是建立对整个链路的直觉。
第二阶段是接入真实计算。选一个你熟悉的小体系,用你平时用的计算软件跑一遍,然后把同样的流程用 Agent 封装。这个阶段会暴露大量工程细节问题:作业提交、状态轮询、输出解析、错误恢复。把这些问题解决掉,系统才算真正可用。
第三阶段是做对比评估。用同一批任务,分别用人工流程和 Agent 流程跑,对比耗时、成本、结果一致性。只有在数据上站得住脚,才值得推广。
需要提前提醒的风险有几个。一是不要把 Agent 的输出当成最终结论,所有数值结果都必须经过独立校验;二是不要在未经验证的情况下让它操作生产计算资源;三是注意数据的合规性,涉及未公开材料的计算数据不要随意传给外部服务。
最后想说,Agent 在计算材料学里的定位,短期内更像是"熟练的科研助理"而不是"独立研究者"。它能把重复劳动接过去,能帮你把流程标准化,能减少低级错误,但科学判断的部分仍然需要人来把关。把这个定位想清楚,落地的时候就不会有太多不切实际的期待,也更容易做出真正有用的东西。