Probabilistic-Path-Pruning

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 pp_v2.py
{
  "pruned_path_graph": {
    "nodes": [
      {
        "id": 0,
        "name": "ProjectStart",
        "required_resource": 13.15,
        "base_feasibility": 0.32,
        "status": "pruned"
      },
      {
        "id": 1,
        "name": "Requirements",
        "required_resource": 6.23,
        "base_feasibility": 0.46,
        "status": "pruned"
      },
      {
        "id": 2,
        "name": "Design",
        "required_resource": 14.99,
        "base_feasibility": 0.77,
        "status": "pruned"
      },
      {
        "id": 3,
        "name": "Prototype",
        "required_resource": 17.95,
        "base_feasibility": 0.36,
        "status": "kept"
      },
      {
        "id": 4,
        "name": "BackendImpl",
        "required_resource": 9.02,
        "base_feasibility": 0.32,
        "status": "pruned"
      },
      {
        "id": 5,

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 220 lines, one file, standard library only.

#!/usr/bin/env python3
"""Probabilistic-Path-Pruning v2: Path-Probability-Score (PPS) calculation and path ranking."""

import math
import random
import sys
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Tuple

class Subtask:
    id: int
    name: str
    required_resource: float
    base_feasibility: float
    depends_on: List[int] = field(default_factory=list)

    def feasibility(self, resource: float) -> float:
        if resource < self.required_resource:
            return 0.0
        surplus = resource - self.required_resource
        return self.base_feasibility * (1.0 - math.exp(-surplus / max(self.required_resource, 1e-9)))


class Particle:
    subtask_id: int
    resource: float
    path: Tuple[int, ...]
    weight: float = 1.0


class TaskGraph:
    nodes: List[Subtask]
    edges: List[Tuple[int, int]]
    successors: Dict[int, List[int]] = field(default_factory=dict)
    predecessors: Dict[int, List[int]] = field(default_factory=dict)

    def __post_init__(self):
        for f, t in self.edges:
            self.successors.setdefault(f, []).append(t)
            self.predecessors.setdefault(t, []).append(f)


class PathPruner:
    def __init(
        self,
        graph: TaskGraph,
        num_particles: int = 500,
        noise_sigma: float = 0.15,
        resample_threshold: float = 0.5,
        seed: Optional[int] = None,
    ) -> None:
        self.graph = graph
        self.num_particles = num_particles
        self.noise_sigma = noise_sigma
        self.resample_threshold = resample_threshold
        self.rng = random.Random(seed)
        self.particles: List[Particle] = []
        self._initialize_particles()

    def _initialize_particles(self):
        entry_nodes = [
            i for i, n in enumerate(self.graph.nodes)
            if not self.graph.predecessors.get(i)
        ]
        if not entry_nodes:
            entry_nodes = [0]
        self.particles = []
        for _ in range(self.num_particles):
            start_id = self.rng.choice(entry_nodes)
            node = self.graph.nodes[start_id]
            initial_resource = self.rng.gauss(
                mu=node.required_resource * 1.5, sigma=node.required_resource * 0.3
            )
            initial_resource = max(0.0, initial_resource)
            self.particles.append(
                Particle(
                    subtask_id=start_id,
                    resource=initial_resource,
                    path=(start_id,),
                    weight=1.0 / self.num_particles,
                )
            )

    def _neff(self) -> float:
        w_sum = sum(p.weight for p in self.particles)
        if w_sum == 0:
            return 0.0
        w_sq_sum = sum(p.weight ** 2 for p in self.particles)
        return w_sum ** 2 / w_sq_sum if w_sq_sum > 0 else 0.0

    def _systematic_resample(self) -> None:
        weights = [p.weight for p in self.particles]
        total = sum(weights)
        if total == 0:
            self._initialize_particles()
            return
        n = self.num_particles
        positions = [self.rng.random() * total / n + (i * total / n) for i in range(n)]
        indices = []
        cumulative = 0.0
        j = 0
        for i, w in enumerate(weights):
            cumulative += w
            while j < n and cumulative > positions[j]:
                indices.append(i)
                j += 1
        new_particles: List[Particle] = []
        for idx in indices:
            src = self.particles[idx]
            new_particles.append(
                Particle(
                    subtask_id=src.subtask_id,
                    resource=src.resource,
                    path=src.path,
                    weight=1.0 / n,
                )
            )
        self.particles = new_particles

    def step(self) -> None:
        self._predict()
        self._update()
        if self._neff() < self.num_particles * self.resample_threshold:
            self._resample()

    def _predict(self) -> None:
        new_particles: List[Particle] = []
        for p in self.particles:
            successors = self.graph.successors.get(p.subtask_id, [])
            if not successors:
                new_particles.append(p)
                continue
            next_id = self.rng.choice(successors)
            next_node = self.graph.nodes[next_id]
            resource_cost = next_node.required_resource
            noise = self.rng.gauss(mu=0.0, sigma=self.noise_sigma * resource_cost)
            new_resource = p.resource - max(0.0, resource_cost + noise)
            if new_resource < 0:
                new_resource = 0.0
            new_path = p.path + (next_id,)
            new_particles.append(
                Particle(
                    subtask_id=next_id,
                    resource=new_resource,
                    path=new_path,
                    weight=p.weight,
                )
            )
        self.particles = new_particles

    def _update(self) -> None:
        for p in self.particles:
            node = self.graph.nodes[p.subtask_id]
            feasibility = node.feasibility(p.resource)
            p.weight *= max(feasibility, 1e-12)

    def _resample(self) -> None:
        self._systematic_resample()

    def run(self, steps: int) -> List[Particle]:
        for _ in range(steps):
            self.step()
        return self.particles

    def path_entropy(self, path: Tuple[int, ...]) -> float:
        entropy = 0.0
        for i in range(len(path) - 1):
            node_id = path[i]
            num_successors = len(self.graph.successors.get(node_id, []))
            if num_successors > 1:
                entropy += math.log2(num_successors)
        return entropy

    def top_paths(self, k: int = 3) -> List[Tuple[Tuple[int, ...], float]]:
        path_weights: Dict[Tuple[int, ...], float] = defaultdict(float)
        for p in self.particles:
            path_weights[p.path] += p.weight

        # Calculate Path-Probability-Score (PPS)
        pps_scores = []
        for path, weight in path_weights.items():
            entropy = self.path_entropy(path)
            pps = weight * math.exp(-entropy * 0.5)
            pps_scores.append((path, pps))

        # Sort by PPS descending
        pps_scores.sort(key=lambda x: x[1], reverse=True)
        return pps_scores[:k]


if __name__ == "__main__":
    # Hard-coded example graph
    nodes = [
        Subtask(id=0, name="Start", required_resource=10.0, base_feasibility=0.8, depends_on=[]),
        Subtask(id=1, name="Task A", required_resource=20.0, base_feasibility=0.7, depends_on=[0]),
        Subtask(id=2, name="Task B", required_resource=15.0, base_feasibility=0.6, depends_on=[0]),
        Subtask(id=3, name="End", required_resource=5.0, base_feasibility=1.0, depends_on=[1, 2]),
    ]
    edges = [
        (0, 1), (0, 2), (1, 3), (2, 3)
    ]
    graph = TaskGraph(nodes=nodes, edges=edges)

    pruner = ProbabilisticPathPruner(graph=graph, num_particles=300, seed=42)
    pruner.run(steps=10)
    top_paths = pruner.top_paths(k=3)

    print("Top Paths by Path-Probability-Score (PPS):")
    for path, score in top_paths:
        print(f"Path: {path}, PPS: {score:.4f}")

    # Original top_paths behavior preserved for comparison
    print("\nTop Paths by Total Weight:")
    path_weights = defaultdict(float)
    for p in pruner.particles:
        path_weights[p.path] += p.weight
    weighted_paths = sorted(path_weights.items(), key=lambda x: x[1], reverse=True)
    for path, weight in weighted_paths[:3]:
        print(f"Path: {path}, Weight: {weight:.4f}")