Emergent’s built-in topologies — complete, cycle, and wheel — cover common cases, but you often need a specific graph structure. Use set_graph() to inject any nx.Graph before running.
See Graph structures for details on the built-in topologies, and Agent model for the full simulation lifecycle.
The critical pattern for custom graphs
initialize_graph() always builds a new graph from num_nodes and graph_type, overwriting whatever was previously set with set_graph(). This means you cannot call set_graph() and then initialize_graph() — doing so discards your custom topology.
The correct pattern is to skip initialize_graph() entirely and seed the nodes manually:
Build your graph
Construct the nx.Graph using any NetworkX generator or by hand.import networkx as nx
from emergent.main import AgentModel
import random
G = nx.barabasi_albert_graph(50, 3)
Define your behavior functions
Write your initial_data_function and timestep_function as normal.def initial_data(model):
return {"opinion": random.random()}
def timestep(model):
graph = model.get_graph()
updates = {}
for node in graph.nodes():
neighbors = list(graph.neighbors(node))
if neighbors:
partner = random.choice(neighbors)
avg = (graph.nodes[node]["opinion"] + graph.nodes[partner]["opinion"]) / 2
updates[node] = avg
for node, opinion in updates.items():
graph.nodes[node]["opinion"] = opinion
Create the model and attach functions
model = AgentModel()
model.set_initial_data_function(initial_data)
model.set_timestep_function(timestep)
Set the graph and seed nodes manually
Set the custom graph, then call your initial_data_function for each node yourself. This replaces the seeding step that initialize_graph() would otherwise handle.model.set_graph(G)
for node in G.nodes():
G.nodes[node].update(initial_data(model))
Do NOT call model.initialize_graph() after this step. It will build a new built-in graph and overwrite G.
Configure convergence parameters and run
model.update_parameters({
"convergence_data_key": "opinion",
"convergence_std_dev": 0.05,
})
steps = model.run_to_convergence()
print(f"Converged in {steps} timesteps")
Examples
Barabási–Albert
Watts–Strogatz
Erdős–Rényi
A scale-free graph where a small number of nodes accumulate most connections — typical of social networks.import networkx as nx
from emergent.main import AgentModel
import random
G = nx.barabasi_albert_graph(n=100, m=3)
def initial_data(model):
return {"opinion": random.random()}
def timestep(model):
graph = model.get_graph()
updates = {}
for node in graph.nodes():
neighbors = list(graph.neighbors(node))
if neighbors:
partner = random.choice(neighbors)
avg = (graph.nodes[node]["opinion"] + graph.nodes[partner]["opinion"]) / 2
updates[node] = avg
for node, opinion in updates.items():
graph.nodes[node]["opinion"] = opinion
model = AgentModel()
model.set_initial_data_function(initial_data)
model.set_timestep_function(timestep)
model.set_graph(G)
for node in G.nodes():
G.nodes[node].update(initial_data(model))
model.update_parameters({
"convergence_data_key": "opinion",
"convergence_std_dev": 0.05,
})
steps = model.run_to_convergence()
print(f"Converged in {steps} timesteps")
A small-world graph with high clustering and short average path lengths.import networkx as nx
from emergent.main import AgentModel
import random
# 80 nodes, each initially connected to 4 nearest neighbors, rewiring probability 0.1
G = nx.watts_strogatz_graph(n=80, k=4, p=0.1)
def initial_data(model):
return {"opinion": random.gauss(model["initial_mean"], 0.2)}
def timestep(model):
graph = model.get_graph()
updates = {}
for node in graph.nodes():
neighbors = list(graph.neighbors(node))
if neighbors:
partner = random.choice(neighbors)
avg = (graph.nodes[node]["opinion"] + graph.nodes[partner]["opinion"]) / 2
updates[node] = avg
for node, opinion in updates.items():
graph.nodes[node]["opinion"] = opinion
model = AgentModel()
model["initial_mean"] = 0.5
model.set_initial_data_function(initial_data)
model.set_timestep_function(timestep)
model.set_graph(G)
for node in G.nodes():
G.nodes[node].update(initial_data(model))
model.update_parameters({
"convergence_data_key": "opinion",
"convergence_std_dev": 0.05,
})
steps = model.run_to_convergence()
A random graph where each edge is included independently with probability p.import networkx as nx
from emergent.main import AgentModel
import random
# 60 nodes, edge probability 0.08
G = nx.erdos_renyi_graph(n=60, p=0.08)
def initial_data(model):
return {"opinion": random.random()}
def timestep(model):
graph = model.get_graph()
updates = {}
for node in graph.nodes():
neighbors = list(graph.neighbors(node))
if neighbors:
partner = random.choice(neighbors)
avg = (graph.nodes[node]["opinion"] + graph.nodes[partner]["opinion"]) / 2
updates[node] = avg
for node, opinion in updates.items():
graph.nodes[node]["opinion"] = opinion
model = AgentModel()
model.set_initial_data_function(initial_data)
model.set_timestep_function(timestep)
model.set_graph(G)
for node in G.nodes():
G.nodes[node].update(initial_data(model))
model.update_parameters({
"convergence_data_key": "opinion",
"convergence_std_dev": 0.05,
})
steps = model.run_to_convergence()
Verifying the graph was set correctly
Call model.get_graph() after set_graph() to confirm the expected graph is in place:
G = nx.barabasi_albert_graph(50, 3)
model.set_graph(G)
check = model.get_graph()
print(check is G) # True
print(check.number_of_nodes()) # 50
print(check.number_of_edges()) # depends on BA model
Error handling
set_graph() raises an exception if the argument is not a nx.Graph instance:
model.set_graph(42) # Exception: The passed parameter is not a graph object.
None to clear the graph without raising an error:
model.set_graph(None) # clears the stored graph
set_graph() accepts any nx.Graph instance or subclass (including nx.DiGraph). The check in the source is isinstance(graph, nx.Graph), so subclasses pass validation. However, Emergent’s built-in topologies and the bundled helper functions are designed for undirected graphs — behaviors relying on graph.neighbors() work correctly on undirected graphs. Test directed graph behavior carefully.