Python API (pyxlog)

pip install pyxlog

export XLOG_CUBIN_DIR=/path/to/xlog/crates/pyxlog/python/pyxlog/kernels
python your_probe.py

python scripts/install_pyxlog_for_python.py --python /usr/local/bin/python --user

python scripts/install_pyxlog_for_python.py --python /usr/local/bin/python
python scripts/install_pyxlog_for_python.py --python /usr/local/bin/python \
  --features extension-module,host-io,arrow-device-import

row = (basis, identity, canonical_record_json, material_pairs, evidence_bytes)
identity = (training_view, training_view_digest, record_sha256,
            source_length, block_size, targets)

(
    ((manifest_hash, (max_entries, max_total_keys,
                      max_records_per_input, max_record_bytes)), ...),
    ((window, pad_id, mask_id, block_size), batch_size, mask_permille,
     training_seed, mask_policy_hash),
)

materials = step.completed_replay_materials(consumer_streams=original_streams)
if materials is not None:
    p_identity, s_identity, full_p, native_provenance, publication_evidence = materials

import pyxlog
import torch

# Compile a deterministic program
program = pyxlog.LogicProgram.compile("""
    pred edge(u32, u32).
    pred reach(u32, u32).

    edge(1, 2). edge(2, 3). edge(3, 4).

    reach(X, Y) :- edge(X, Y).
    reach(X, Z) :- reach(X, Y), edge(Y, Z).

    ?- reach(1, N).
""")

# Execute and get results
result = program.evaluate()

# Results are a list of query outputs (relations) with per-column DLPack tensors
for q in result.queries:
    print(q.relation_name, q.columns, q.num_rows, q.is_true)
    cols = [torch.from_dlpack(t) for t in q.tensors]
    print(cols)

import pyxlog
import torch

program = pyxlog.LogicProgram.compile("""
    pred edge(u32, u32).
    pred reach(u32, u32).
    reach(X, Y) :- edge(X, Y).
    ?- reach(1, N).
""")

# Two 1D columns, not a 2D tensor.
edge_a = torch.tensor([1, 2, 3], device="cuda", dtype=torch.int32)
edge_b = torch.tensor([2, 3, 4], device="cuda", dtype=torch.int32)

result = program.evaluate(dlpack_inputs={"edge": [edge_a, edge_b]})

import pyxlog
import torch

program = pyxlog.LogicProgram.compile("""
    pred edge(i32, i32).
    pred reach(i32, i32).
    reach(X, Y) :- edge(X, Y).
    ?- reach(X, Y).
""")

session = program.session()

edge_a = torch.tensor([1, 2, 3], device="cuda", dtype=torch.int32)
edge_b = torch.tensor([2, 3, 4], device="cuda", dtype=torch.int32)

session.put_relation("edge", [edge_a, edge_b])   # register or replace
result = session.evaluate()                      # reuse stored relations
exported = session.export_relation("edge")       # DLPack columns

session.remove_relation("edge")
session.clear_relations()

session.put_relation("edge", [row_id, parent_id])
session.evaluate()

delta = session.insert_relation("edge", [new_row_id, new_parent_id])
result = session.evaluate()          # returns the delta-updated cached store
print(session.delta_stats(), delta)

session.apply_relation_delta(
    "edge",
    insert_columns=[added_row_id, added_parent_id],
    delete_columns=[removed_row_id, removed_parent_id],
)

session.apply_relation_delta_batch([
    {"name": "edge", "insert_columns": [row_a, parent_a]},
    {"name": "edge", "delete_columns": [row_b, parent_b]},
])

debug = session.apply_relation_delta_debug(
    [{"name": "edge", "insert_columns": [row_c, parent_c]}],
    check_equivalence=True,
)

def register_relation_callback(callback) -> int: ...
def unregister_relation_callback(callback_id: int) -> bool: ...

events = []
callback_id = session.register_relation_callback(events.append)
session.apply_relation_delta_batch([
    {"name": "edge", "insert_columns": [row_a, parent_a]},
])
session.unregister_relation_callback(callback_id)

def rule_provenance() -> list[dict]: ...
def proof_traces() -> list[dict]: ...

