The GISMOL Developer Guide
A comprehensive reference for building intelligent systems with Constrained Object Hierarchies and the GISMOL library. Every module, every class, every contract.
01Introduction
What GISMOL is
GISMOL — General Intelligent Systems Modelling Language — is a Python-embedded DSL for modelling intelligent systems. It implements Constrained Object Hierarchies, a mathematically grounded framework unifying symbolic reasoning, neural learning, dynamical systems, and constraint-driven regulation into a single architecture.
When you write GISMOL code, you are not writing an application. You are declaring the mathematical structure of an intelligent system. The toolkit then gives you, for free:
- Hybrid simulation (continuous flow + discrete events)
- Constraint enforcement with real-time monitors
- Stability certificates
- Compositional algebra (products, coproducts, exponentials)
- Learning loops (constrained RL, hierarchical credit assignment)
- Reasoning engines
- Natural-language input and output
- Validation and soundness guarantees
Who this guide is for
- Comfortable with Python (functions, classes, lambdas, decorators)
- Have a system you want to make intelligent
- Understand state, action, objective, and constraint
You do not need prior exposure to category theory, Lyapunov stability, or free-energy minimisation.
How to read this guide
- §2–§4 get you running in fifteen minutes.
- §5–§11 are the core reference: one chapter per component.
- §12–§17 cover advanced capabilities.
- §18–§23 are operational.
- §24 builds a complete system.
- §25–§26 are quick reference.
02Installation and project setup
Requirements
GISMOL requires Python 3.9 or newer and has zero mandatory runtime dependencies. Everything you need is in the standard library.
pip install -e ".[fast]" # numpy acceleration
pip install -e ".[plot]" # networkx + matplotlib
pip install -e ".[nlp]" # sentence-transformers
Installing
pip install gismol
# or from source
git clone https://github.com/hongxueharriswang/gismol-lib.git
cd gismol
pip install -e ".[dev]"
python -c "import gismol; print(gismol.__version__)"
# 0.1.0
Recommended project layout
my_intelligent_system/
├── systems/ # One module per intelligent system
│ ├── __init__.py
│ ├── robot.py
│ └── scheduler.py
├── constraints/ # Reusable constraint factories
│ └── safety.py
├── daemons/ # Custom daemon implementations
├── neural/ # Neural architectures
├── configs/ # JSON/YAML configuration
├── tests/
├── requirements.txt
└── README.md
build_<name>() -> COHObject
function. This makes systems testable, serialisable, and composable.
Recommended import style
from gismol import COHObject, Attribute, Method
from gismol.core.constraints import (
IdentityConstraint, TriggerConstraint, GoalConstraint,
)
from gismol.neural.components import MLPComponent, PolicyComponent
from gismol.reasoners import GeneralReasoner, Query
Never use from gismol import * — the framework is deliberately explicit.
03Mental model — how to think in COH
Intelligence is constrained optimisation
An intelligent system is not a program that computes outputs from inputs. It is an optimiser that maximises objectives while satisfying invariants:
intelligence = max_π E[Σ γ^t r_t] − λ · KL(p_π ‖ p_constraints)
Pursue goals, but stay within bounds. Everything in GISMOL serves this equation.
Structure precedes behaviour
- What are the parts? — the hierarchy (C)
- What do the parts measure? — attributes (A)
- What can they do? — methods (M)
- What must always be true? — identity constraints (I)
- What reacts to events? — triggers (T)
- What is being optimised? — goals (G)
Only then add neural components (N), embeddings (E), and daemons (D).
Everything is a COHObject
A COHObject can represent a physical device, a logical subsystem, an abstract concept, or a whole society. Whether the system is 3 objects deep or 30, the interface is identical.
Constraints are first-class
Most frameworks treat constraints as afterthoughts. In GISMOL, constraints are declarative objects that participate in simulation, learning, reasoning, and validation.
04Your first intelligent system
Let's build a smart light that maintains target brightness while never exceeding power limits.
from gismol import COHObject, Attribute, Method
from gismol.core.constraints import (
IdentityConstraint, TriggerConstraint, GoalConstraint,
)
from gismol.core.daemons import InvariantDaemon, BoundaryDaemon
def build_smart_light() -> COHObject:
light = COHObject("smart_light", description="Adaptive LED controller")
# A — attributes
light.add_attribute(Attribute("brightness", 0.5, bounds=(0.0, 1.0)))
light.add_attribute(Attribute("power_watts", 5.0, bounds=(0.0, 20.0)))
light.add_attribute(Attribute("target", 0.8))
# M — methods
light.add_method(Method(
"increase",
lambda s, a=None: ({**s,
"brightness": min(1.0, s["brightness"] + 0.1),
"power_watts": 20.0 * min(1.0, s["brightness"] + 0.1)},
0.1)))
# I — invariants
light.add_identity_constraint(IdentityConstraint(
"power_limit", lambda s: s["power_watts"] <= 15.0,
))
# T — triggers
light.add_trigger_constraint(TriggerConstraint(
"night_boost", "ambient_dark",
condition=lambda s: s["brightness"] < 0.9,
action="increase", priority=1,
))
# G — goals
light.add_goal_constraint(GoalConstraint(
"brightness_match",
lambda s: -abs(s["brightness"] - s["target"]),
direction="max", weight=2.0,
))
# D — daemons
light.add_daemon(InvariantDaemon(constraints=list(light.I)))
light.add_daemon(BoundaryDaemon({
"brightness": (0.0, 1.0),
"power_watts": (0.0, 20.0),
}))
return light
light = build_smart_light()
print(light.component_report())
traj = light.simulate(steps=20, dt=0.1)
What just happened
- Trigger phase — check every trigger whose condition matches; execute the highest-priority action.
- Daemon phase — run every daemon; apply corrections.
- Continuous phase — integrate the flow
ds/dt = f(s, a, θ). - Record — append the resulting state and reward to a trajectory.
This example has no neural components, no embedding, no hierarchy, and no dynamics. That's fine — GISMOL is pay-for-what-you-need.
05Building hierarchies (Component C)
The composition operator
arm = COHObject("arm")
joint1 = arm.add_child(COHObject("joint1"))
joint2 = arm.add_child(COHObject("joint2"))
gripper = arm.add_child(COHObject("gripper"))
Depth guidance
| Depth | Role | Example |
|---|---|---|
| 0 | System | A whole warehouse |
| 1 | Subsystem | Picking zone, packing zone |
| 2 | Unit | A single robot |
| 3 | Component | Navigation, gripper, battery |
| 4 | Micro-component | Kalman filter, force controller |
Traversal
fleet = COHObject("fleet")
zone = fleet.add_child(COHObject("zone"))
robot = zone.add_child(COHObject("robot"))
fleet.C # {'zone': <COHObject>}
fleet.children() # same
zone.child("robot") # <COHObject robot>
fleet.descendants() # [zone, robot]
fleet.topological_order() # [robot, zone, fleet]
fleet.component_dag() # {…}
Cycle safety
a = COHObject("a"); b = COHObject("b")
a.add_child(b)
b.add_child(a) # raises CyclicHierarchyError
Naming
Use descriptive names, lower-case with underscores. Names appear in serialised output, log messages, and reasoning traces.
Removing children
arm.remove_child("j1")
06Attributes and state (Component A)
Declaring attributes
from gismol import Attribute
Attribute("temperature", 20.0, description="°C",
bounds=(-40.0, 150.0), space="real", unit="°C")
| Parameter | Meaning |
|---|---|
name | Identifier, used as key in state dict |
default | Initial value |
space | "real", "bool", "categorical", "vector" |
bounds | (low, high) used by BoundaryDaemon |
description | Human-readable note |
unit | Physical unit, for documentation |
Hierarchical state
arm = COHObject("arm")
arm.add_attribute(Attribute("reach", 1.0))
joint = arm.add_child(COHObject("joint"))
joint.add_attribute(Attribute("angle", 0.0))
arm.get("joint.angle") # 0.0
arm.set("joint.angle", 0.5)
arm.get("joint.angle") # 0.5
metadata.
07Methods and actions (Component M)
Signatures
Method("name", fn, description="",
preconditions=[], postconditions=[])
# Accepted signatures
fn(state) -> new_state | (new_state, reward)
fn(state, action) -> new_state | (new_state, reward)
Building a method
import math
def move_forward(state, action=None):
new = state.copy()
new["x"] += 0.5 * math.cos(new["heading"])
new["y"] += 0.5 * math.sin(new["heading"])
return new, 0.1
robot.add_method(Method("move_forward", move_forward,
description="advance 0.5 m along heading"))
Preconditions and postconditions
robot.add_method(Method(
"grasp",
lambda s, a=None: ({**s, "holding": True}, 1.0),
preconditions=[
lambda s: s["payload_kg"] <= 20.0,
lambda s: not s["holding"],
],
postconditions=[lambda s: s["holding"] is True],
))
MethodExecutionError if violated;
postconditions likewise.
Invoking methods
new_state, reward = robot.invoke("move_forward")
# From a trigger
robot.add_trigger_constraint(TriggerConstraint(
"auto_advance", "tick",
condition=lambda s: s["x"] < 10.0,
action="move_forward",
))
Purity
Prefer pure methods that do not mutate the input state. The toolkit never relies on mutation and copying is cheap for small states.
08Constraint language in depth (I, T, G)
Writing constraints — three ways
# (a) Lambda
IdentityConstraint("positive", lambda s: s["x"] > 0)
# (b) Expression tree
from gismol.core import expression as E
IdentityConstraint("positive", E.Var("x") > 0)
# (c) String — parsed safely
IdentityConstraint("positive", "x > 0")
Identity constraints (I)
IdentityConstraint(
name="temp_bounds",
predicate=lambda s: 0.0 <= s["temperature"] <= 100.0,
description="Operating temperature must stay within safe envelope",
weight=2.0,
margin=None,
)
When to use: physical limits, logical invariants, safety requirements, legal boundaries.
When not to use: for soft preferences use goals; for event responses use triggers.
Trigger constraints (T)
TriggerConstraint(
name="emergency_shutdown",
event="overheat",
condition=lambda s: s["temp"] > 90.0,
action="shutdown",
priority=10,
preserves=lambda s: s["temp"] < 100.0,
)
Multiple triggers may match the same event. They fire in descending priority order.
Goal constraints (G)
G_total(s, a, t) = Σ_j w_j · g_j(s, a, t) − λ R(θ)
GoalConstraint(
name="throughput",
objective=lambda s: s["items_per_hour"],
direction="max",
weight=2.0,
)
GoalConstraint(
name="energy_cost",
objective=lambda s: s["watts"],
direction="min",
weight=1.0,
)
The ConstraintSet container
from gismol.core.constraints import ConstraintSet
cs = ConstraintSet()
cs.add_identity(IdentityConstraint("i1", "x > 0"))
cs.add_trigger(TriggerConstraint("t1", "evt", "x > 5", "increase"))
cs.add_goal(GoalConstraint("g1", "x", direction="max"))
cs.check_identity({"x": 1}) # passes
cs.check_identity({"x": -1}) # raises ConstraintViolationError
cs.satisfies_identity({"x": 1}) # True
cs.violation_score({"x": -5}) # positive magnitude
cs.total_goal({"x": 3}) # -3.0
Expression language reference
| Category | Syntax | Example |
|---|---|---|
| Variables | identifiers, dotted paths | x, arm.joint.angle |
| Constants | numbers, strings, booleans | 1.5, "on", True |
| Arithmetic | + - * / % ** | (x + y) / 2 |
| Comparison | == != < <= > >= in not in | x >= 0 and x < 10 |
| Boolean | and or not | (x > 0) & (y < 5) |
| Functions | whitelisted calls | abs(x), min(a, b), sqrt(x) |
Not supported: attribute access, subscripting, lambdas, comprehensions, imports, assignment, eval, exec.
Register your own functions
from gismol.core import expression as E
@E.register_function("clamp")
def _clamp(x, lo, hi):
return max(lo, min(hi, x))
expr = E.parse_expression("clamp(x, 0, 100) > 50")
expr.evaluate({"x": 75}) # True
09Daemons in depth (Component D)
What daemons do
A daemon continuously monitors state and emits corrective actions. Unlike triggers, daemons are not tied to events — they run every step.
The daemon interface
class Daemon:
def __init__(self, name, description="", enabled=True): ...
def observe(self, state, t=0.0) -> Optional[CorrectiveAction]: ...
def monitor(self, trajectory) -> List[CorrectiveAction]: ...
Built-in daemons
| Daemon | Purpose |
|---|---|
InvariantDaemon | Reports identity violations |
BoundaryDaemon | Projects state into a box |
RateDaemon | Limits rate of change |
SafetyDaemon | Hazard-margin monitoring |
HomeostasisDaemon | Set-point regulation |
ProjectionDaemon | Custom projection onto feasible region |
BarrierDaemon | Control-barrier forward invariance |
PrecisionWeightedDaemon | Prediction-error minimisation |
EventDaemon | Edge-triggered action |
DaemonNetwork | Composite; highest severity wins |
Attaching daemons
from gismol.core.daemons import InvariantDaemon, BoundaryDaemon
robot.add_daemon(InvariantDaemon(constraints=list(robot.I)))
robot.add_daemon(BoundaryDaemon({
"x": (-50.0, 50.0),
"y": (-50.0, 50.0),
"payload_kg": (0.0, 25.0),
}))
robot.D # DaemonNetwork
robot.D.by_name("invariant") # the InvariantDaemon
robot.monitor() # highest-severity action
Homeostatic regulation
from gismol.core.daemon_impl import HomeostasisDaemon
battery.add_daemon(HomeostasisDaemon(
attribute="temperature",
setpoint=25.0,
gain=0.5,
tolerance=0.5,
))
Control-barrier supervision
ḣ(s) + α(h(s)) ≥ 0
from gismol.core.daemon_impl import BarrierDaemon
robot.add_daemon(BarrierDaemon(
barrier=lambda s: 1.0 - (s["x"] ** 2 + s["y"] ** 2) / 100.0,
alpha=lambda h: 0.5 * h,
intervention="emergency_stop",
))
Writing a custom daemon
from gismol.core.daemons import Daemon, CorrectiveAction
class ThermalThrottleDaemon(Daemon):
"""Reduce power when temperature approaches a limit."""
def __init__(self, attribute="temperature", limit=80.0, margin=10.0):
super().__init__("thermal_throttle")
self.attribute = attribute
self.limit = limit
self.margin = margin
def observe(self, state, t=0.0):
temp = state.get(self.attribute, 0.0)
if temp < self.limit - self.margin:
return None
severity = (temp - (self.limit - self.margin)) / self.margin
throttle = max(0.0, 1.0 - severity)
return self._emit(CorrectiveAction(
name="throttle",
payload={"power_limit": throttle},
daemon=self.name,
severity=severity,
message=f"temp={temp:.1f} → throttle={throttle:.2f}",
))
reactor.add_daemon(ThermalThrottleDaemon("core_temp", limit=900.0))
Severity and priority
When multiple daemons fire, the network returns the one with the highest severity. For critical safety daemons, use severity values well above 1.0 to guarantee dominance.
10Neural components (Component N)
Why neural components?
Identity constraints and triggers capture known rules. Neural components capture learned mappings. They live inside COHObjects so that learning is scoped and composable.
Neural component types
| Class | Role | Typical size |
|---|---|---|
MLPComponent | Generic f_θ : X → Y | 2–3 layers |
PerceptionComponent | Encode observations | input → 4–32 dims |
PredictorComponent | Predict next state | input → 1–4 outputs |
PolicyComponent | Softmax over discrete actions | input → n_actions |
ValueComponent | Approximate V(s) | input → 1 |
from gismol.neural.components import (
MLPComponent, PolicyComponent, ValueComponent, PredictorComponent,
)
robot.add_neural(
PolicyComponent("pick_policy", input_dim=6, n_actions=4, seed=42),
name="policy")
robot.add_neural(
ValueComponent("V", input_dim=6, seed=43),
name="value")
robot.add_neural(
PredictorComponent("next_pos", input_dim=6, output_dim=2, seed=44),
name="predictor")
Training
import random
from gismol.neural.optimizers import Adam
X = [[random.random() for _ in range(6)] for _ in range(500)]
Y = [[sum(x[:3]) - sum(x[3:])] for x in X]
predictor = robot.N["predictor"]
history = predictor.fit(
X, Y,
epochs=200, lr=0.05, batch_size=32,
optimizer=Adam(0.01), verbose=True,
)
print(f"final loss: {history[-1]:.6f}")
Optimisers
from gismol.neural.optimizers import (
SGD, Adam, RMSProp, AdaGrad, Momentum,
)
opt = Adam(lr=0.01, beta1=0.9, beta2=0.999)
For constrained problems, LagrangianOptimizer performs dual ascent on multipliers.
Parameter access
comp = robot.N["policy"]
flat = comp.theta # list[float]
comp.set_theta([0.0] * len(flat)) # reset
comp.n_parameters() # count
robot.parameters() # all params recursively
Best practices
- Seed everything. Neural components accept a
seed. - Keep networks small. Neural components live inside hierarchies.
- Train offline, deploy frozen. Freeze
thetaafter training. - Never let N override I. Neural outputs must respect identity constraints.
11Embeddings (Component E)
What embeddings are for
vec = coh.embed() # tuple[float, ...]
sim = coh1.similarity(coh2) # in [-1, 1]
dist = coh1.E.space.distance(coh1.embed(), coh2.embed())
Built-in embedders
| Embedder | Purpose |
|---|---|
HashEmbedder | Deterministic structural hashing (default) |
FeatureEmbedder | Selected attributes with normalisation |
CompositeEmbedder | Combines own + children embeddings |
NeuralEmbedder | Learned MLP-based projection |
AutoencoderEmbedder | Encoder/decoder pair |
SiameseEmbedder | Contrastive pairs |
ConstantEmbedder | Trivial zero vector |
Choosing an embedder
- Structural similarity →
HashEmbedderorCompositeEmbedder - Semantic similarity →
FeatureEmbedderwith normalisation - Learned task-specific →
NeuralEmbedder
FeatureEmbedder with ranges
from gismol.core.embedding import FeatureEmbedder
robot.set_embedder(FeatureEmbedder(
features=["x", "y", "heading", "payload_kg"],
ranges={
"x": (-100, 100),
"y": (-100, 100),
"heading": (-3.14, 3.14),
"payload_kg": (0, 25),
},
))
Composite embeddings
from gismol.core.embedding import CompositeEmbedder
fleet.set_embedder(CompositeEmbedder(
self_embedder=FeatureEmbedder(["total_throughput", "fleet_health"]),
child_weight=0.5,
mode="concat",
))
Similarity in practice
from gismol.nlp.similarity import SimilarityIndex
index = SimilarityIndex()
for coh in systems:
index.add(coh.name, coh.embed(), payload=coh)
for name, score, payload in index.query(target.embed(), k=5):
print(f"{name}: {score:.3f}")
12Hybrid dynamics — simulation and control
The hybrid model
- Continuous flow —
ds/dt = f(s, a, θ) - Discrete jumps —
s_{t+1} = g(s_t, a_t, θ) - Switching conditions — triggers decide when jumps occur
Attaching a flow
from gismol.core.dynamics import ContinuousFlow
reactor.flow = ContinuousFlow(
f=lambda s, a, th: {
"temperature": -0.01 * (s["temperature"] - 20.0) + 0.5 * s["power"],
"pressure": 0.05 * s["temperature"],
},
variables=["temperature", "pressure"],
)
The flow is integrated with classical Runge-Kutta 4.
Attaching a jump
from gismol.core.dynamics import DiscreteTransition
reactor.jump = DiscreteTransition(
g=lambda s, a, th: {**s, "temperature": 0.9 * s["temperature"]},
)
Stepping and simulating
state, info = reactor.step(action=None, dt=0.05, event="tick")
print(info)
# {'events': ['overheat'], 'corrections': ['project'], 'admissible': True}
traj = reactor.simulate(steps=200, dt=0.01,
policy=lambda s, t: "regulate")
Trajectory object
traj.times # [0.0, 0.01, 0.02, ...]
traj.states # [State, State, ...]
traj.actions
traj.rewards
traj.events
traj.final # last State
traj.cumulative_reward()
traj.discounted_return(gamma=0.99)
traj.to_dict() # serialisable
Well-posedness
system = reactor.to_hybrid_system()
report = system.lipschitz_report(reactor.state)
# {'well_posed': True, 'flow_lipschitz': 0.62, 'jump_lipschitz': None}
Policies
def bang_bang(state, t):
return "heat" if state["temperature"] < state["target"] else "cool"
traj = reactor.simulate(steps=100, policy=bang_bang)
Discrete events
traj = reactor.simulate(
steps=100,
events=["tick", None, "tick", "fault", None],
)
13Categorical composition
The three compositions
arm = build_arm()
camera = build_camera()
robot_head = arm.product(camera) # parallel
either = arm.coproduct(camera) # alternative
transformer = camera.exponential(arm) # function object
Semantics of the product
- Merges attributes, methods, neural components
- Conjoins identity constraints (
I₁ ∧ I₂) - Concatenates triggers and goals
- Attaches both as named children
The COHCategory
from gismol.categorical import COHCategory
cat = COHCategory()
cat.register(arm)
cat.register(camera)
prod = cat.product(arm, camera)
coprod = cat.coproduct(arm, camera)
exp = cat.exponential(arm, camera)
cat.is_cartesian_closed() # True
Morphisms
from gismol.categorical import COHMorphism
mor = COHMorphism(source=small_robot, target=big_robot, name="upgrade")
mor.preserves() # True if the map preserves all structure
mor.violations # list of problems
Functors
from gismol.categorical import ComponentFunctor
F = ComponentFunctor()
dag = F.map_object(robot)
dag.nodes # ['robot', 'robot/nav', ...]
dag.edges # [('robot', 'robot/nav'), ...]
14Stability analysis
Lyapunov functions
from gismol.stability import lyapunov_from_constraints, goal_lyapunov
V_constraints = lyapunov_from_constraints(reactor) # sum of violations
V_goal = goal_lyapunov(reactor) # negative goal value
The certificate
from gismol.stability import ConstraintStabilityAnalyzer
analyzer = ConstraintStabilityAnalyzer(reactor)
cert = analyzer.analyze(states=sample_states)
print(cert)
# LyapunovCertificate(asymptotically stable, rate=0.0413, daemons=True)
print(cert.to_dict())
# {'stable': True, 'asymptotic': True, 'rate': 0.0413,
# 'admissible_fraction': 1.0, 'samples': 100,
# 'worst_derivative': -0.0413, 'daemons_enforcing': True, 'notes': []}
Sampling states
import random
states = []
for _ in range(200):
states.append({
"temperature": random.uniform(20.0, 90.0),
"pressure": random.uniform(0.5, 5.0),
})
cert = analyzer.analyze(states=states)
Interpreting failures
- Look at
worst_derivative— it is positive. - Inspect which states contributed.
- Common causes:
- Goal sign is wrong
- Flow pushes state out of the admissible region
- Constraints are mutually inconsistent
15Free-energy correspondence
Generative models
from gismol.dynamical.free_energy import (
GenerativeModel, VariationalPosterior, FreeEnergy,
)
model = GenerativeModel(
prior={"cold": 0.4, "warm": 0.6},
likelihood={
"cold": {"sensor_low": 0.9, "sensor_high": 0.1},
"warm": {"sensor_low": 0.2, "sensor_high": 0.8},
},
)
q = VariationalPosterior({"cold": 0.3, "warm": 0.7})
fe = FreeEnergy(model)
print(fe.variational("sensor_low", q))
The COH ↔ FEP correspondence
from gismol.dynamical.free_energy import coh_free_energy_correspondence
mapping = coh_free_energy_correspondence(reactor)
# {'identity_constraints': [...], 'identity_role': 'structural priors p(s)',
# 'trigger_constraints': [...], 'trigger_role': 'conditional dependencies',
# 'goal_constraints': [...], 'goal_role': 'expected log-likelihood terms',
# 'daemons': [...], 'daemon_role': 'precision-weighted prediction-error minimisation'}
Scalar proxy
from gismol.dynamical.free_energy import free_energy_from_coh
fe = free_energy_from_coh(reactor, reactor.state)
print(f"free energy: {fe:.4f}")
16The natural-language front-end
What it does
from gismol.nlp import text_to_coh
reactor = text_to_coh(
"Define a thermal plant called reactor. "
"The core temperature must be less than 900. "
"If core temperature exceeds 800 then increase coolant. "
"Maximize energy output. Minimize coolant usage.",
name="reactor",
)
Pipeline stages
| Stage | Module | What it does |
|---|---|---|
| Normalise | normalizers.py | Unicode + whitespace + case |
| Parse | parsers.py | Tokens, sentences, POS |
| Extract entities | entity_extractor.py | Numbers, dates, identifiers |
| Mine relations | relations_miner.py | is_a, has, controls, … |
| Recognise intent | intent_recognizer.py | define, constrain, … |
| Compile constraints | constraint_parser.py | Text → I/T/G |
| Embed | embedders.py | Bag-of-words, TF-IDF, hashing |
| Validate | validation.py | Structural and semantic checks |
Explicit constraint syntax
text = """
constraint: temperature > 0 and temperature < 900
constraint: pressure <= 10
if pressure > 8 then vent
maximize energy_output
minimize coolant_flow
"""
coh = text_to_coh(text, name="reactor")
The chat pipeline
from gismol.nlp import ChatModule
chat = ChatModule()
print(chat.process("define a system called pump").text)
print(chat.process("the pressure must be less than 200").text)
print(chat.process("describe it").text)
Caveats
- The NL pipeline is rule-based, not ML-based.
- Numbers are parsed greedily.
- Relations are heuristic.
17Reasoners
The reasoning menu
| Reasoner | Question |
|---|---|
ForwardChainer | What facts follow from the current state? |
ConstraintReasoner | Which invariants are violated? How to repair? |
GoalSolver | What method sequence maximises the goals? |
AbductiveReasoner | What would have to be true for constraints to hold? |
CausalReasoner | What are the causes and effects? |
AnalogicalReasoner | Which systems are similar? |
CategoricalReasoner | What is the categorical structure? |
DomainReasoner | How does it behave under domain rules? |
MetaReasoner | Which reasoner should I use? |
Querying
from gismol.reasoners import Query, GeneralReasoner
result = GeneralReasoner().infer(reactor, Query("maximize energy output"))
print(result.answer)
print(result.confidence)
print(result.trace)
Constraint repair
from gismol.reasoners import ConstraintReasoner
reactor.set("core_temperature", 1500.0)
result = ConstraintReasoner().infer(reactor, Query("violations"))
print(result.answer)
# {'violations': ['inv_core_temperature_10'],
# 'repairs': [{'core_temperature': 900.0}]}
Planning
from gismol.reasoners import GoalSolver
result = GoalSolver(depth=4, beam=6).infer(reactor, Query("best plan"))
print(result.answer["plan"]) # ['increase_coolant', 'vent', 'regulate']
print(result.answer["value"]) # total goal score
Causal reasoning
from gismol.reasoners import CausalReasoner
result = CausalReasoner().infer(
reactor,
Query("coolant_flow", parameters={"intervene": True}),
)
print(result.answer["effects"])
Domain reasoning
from gismol.reasoners import DomainReasoner, get_domain
reasoner = DomainReasoner(profile=get_domain("finance"))
result = reasoner.infer(trading_desk, Query("analyse"))
print(result.answer["domain"])
Composing reasoners
from gismol.reasoners import MetaReasoner
meta = MetaReasoner()
result = meta.infer(reactor, Query("check violations and plan a fix"))
print(result.answer)
18Validation and soundness
Two kinds of correctness
- Soundness — internally consistent
- Validation — matches your intent
The soundness report
report = reactor.soundness_report()
print(report)
# {'hierarchical_sound': True,
# 'constraints_satisfied_at_sample': True,
# 'violations': [],
# 'triggers_breaking_invariants': [],
# 'sound': True}
The validator
from gismol.nlp.validation import COHValidator
validator = COHValidator(require_all_components=False, strict_goals=True)
for issue in validator.validate(reactor):
print(issue)
# [WARNING] reactor: goal 'comfort' references unknown attributes ['x']
Full component check
report = reactor.component_report()
# {'C': 3, 'A': 12, 'M': 5, 'N': 4, 'E': 'FeatureEmbedder',
# 'I': 6, 'T': 3, 'G': 2, 'D': 3}
reactor.is_minimal() # True if all nine components are populated
Raise on invalid
validator.raise_if_invalid(reactor) # raises ValidationError
19Persistence and repositories
Serialising a single object
data = reactor.to_dict()
restored = COHObject.from_dict(data)
comp.theta,
and method source code with inspect.getsource.
The repository
from gismol.core.repository import COHRepository
repo = COHRepository("production")
oid = repo.add(reactor, tags=["thermal", "critical"])
repo.add(other_reactor, tags=["thermal"])
repo.find(tag="critical")
repo.find(name="reactor")
repo.get(oid)
repo.tags_of(oid)
repo.index_hierarchy(fleet)
repo.save("production.json")
loaded = COHRepository.load("production.json")
Storing trained weights
import json
weights = {
name: comp.theta
for name, comp in reactor.N.items()
if comp.n_parameters() > 0
}
with open("weights.json", "w") as fh:
json.dump(weights, fh)
# Reload
with open("weights.json") as fh:
weights = json.load(fh)
for name, theta in weights.items():
reactor.N[name].set_theta(theta)
Best practices
- Save after every structural change.
- Tag systems by domain, criticality, and owner.
- Version your weights — store the git SHA alongside them.
- Never trust a reloaded object without calling
soundness_report().
20Design patterns and recipes
The standard builder pattern
def build_system() -> COHObject:
sys = COHObject("system")
_add_attributes(sys)
_add_methods(sys)
_add_constraints(sys)
_add_daemons(sys)
_add_neural(sys)
return sys
Sensor–actuator split
system.add_child(build_sensors()) # A-heavy, no M
system.add_child(build_actuators()) # M-heavy, no A
system.add_child(build_controller()) # N-heavy, bridges them
Safety envelope
def add_safety_envelope(coh, var, lo, hi):
coh.add_identity_constraint(IdentityConstraint(
f"{var}_bounds", lambda s: lo <= s[var] <= hi))
coh.add_daemon(BoundaryDaemon({var: (lo, hi)}))
coh.add_daemon(BarrierDaemon(
barrier=lambda s: min(s[var] - lo, hi - s[var]),
alpha=lambda h: 0.5 * h,
))
Two-level control
fast = COHObject("fast_loop")
slow = COHObject("slow_loop")
system = slow.product(fast)
Event-driven architecture
for transition in [("idle", "start", "running"), ...]:
src, evt, dst = transition
coh.add_trigger_constraint(TriggerConstraint(
f"{src}_to_{dst}", evt,
condition=lambda s, src=src: s["mode"] == src,
action={"mode": dst},
))
Learned policy with safety net
coh.add_neural(PolicyComponent("policy", input_dim=6, n_actions=4))
coh.add_daemon(BarrierDaemon(
barrier=lambda s: 1.0 - (s["x"] ** 2 + s["y"] ** 2) / 100.0,
intervention="safe_stop",
))
Config-driven construction
def build_from_config(cfg: dict) -> COHObject:
coh = COHObject(cfg["name"])
for a in cfg["attributes"]:
coh.add_attribute(Attribute(**a))
for i in cfg["invariants"]:
coh.add_identity_constraint(
IdentityConstraint(i["name"], i["expression"]))
return coh
21Anti-patterns to avoid
1 · God objects
Bad: one COHObject with 200 attributes and 50 methods.
Good: nested objects with focused responsibilities. Split if an object has more than ~15 attributes or ~10 methods.
2 · Constraints as assertions
Bad: checking constraints inside methods with assert.
Good: declaring them as IdentityConstraints.
3 · Mutable state
Bad: methods that mutate state in place.
Good: methods that return a new state.
4 · Magic values
Bad: if s["temp"] > 87.3: scattered through code.
Good: declare an Attribute("max_temp", 87.3).
5 · Overly deep hierarchies
Bad: 8-level nesting for a simple controller.
Good: 3–5 levels.
6 · Neural components without constraints
Bad: a policy that can output any action.
Good: a policy gated by identity constraints and daemons.
7 · Skipping validation
Bad: deploying without calling soundness_report().
Good: run it in CI on every commit.
8 · Over-using triggers
Bad: every state change is a trigger.
Good: triggers for events, goals for preferences, methods for actions.
9 · Confusing goals and constraints
Bad: using GoalConstraint for safety.
Good: safety is an IdentityConstraint with a daemon.
10 · Ignoring embeddings
Bad: never setting an embedder.
Good: set a FeatureEmbedder or CompositeEmbedder.
22Performance and scaling
Where time goes
| Phase | Typical share |
|---|---|
| Constraints | 30–50% |
| Daemons | 20–40% |
| Neural forward | 10–30% |
| Flow integration | 5–15% |
| Bookkeeping | 5% |
Constraint caching
cached = {}
def get_constraint_value(state, key):
if key not in cached or cached[key + "_state"] is not state:
cached[key] = constraint.satisfied(state)
cached[key + "_state"] = state
return cached[key]
Daemon throttling
class ThrottledDaemon(Daemon):
def __init__(self, inner, period=10):
super().__init__(f"throttled_{inner.name}")
self.inner = inner
self.period = period
self.tick = 0
def observe(self, state, t=0.0):
self.tick += 1
if self.tick % self.period != 0:
return None
return self.inner.observe(state, t)
Parallel simulation
from concurrent.futures import ProcessPoolExecutor
def run_one(seed):
system = build_system()
return system.simulate(steps=100).cumulative_reward()
with ProcessPoolExecutor() as ex:
results = list(ex.map(run_one, range(100)))
Numpy acceleration
pip install -e ".[fast]"
10–50× speedup on large networks with no code changes.
Avoiding redundant embeddings
@functools.lru_cache(maxsize=1024)
def embed_cached(uid):
return coh.embed()
23Debugging
The describe method
print(reactor.describe())
The soundness report
print(reactor.soundness_report())
The component report
print(reactor.component_report())
Simulating with traces
traj = reactor.simulate(steps=10, dt=0.1)
for t, state in traj:
print(f"t={t:.2f} temp={state['temperature']:.2f} "
f"admissible={reactor.satisfies_constraints(state)}")
Per-step info
state, info = reactor.step(dt=0.1)
print(info)
# {'events': ['overheat'], 'corrections': ['project'], 'admissible': True}
Daemon history
for d in reactor.D:
print(f"{d.name}: {len(d.history)} actions")
for a in d.history[-3:]:
print(f" {a.name} severity={a.severity:.3f} {a.message}")
Constraint diagnostics
for c in reactor.I:
if not c.satisfied(reactor.state):
print(f"VIOLATED: {c.name} margin={c.violation(reactor.state):.3f}")
Common errors and fixes
| Error | Cause | Fix |
|---|---|---|
CyclicHierarchyError | Parent/child loop | Restructure |
ConstraintViolationError | Admissibility broken | Fix state or constraint |
MethodExecutionError | Precondition failed | Check preconditions |
UnsatisfiableConstraintError | Contradictory constraints | Reconcile or relax |
ParseError | Invalid expression | Check the whitelist |
MorphismError | Structure dropped | Review the mapping |
ValidationError | Validator found errors | Fix before deploying |
24Case study — a full drone inspection system
Requirements
- A drone flies a patrol route over a solar farm
- It must not fly in high wind or low battery
- It must detect thermal anomalies on panels
- It must return to base when anomalous or out of bounds
Hierarchy
drone_system (level 0)
├── flight_controller (level 1)
│ ├── attitude_control (level 2)
│ │ ├── roll_ctrl (level 3)
│ │ ├── pitch_ctrl
│ │ └── yaw_ctrl
│ └── navigation
│ └── gps
├── power
│ └── battery
│ └── cell_monitor
└── payload
├── thermal_camera
│ └── image_processor
└── anomaly_reporter
Implementation sketch
import math
from gismol import COHObject, Attribute, Method
from gismol.core.constraints import (
IdentityConstraint, TriggerConstraint, GoalConstraint,
)
from gismol.core.daemons import BoundaryDaemon, InvariantDaemon
from gismol.core.daemon_impl import BarrierDaemon
from gismol.core.embedding import FeatureEmbedder
from gismol.neural.components import MLPComponent, PolicyComponent
from gismol.nlp.validation import COHValidator
def build_drone_system() -> COHObject:
def axis_ctrl(axis):
c = COHObject(f"{axis}_ctrl")
c.add_attribute(Attribute(f"angle_{axis}", 0.0, bounds=(-1.5, 1.5)))
c.add_attribute(Attribute(f"rate_{axis}", 0.0, bounds=(-3.0, 3.0)))
c.add_identity_constraint(IdentityConstraint(
f"{axis}_angle_ok", lambda s, ax=axis: abs(s[f"angle_{ax}"]) < 1.4))
return c
# ... (build attitude, navigation, battery, thermal_camera, etc.)
drone = COHObject("drone_system",
description="Autonomous solar farm inspection UAV")
drone.add_attribute(Attribute("mission_time_s", 0.0, bounds=(0.0, 3600.0)))
drone.add_attribute(Attribute("coverage_pct", 0.0, bounds=(0.0, 100.0)))
drone.add_attribute(Attribute("wind_speed_mps", 5.0, bounds=(0.0, 30.0)))
drone.add_child(axis_ctrl("roll"), key="roll")
drone.add_child(axis_ctrl("pitch"), key="pitch")
drone.add_identity_constraint(IdentityConstraint(
"mission_time_limit", lambda s: s["mission_time_s"] <= 3000.0))
drone.add_trigger_constraint(TriggerConstraint(
"high_wind_rtb", "wind_alert",
condition=lambda s: s["wind_speed_mps"] > 15.0,
action={"mission_status": "returning"}, priority=9))
drone.add_goal_constraint(GoalConstraint(
"max_coverage", lambda s: s["coverage_pct"],
direction="max", weight=3.0))
drone.add_daemon(InvariantDaemon(constraints=list(drone.I)))
drone.add_daemon(BoundaryDaemon({
"wind_speed_mps": (0.0, 25.0),
"coverage_pct": (0.0, 100.0),
}))
drone.add_daemon(BarrierDaemon(
barrier=lambda s: 25.0 - s["wind_speed_mps"],
intervention={"mission_status": "returning"},
))
drone.add_neural(
PolicyComponent("mission_policy", input_dim=5, n_actions=4, seed=42),
name="mission_policy")
drone.set_embedder(FeatureEmbedder(
["mission_time_s", "coverage_pct", "wind_speed_mps"],
ranges={"mission_time_s": (0, 3600), "coverage_pct": (0, 100),
"wind_speed_mps": (0, 30)}))
return drone
drone = build_drone_system()
print(f"Descendants: {len(drone.descendants())}")
print(f"Components: {drone.component_report()}")
print(f"Sound: {drone.soundness_report()['sound']}")
validator = COHValidator()
errors = [i for i in validator.validate(drone) if i.severity == "error"]
assert not errors, errors
traj = drone.simulate(steps=200, dt=0.1)
print(f"Final coverage: {traj.final['coverage_pct']:.1f}%")
What we achieved
- Four levels of hierarchy
- All nine components populated
- Hybrid dynamics — continuous state + trigger-driven mode switching
- Layered safety — invariants + boundary daemons + barrier daemon
- Learned policy ready to be trained
- Full validation — no errors, soundness confirmed
25API cheatsheet
The 9-tuple at a glance
coh = COHObject("name", description="...")
coh.C # children (dict)
coh.A # attributes (dict)
coh.M # methods (dict)
coh.N # neural components (dict)
coh.E # embedder
coh.I # identity constraints (list)
coh.T # trigger constraints (list)
coh.G # goal constraints (list)
coh.D # daemon network
Construction
COHObject(name, children=[], attributes=[], methods=[],
neural={}, embedder=None, identity=[], triggers=[],
goals=[], daemons=None, description="", metadata={},
flow=None, jump=None)
coh.add_child(COHObject("child"))
coh.add_attribute(Attribute("x", 0.0, bounds=(0, 10)))
coh.add_method(Method("do", lambda s, a=None: (s, 1.0)))
coh.add_identity_constraint(IdentityConstraint("i1", "x > 0"))
coh.add_trigger_constraint(TriggerConstraint("t1", "evt", "x > 5", "do"))
coh.add_goal_constraint(GoalConstraint("g1", "x", direction="max"))
coh.add_daemon(BoundaryDaemon({"x": (0, 10)}))
coh.add_neural(MLPComponent("nn", [3, 4, 1]))
coh.set_embedder(FeatureEmbedder(["x"]))
State
coh.state # State (dict-like)
coh.get("x") # top-level
coh.get("child.x") # dotted
coh.set("child.x", 5.0)
coh.initial_state()
coh.reset()
coh.snapshot()
Methods
coh.invoke("method_name") # -> (new_state, reward)
coh.method("method_name") # -> Method object
Simulation
coh.step(action=None, dt=0.01, event=None)
coh.simulate(steps=100, dt=0.01, policy=None)
traj.final
traj.cumulative_reward()
traj.discounted_return(gamma=0.99)
traj.to_dict()
Constraints
coh.constraints.check_identity(state)
coh.constraints.satisfies_identity(state)
coh.constraints.violation_score(state)
coh.constraints.total_goal(state, action, t)
Daemons
coh.monitor(state=None)
coh.D
coh.D.by_name("...")
for d in coh.D: ...
Embeddings
coh.embed()
coh.similarity(other)
coh.canonical_form()
Reasoning
from gismol.reasoners import GeneralReasoner, Query
GeneralReasoner().infer(coh, Query("target"))
ConstraintReasoner().infer(coh, Query("check"))
GoalSolver(depth=3, beam=4).infer(coh, Query("plan"))
NLP
from gismol.nlp import text_to_coh, ChatModule
coh = text_to_coh("...text...", name="sys")
chat = ChatModule()
chat.process("...")
Validation
from gismol.nlp.validation import COHValidator
COHValidator().validate(coh)
COHValidator().is_valid(coh)
coh.soundness_report()
coh.component_report()
coh.is_minimal()
Persistence
data = coh.to_dict()
coh2 = COHObject.from_dict(data)
repo = COHRepository()
repo.add(coh, tags=["x"])
repo.save("repo.json")
COHRepository.load("repo.json")
26Glossary
| Term | Definition |
|---|---|
| Admissible region | The set of states where all identity constraints hold |
| Attribute (A) | A measurable state variable of a COHObject |
| Barrier function | A scalar function whose non-negativity certifies safety |
| COH | Constrained Object Hierarchy |
| COHObject | The base class implementing the 9-tuple |
| Constraint | A predicate that restricts admissible behaviour |
| Coproduct | Categorical alternative composition |
| Corrective action | An action emitted by a daemon |
| Daemon (D) | A continuous monitor enforcing invariants |
| Embedding (E) | A map O → H into a Hilbert space |
| Exponential | Categorical function object O₂^O₁ |
| Free energy | An information-theoretic bound on surprise |
| Goal constraint (G) | An optimisation objective |
| GISMOL | General Intelligent Systems Modelling Language |
| Hybrid system | Combining continuous and discrete dynamics |
| Identity constraint (I) | An invariant predicate |
| Lyapunov function | A scalar function decreasing along trajectories |
| Method (M) | An executable state transformation |
| Morphism | A structure-preserving map between COHObjects |
| Neural component (N) | A parameterised adaptive function |
| Product | Categorical parallel composition |
| Soundness | Internal consistency of a COHObject |
| State | A hierarchical assignment of values to attributes |
| Trigger constraint (T) | An event-condition-action rule |
| 9-tuple | (C, A, M, N, E, I, T, G, D) |