Example Workflows | NVIDIA Alchemi NIMs

Use the Quickstart Guide to launch the NIM before running the following examples.

For the request format and required atomic data fields, refer to API Reference. Note the BMDAtomicData fields (atoms) and MDConfig fields (config) tables.

Install the Python libraries used in the following code samples:

pip install requests aiohttp numpy

Supported Ensembles

Configure the ensemble using the config object:

NVE (Microcanonical) - Constant Energy:

config = {
    "nvt": False,
    "npt": False,
    "dt": 0.5,           # Use a smaller timestep for NVE
    "md_time_max": 1.0
}

NVT (Canonical) - Langevin Thermostat:

config = {
    "temperature": 300.0,  # Target temperature in K
    "nvt": True,
    "friction": 1.0,       # Langevin friction in ps^-1
    "dt": 1.0,             # fs
    "md_time_max": 10.0
}

NPT (Isothermal-Isobaric) - MC Barostat:

config = {
    "temperature": 300.0,
    "nvt": True,
    "npt": True,
    "pressure": 1.0,                # Target pressure in kBar
    "barostat_every": 25,           # Update barostat every N steps
    "barostat_anisotropic": False,  # Use isotropic barostat
    "dt": 1.0,                      # fs
    "md_time_max": 10.0
}

Restart Simulations

Continue a simulation from a previous state by using the final snapshot as shown in the following code sample:

import requests
server_url = "http://localhost:8000"
# First simulation run
initial_request = {
    "atoms": {
        "coord": [0.0, 0.0, 0.0, 1.5, 0.0, 0.0],
        "numbers": [6, 6],
        "cell": [10.0, 0.0, 0.0, 0.0, 10.0, 0.0, 0.0, 0.0, 10.0],
        "pbc": [True, True, True]
    },
    "config": {
        "temperature": 300.0,
        "dt": 1.0,
        "md_time_max": 1.0,
        "nvt": True,
        "save_interval": 10
    }
}
response = requests.post(f"{server_url}/v1/infer", json=initial_request)
result = response.json()
# Get final state from first run
final_config = result["config"]
final_snapshot = result["trajectory"][-1]
# Create restart request
restart_request = {
    "atoms": {
        "coord": final_snapshot["coord"],
        "velocity": final_snapshot["velocity"],  # Include velocities for restart
        "numbers": [6, 6],
        "cell": final_snapshot.get("cell", initial_request["atoms"]["cell"]),
        "pbc": [True, True, True]
    },
    "config": {
        "temperature": 300.0,
        "dt": 1.0,
        "nvt": True,
        "save_interval": 10,
        "istep": final_config["istep"],             # Continue step count
        "md_time": final_config["md_time"],         # Continue from last time
        "md_time_max": final_config["md_time"] + 1.0  # Run 1 more ps
    }
}
response = requests.post(f"{server_url}/v1/infer", json=restart_request)
continuation = response.json()
print(f"Continued simulation: {len(continuation['trajectory'])} new snapshots")

Concurrent Requests

The inference API accepts multiple concurrent requests; the server processes them and returns responses in order. There is no dedicated async or polling API. You can achieve concurrent behavior by issuing multiple requests (for example, from separate threads or processes).

Concurrent behavior example (Python)

import concurrent.futures
import requests
def run_md(payload):
    return requests.post("http://localhost:8000/v1/infer", json=payload)
# Example: run three MD requests concurrently
base = {
    "atoms": {"coord": [0,0,0,1.5,0,0], "numbers": [6,6], "cell": [10,0,0,0,10,0,0,0,10], "pbc": [True,True,True]},
    "config": {"temperature": 300, "dt": 1.0, "md_time_max": 0.5, "nvt": True, "save_interval": 5}
}
payloads = [base for _ in range(3)]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(run_md, payloads))
for r in results:
    r.raise_for_status()
    print(r.json()["status"])

Asynchronous example (Python)

Use asyncio with an async HTTP client. Install aiohttp (for example, pip install aiohttp) and run the following:

import asyncio
import aiohttp
async def run_md(session, payload):
    async with session.post("http://localhost:8000/v1/infer", json=payload) as resp:
        resp.raise_for_status()
        return await resp.json()
async def main():
    base = {
        "atoms": {"coord": [0,0,0,1.5,0,0], "numbers": [6,6], "cell": [10,0,0,0,10,0,0,0,10], "pbc": [True,True,True]},
        "config": {"temperature": 300, "dt": 1.0, "md_time_max": 0.5, "nvt": True, "save_interval": 5}
    }
    payloads = [base for _ in range(3)]
    async with aiohttp.ClientSession() as session:
        results = await asyncio.gather(*[run_md(session, p) for p in payloads])
    for r in results:
        print(r["status"])
asyncio.run(main())

Analyzing Trajectories

The trajectory output can be analyzed to compute thermodynamic properties, as shown in the following Python code sample:

import numpy as np
# Physical constants
KE_CONV = 103.64269667160806  # amu·Å²/fs² → eV
BOLTZ_EV_K = 8.617333262145179e-05  # eV/K
def compute_temperature(snapshot):
    """Compute instantaneous temperature from velocities."""
    masses = np.asarray(snapshot["mass"], dtype=float)
    velocity = np.asarray(snapshot["velocity"], dtype=float).reshape(-1, 3)
    e_kin = 0.5 * np.sum(masses[:, None] * velocity**2) * KE_CONV
    dof = 3 * len(masses) - 3  # Degrees of freedom (minus COM motion)
    return 2.0 * e_kin / (dof * BOLTZ_EV_K)
# Example usage with trajectory
for snap in result["trajectory"]:
    T = compute_temperature(snap)
    print(f"Step {snap['istep']}: T = {T:.1f} K, E_pot = {snap['energy']:.4f} eV")