session.put_temporal_relation(
    "stream_row",
    columns,
    timestamp_column="event_ts",
    dataset_id="hf-live",
    row_hashes=row_hashes,
    field_hashes=field_hashes,
    uncertainty=uncertainty,
    stream_id="camera-a",
    order_column="seq",
    source="hf://dataset/split",
    process_boundary="observation_process",
    temporal_order=["seq"],
)
session.temporal_provenance("stream_row")

pyxlog.put_temporal_relation(
    session,
    "stream_row_copy",
    columns,
    timestamp_column="event_ts",
    dataset_id="hf-live",
    row_hashes=row_hashes,
    field_hashes=field_hashes,
    uncertainty=uncertainty,
    stream_id="camera-a",
    order_column="seq",
    source="hf://dataset/split",
    process_boundary="observation_process",
    temporal_order=["seq"],
)
pyxlog.temporal_provenance(session, "stream_row_copy")

program = pyxlog.LogicProgram.compile("""
    domain party: u32.
    domain asset: u32.
    pred transfer(giver: party, receiver: party, asset: asset, event_time: i64).
""")
session = program.session()

snapshot = session.put_relation_with_provenance(
    "transfer",
    [giver, receiver, asset, event_time],
    roles=[
        {"name": "giver", "sort": "party", "type": "u32"},
        {"name": "receiver", "sort": "party", "type": "u32"},
        {"name": "asset", "sort": "asset", "type": "u32"},
        {"name": "event_time", "type": "i64"},
    ],
    facts=[{
        "tuple": [10, 20, 7, 1_700_000_000],
        "provenance": [
            {
                "source": "extractor-output",
                "document": "document-42",
                "span": {"start": 18, "end": 41},
                "content_hash": "sha256:...",
                "kind": "assertion",
                "polarity": "positive",
            },
            {"source": "manual-review", "kind": "confirmation"},
        ],
    }],
)
same_snapshot = session.relation("transfer").provenance()
all_evidence = session.evidence()

exported = session.export_relation_with_provenance("transfer")

fresh = program.session()
restored = fresh.put_relation_from_manifest(
    "transfer",
    exported["columns"],
    exported["manifest"],
)

handle = session.evaluate_async(memory_mb=512)
result = handle.result(timeout=30)

for chunk in session.evaluate_stream(memory_mb=512, chunk_rows=1024):
    cols = chunk.tensors  # torch CUDA tensor views, DLPack-compatible
    print(chunk.relation_name, chunk.offset, chunk.num_rows, cols)

result = session.evaluate()
for chunk in result.iter_query_chunks(chunk_rows=1024):
    ...

session.progress_stats()
session.memory_stats()
session.host_transfer_stats()
session.cuda_graph_stats()

program.progress_stats()
program.memory_stats()
program.host_transfer_stats()
program.cuda_graph_stats()
def neural_hot_loop_diagnostics() -> dict: ...
program.neural_hot_loop_diagnostics()

program.register_network(
    "mnist_net",
    net,
    optimizer,
    checkpoint_hash="sha256:...",
    split_hashes={"train": "sha256:...", "validation": "sha256:..."},
    calibration_metrics={"ece": 0.03},
    cuda_device=0,
    influence_audit={"calibration_set": "heldout-a"},
)
program.record_nn4_influence(
    "mnist_net",
    query="addition(0, 1, 1)",
    changed_acceptance=True,
    before=False,
    after=True,
)
program.nn4_lineage()
program.neural_hot_loop_diagnostics()["nn4_lineage"]

program.register_network(
    "mnist_net",
    net,
    optimizer,
    arity=2,
    arg_sorts=[3, 9],
    artifact_hash="sha256:...",
)

program.network_metadata("mnist_net")
# {"arity": ..., "arg_sorts": [...], "artifact_hash": ...,
#  "declared": [{"predicate": ..., "predicate_arity": ..., "input_arity": ..., "labels": [...]}]}

RuntimeError: Unsupported epistemic construct: accepted GPU world-view evidence
(probabilistic evidence requires non-empty accepted GPU final output)

import pyxlog

known = pyxlog.LogicProgram.compile("""
pred fact().
pred accepted().

fact().

accepted() :- know fact().
""")

