Stateful-Rate-Limited Path Scorer

The problem

Complex problem-solving often involves exploring many different logical paths, which can quickly exhaust available computing resources or lead to dead ends.

What it does

It evaluates different lines of reasoning by scoring them based on logical progress while simultaneously tracking how many resources are left to spend.

Why it matters

It allows for more efficient problem-solving by pruning unproductive ideas before they waste resources.

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 script.py
Path Scorer — Results
==================================================

Path 1  (score=0.877  status=viable)
  - How to optimize a sorting network
  - Decompose 'How to optimize' into smaller parts
  - Decompose 'Decompose 'How ' into smaller parts
  - Find an analogy for 'Decompose 'Deco' in physics
  - Decompose 'Find an analogy' into smaller parts
  - Decompose 'Decompose 'Find' into smaller parts

Path 2  (score=0.877  status=viable)
  - How to optimize a sorting network
  - Test 'How to optimize' via a numeric example
  - Decompose 'Test 'How to op' into smaller parts
  - Find an analogy for 'Decompose 'Test' in physics
  - Find an analogy for 'Find an analogy' in physics
  - Decompose 'Find an analogy' into smaller parts

Path 3  (score=0.877  status=viable)
  - How to optimize a sorting network
  - Decompose 'How to optimize' into smaller parts
  - Decompose 'Decompose 'How ' into smaller parts
  - Test 'Decompose 'Deco' via a numeric example
  - Find an analogy for 'Test 'Decompose' in physics
  - Decompose 'Find an analogy' into smaller parts

Path 4  (score=0.877  status=viable)
  - How to optimize a sorting network
  - Find an analogy for 'How to optimize' in physics

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 — 271 lines, one file, standard library only.

#!/usr/bin/env python3

import heapq
import math
import random
import time
from dataclasses import dataclass, field
from typing import Callable, Optional


# ─── Token Bucket ────────────────────────────────────────────────────────────

@dataclass
class TokenBucket:
    max_tokens: float
    refill_rate: float
    tokens: float = field(init=False)
    last_refill: float = field(default_factory=time.monotonic)

    def __post_init__(self) -> None:
        self.tokens = float(self.max_tokens)

    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.max_tokens, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

    def consume(self, tokens: int = 1) -> bool:
        self._refill()
        if self.tokens >= tokens:
            self.tokens -= tokens
            return True
        return False

    @property
    def available(self) -> float:
        self._refill()
        return self.tokens

    @property
    def depleted(self) -> bool:
        return self.available < 1.0


# ─── Thought Tree ────────────────────────────────────────────────────────────

@dataclass
class Thought:
    text: str
    score: float = 0.0
    parent: Optional["Thought"] = None
    children: list["Thought"] = field(default_factory=list)
    bucket: Optional[TokenBucket] = None
    depth: int = 0

    def __post_init__(self) -> None:
        if self.parent is not None:
            self.depth = self.parent.depth + 1

    def __lt__(self, other: "Thought") -> bool:
        return self.score > other.score


@dataclass
class ThoughtTree:
    root: Thought
    evaluate_fn: Callable[[Thought], float]
    branch_fn: Callable[[Thought], list[str]]
    width: int = 3
    max_depth: int = 5
    bucket_capacity: int = 10
    bucket_fill_rate: float = 2.0
    cost_per_thought: int = 1
    nodes_visited: int = 0
    nodes_pruned: int = 0

    def run(self) -> list[Thought]:
        results: list[Thought] = []
        frontier: list[tuple[int, Thought]] = [(0, self.root)]
        visited: set[int] = {id(self.root)}
        self.nodes_visited = 1

        while frontier:
            _, node = frontier.pop(0)

            if node.depth >= self.max_depth:
                results.append(node)
                continue

            candidates = self.branch_fn(node)
            if not candidates:
                if node.depth > 0:
                    results.append(node)
                continue

            scored: list[Thought] = []
            for child in candidates:
                self.nodes_visited += 1
                if id(child) in visited:
                    continue
                visited.add(id(child))

                if node.bucket is not None and not node.bucket.consume(self.cost_per_thought):
                    self.nodes_pruned += 1
                    continue

                quality = self.evaluate_fn(child)
                bucket_health = node.bucket.available / node.bucket.max_tokens if node.bucket else 1.0
                child.score = quality * 0.6 + bucket_health * 0.4
                child.bucket = TokenBucket(
                    max_tokens=self.bucket_capacity,
                    refill_rate=self.bucket_fill_rate,
                )
                scored.append(child)

            scored.sort(key=lambda t: t.score, reverse=True)
            frontier.extend([(None, s) for s in scored[:self.width]])

        return results

    def _best_path(self, leaf: Thought) -> list[str]:
        path: list[str] = []
        node: Optional[Thought] = leaf
        while node:
            path.append(node.text)
            node = node.parent
        path.reverse()
        return path

    def best_paths(self, k: Optional[int] = None) -> list[tuple[list[str], float]]:
        leaves = self.run()
        leaves.sort(key=lambda t: t.score, reverse=True)
        if k:
            leaves = leaves[:k]
        return [(self._best_path(leaf), leaf.score) for leaf in leaves]


