Python API

The BasinSurgeon class is the primary Python interface for Energy Language.

Construction

from qriton_hlm import BasinSurgeon

# From a checkpoint file
surgeon = BasinSurgeon.from_checkpoint("model.pt", device="cuda")

# From a raw weight matrix
surgeon = BasinSurgeon.from_W(W_tensor, device="cpu")

Basin Operations

# Survey all basins in a layer
survey = surgeon.survey(layer=0)
# Returns: {num_basins, basins, energies, populations, ...}

# Inject a new basin
result = surgeon.inject(layer=0, seed=42, strength=0.1)

# Remove a basin
result = surgeon.remove(layer=0, seed=17, strength=0.1)

# Move a basin
result = surgeon.move(layer=0, seed=42, strength=0.1)

# Verify a basin exists
v = surgeon.verify(layer=0, seed=42)
# Returns: {is_basin, cos, energy, iters}

# Strengthen / weaken
surgeon.strengthen(layer=0, seed=42, factor=1.5)
surgeon.weaken(layer=0, seed=42, factor=0.5)

# Measure energy at a point
e = surgeon.energy(layer=0, seed=42)

Concept Operations

# Capture concept from text (requires full model)
surgeon.capture(layer=5, text="Thank you so much", concept_name="polite")
surgeon.capture(layer=5, text="I really appreciate it", concept_name="polite")

# Batch capture
surgeon.batch_capture(layer=5, concept_name="polite", texts=[
    "Thank you so much",
    "I really appreciate it",
    "That's very kind of you",
])

# Inject captured concept
result = surgeon.inject_concept(layer=5, concept_name="polite", strength=0.1)

# Remove concept
surgeon.remove_concept(layer=5, concept_name="polite", strength=0.1)

# Blend two concepts
surgeon.blend("polite", "formal", "professional", ratio=0.6)

# List all concepts
surgeon.list_concepts()

# Export / Import
surgeon.export_concept("polite", "polite.concept")
surgeon.import_concept("polite.concept")

# Transplant from another model
other = BasinSurgeon.from_checkpoint("other_model.pt")
surgeon.transplant(other, layer=5, concept_name="humor")

Concept Algebra v0.10

# Similarity
sim = surgeon.concept_similarity("polite", "formal")
print(f"cos(polite, formal) = {sim['similarity']:.4f}")

# Vector arithmetic
surgeon.concept_add("polite", "formal", "professional")        # A + B
surgeon.concept_subtract("polite", "casual", "formality")      # A - B

# Analogy: A is to B as C is to ?
surgeon.concept_analogy("man", "king", "woman", "queen")

# Weighted composition of N concepts
surgeon.concept_compose(
    ["polite", "formal", "warm"],
    [0.5, 0.3, 0.2],
    "ideal_tone"
)

# Interpolation path from A to B
path = surgeon.concept_interpolate("polite", "rude", num_steps=10)
for step in path['steps']:
    print(f"  ratio={step['ratio']:.1f}  sim_A={step['similarity_to_a']:.3f}")

Consolidation (Sleep Cycle) v0.10

# Single layer: strengthen popular basins, prune weak
report = surgeon.consolidate(layer=0, strengthen_factor=1.5, prune_threshold=0.3)
print(f"Basins: {report['basins_before']} -> {report['basins_after']}")
print(f"+{report['strengthened']} strengthened, -{report['pruned']} pruned")

# Full sleep: all layers, multiple passes
dream = surgeon.dream(cycles=3, strengthen_factor=1.5, prune_threshold=0.3)
print(f"+{dream['total_strengthened']} -{dream['total_pruned']} across {dream['layers']} layers")

Watermark (IP Protection) v0.10

# Sign your model
surgeon.watermark_inject("my-secret-key", num_marks=5, strength=0.05)

# Verify on any model
result = surgeon.watermark_verify("my-secret-key", num_marks=5)
print(f"Verified: {result['verified']} ({result['confidence']:.0%})")

# Test robustness
strip = surgeon.watermark_strip_attempt("my-secret-key", num_marks=5)
print(f"Survived: {strip['survived']}/{strip['total']}")

Multi-Model v0.10

other = BasinSurgeon.from_checkpoint("other_model.pt")

# Compare basin structure
diff = surgeon.compare(other, layer=0)
print(f"Shared: {diff['shared']}, Only-self: {diff['only_self']}, Only-other: {diff['only_other']}")

# Transplant concept from another model
surgeon.transplant(other, layer=0, concept_name="humor", strength=0.1)

Benchmark v0.10

# Measure perplexity (requires loaded model + tokenizer)
result = surgeon.benchmark(max_tokens=50)
print(f"PPL: {result['perplexity']:.2f}, Loss: {result['avg_loss']:.4f}")

# With custom texts
result = surgeon.benchmark(texts=["The capital of France is", "Once upon a time"])

Causal Operations

# Discover causal graph: which basins cause which?
graph = surgeon.causal_scan(layer=5, threshold=0.15, num_inits=50)
for edge in graph['edges']:
    print(f"B{edge['source']} -> B{edge['target']}  drift={edge['drift']:.4f}")

# do(X) — intervene on a basin, measure downstream effects
result = surgeon.causal_intervene(layer=5, basin_idx=0, operation='remove')
print(f"Affected: {result['num_affected']} basins")

# Counterfactual: what if basin 0 had been inverted? (non-destructive)
cf = surgeon.causal_counterfactual(layer=5, basin_idx=0, modification='invert')
print(f"Would affect: {cf['num_affected']} basins")

See the full Causal Programming guide for parameters, return values, and pipeline examples.

Safety

# Set guards — pre-execution checks that block unsafe operations
surgeon.guard(layer=5, max_remove_pct=10.0)

# Trace convergence trajectory (useful for debugging basin dynamics)
trajectory = surgeon.trace(layer=5, text="Hello world")
# Returns convergence path: states, energies, iterations

Persistence

# Apply modified W to live model
surgeon.apply(layer=5)

# Restore original W
surgeon.restore(layer=5)
surgeon.restore_all()

# Check what changed
diff = surgeon.diff(layer=5)
# Returns: {frobenius_norm, relative_pct, max_change, ...}

# Compare with another model
diff = surgeon.compare(other_surgeon, layer=5)

# View operation history
surgeon._history

# Save everything to disk (W matrices + concepts + history)
surgeon.save_checkpoint("session.pt")

# Load a previous session
surgeon.load_session("session.pt")
# Restores: W matrices, backups, captured concepts, history

Database Sync

Sync SQL database rows into HLM as named concept basins:

from qriton_hlm.db import HLMSync

with HLMSync.from_config("hlm_sync_config.json") as syncer:
    syncer.sync_row("Products", {"id": 1, "name": "Widget", "price": 9.99})
    syncer.full_sync_table("Products")
    syncer.delete_row("Products", 1)

See the full Database Sync guide for configuration, strategies, and API reference.

Low-Level Primitives

For direct tensor operations:

from qriton_hlm import (
    poly_interaction,
    compute_energy,
    find_basins,
    inject_basin,
    remove_basin,
    move_basin,
    verify_basin_exists,
    load_W_from_checkpoint,
)

# Example: manual basin injection
W_new = inject_basin(W, target_vector, strength=0.1, beta=7.0)
exists, state, cos, iters = verify_basin_exists(W_new, target_vector)