result = known.evaluate_conditioned("0.6::fact().\nquery(fact()).\n")
result.log_z_e                                       # ln(0.6) ~= -0.5108256, log P(evidence)
result.trace["gpu_conditioned_evidence_facts"]        # >= 1 when evidence conditioned the circuit
result.trace["gpu_conditioned_know_evidence_facts"]   # the `know` share of that total
result.trace["accepted_gpu_production_path_events"]   # positive observed GPU work
result.trace["gpu_exact_query_evaluations"]            # positive GPU exact execution evidence

prepared = known.prepare_conditioned(
    "0.5::target().\n0.6::fact().\nquery(target()).\nquery(fact()).\n"
)
target_var = next(
    var for var, info in enumerate(prepared.prob_var_map())
    if info.get("atom") == "target()"
)

for prior in (0.5, 0.9, 0.1):
    prepared.set_fact_probabilities({target_var: prior})
    result = prepared.evaluate()
    assert result.trace["gpu_exact_source_compiles"] == 0
    assert result.trace["gpu_exact_program_compiles"] == 0
    assert result.trace["gpu_conditioned_circuit_reuses"] == 1
    assert result.trace["gpu_conditioned_circuit_materializations"] == 1
    assert (
        result.trace["gpu_conditioned_circuit_preparation_compiles"]
        + result.trace["gpu_conditioned_circuit_disk_cache_restores"]
        + result.trace["gpu_conditioned_circuit_gpu_cache_hits"]
    ) == 1

import pyxlog

# Compile with exact inference
program = pyxlog.Program.compile("""
    0.3::rain.
    0.7::sprinkler.

    wet :- rain.
    wet :- sprinkler.

    evidence(sprinkler, false).
    query(wet).
""", prob_engine="exact_ddnnf")

result = program.evaluate()
import torch
prob = torch.from_dlpack(result.prob)       # f64 CUDA tensor, shape [num_queries]
log_prob = torch.from_dlpack(result.log_prob)
print(list(zip(result.atoms, prob.tolist())))  # host read for printing

# If you need a single host scalar (e.g., for logging), read it explicitly:
p0 = float(prob[0].item())  # host read
print(f"P(wet | not sprinkler) = {p0}")

# With gradients (exact engine only; per-query grad vectors are DLPack too)
result = program.evaluate(return_grads=True)
grad_true0 = torch.from_dlpack(result.grad_true[0])   # f64 CUDA tensor, shape [num_vars]
grad_false0 = torch.from_dlpack(result.grad_false[0]) # f64 CUDA tensor, shape [num_vars]

program = pyxlog.Program.compile(source, prob_engine="mc")

device_result = program.evaluate_device(
    samples=10000,
    seed=42,
    confidence=0.95,
)

import torch
query_counts = torch.from_dlpack(device_result.query_counts)       # int32 CUDA tensor, shape [num_queries]
evidence_count = torch.from_dlpack(device_result.evidence_count)   # int32 CUDA tensor, shape [1]
print(device_result.total_samples, device_result.seed, device_result.confidence)

program = pyxlog.Program.compile(source, prob_engine="mc")

result = program.evaluate(
    samples=10000,
    seed=42,
    confidence=0.95
)

import torch
prob = torch.from_dlpack(result.prob)
stderr = torch.from_dlpack(result.stderr)
ci_low = torch.from_dlpack(result.ci_low)
ci_high = torch.from_dlpack(result.ci_high)
print(f"P(query) = {float(prob[0].item())} ± {float(stderr[0].item())}")  # host reads
print(f"95% CI: [{float(ci_low[0].item())}, {float(ci_high[0].item())}]") # host reads

import pyxlog

carrier = pyxlog.JointConstraintCarrier(
    device=0,
    entities=1024,
    domain_lanes=4,
    candidates=16,
    labels=8,
    fuel_limit=1_000_000,
)
carrier.register_schema(catalog_sha, pyxlog.SOLVER_ABI_IDENTITY)
carrier.bind_signatures(head_masks, tail_masks)

carrier.solve_label_feasibility(abstain_label=7)
carrier.solve_label_map_top2()
carrier.solve_components_exact(comp_offsets, comp_indices)

