khalilbenaz/claude-skills-collection/agent-consensus-builder

def consensus_necessity_score(decision: dict) -> int:
    score = 0
    if decision["is_reversible"] is False: score += 2
    if decision["uncertainty"] == "high": score += 2
    if decision["perspectives_needed"] > 1: score += 1
    # score >= 3 → consensus justifié
    # score < 3 → agent unique suffisant
    return score

CONSENSUS_PROMPT_TEMPLATE = """
Tu es un agent spécialisé en {role}.
Question : {question}
Options disponibles : {options}
Contexte : {context}
Réponds en JSON :
{{
  "choice": "<une des options>",
  "confidence": <float 0.0-1.0>,
  "rationale": "<explication concise>",
  "risks": ["<risque 1>", "<risque 2>"]
}}
"""

from collections import Counter
from typing import Any
def majority_vote(votes: list[str]) -> str:
    counts = Counter(votes)
    winner, count = counts.most_common(1)[0]
    return winner if count > len(votes) / 2 else "no_consensus"
def confidence_weighted_consensus(votes: list[dict[str, Any]]) -> dict:
    # votes = [{"choice": "A", "confidence": 0.8}, ...]
    scores: dict[str, float] = {}
    for v in votes:
        scores[v["choice"]] = scores.get(v["choice"], 0) + v["confidence"]
    total = sum(scores.values())
    normalized = {k: round(v / total, 3) for k, v in scores.items()}
    winner = max(normalized, key=normalized.get)
    return {"winner": winner, "scores": normalized, "strength": normalized[winner]}
def borda_count(rankings: list[list[str]]) -> str:
    # rankings = [["A","B","C"], ["B","A","C"], ...]
    n = len(rankings[0])
    scores: dict[str, int] = {}
    for ranking in rankings:
        for i, option in enumerate(ranking):
            scores[option] = scores.get(option, 0) + (n - i - 1)
    return max(scores, key=scores.get)

AGENT_CONFIGS = [
    {"role": "risk_analyst",   "temperature": 0.2, "framing": "Quels sont les risques ?"},
    {"role": "optimist",       "temperature": 0.7, "framing": "Quels sont les gains potentiels ?"},
    {"role": "devils_advocate","temperature": 0.5, "framing": "Pourquoi cette option échouerait-elle ?"},
    {"role": "neutral_analyst","temperature": 0.3, "framing": "Évalue objectivement chaque option."},
]
# Règle : jamais 2 agents avec le même role + temperature + framing

import asyncio
async def run_vote(agents: list, question: str, options: list[str]) -> list[dict]:
    tasks = [agent.vote(question, options) for agent in agents]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # Filtrer les erreurs sans bloquer le vote
    valid = [r for r in results if isinstance(r, dict)]
    if len(valid) < len(agents) // 2 + 1:
        raise RuntimeError("Trop d'agents en échec pour un consensus fiable")
    return valid

async def run_debate(agents: list, question: str, options: list[str]) -> dict:
    # Round 1 : positions initiales (parallèle)
    positions = await run_vote(agents, question, options)
    # Round 2 : cross-examination (chaque agent voit les autres)
    critiques = await asyncio.gather(*[
        agent.critique(question, positions, own_idx=i)
        for i, agent in enumerate(agents)
    ])
    # Round 3 : vote final avec toutes les informations
    final_votes = await asyncio.gather(*[
        agent.final_vote(question, options, positions, critiques)
        for agent in agents
    ])
    return {"votes": final_votes, "positions": positions, "critiques": critiques}

def resolve_deadlock(votes: list[str], options: list[str], safety_order: list[str]) -> str:
    """safety_order = options classées de la plus sûre à la plus risquée"""
    counts = Counter(votes)
    max_count = max(counts.values())
    tied = [o for o, c in counts.items() if c == max_count]
    # Préférer l'option la plus sûre parmi les ex-aequo
    for safe_option in safety_order:
        if safe_option in tied:
            return safe_option
    return tied[0]  # fallback

def validate_consensus(votes: list[dict], winner: str) -> dict:
    total = len(votes)
    winner_votes = [v for v in votes if v["choice"] == winner]
    minority_votes = [v for v in votes if v["choice"] != winner]
    strength = len(winner_votes) / total
    return {
        "winner": winner,
        "consensus_strength": round(strength, 3),
        "is_strong": strength >= 0.66,
        "requires_human_review": strength < 0.51,
        "minority_report": {
            "options": list({v["choice"] for v in minority_votes}),
            "rationales": [v.get("rationale") for v in minority_votes],
            "risks_raised": [r for v in minority_votes for r in v.get("risks", [])],
        },
    }

async def adaptive_consensus(agents_pool: list, question: str, options: list) -> dict:
    # Étape 1 : vote rapide avec 2 agents (modèle léger possible)
    quick_votes = await run_vote(agents_pool[:2], question, options)
    result = confidence_weighted_consensus(quick_votes)
    if result["strength"] >= 0.85:
        return {**result, "method": "quick_vote", "agents_used": 2}
    # Étape 2 : vote étendu si pas de consensus fort
    full_votes = await run_vote(agents_pool, question, options)
    result = confidence_weighted_consensus(full_votes)
    if result["strength"] >= 0.66:
        return {**result, "method": "full_vote", "agents_used": len(agents_pool)}
    # Étape 3 : débat structuré si toujours pas de consensus
    debate_result = await run_debate(agents_pool, question, options)
    final = confidence_weighted_consensus(debate_result["votes"])
    return {**final, "method": "full_debate", "agents_used": len(agents_pool)}

import time
def build_audit_log(session_id: str, question: str, options: list,
                    agents_config: list, result: dict, duration_ms: int) -> dict:
    return {
        "session_id": session_id,
        "timestamp": time.time(),
        "question": question,
        "options": options,
        "agents": agents_config,           # roles, temperatures, framings
        "method": result["method"],
        "winner": result["winner"],
        "consensus_strength": result["strength"],
        "minority_report": result.get("minority_report"),
        "duration_ms": duration_ms,
    }
# Stocke en base ou fichier JSON — indispensable pour débugger les désaccords récurrents