cegraph 0.1.0

Causal-aware execution runtime for production Python systems.

cegraph

Causal-aware execution runtime for production Python systems.

cegraph tracks causal dependencies between Python functions, records execution traces, runs lightweight counterfactual simulations, and performs adaptive fallback when constraints are violated. Built for MLOps, dynamic decisioning, and root-cause analysis -- no heavy dependencies beyond numpy.

Features

• @causal_node decorator -- Annotate functions with sensitivity flags and constraint checks, with under 7% overhead vs native calls
• CausalGraph -- Directed graph with DFS cycle detection and type-level edge validation. Pure Python, no networkx
• CausalTracer -- Lock-free ring buffer tracer with adaptive sampling. O(1) append, lossy on overflow
• Counterfactual engine -- Deterministic perturbation simulation (seed=42) with per-node impact scores and confidence
• Fallback optimizer -- Evaluate constraints (latency, confidence, critical) and dispatch cache/bypass/error fallback in order

Quick Start

from cegraph import causal_node, CausalGraph, Context, counterfactual, optimize

@causal_node(sensitivity=["price"])
def fetch_price(symbol: str) -> dict:
    return {"price": 100.0, "symbol": symbol}

@causal_node(sensitivity=["price"])
def compute_markup(data: dict) -> float:
    return data["price"] * 1.1

@causal_node(constraint=lambda x: x > 0)
def apply_strategy(base: float) -> float:
    return base * 1.15

graph = CausalGraph()
graph.connect(fetch_price, compute_markup)
graph.connect(compute_markup, apply_strategy)
graph.validate()

with Context(graph=graph, buffer_size=1000) as ctx:
    data = fetch_price("AAPL")
    base = compute_markup(data)
    result = apply_strategy(base)

cf = counterfactual(
    base_trace=ctx.tracer.records,
    interventions={"fetch_price": {"price": 150.0}},
)
print(f"Impact: {cf.overall_impact:.2f}, Confidence: {cf.confidence:.2f}")

result = optimize(ctx, constraints={"max_latency_ms": 50.0})
print(f"Status: {result.status}")

Installation

pip install cegraph

Requires Python 3.10+ and numpy 1.24+.

Use Cases

MLOps & Model Drift

Production models degrade silently. Isolate the causal node responsible, run counterfactual simulations, and route to fallback automatically.

@causal_node(sensitivity=["feature_distribution"])
def encode_features(raw: dict) -> dict:
    return preprocessor.transform(raw)

@causal_node(sensitivity=["feature_distribution"], constraint=lambda x: x > 0.5)
def predict(features: dict) -> float:
    return model.predict(features)

@causal_node()
def fallback_predict(features: dict) -> float:
    return ensemble_fallback(features)

with Context(buffer_size=5000) as ctx:
    score = predict(encode_features(input_data))

When predict violates its confidence constraint, optimize() returns fallback_cache or fallback_bypass with recommendations.

Dynamic Decisioning

Real-time what-if simulation. "If I raise price 5%, what's the causal impact?"

cf = counterfactual(
    base_trace=ctx.tracer.records,
    interventions={"pricing_model": {"price_multiplier": 1.05}},
    n_perturbations=100,
)
print(f"Estimated impact: {cf.overall_impact:.3f}")
print(f"Sensitivity ranking: {cf.node_impacts}")

Root-Cause Analysis

Replace correlation-timing debugging with explicit causal traces.

summary = ctx.tracer.summary()
for node, stats in summary.items():
    print(f"{node}: count={stats['count']}, "
          f"mean={stats['mean_latency']:.2f}ms, "
          f"p95={stats['p95_latency']:.2f}ms")

API Reference

API	Description
@causal_node(sensitivity, constraint, low_sensitivity)	Decorate a function for causal tracking. constraint raises CausalConstraintViolation on failure.
NodeMetadata	Attached to decorated functions as __cegraph_meta__.
CausalGraph.connect(src, dst)	Register a causal edge. validate() runs DFS cycle detection + type checking.
CausalGraph.ancestors(fn) / descendants(fn)	Traverse upstream/downstream causal dependencies.
Context(graph, buffer_size, sample_rate)	Session scope. Binds tracer to all @causal_node calls via threading.local.
CausalTracer	Ring buffer (deque(maxlen=N)) with adaptive sampling. Overflow warning once.
TraceRecord	__slots__ dataclass per node execution. Fields: node_name, input_hash, output_summary, latency_ms, sensitivity_flags, timestamp, constraint_passed.
counterfactual(base_trace, interventions, ...)	Deterministic perturbation engine. Returns CounterfactualResult with per-node ImpactScore.
optimize(context, objective, constraints)	Evaluate constraints in order: max_latency_ms -> min_confidence -> critical. Returns OptimizationResult with fallback status.

Exceptions

Exception	Raised when
CegraphError	Base class for all cegraph errors.
CausalCycleError	A cycle is detected in the causal graph.
CausalTypeError	Output type of src node mismatches input type of dst node.
CausalConstraintViolation	A node's constraint function returns False.
TracerOverflowWarning	Ring buffer is full and overwriting old records (warning, not exception).

Contributing

See CONTRIBUTING.md for development setup, code style, and pull request process. All contributions are welcome.

License

MIT -- see LICENSE for details.