buffer = carrier.export_buffer("map_results") # zero-copy DLPack capsule
print(carrier.fuel_spent)                     # property, not a call

program = pyxlog.Program.compile("""
    nn(entity_embed, [X], E) :: embed(X, E).
""")

# Trainable nn.Embedding — autograd graph preserved
embedding = torch.nn.Embedding(100, 64).cuda()
program.register_embedding("entity_embed", embedding, trainable=True)

# Frozen torch.Tensor — detached at registration, no gradient flow
weights = torch.randn(100, 64).cuda()
program.register_embedding("entity_embed", weights, trainable=False)

# Returns [n, dim] tensor on same device as embedding
vectors = program.forward_embedding("entity_embed", [0, 5, 42])

# For trainable nn.Embedding: vectors.requires_grad == True
# For frozen torch.Tensor: vectors.requires_grad == False

loss = program.nll_loss("addition(0, 1, 7)")
loss = program.nll_loss_batch(queries)
loss = program.nll_loss_mean(queries)

loss_t = program.nll_loss_tensor("addition(0, 1, 7)")
batch_t = program.nll_loss_batch_tensor(queries)
avg_loss = program.evaluate_loss(queries)

top = program.deterministic_topk(scores, k=4)
stats = program.neural_cache_stats()

terms = program.belnap_loss(
    pro=pro_scores,
    contra=contra_scores,
    quarantine=quarantine_scores,
    pro_reward=1.0,
    contra_penalty=2.0,
    quarantine_penalty=0.5,
)

semantic = program.semantic_loss_tensor(violations, weight=1.5)
mse = program.mse_loss_tensor(pred, target)
info = program.infoloss_tensor(prob)

program.zero_grad()
program.optimizer_step()
program.clip_grad_norms(max_norm=1.0)

program.scheduler_step()
program.scheduler_step(network_name="mnist_net")

lr = program.get_lr("mnist_net")
program.set_lr("mnist_net", 1e-4)

stats = program.train_epoch(queries, batch_size=32, max_grad_norm=1.0)
stats = program.train_epoch_tensor(queries, batch_size=32, max_grad_norm=1.0)

profile = program.warmup_breakdown()

from pyxlog.ilp import train_only, train_and_promote, TrainConfig, LearnedArtifact

# Define a learnable program
source = """
    edge(1, 2). edge(2, 3). edge(3, 4). edge(4, 5).
    learnable(W) :: reach(X, Y) :- bL(X, Z), bR(Z, Y).
"""
pos = [("reach", [1, 3]), ("reach", [2, 4])]
neg = [("reach", [1, 1])]

# Configure training
config = TrainConfig(
    step_budget_per_attempt=150,   # steps per attempt
    max_attempts=5,                # multi-start attempts
    tau_start=2.0,                 # initial temperature
    tau_floor=0.05,                # minimum temperature
    seed=42,                       # reproducibility
)

# Train only (no promotion gates)
result = train_only(source, "W", pos, neg, config)
assert result.converged
print(result.discovered_rule)      # e.g., "reach(X,Y) :- edge(X,Z), edge(Z,Y)."

# Train and promote (with gates)
config = TrainConfig(check_ambiguity=True, max_novel_rate=0.05)
promotion = train_and_promote(source, "W", pos, neg, config)
print(promotion.status)            # PromotionStatus.PROMOTED

# Save learned artifact
result.artifact.save("artifact.json")

# Load with hash verification
loaded = LearnedArtifact.load("artifact.json", verify_hash=True)
print(loaded.discovered_rule)
print(loaded.logits)

# TrainResult
result.converged          # bool
result.discovered_rule    # str | None
result.attempt_count      # int
result.total_steps        # int
result.precision          # float
result.recall             # float
result.holdout_f1         # float | None
result.artifact           # LearnedArtifact

# PromotionResult
promotion.status          # PromotionStatus (PROMOTED, GATE_FAILED, etc.)
promotion.gates           # list[GateResult]
promotion.novel_count     # int | None
promotion.novel_rate      # float | None
promotion.committed_source # str | None
promotion.rule_inventory  # RuleInventory | None

