76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""Weighted consensus engine for multi-agent decisions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_WEIGHTS = {
|
|
"custom_gpt": 2,
|
|
"windows_codex": 2,
|
|
"server_agent": 1,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class Vote:
|
|
voter_id: str
|
|
voter_type: str
|
|
choice: str
|
|
confidence: float = 1.0
|
|
rationale: str = ""
|
|
|
|
def weight(self, weights: dict[str, int]) -> float:
|
|
return float(weights.get(self.voter_type, weights.get("server_agent", 1))) * max(0.0, min(1.0, self.confidence))
|
|
|
|
|
|
class ConsensusEngine:
|
|
def __init__(self, weights: dict[str, int] | None = None) -> None:
|
|
self.weights = dict(DEFAULT_WEIGHTS)
|
|
if weights:
|
|
self.weights.update(weights)
|
|
|
|
def weighted_vote(self, votes: list[Vote]) -> dict[str, Any]:
|
|
totals: dict[str, float] = {}
|
|
for vote in votes:
|
|
totals[vote.choice] = totals.get(vote.choice, 0.0) + vote.weight(self.weights)
|
|
if not totals:
|
|
return {"decision": "owner_escalation_required", "totals": {}, "tie": True}
|
|
ordered = sorted(totals.items(), key=lambda item: (-item[1], item[0]))
|
|
top_score = ordered[0][1]
|
|
winners = [choice for choice, score in ordered if score == top_score]
|
|
if len(winners) > 1:
|
|
return self.owner_escalation(votes, totals, reason="weighted_tie")
|
|
return {"decision": winners[0], "totals": totals, "tie": False}
|
|
|
|
def conflict_detection(self, votes: list[Vote]) -> dict[str, Any]:
|
|
choices = sorted({vote.choice for vote in votes})
|
|
return {"conflict": len(choices) > 1, "choices": choices, "vote_count": len(votes)}
|
|
|
|
def owner_escalation(self, votes: list[Vote], totals: dict[str, float] | None = None, reason: str = "conflict") -> dict[str, Any]:
|
|
return {
|
|
"decision": "owner_escalation_required",
|
|
"reason": reason,
|
|
"tie": True,
|
|
"totals": totals or {},
|
|
"review_required": "owner/admin",
|
|
"votes": [
|
|
{
|
|
"voter_id": vote.voter_id,
|
|
"voter_type": vote.voter_type,
|
|
"choice": vote.choice,
|
|
"confidence": vote.confidence,
|
|
"rationale": vote.rationale,
|
|
}
|
|
for vote in votes
|
|
],
|
|
}
|
|
|
|
def decide(self, votes: list[Vote]) -> dict[str, Any]:
|
|
conflict = self.conflict_detection(votes)
|
|
result = self.weighted_vote(votes)
|
|
result["conflict"] = conflict["conflict"]
|
|
result["choices"] = conflict["choices"]
|
|
return result
|