# ─── Pruning & Scoring Helpers ───────────────────────────────────────────────

def phoenix_logical_progression_score(thought: Thought) -> float:
    txt = thought.text.lower()
    base = 0.0

    words = len(txt.split())
    if words < 3:
        base += 0.1
    elif words > 50:
        base += 0.3
    else:
        base += 0.7

    reasoning_terms = ["therefore", "because", "since", "if", "then", "implies", "infer"]
    hits = sum(1 for t in reasoning_terms if t in txt)
    base += min(hits * 0.1, 0.3)

    contradiction_terms = ["however", "but", "although", "nevertheless", "contradict"]
    penalty = sum(1 for t in contradiction_terms if t in txt)
    base -= min(penalty * 0.05, 0.25)

    base += min(thought.depth * 0.03, 0.15)

    return min(max(base, 0.0), 1.0)


class PathScorer:
    """Top-level API: scores and prunes reasoning paths with resource-awareness."""

    def __init__(
        self,
        *,
        bucket_capacity: int = 15,
        fill_rate: float = 3.0,
        cost_per_thought: int = 1,
        width: int = 3,
        max_depth: int = 6,
    ) -> None:
        self.bucket_capacity = bucket_capacity
        self.fill_rate = fill_rate
        self.cost_per_thought = cost_per_thought
        self.width = width
        self.max_depth = max_depth

    def score_paths(
        self,
        *,
        root_text: str = "Problem: solve the given task",
        branch_fn: Callable[[Thought], list[str]] | None = None,
        evaluate_fn: Callable[[Thought], float] | None = None,
    ) -> list[tuple[list[str], float, str]]:
        eval_fn = evaluate_fn or phoenix_logical_progression_score
        wrapped_branch_fn = self._wrap_branch_fn(branch_fn or self._default_branch_fn)

        root_thought = Thought(text=root_text)
        root_thought.bucket = TokenBucket(
            max_tokens=self.bucket_capacity,
            refill_rate=self.fill_rate,
        )

        tree = ThoughtTree(
            root=root_thought,
            evaluate_fn=eval_fn,
            branch_fn=wrapped_branch_fn,
            width=self.width,
            max_depth=self.max_depth,
            bucket_capacity=self.bucket_capacity,
            bucket_fill_rate=self.fill_rate,
            cost_per_thought=self.cost_per_thought,
        )

        paths = tree.best_paths()

        results: list[tuple[list[str], float, str]] = []
        for path, score in paths:
            bucket = TokenBucket(max_tokens=self.bucket_capacity, refill_rate=self.fill_rate)
            cost = len(path) * self.cost_per_thought
            resource_ok = bucket.available / bucket.max_tokens >= 0.1
            label = "viable" if resource_ok else "starved"
            results.append((path, score, label))

        return results

    @staticmethod
    def _wrap_branch_fn(raw_fn: Callable[[Thought], list[str]]) -> Callable[[Thought], list[Thought]]:
        def wrapped(thought: Thought) -> list[Thought]:
            texts = raw_fn(thought)
            return [Thought(text=t, parent=thought) for t in texts]
        return wrapped

    @staticmethod
    def _default_branch_fn(thought: Thought) -> list[str]:
        templates = [
            f"Extend {thought.text[:20]} via analogy to {random.choice(['physics','biology','economics'])}",
            f"Decompose {thought.text[:15]} into sub-steps {random.randint(2,5)}",
            f"Hypothesize a counter-example to {thought.text[:10]}",
            f"Validate {thought.text[:12]} against first principles",
        ]
        random.shuffle(templates)
        return templates[:3]


# ─── Demo ─────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    scorer = PathScorer(
        bucket_capacity=12,
        fill_rate=2.0,
        max_depth=5,
        width=3,
    )

    def demo_branch(thought: Thought) -> list[str]:
        directions = [
            f"Decompose '{thought.text[:15]}' into smaller parts",
            f"Find an analogy for '{thought.text[:15]}' in physics",
            f"Test '{thought.text[:15]}' via a numeric example",
            f"Refute '{thought.text[:15]}' by contradiction",
            f"Generalize '{thought.text[:15]}' to a broader class",
        ]
        return directions[:3]

    results = scorer.score_paths(root_text="How to optimize a sorting network", branch_fn=demo_branch)

    print("Path Scorer — Results\n" + "=" * 50)
    for i, (path, score, health) in enumerate(results[:5], 1):
        print(f"\nPath {i}  (score={score:.3f}  status={health})")
        for step in path:
            print(f"  - {step}")

    print("\n" + "=" * 50)
    print(f"Built {len(results)} complete path(s).")