from pyxlog.ilp.neurosymbolic import (
    NeuroSymbolicTrainingConfig,
    train_neurosymbolic_program,
)
from pyxlog.runtime_audit import CudaExecutionAudit
from pyxlog.transfer_diagnostics import PredictionRecord, compute_transfer_diagnostics

trained = train_neurosymbolic_program(
    source,
    networks={"ranker": model},
    examples=rows,
    config=NeuroSymbolicTrainingConfig(steps=16),
)
inventory = trained.learned_rule_inventory

with CudaExecutionAudit(forbid_host_materialization=True) as audit:
    scores = model(batch)
    audit.record_nn4_scores("ranker", scores, device_resident=True)

diagnostics = compute_transfer_diagnostics(
    [PredictionRecord(domain="d0", variant="clean", y_true=1, y_pred=1)],
    required_domains=("d0",),
    required_variants=("clean",),
)
assert diagnostics.passed

import torch

prog = pyxlog.IlpProgramFactory.compile(source, device=0, memory_mb=512)

# Device membership: bool CUDA tensor, one row per queried fact.
mask = torch.from_dlpack(
    prog.batch_fact_membership_device("edge", [[1, 2], [9, 9], [2, 3]])
)
assert mask.device.type == "cuda"
assert mask.dtype == torch.bool

# Device tagged credit: CSR-style CUDA outputs.
credit = prog.batch_tagged_credit_device("reach", [[1, 3], [2, 4]])
row_offsets = torch.from_dlpack(credit.fact_row_offsets)   # int32 CUDA tensor
entry_indices = torch.from_dlpack(credit.entry_indices)    # int32 CUDA tensor
entry_i = torch.from_dlpack(credit.entry_i)                # int32 CUDA tensor
entry_j = torch.from_dlpack(credit.entry_j)                # int32 CUDA tensor
entry_k = torch.from_dlpack(credit.entry_k)                # int32 CUDA tensor

from pyxlog.ilp import induce_exact

result = induce_exact(
    prog,
    head_relation="p_A",
    candidate_relations=["p_B", "p_C", "p_D"],
    positive_arg0=pos_a0,
    positive_arg1=pos_a1,
    negative_arg0=neg_a0,
    negative_arg1=neg_a1,
    k_per_topology=2,
    deterministic=True,
    backend="native",
)

import torch
import pyxlog

# Create GPU columns
edge_a = torch.tensor([1, 2, 3], device="cuda", dtype=torch.int32)
edge_b = torch.tensor([2, 3, 4], device="cuda", dtype=torch.int32)

# Pass as input (relation name -> sequence of columns)
program = pyxlog.LogicProgram.compile(source)
result = program.evaluate(dlpack_inputs={"edge": [edge_a, edge_b]})

result = program.evaluate()

# Convert to PyTorch
import torch
for q in result.queries:
    cols = [torch.from_dlpack(t) for t in q.tensors]

# Convert to CuPy
import cupy
for q in result.queries:
    cols = [cupy.from_dlpack(t) for t in q.tensors]

import torch
from pyxlog import dlpack_roundtrip

tensor = torch.arange(8, dtype=torch.int32, device="cuda")
capsule = dlpack_roundtrip(tensor, device=0, memory_mb=1024)
restored = torch.from_dlpack(capsule)
assert torch.equal(tensor, restored)

program = pyxlog.LogicProgram.compile(
    source,                    # str: Datalog source code
    device=0,                  # int: CUDA device index
    memory_mb=32768,          # int: GPU memory limit in megabytes
)

program = pyxlog.Program.compile(
    source,                    # str: Probabilistic Datalog source
    prob_engine=None,          # Optional[str]: explicit "exact_ddnnf"/"mc" override;
                               # otherwise use #pragma prob_engine, then exact_ddnnf
    device=0,                  # int: CUDA device index
    memory_mb=32768,          # int: GPU memory limit in megabytes
)

result = program.evaluate()

result.queries             # list[LogicQueryResult]
result.queries[0].tensors  # list[PyCapsule] (DLPack), one per column
result.queries[0].columns  # list[str]
result.queries[0].num_rows # int
result.queries[0].is_true  # bool

