5장: 에이전트 협업과 합의 메커니즘

from dataclasses import dataclass
 
@dataclass
class DebateRound:
    speaker: str
    position: str
    arguments: list[str]
    counter_arguments: list[str]
    confidence: float
 
class DebateOrchestrator:
    def __init__(
        self,
        agents: list[Agent],
        moderator: Agent,
        max_rounds: int = 3
    ):
        self.agents = agents
        self.moderator = moderator
        self.max_rounds = max_rounds
        self.transcript: list[DebateRound] = []
 
    async def run_debate(self, topic: str) -> str:
        # 초기 입장 수집
        positions = await asyncio.gather(*[
            self._get_initial_position(agent, topic)
            for agent in self.agents
        ])
        self.transcript.extend(positions)
 
        # 토론 라운드
        for round_num in range(self.max_rounds):
            new_positions = []
            for i, agent in enumerate(self.agents):
                # 다른 에이전트의 입장을 전달하여 반론 요청
                others = [
                    p for j, p in enumerate(positions)
                    if j != i
                ]
                response = await self._get_rebuttal(
                    agent, topic, others
                )
                new_positions.append(response)
            positions = new_positions
            self.transcript.extend(positions)
 
            # 합의 도달 여부 확인
            if self._check_convergence(positions):
                break
 
        # 모더레이터가 최종 종합
        return await self._moderate(topic)
 
    async def _get_initial_position(
        self, agent: Agent, topic: str
    ) -> DebateRound:
        prompt = f"""다음 주제에 대한 당신의 입장을 제시하세요.
        주제: {topic}
        근거를 포함하여 논리적으로 설명하세요."""
        result = await agent.run(prompt)
        return DebateRound(
            speaker=agent.name,
            position=result.position,
            arguments=result.arguments,
            counter_arguments=[],
            confidence=result.confidence
        )
 
    async def _get_rebuttal(
        self, agent: Agent, topic: str,
        other_positions: list[DebateRound]
    ) -> DebateRound:
        others_text = "\n".join([
            f"{p.speaker}: {p.position}\n근거: {p.arguments}"
            for p in other_positions
        ])
        prompt = f"""주제: {topic}
        
다른 참여자의 입장:
{others_text}
 
위 입장들을 검토하고:
1. 동의하는 부분과 이유
2. 반박하는 부분과 이유
3. 수정된 당신의 최종 입장
을 제시하세요."""
        return await agent.run(prompt)
 
    def _check_convergence(
        self, positions: list[DebateRound]
    ) -> bool:
        """입장이 수렴했는지 확인"""
        confidences = [p.confidence for p in positions]
        # 모든 에이전트의 확신도가 0.8 이상이면 합의 도달
        return all(c >= 0.8 for c in confidences)
 
    async def _moderate(self, topic: str) -> str:
        transcript_text = "\n\n".join([
            f"[{r.speaker}] 입장: {r.position}\n"
            f"근거: {r.arguments}\n"
            f"반론: {r.counter_arguments}"
            for r in self.transcript
        ])
        prompt = f"""다음 토론을 종합하여 최종 결론을 도출하세요.
 
주제: {topic}
 
토론 기록:
{transcript_text}
 
가장 강력한 근거에 기반하여 균형 잡힌 결론을 제시하세요.
소수 의견이더라도 타당한 근거가 있다면 반영하세요."""
        return await self.moderator.run(prompt)

from collections import Counter
 
class MajorityVoting:
    def __init__(self, agents: list[Agent], min_agreement: float = 0.6):
        self.agents = agents
        self.min_agreement = min_agreement
 
    async def vote(self, question: str, options: list[str]) -> dict:
        """에이전트들의 독립적 투표를 수집하고 집계"""
        prompt = f"""다음 질문에 대해 주어진 선택지 중 하나를 선택하세요.
 
질문: {question}
선택지: {', '.join(options)}
 
반드시 선택지 중 하나만 응답하세요."""
 
        # 독립적 병렬 투표
        votes = await asyncio.gather(*[
            agent.run(prompt) for agent in self.agents
        ])
 
        # 집계
        vote_counts = Counter(votes)
        total = len(votes)
        winner, count = vote_counts.most_common(1)[0]
        agreement = count / total
 
        return {
            "decision": winner if agreement >= self.min_agreement else None,
            "agreement_rate": agreement,
            "vote_distribution": dict(vote_counts),
            "quorum_met": agreement >= self.min_agreement
        }

