khalilbenaz/claude-skills-collection/agent-conflict-resolver

from dataclasses import dataclass
from typing import Any
@dataclass
class ConflictDetectionResult:
    has_conflict: bool
    conflict_type: str   # "direct_contradiction" | "numerical_inconsistency" | "logical_conflict" | "priority_conflict"
    severity: str        # "critical" | "major" | "minor"
    agent_a: str
    agent_b: str
    output_a: Any
    output_b: Any
    description: str
def detect_conflict(output_a: Any, agent_a: str, output_b: Any, agent_b: str) -> ConflictDetectionResult:
    # Détection numérique
    if isinstance(output_a, (int, float)) and isinstance(output_b, (int, float)):
        diff_pct = abs(output_a - output_b) / max(abs(output_a), abs(output_b), 1)
        if diff_pct > 0.10:
            return ConflictDetectionResult(
                has_conflict=True, conflict_type="numerical_inconsistency",
                severity="major" if diff_pct > 0.30 else "minor",
                agent_a=agent_a, agent_b=agent_b,
                output_a=output_a, output_b=output_b,
                description=f"Écart de {diff_pct:.1%} entre {agent_a} et {agent_b}"
            )
    # Détection booléenne
    if isinstance(output_a, bool) and isinstance(output_b, bool) and output_a != output_b:
        return ConflictDetectionResult(
            has_conflict=True, conflict_type="direct_contradiction", severity="critical",
            agent_a=agent_a, agent_b=agent_b, output_a=output_a, output_b=output_b,
            description=f"{agent_a} dit {output_a}, {agent_b} dit {output_b}"
        )
    return ConflictDetectionResult(
        has_conflict=False, conflict_type="none", severity="none",
        agent_a=agent_a, agent_b=agent_b, output_a=output_a, output_b=output_b, description=""
    )

async def gather_evidence(conflict: ConflictDetectionResult, agents: dict) -> dict:
    prompt = (
        "Explique ton raisonnement étape par étape, liste tes sources, "
        "donne ton score de confiance (0–1) et identifie tes hypothèses."
    )
    evidence_a = await agents[conflict.agent_a].justify(output=conflict.output_a, prompt=prompt)
    evidence_b = await agents[conflict.agent_b].justify(output=conflict.output_b, prompt=prompt)
    return {
        "agent_a": {"output": conflict.output_a, "evidence": evidence_a},
        "agent_b": {"output": conflict.output_b, "evidence": evidence_b},
    }

class ConflictResolutionStrategy:
    @staticmethod
    def confidence_based(evidence: dict) -> dict:
        conf_a = evidence["agent_a"]["evidence"].get("confidence", 0.5)
        conf_b = evidence["agent_b"]["evidence"].get("confidence", 0.5)
        if abs(conf_a - conf_b) < 0.1:
            return {"winner": None, "method": "confidence_tie", "needs_escalation": True}
        winner = "agent_a" if conf_a > conf_b else "agent_b"
        return {"winner": winner, "method": "confidence_based", "confidence_delta": abs(conf_a - conf_b)}
    @staticmethod
    def source_verification(conflict: ConflictDetectionResult, external_source) -> dict:
        ground_truth = external_source.lookup(conflict.output_a, conflict.output_b)
        winner = "agent_a" if ground_truth == conflict.output_a else "agent_b"
        return {"winner": winner, "method": "source_verification", "ground_truth": ground_truth}

ARBITRATOR_PROMPT = """Tu es un arbitre neutre. Voici deux réponses contradictoires à la même question.
Question : {question}
Réponse A (de {agent_a}) : {output_a}
Justification A : {evidence_a}
Réponse B (de {agent_b}) : {output_b}
Justification B : {evidence_b}
Critères d'évaluation :
1. Exactitude factuelle (sources vérifiables)
2. Cohérence du raisonnement
3. Complétude de la réponse
4. Score de confiance déclaré
Réponds en JSON strict :
{{"winner": "A"|"B"|"neither", "reasoning": "...", "confidence": 0.0-1.0}}
"""

async def weighted_merge(evidence: dict, llm) -> dict:
    """Prend chaque champ de l'agent le plus confiant sur ce champ."""
    merged = {}
    for field in set(evidence["agent_a"]["output"]) | set(evidence["agent_b"]["output"]):
        conf_a = evidence["agent_a"]["evidence"].get(f"confidence_{field}", 0.5)
        conf_b = evidence["agent_b"]["evidence"].get(f"confidence_{field}", 0.5)
        source = evidence["agent_a"]["output"] if conf_a >= conf_b else evidence["agent_b"]["output"]
        merged[field] = source.get(field)
    return merged

def should_escalate(resolution: dict, conflict: ConflictDetectionResult) -> bool:
    return (
        resolution.get("confidence", 1.0) < 0.5 or               # Arbitre incertain
        (conflict.severity == "critical"
         and resolution.get("method") != "source_verification") or  # Critique sans vérif externe
        resolution.get("needs_escalation", False) or               # Tie sur confiance
        resolution.get("arbitration_round", 0) >= 2               # Boucle d'arbitrage
    )

{
  "conflict_id": "uuid",
  "timestamp": "ISO8601",
  "agents_involved": ["agent_a", "agent_b"],
  "conflict_type": "numerical_inconsistency",
  "severity": "major",
  "resolution_strategy": "confidence_based",
  "winner": "agent_a",
  "confidence_in_resolution": 0.82,
  "escalated": false,
  "resolution_duration_ms": 340
}

class AgentPerformanceTracker:
    def __init__(self):
        self.wins: dict[str, int] = {}
        self.losses: dict[str, int] = {}
        self.by_task_type: dict[str, dict[str, int]] = {}
    def record_resolution(self, winner: str, loser: str, task_type: str):
        self.wins[winner] = self.wins.get(winner, 0) + 1
        self.losses[loser] = self.losses.get(loser, 0) + 1
        self.by_task_type.setdefault(task_type, {})
        self.by_task_type[task_type][winner] = self.by_task_type[task_type].get(winner, 0) + 1
    def win_rate(self, agent_id: str) -> float:
        wins = self.wins.get(agent_id, 0)
        losses = self.losses.get(agent_id, 0)
        total = wins + losses
        return wins / total if total > 0 else 0.5