result = program.evaluate()  # requires host-io
result.atoms         # list[str]: query atoms (stringified)
result.prob          # PyCapsule: DLPack f64 vector of probabilities (len = num_queries)
result.log_prob      # PyCapsule: DLPack f64 vector of log-probabilities (len = num_queries)
result.num_vars      # int: number of CNF variables in the compiled program
result.log_z_e       # Optional[float]: exact log-evidence log Z_E (None for Monte Carlo)

# Exact-only, independent of return_grads: what each CNF variable stands for.
# Length equals result.num_vars, i.e. the encoder's variable capacity, so entry i lines up
# with grad_true/grad_false position i when return_grads=True. Index 0 and every slot that
# holds no assigned variable are padding ("other"); the length is NOT a count of random vars.
# "choice" entries carry both "probs" (the disjunction's declared, marginal probabilities —
# display context only) and "prob" (the conditional Bernoulli parameter actually assigned to
# this variable's weight; use prob*(1-prob), not probs[choice_index]*(1-probs[choice_index]),
# as the grad_true/grad_false Jacobian). Raises ValueError for MC programs and for exact
# programs compiled through the GPU count-lift fast path (no CNF encoding is built there).
program.prob_var_map() # list[dict]: {"kind": "fact"|"choice"|"other", ...} per CNF variable

# Exact-only (when return_grads=True):
result.grad_true     # Optional[list[PyCapsule]]: per-query DLPack f64 vector (len = num_vars)
result.grad_false    # Optional[list[PyCapsule]]: per-query DLPack f64 vector (len = num_vars)

# Monte Carlo only:
result.stderr        # Optional[PyCapsule]: DLPack f64 vector (len = num_queries)
result.ci_low        # Optional[PyCapsule]: DLPack f64 vector (len = num_queries)
result.ci_high       # Optional[PyCapsule]: DLPack f64 vector (len = num_queries)
result.samples       # Optional[int]
result.evidence_samples # Optional[int]
result.seed          # Optional[int]
result.confidence    # Optional[float]

device_result = program.evaluate_device(...)
device_result.query_counts    # PyCapsule: DLPack int32 vector (len = num_queries)
device_result.evidence_count  # PyCapsule: DLPack int32 vector (len = 1)
device_result.total_samples   # int
device_result.seed            # int
device_result.confidence      # float

try:
    session.put_relation_with_provenance(
        "transfer", columns, roles=roles, facts=facts
    )
except pyxlog.RelationMetadataError as e:
    print(f"Relation metadata rejected: {e}")
except KeyError as e:
    print(f"Stored relation not found: {e}")
except ValueError as e:
    print(f"Invalid input: {e}")
except RuntimeError as e:
    print(f"XLOG error: {e}")

import torch
import pyxlog

# Neural-symbolic training loop:
# - neural predicate outputs (CUDA tensors) are imported via DLPack
# - XLOG computes NLL gradients on GPU and calls output.backward(grad) internally

source = """
nn(mnist_net, [X], Y, [0,1,2,3,4,5,6,7,8,9]) :: digit(X, Y).
addition(X, Y, Z) :- digit(X, LeftDigit), digit(Y, RightDigit), Z is LeftDigit + RightDigit.
"""
program = pyxlog.Program.compile(source, prob_engine="exact_ddnnf")

net = torch.nn.Sequential(
    torch.nn.Flatten(),
    torch.nn.Linear(28 * 28, 10),
    torch.nn.Softmax(dim=-1),
).cuda()
optimizer = torch.optim.Adam(net.parameters(), lr=1e-3)
program.register_network("mnist_net", net, optimizer)

images = torch.randn(128, 1, 28, 28, device="cuda")
program.add_tensor_source("train", images)

program.zero_grad()
loss = program.forward_backward_tensor("addition(0, 1, 7)")  # CUDA scalar tensor (no host reads required)
program.optimizer_step()

# Optional host read for logging:
print(float(loss.item()))

# Process multiple inputs
for batch in data_loader:
    edge_rows = batch["edges"].to(device="cuda")
    edge_src = edge_rows[:, 0].contiguous()
    edge_dst = edge_rows[:, 1].contiguous()
    results = program.evaluate(dlpack_inputs={
        "edge": [edge_src, edge_dst],
    })
    # Process results...