class WeightedVoting:
    def __init__(self):
        self.agents: list[Agent] = []
        self.weights: dict[str, float] = {}
        self.accuracy_history: dict[str, list[bool]] = {}
 
    def register_agent(
        self, agent: Agent, initial_weight: float = 1.0
    ):
        self.agents.append(agent)
        self.weights[agent.name] = initial_weight
        self.accuracy_history[agent.name] = []
 
    async def vote(self, question: str, options: list[str]) -> dict:
        votes = {}
        for agent in self.agents:
            result = await agent.run(question)
            votes[agent.name] = {
                "choice": result.choice,
                "confidence": result.confidence,
                "reasoning": result.reasoning
            }
 
        # 가중 점수 계산
        option_scores: dict[str, float] = {opt: 0.0 for opt in options}
        for agent_name, vote in votes.items():
            weight = self.weights[agent_name]
            confidence = vote["confidence"]
            # 가중치 * 확신도를 점수에 반영
            option_scores[vote["choice"]] += weight * confidence
 
        winner = max(option_scores, key=option_scores.get)
        return {
            "decision": winner,
            "scores": option_scores,
            "individual_votes": votes
        }
 
    def update_weights(self, agent_name: str, was_correct: bool):
        """결과 피드백에 따라 가중치 조정"""
        self.accuracy_history[agent_name].append(was_correct)
        recent = self.accuracy_history[agent_name][-20:]
        accuracy = sum(recent) / len(recent)
        self.weights[agent_name] = 0.5 + accuracy  # 0.5 ~ 1.5 범위

class GeneratorCriticPair:
    def __init__(
        self,
        generator: Agent,
        critic: Agent,
        max_iterations: int = 3
    ):
        self.generator = generator
        self.critic = critic
        self.max_iterations = max_iterations
 
    async def produce(self, task: str) -> dict:
        current_output = None
        history = []
 
        for iteration in range(self.max_iterations):
            # 생성 (이전 피드백 포함)
            if current_output is None:
                gen_prompt = task
            else:
                gen_prompt = f"""{task}
 
이전 출력:
{current_output}
 
검토 피드백:
{history[-1]['feedback']}
 
피드백을 반영하여 개선된 버전을 작성하세요."""
 
            current_output = await self.generator.run(gen_prompt)
 
            # 검토
            review = await self.critic.run(f"""다음 출력을 검토하세요.
 
원래 작업: {task}
출력:
{current_output}
 
다음 기준으로 평가하세요:
1. 정확성 (1-10)
2. 완성도 (1-10)
3. 개선이 필요한 부분 (구체적으로)
4. 승인 여부 (approved / needs_revision)""")
 
            history.append({
                "iteration": iteration,
                "output": current_output,
                "feedback": review.feedback,
                "scores": review.scores,
                "approved": review.approved
            })
 
            if review.approved:
                break
 
        return {
            "final_output": current_output,
            "iterations": len(history),
            "approved": history[-1]["approved"],
            "history": history
        }

class MultiCriticChain:
    def __init__(self, generator: Agent, critics: list[Agent]):
        self.generator = generator
        self.critics = critics
 
    async def produce(self, task: str) -> dict:
        output = await self.generator.run(task)
        all_feedback = []
 
        # 각 Critic이 독립적으로 검토
        reviews = await asyncio.gather(*[
            critic.run(
                f"다음을 {critic.name}의 관점에서 검토하세요:\n{output}"
            )
            for critic in self.critics
        ])
 
        for critic, review in zip(self.critics, reviews):
            all_feedback.append({
                "critic": critic.name,
                "issues": review.issues,
                "severity": review.severity,
                "suggestions": review.suggestions
            })
 
        # 심각한 이슈가 있으면 재생성
        critical_issues = [
            f for f in all_feedback
            if f["severity"] == "critical"
        ]
 
        if critical_issues:
            feedback_text = "\n".join([
                f"[{f['critic']}] {f['issues']}: {f['suggestions']}"
                for f in critical_issues
            ])
            output = await self.generator.run(
                f"{task}\n\n치명적 이슈 수정 필요:\n{feedback_text}"
            )
 
        return {
            "output": output,
            "reviews": all_feedback,
            "had_critical_issues": len(critical_issues) > 0
        }
 
