Consensus-Driven Path Ranking

Validation

It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.

$ python3 consensus_path_ranking.py
Traceback (most recent call last):
  File "/work/consensus_path_ranking.py", line 118, in <module>
    main()
  File "/work/consensus_path_ranking.py", line 95, in main
    input_data = json.load(sys.stdin)
                 ^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/json/__init__.py", line 293, in load
    return loads(fp.read(),
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/json/__init__.py", line 346, in loads
    return _default_decoder.decode(s)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/json/decoder.py", line 338, in decode
    obj, end = self.raw_decode(s, idx=_w(s, 0).end())
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/json/decoder.py", line 356, in raw_decode
    raise JSONDecodeError("Expecting value", s, err.value) from None
json.decoder.JSONDecodeError: Expecting value: line 1 column 1 (char 0)

No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.

The code

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

# Consensus-Driven Path Ranking Script
import json
import os
import random
import sys
from typing import List, Dict, Tuple, Optional

class Agent:
    def __init__(self, name: str, weights: Dict[str, float]) -> None:
        self.name = name
        self.weights = weights
        self.bias = random.random()

    def score_path(self, path: dict, criteria: Dict[str, float]) -> float:
        path_data = isinstance(path, dict) and path or {}
        score = 0.0
        for c_key in criteria:
            if c_key in self.weights:
                val = path_data.get(c_key, 0.0)
                if isinstance(val, dict):
                    val = val.get("ratio", val.get("value", 0.0))
                score += float(val) * self.weights[c_key]
        return score

    def debate(self, other_agent: 'Agent', path: dict, criteria: Dict[str, float]) -> float:
        my_score = self.score_path(path, criteria)
        their_score = other_agent.score_path(path, criteria)
        return (my_score + their_score + random.uniform(-0.1, 0.1)) / 2

class PathRanker:
    def __init__(self, agents: List[Agent]) -> None:
        self.agents = agents

    def get_path_id(self, path: dict) -> str:
        return path.get("id", path.get("task", path.get("name", id(path))))

    def rank_paths(self, paths: list, criteria: Dict[str, float]) -> List[Dict]:
        results = []
        self.criteria = criteria
        for path in paths:
            path_dict = {}
            if isinstance(path, dict):
                path_dict = path
            elif isinstance(path, str):
                path_dict = {"id": path}
            else:
                path_dict = {"id": path.get("id") if isinstance(path, dict) else str(path)}

            consensus_score = sum(
                a.debate(random.choice(self.agents), path_dict, criteria)
                for a in self.agents
            ) / len(self.agents)
            path_id = path_dict.get("id", path_dict.get("task", ""))
            entry = {"id": path_id, "score": round(consensus_score, 4)}
            if path_id and path_id != entry.get("task"):
                pass
            results.append(entry)
        results.sort(key=lambda x: x["score"], reverse=True)
        return results

def resolve_input_paths(input_data: dict) -> list:
    for key in ["path_scenarios", "paths", "tasks"]:
        val = input_data.get(key)
        if val and isinstance(val, list):
            return val
    return input_data.get("paths", input_data.get("tasks", []))

def resolve_input_criteria(input_data: dict) -> Dict[str, float]:
    raw = input_data.get("criteria", input_data.get("path_scores", input_data.get("path_score", {})))
    if isinstance(raw, dict):
        return {k: float(v.get("ratio", v) if isinstance(v, dict) else v) for k, v in raw.items() if float(v) > 0}
    return {"cost": 0.5, "risk": 0.5}

def resolve_agents_config(input_data: dict) -> list:
    for key in ["agents", "agent_config"]:
        val = input_data.get(key)
        if val and isinstance(val, list):
            return val
    return [
        {"name": "CostOptimist", "weights": {"cost": 0.7, "risk": 0.3}},
        {"name": "RiskAverter", "weights": {"cost": 0.4, "risk": 0.6}},
        {"name": "Innovator", "weights": {"cost": 0.3, "risk": 0.7}},
    ]

def main():
    if len(sys.argv) > 1 and sys.argv[1] == "input.json":
        input_path = sys.argv[1]
    else:
        input_path = "input.json"

    if os.path.exists(input_path):
        with open(input_path, "r") as f:
            input_data = json.load(f)
    else:
        input_data = json.load(sys.stdin)

    paths = resolve_input_paths(input_data)
    criteria = resolve_input_criteria(input_data)
    agents_config = resolve_agents_config(input_data)

    agents = [Agent(a["name"], a["weights"]) for a in agents_config]

    ranker = PathRanker(agents)
    ranked = ranker.rank_paths(paths, criteria)

    output = {
        "ranked_paths": ranked,
        "agents": [{"name": a.name, "weights": a.weights} for a in agents],
    }

    output_path = "output.json"
    with open(output_path, "w") as f:
        json.dump(output, f, indent=2)
    json.dump(output, sys.stdout, indent=2)
    sys.stdout.write("\n")

if __name__ == '__main__':
    main()