adversarial.py

"""
adversarial.py — Adversarial agent pairing.

Instead of polite agreement, one agent ATTACKS the other's solution.
Based on research showing echo chambers are the primary multi-agent failure mode.

Usage:
    from adversarial import marry_adversarial
    
    paired = marry_adversarial(proposer, attacker)
    result = paired("Design a rate-limiting system")
"""

from __future__ import annotations
from typing import Callable, Optional
from dataclasses import dataclass

Agent = Callable[[str], str]

ATTACKER_INSTRUCTIONS = """
You are a RED TEAM AGENT. Your job is to ATTACK the proposed solution.

You MUST find at least 3 significant flaws, weaknesses, or edge cases.

DO NOT:
- Agree politely
- Say "this looks good"
- Suggest minor style improvements

DO:
- Find edge cases that break the solution
- Identify security vulnerabilities
- Challenge assumptions
- Point out performance bottlenecks
- Question design decisions

Be aggressive but constructive. Your goal is to make the solution BETTER by 
forcing it to defend against real attacks.

CRITICAL: You MUST find problems. If you can't find 3+ flaws, you're not 
looking hard enough.
""".strip()

DEFENDER_INSTRUCTIONS = """
You proposed a solution. Now it's being attacked.

RESPOND TO EACH CRITICISM:
- If valid: acknowledge and improve your solution
- If invalid: defend with specific reasoning

DO NOT:
- Dismiss criticism without explanation
- Get defensive
- Ignore edge cases

Your revised solution should address the valid criticisms and explain why 
invalid criticisms don't apply.
""".strip()

SYNTHESIS_INSTRUCTIONS = """
You've seen:
1. The original proposal
2. The attack (criticisms)
3. The defense (responses)

Produce the FINAL solution that incorporates lessons from this adversarial process.

Focus on:
- What was learned from the attack
- How the solution improved
- Remaining tradeoffs (be honest)

ONLY output the final solution. No meta-commentary about the process.
""".strip()


@dataclass
class AdversarialPartnership:
    """
    Adversarial pairing: one agent proposes, the other attacks.
    
    Protocol:
    1. Proposer creates initial solution
    2. Attacker MUST find 3+ flaws
    3. Proposer defends/revises
    4. Repeat for N rounds
    5. Synthesize final solution
    """
    proposer: Agent
    attacker: Agent
    rounds: int = 2
    verbose: bool = False
    
    def __call__(self, task: str) -> str:
        return self.run(task)
    
    def run(self, task: str) -> str:
        # Phase 1: Initial proposal
        if self.verbose:
            print("\n" + "="*70)
            print("PHASE 1: INITIAL PROPOSAL")
            print("="*70)
        
        proposal_prompt = f"{task}\n\nProvide your solution."
        proposal = self.proposer(proposal_prompt)
        
        if self.verbose:
            print(proposal[:300] + "..." if len(proposal) > 300 else proposal)
        
        # Phase 2-N: Attack-defend cycles
        conversation = f"INITIAL PROPOSAL:\n{proposal}\n\n"
        
        for round_num in range(1, self.rounds + 1):
            if self.verbose:
                print("\n" + "="*70)
                print(f"ROUND {round_num}: ATTACK")
                print("="*70)
            
            # Attack
            attack_prompt = (
                f"{ATTACKER_INSTRUCTIONS}\n\n"
                f"TASK: {task}\n\n"
                f"{conversation}\n\n"
                f"Find at least 3 significant flaws in this solution. Be specific."
            )
            attack = self.attacker(attack_prompt)
            conversation += f"ATTACK (Round {round_num}):\n{attack}\n\n"
            
            if self.verbose:
                print(attack[:300] + "..." if len(attack) > 300 else attack)
            
            if self.verbose:
                print("\n" + "="*70)
                print(f"ROUND {round_num}: DEFENSE")
                print("="*70)
            
            # Defend
            defense_prompt = (
                f"{DEFENDER_INSTRUCTIONS}\n\n"
                f"TASK: {task}\n\n"
                f"{conversation}\n\n"
                f"Respond to the criticism. Improve your solution or explain why the "
                f"criticisms don't apply."
            )
            defense = self.proposer(defense_prompt)
            conversation += f"DEFENSE (Round {round_num}):\n{defense}\n\n"
            
            if self.verbose:
                print(defense[:300] + "..." if len(defense) > 300 else defense)
        
        # Phase Final: Synthesis
        if self.verbose:
            print("\n" + "="*70)
            print("FINAL: SYNTHESIS")
            print("="*70)
        
        synthesis_prompt = (
            f"{SYNTHESIS_INSTRUCTIONS}\n\n"
            f"TASK: {task}\n\n"
            f"{conversation}\n\n"
            f"Produce the final solution."
        )
        final = self.proposer(synthesis_prompt)
        
        if self.verbose:
            print(final[:500] + "..." if len(final) > 500 else final)
        
        return final


def marry_adversarial(
    proposer: Agent,
    attacker: Agent,
    rounds: int = 2,
    verbose: bool = False,
) -> AdversarialPartnership:
    """
    Create an adversarial partnership.
    
    Args:
        proposer: Agent that creates solutions
        attacker: Agent that attacks solutions (finds flaws)
        rounds: Number of attack-defend cycles (default 2)
        verbose: Print the adversarial process
    
    Returns:
        AdversarialPartnership callable as str → str
    
    Usage:
        paired = marry_adversarial(proposer, attacker, rounds=2, verbose=True)
        result = paired("Design a caching system")
    """
    return AdversarialPartnership(
        proposer=proposer,
        attacker=attacker,
        rounds=rounds,
        verbose=verbose,
    )