摘要本文系统介绍 Python 包 aiida-workgraph 的定位与核心功能讲解安装方式、WorkGraph 的语法与常用参数并通过 16 个由浅入深的实际案例演示任务编排、数据流转、条件分支、动态循环、错误处理与结果导出等典型用法最后总结常见错误与使用注意事项。一、aiida-workgraph 是什么aiida-workgraph 是 AiiDA 生态中用于构建和运行工作流图的 Python 库。它把工作流抽象为「节点 边」的有向图节点代表一个计算任务、一个 Python 函数或一个数据对象边代表数据依赖关系。相比 AiiDA 原生的 WorkChainaiida-workgraph 更强调动态构建与交互式编排适合需要根据中间结果动态决定后续步骤的场景。1.1 核心设计理念图即工作流用 Python 代码描述节点与依赖图结构在运行时才最终确定。数据流驱动节点之间的连接通过输入输出端口socket建立数据自动流转。可持久化基于 AiiDA 的 provenance 机制每个节点的执行记录都可追溯。可交互支持在 Jupyter Notebook 中逐步构建、运行和检查工作流。1.2 主要功能特性功能说明任务节点支持 AiiDA CalcJob、WorkChain、Python 函数作为节点数据节点支持 AiiDA Data 类型Int、Float、Dict、Structure 等动态分支根据运行时结果决定后续节点是否执行循环与映射支持对列表批量展开map与 while 循环嵌套工作流WorkGraph 可作为另一个 WorkGraph 的节点结果收集通过 outputs 统一收集最终结果可视化可导出图结构用于检查依赖关系二、安装与环境准备2.1 前置条件aiida-workgraph 依赖 AiiDA 核心因此需要先安装并初始化 AiiDA 环境。# 建议使用独立虚拟环境 python -m venv aiida-env source aiida-env/bin/activate 安装 AiiDA 核心 pip install aiida-core 安装 aiida-workgraph pip install aiida-workgraph2.2 初始化 AiiDA 配置# 创建并启动 AiiDA 配置文件 verdi quicksetup --profile workgraph-demo 验证安装 verdi status python -c import aiida_workgraph; print(aiida_workgraph.version)2.3 可选依赖aiida-quantumespresso如需运行 Quantum ESPRESSO 计算。jupyter如需在 Notebook 中交互式构建工作流。graphviz如需导出图结构图片。三、核心语法与参数3.1 创建 WorkGraphfrom aiida_workgraph import WorkGraph wg WorkGraph(namemy_first_workgraph)常用参数name工作流名称用于标识和查询。3.2 添加节点# 添加 Python 函数节点 wg.add_task(my_function, nametask1, x1, y2) 添加 AiiDA 计算节点 wg.add_task(SomeCalcJob, namecalc1, parametersparams)常用参数name节点唯一标识。其余关键字参数对应节点的输入端口。3.3 建立依赖关系# 用输出端口连接输入端口 wg.add_link(task1.outputs.result, task2.x)3.4 运行与获取结果wg.run() result wg.outputs.result.value3.5 常用参数速查表参数作用name节点或工作流名称outputs声明节点输出端口metadata节点元信息如描述、标签wait是否等待节点完成max_iterations循环最大迭代次数四、16 个实际应用案例案例 1最简单的两节点串联from aiida_workgraph import WorkGraph def add(a, b): return a b def multiply(x, y): return x * y wg WorkGraph(namecase1) wg.add_task(add, nameadd, a1, b2) wg.add_task(multiply, namemul, y10) wg.add_link(add.result, mul.x) wg.run() print(wg.outputs.mul.result.value) # 30案例 2并行节点wg WorkGraph(namecase2) wg.add_task(add, namet1, a1, b1) wg.add_task(add, namet2, a2, b2) wg.add_task(add, namet3, a3, b3) wg.run() print(wg.outputs.t1.result.value, wg.outputs.t2.result.value, wg.outputs.t3.result.value)案例 3数据节点作为输入from aiida.orm import Int wg WorkGraph(namecase3) wg.add_task(add, nameadd, aInt(5), bInt(7)) wg.run() print(wg.outputs.add.result.value) # 12案例 4条件分支def check(value): return value 10 wg WorkGraph(namecase4) wg.add_task(check, namecheck, value15) wg.add_task(add, namebig, a100, b1) wg.add_task(add, namesmall, a1, b1) wg.add_link(check.result, big._condition) wg.add_link(check.result, small._condition, invertTrue) wg.run()案例 5列表映射mapwg WorkGraph(namecase5) wg.add_task(add, nameadd, a[1, 2, 3], b10) wg.run() print(wg.outputs.add.result.value) # [11, 12, 13]案例 6while 循环def increment(x): return x 1 wg WorkGraph(namecase6) wg.add_task(increment, nameinc, x0) wg.add_task(increment, nameloop, x0) wg.add_link(loop.result, loop.x) wg.run(max_iterations5)案例 7嵌套 WorkGraphinner WorkGraph(nameinner) inner.add_task(add, nameadd, a1, b2) outer WorkGraph(nameouter) outer.add_task(inner, namesub) outer.run() print(outer.outputs.sub.add.result.value)案例 8调用 AiiDA CalcJobfrom aiida.calculations.arithmetic.add import ArithmeticAddCalculation wg WorkGraph(namecase8) wg.add_task(ArithmeticAddCalculation, nameadd_calc, xInt(2), yInt(3)) wg.run() print(wg.outputs.add_calc.sum.value) # 5案例 9多输入聚合def aggregate(*args): return sum(args) wg WorkGraph(namecase9) wg.add_task(add, namea, a1, b1) wg.add_task(add, nameb, a2, b2) wg.add_task(aggregate, nameagg) wg.add_link(a.result, agg.args) wg.add_link(b.result, agg.args) wg.run()案例 10错误处理与重试def risky(x): if x 0: raise ValueError(negative input) return x * 2 wg WorkGraph(namecase10) wg.add_task(risky, namerisky, x5, metadata{retry: 3}) wg.run()案例 11动态添加节点wg WorkGraph(namecase11) wg.add_task(add, namebase, a1, b1) for i in range(3): wg.add_task(add, namefextra_{i}, ai, b10) wg.run()案例 12结果导出到文件import json wg WorkGraph(namecase12) wg.add_task(add, nameadd, a3, b4) wg.run() with open(result.json, w) as f: json.dump({result: wg.outputs.add.result.value}, f)案例 13与 StructureData 结合from aiida.orm import StructureData structure StructureData(cell[[3.0, 0, 0], [0, 3.0, 0], [0, 0, 3.0]]) structure.append_atom(position(0, 0, 0), symbolsSi) wg WorkGraph(namecase13) wg.add_task(SomeCalcJob, namecalc, structurestructure) wg.run()案例 14多工作流批量执行results [] for i in range(5): wg WorkGraph(namefbatch_{i}) wg.add_task(add, nameadd, ai, bi) wg.run() results.append(wg.outputs.add.result.value) print(results)案例 15图结构可视化wg WorkGraph(namecase15) wg.add_task(add, namet1, a1, b2) wg.add_task(multiply, namet2, y3) wg.add_link(t1.result, t2.x) # 导出为 graphviz 格式 wg.to_graphviz(workgraph.dot)案例 16完整数据处理流水线def load_data(path): with open(path) as f: return [float(line) for line in f] def normalize(data): total sum(data) return [x / total for x in data] def summarize(data): return {count: len(data), max: max(data), min: min(data)} wg WorkGraph(namecase16) wg.add_task(load_data, nameload, pathdata.txt) wg.add_task(normalize, namenorm) wg.add_task(summarize, namesum) wg.add_link(load.result, norm.data) wg.add_link(norm.result, sum.data) wg.run() print(wg.outputs.sum.result.value)五、常见错误与使用注意事项5.1 常见错误错误信息原因解决方法ProfileNotSet未初始化 AiiDA 配置执行verdi quicksetupValueError: socket not found端口名拼写错误检查节点输出端口名称KeyError: outputs节点尚未运行完成先调用wg.run()TypeError: unhashable输入类型不被支持转换为 AiiDA Data 类型RuntimeError: max iterations循环未收敛增大max_iterations或检查终止条件5.2 使用注意事项先初始化 AiiDA所有操作都依赖 AiiDA profile务必先完成配置。节点命名唯一同一 WorkGraph 内节点名不能重复。端口类型匹配连接时确保输出类型与输入类型兼容。循环需设上限while 循环必须设置max_iterations避免死循环。大任务注意资源并行节点过多时需关注计算资源与数据库压力。结果及时导出运行结束后尽快导出结果避免数据丢失。版本兼容aiida-workgraph 与 aiida-core 版本需匹配升级前查阅发布说明。六、总结aiida-workgraph 以图结构为核心把任务编排、数据流转和动态控制统一在 Python 代码中既适合快速原型验证也能支撑较复杂的科学计算流水线。掌握节点添加、端口连接、条件分支与循环这几类基本模式后大多数工作流都可以用少量代码清晰表达。建议从本文的简单案例入手逐步过渡到嵌套工作流与真实 CalcJob 集成。《AI提示工程必知必会》为读者提供了丰富的AI提示工程知识与实战技能主要包括各类提示词的应用如问答式、指令式、状态类、建议式、安全类和感谢类提示词以及如何通过实战演练掌握提示词的使用技巧使用提示词进行文本摘要、改写重述、语法纠错、机器翻译等语言处理任务以及在数据挖掘、程序开发等领域的应用AI在绘画创作上的应用百度文心一言和阿里通义大模型这两大智能平台的特性与功能以及市场调研中提示词的实战应用。通过阅读《AI提示工程必知必会》读者可掌握如何有效利用AI提示工程提升工作效率创新工作流程并在职场中脱颖而出。 
长河君 · 2026/11/3 10:45:38 · 阅读 1.2k aiida-workgraph 完全指南及实战案例
摘要本文系统介绍 Python 包 aiida-workgraph 的定位与核心功能讲解安装方式、WorkGraph 的语法与常用参数并通过 16 个由浅入深的实际案例演示任务编排、数据流转、条件分支、动态循环、错误处理与结果导出等典型用法最后总结常见错误与使用注意事项。一、aiida-workgraph 是什么aiida-workgraph 是 AiiDA 生态中用于构建和运行工作流图的 Python 库。它把工作流抽象为「节点 边」的有向图节点代表一个计算任务、一个 Python 函数或一个数据对象边代表数据依赖关系。相比 AiiDA 原生的 WorkChainaiida-workgraph 更强调动态构建与交互式编排适合需要根据中间结果动态决定后续步骤的场景。1.1 核心设计理念图即工作流用 Python 代码描述节点与依赖图结构在运行时才最终确定。数据流驱动节点之间的连接通过输入输出端口socket建立数据自动流转。可持久化基于 AiiDA 的 provenance 机制每个节点的执行记录都可追溯。可交互支持在 Jupyter Notebook 中逐步构建、运行和检查工作流。1.2 主要功能特性功能说明任务节点支持 AiiDA CalcJob、WorkChain、Python 函数作为节点数据节点支持 AiiDA Data 类型Int、Float、Dict、Structure 等动态分支根据运行时结果决定后续节点是否执行循环与映射支持对列表批量展开map与 while 循环嵌套工作流WorkGraph 可作为另一个 WorkGraph 的节点结果收集通过 outputs 统一收集最终结果可视化可导出图结构用于检查依赖关系二、安装与环境准备2.1 前置条件aiida-workgraph 依赖 AiiDA 核心因此需要先安装并初始化 AiiDA 环境。# 建议使用独立虚拟环境 python -m venv aiida-env source aiida-env/bin/activate 安装 AiiDA 核心 pip install aiida-core 安装 aiida-workgraph pip install aiida-workgraph2.2 初始化 AiiDA 配置# 创建并启动 AiiDA 配置文件 verdi quicksetup --profile workgraph-demo 验证安装 verdi status python -c import aiida_workgraph; print(aiida_workgraph.version)2.3 可选依赖aiida-quantumespresso如需运行 Quantum ESPRESSO 计算。jupyter如需在 Notebook 中交互式构建工作流。graphviz如需导出图结构图片。三、核心语法与参数3.1 创建 WorkGraphfrom aiida_workgraph import WorkGraph wg WorkGraph(namemy_first_workgraph)常用参数name工作流名称用于标识和查询。3.2 添加节点# 添加 Python 函数节点 wg.add_task(my_function, nametask1, x1, y2) 添加 AiiDA 计算节点 wg.add_task(SomeCalcJob, namecalc1, parametersparams)常用参数name节点唯一标识。其余关键字参数对应节点的输入端口。3.3 建立依赖关系# 用输出端口连接输入端口 wg.add_link(task1.outputs.result, task2.x)3.4 运行与获取结果wg.run() result wg.outputs.result.value3.5 常用参数速查表参数作用name节点或工作流名称outputs声明节点输出端口metadata节点元信息如描述、标签wait是否等待节点完成max_iterations循环最大迭代次数四、16 个实际应用案例案例 1最简单的两节点串联from aiida_workgraph import WorkGraph def add(a, b): return a b def multiply(x, y): return x * y wg WorkGraph(namecase1) wg.add_task(add, nameadd, a1, b2) wg.add_task(multiply, namemul, y10) wg.add_link(add.result, mul.x) wg.run() print(wg.outputs.mul.result.value) # 30案例 2并行节点wg WorkGraph(namecase2) wg.add_task(add, namet1, a1, b1) wg.add_task(add, namet2, a2, b2) wg.add_task(add, namet3, a3, b3) wg.run() print(wg.outputs.t1.result.value, wg.outputs.t2.result.value, wg.outputs.t3.result.value)案例 3数据节点作为输入from aiida.orm import Int wg WorkGraph(namecase3) wg.add_task(add, nameadd, aInt(5), bInt(7)) wg.run() print(wg.outputs.add.result.value) # 12案例 4条件分支def check(value): return value 10 wg WorkGraph(namecase4) wg.add_task(check, namecheck, value15) wg.add_task(add, namebig, a100, b1) wg.add_task(add, namesmall, a1, b1) wg.add_link(check.result, big._condition) wg.add_link(check.result, small._condition, invertTrue) wg.run()案例 5列表映射mapwg WorkGraph(namecase5) wg.add_task(add, nameadd, a[1, 2, 3], b10) wg.run() print(wg.outputs.add.result.value) # [11, 12, 13]案例 6while 循环def increment(x): return x 1 wg WorkGraph(namecase6) wg.add_task(increment, nameinc, x0) wg.add_task(increment, nameloop, x0) wg.add_link(loop.result, loop.x) wg.run(max_iterations5)案例 7嵌套 WorkGraphinner WorkGraph(nameinner) inner.add_task(add, nameadd, a1, b2) outer WorkGraph(nameouter) outer.add_task(inner, namesub) outer.run() print(outer.outputs.sub.add.result.value)案例 8调用 AiiDA CalcJobfrom aiida.calculations.arithmetic.add import ArithmeticAddCalculation wg WorkGraph(namecase8) wg.add_task(ArithmeticAddCalculation, nameadd_calc, xInt(2), yInt(3)) wg.run() print(wg.outputs.add_calc.sum.value) # 5案例 9多输入聚合def aggregate(*args): return sum(args) wg WorkGraph(namecase9) wg.add_task(add, namea, a1, b1) wg.add_task(add, nameb, a2, b2) wg.add_task(aggregate, nameagg) wg.add_link(a.result, agg.args) wg.add_link(b.result, agg.args) wg.run()案例 10错误处理与重试def risky(x): if x 0: raise ValueError(negative input) return x * 2 wg WorkGraph(namecase10) wg.add_task(risky, namerisky, x5, metadata{retry: 3}) wg.run()案例 11动态添加节点wg WorkGraph(namecase11) wg.add_task(add, namebase, a1, b1) for i in range(3): wg.add_task(add, namefextra_{i}, ai, b10) wg.run()案例 12结果导出到文件import json wg WorkGraph(namecase12) wg.add_task(add, nameadd, a3, b4) wg.run() with open(result.json, w) as f: json.dump({result: wg.outputs.add.result.value}, f)案例 13与 StructureData 结合from aiida.orm import StructureData structure StructureData(cell[[3.0, 0, 0], [0, 3.0, 0], [0, 0, 3.0]]) structure.append_atom(position(0, 0, 0), symbolsSi) wg WorkGraph(namecase13) wg.add_task(SomeCalcJob, namecalc, structurestructure) wg.run()案例 14多工作流批量执行results [] for i in range(5): wg WorkGraph(namefbatch_{i}) wg.add_task(add, nameadd, ai, bi) wg.run() results.append(wg.outputs.add.result.value) print(results)案例 15图结构可视化wg WorkGraph(namecase15) wg.add_task(add, namet1, a1, b2) wg.add_task(multiply, namet2, y3) wg.add_link(t1.result, t2.x) # 导出为 graphviz 格式 wg.to_graphviz(workgraph.dot)案例 16完整数据处理流水线def load_data(path): with open(path) as f: return [float(line) for line in f] def normalize(data): total sum(data) return [x / total for x in data] def summarize(data): return {count: len(data), max: max(data), min: min(data)} wg WorkGraph(namecase16) wg.add_task(load_data, nameload, pathdata.txt) wg.add_task(normalize, namenorm) wg.add_task(summarize, namesum) wg.add_link(load.result, norm.data) wg.add_link(norm.result, sum.data) wg.run() print(wg.outputs.sum.result.value)五、常见错误与使用注意事项5.1 常见错误错误信息原因解决方法ProfileNotSet未初始化 AiiDA 配置执行verdi quicksetupValueError: socket not found端口名拼写错误检查节点输出端口名称KeyError: outputs节点尚未运行完成先调用wg.run()TypeError: unhashable输入类型不被支持转换为 AiiDA Data 类型RuntimeError: max iterations循环未收敛增大max_iterations或检查终止条件5.2 使用注意事项先初始化 AiiDA所有操作都依赖 AiiDA profile务必先完成配置。节点命名唯一同一 WorkGraph 内节点名不能重复。端口类型匹配连接时确保输出类型与输入类型兼容。循环需设上限while 循环必须设置max_iterations避免死循环。大任务注意资源并行节点过多时需关注计算资源与数据库压力。结果及时导出运行结束后尽快导出结果避免数据丢失。版本兼容aiida-workgraph 与 aiida-core 版本需匹配升级前查阅发布说明。六、总结aiida-workgraph 以图结构为核心把任务编排、数据流转和动态控制统一在 Python 代码中既适合快速原型验证也能支撑较复杂的科学计算流水线。掌握节点添加、端口连接、条件分支与循环这几类基本模式后大多数工作流都可以用少量代码清晰表达。建议从本文的简单案例入手逐步过渡到嵌套工作流与真实 CalcJob 集成。《AI提示工程必知必会》为读者提供了丰富的AI提示工程知识与实战技能主要包括各类提示词的应用如问答式、指令式、状态类、建议式、安全类和感谢类提示词以及如何通过实战演练掌握提示词的使用技巧使用提示词进行文本摘要、改写重述、语法纠错、机器翻译等语言处理任务以及在数据挖掘、程序开发等领域的应用AI在绘画创作上的应用百度文心一言和阿里通义大模型这两大智能平台的特性与功能以及市场调研中提示词的实战应用。通过阅读《AI提示工程必知必会》读者可掌握如何有效利用AI提示工程提升工作效率创新工作流程并在职场中脱颖而出。