# 코드 리뷰 예시
code_generator = Agent(name="coder", instructions="코드를 작성합니다")
security_critic = Agent(
    name="security-reviewer",
    instructions="보안 관점에서 코드를 검토합니다. SQL 인젝션, XSS 등을 확인합니다."
)
performance_critic = Agent(
    name="performance-reviewer",
    instructions="성능 관점에서 코드를 검토합니다. N+1 쿼리, 메모리 누수 등을 확인합니다."
)
style_critic = Agent(
    name="style-reviewer",
    instructions="코드 스타일과 가독성을 검토합니다."
)
 
chain = MultiCriticChain(
    generator=code_generator,
    critics=[security_critic, performance_critic, style_critic]
)

class DelphiConsensus:
    def __init__(
        self, agents: list[Agent],
        max_rounds: int = 3,
        convergence_threshold: float = 0.7
    ):
        self.agents = agents
        self.max_rounds = max_rounds
        self.convergence_threshold = convergence_threshold
 
    async def reach_consensus(self, question: str) -> dict:
        round_results = []
 
        for round_num in range(self.max_rounds):
            if round_num == 0:
                # 1라운드: 독립적 의견 수집
                responses = await asyncio.gather(*[
                    self._collect_opinion(agent, question, None)
                    for agent in self.agents
                ])
            else:
                # 후속 라운드: 이전 라운드 요약 공유 후 재검토
                summary = self._summarize_round(round_results[-1])
                responses = await asyncio.gather(*[
                    self._collect_opinion(agent, question, summary)
                    for agent in self.agents
                ])
 
            round_results.append(responses)
 
            # 수렴 확인
            if self._has_converged(responses):
                break
 
        return {
            "consensus": self._extract_consensus(round_results[-1]),
            "rounds": len(round_results),
            "convergence_achieved": self._has_converged(
                round_results[-1]
            ),
            "history": round_results
        }
 
    async def _collect_opinion(
        self, agent: Agent, question: str,
        previous_summary: str | None
    ) -> dict:
        if previous_summary:
            prompt = f"""질문: {question}
 
이전 라운드의 전체 의견 분포:
{previous_summary}
 
다른 참여자들의 의견을 고려한 후, 
당신의 수정된 견해를 제시하세요.
확신도(0-1)와 근거를 포함하세요."""
        else:
            prompt = f"""질문: {question}
 
당신의 독립적인 견해를 제시하세요.
확신도(0-1)와 근거를 포함하세요."""
 
        result = await agent.run(prompt)
        return {
            "agent": agent.name,
            "opinion": result.opinion,
            "confidence": result.confidence,
            "reasoning": result.reasoning
        }
 
    def _summarize_round(self, responses: list[dict]) -> str:
        opinions = [r["opinion"] for r in responses]
        confidences = [r["confidence"] for r in responses]
        return f"""참여자 {len(responses)}명의 의견:
평균 확신도: {sum(confidences)/len(confidences):.2f}
의견 분포: {Counter(opinions).most_common()}
주요 근거들: {[r['reasoning'][:100] for r in responses]}"""
 
    def _has_converged(self, responses: list[dict]) -> bool:
        opinions = [r["opinion"] for r in responses]
        most_common = Counter(opinions).most_common(1)[0]
        return most_common[1] / len(opinions) >= self.convergence_threshold

class CollaborationCostAnalyzer:
    @staticmethod
    def estimate_cost(
        pattern: str,
        num_agents: int,
        avg_tokens_per_call: int,
        cost_per_1k_tokens: float,
        rounds: int = 1
    ) -> dict:
        if pattern == "voting":
            total_calls = num_agents  # 1회 투표
            interaction_overhead = 0
        elif pattern == "debate":
            total_calls = num_agents * rounds + 1  # 라운드별 + 모더레이터
            interaction_overhead = 0.3  # 이전 의견 참조로 토큰 증가
        elif pattern == "critic_chain":
            total_calls = 1 + num_agents  # 생성 + 검토
            interaction_overhead = 0.5  # 전체 출력을 검토에 포함
        elif pattern == "delphi":
            total_calls = num_agents * rounds
            interaction_overhead = 0.2 * rounds  # 라운드마다 요약 추가
 
        effective_tokens = avg_tokens_per_call * (1 + interaction_overhead)
        total_tokens = total_calls * effective_tokens
        total_cost = (total_tokens / 1000) * cost_per_1k_tokens
 
        return {
            "pattern": pattern,
            "total_llm_calls": total_calls,
            "estimated_total_tokens": int(total_tokens),
            "estimated_cost_usd": round(total_cost, 4),
            "latency_factor": f"{total_calls}x (순차) / ~{rounds}x (병렬)"
        }