6장: 멀티 에이전트 패턴 - 협업과 조율의 아키텍처

import anthropic
import json
 
client = anthropic.Anthropic()
 
class Agent:
    def __init__(self, name: str, role: str, system_prompt: str, tools: list[dict] = None):
        self.name = name
        self.role = role
        self.system_prompt = system_prompt
        self.tools = tools or []
 
    def run(self, task: str, context: str = "") -> str:
        prompt = task
        if context:
            prompt = f"배경 정보:\n{context}\n\n작업:\n{task}"
 
        messages = [{"role": "user", "content": prompt}]
 
        # 도구가 있으면 ReAct 루프 실행
        if self.tools:
            return self._run_with_tools(messages)
 
        response = client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            system=self.system_prompt,
            messages=messages,
        )
        return response.content[0].text
 
    def _run_with_tools(self, messages: list) -> str:
        for _ in range(5):
            response = client.messages.create(
                model="claude-sonnet-4-20250514",
                max_tokens=4096,
                system=self.system_prompt,
                tools=self.tools,
                messages=messages,
            )
            if response.stop_reason == "end_turn":
                for block in response.content:
                    if hasattr(block, "text"):
                        return block.text
                return "완료"
 
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})
 
        return "최대 반복 도달"
 
 
class Supervisor:
    def __init__(self, agents: list[Agent]):
        self.agents = {agent.name: agent for agent in agents}
        self.client = client
 
    def delegate(self, task: str) -> str:
        """작업을 분석하고 적절한 에이전트에 위임합니다."""
        agent_descriptions = "\n".join(
            f"- {name}: {agent.role}"
            for name, agent in self.agents.items()
        )
 
        plan_prompt = f"""다음 작업을 수행하기 위해 팀원에게 작업을 위임하십시오.
 
작업: {task}
 
사용 가능한 팀원:
{agent_descriptions}
 
각 팀원에게 위임할 구체적인 작업을 JSON으로 계획하십시오:
[
  {{"agent": "에이전트 이름", "task": "위임할 작업", "order": 1}},
  ...
]
 
작업 간 의존성이 있으면 order를 활용하십시오.
같은 order의 작업은 병렬 실행됩니다."""
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            messages=[{"role": "user", "content": plan_prompt}],
        )
 
        try:
            delegations = json.loads(response.content[0].text)
        except json.JSONDecodeError:
            return "위임 계획 생성에 실패했습니다."
 
        # 순서대로 실행
        results = {}
        delegations.sort(key=lambda x: x["order"])
 
        current_order = None
        for delegation in delegations:
            agent_name = delegation["agent"]
            if agent_name not in self.agents:
                results[agent_name] = f"에이전트 '{agent_name}'을 찾을 수 없습니다."
                continue
 
            # 이전 단계 결과를 컨텍스트로 제공
            context = "\n".join(
                f"[{name}의 결과]: {result}"
                for name, result in results.items()
            )
 
            agent = self.agents[agent_name]
            print(f"[{agent_name}] 작업 시작: {delegation['task']}")
            result = agent.run(delegation["task"], context)
            results[agent_name] = result
            print(f"[{agent_name}] 완료")
 
        # 결과 종합
        return self._synthesize(task, results)
 
    def _synthesize(self, task: str, results: dict) -> str:
        synthesis_prompt = f"""다음은 팀원들의 작업 결과입니다.
 
원래 작업: {task}
 
결과:
{json.dumps(results, ensure_ascii=False, indent=2)}
 
모든 결과를 종합하여 최종 답변을 작성하십시오."""
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            messages=[{"role": "user", "content": synthesis_prompt}],
        )
        return response.content[0].text

# 에이전트 팀 구성
researcher = Agent(
    name="researcher",
    role="최신 기술 동향을 조사하고 자료를 수집합니다.",
    system_prompt="당신은 기술 리서처입니다. 주어진 주제에 대해 핵심 정보를 체계적으로 조사하십시오.",
    tools=[search_tool],
)
 
analyst = Agent(
    name="analyst",
    role="수집된 자료를 분석하고 인사이트를 도출합니다.",
    system_prompt="당신은 데이터 분석가입니다. 제공된 정보에서 패턴과 인사이트를 찾아내십시오.",
)
 
writer = Agent(
    name="writer",
    role="분석 결과를 바탕으로 보고서를 작성합니다.",
    system_prompt="당신은 기술 테크니컬 라이터입니다. 전문적이면서도 이해하기 쉬운 보고서를 작성하십시오.",
)
 
# 감독자 생성 및 실행
supervisor = Supervisor([researcher, analyst, writer])
result = supervisor.delegate(
    "2026년 AI 에이전트 프레임워크의 현황을 분석하고 보고서를 작성하십시오."
)

class DebateSystem:
    def __init__(self, perspectives: list[dict]):
        """
        perspectives: [
            {"name": "낙관론자", "system_prompt": "..."},
            {"name": "비관론자", "system_prompt": "..."},
            {"name": "현실론자", "system_prompt": "..."},
        ]
        """
        self.perspectives = perspectives
        self.client = anthropic.Anthropic()
 
    def debate(self, topic: str, rounds: int = 3) -> str:
        """여러 관점에서 토론을 진행합니다."""
        history = []
 
        for round_num in range(rounds):
            print(f"\n--- 라운드 {round_num + 1} ---")
 
            round_responses = {}
            for perspective in self.perspectives:
                history_text = self._format_history(history)
 
                prompt = f"""주제: {topic}
 
{f'이전 토론 내용:{history_text}' if history_text else '첫 번째 발언입니다.'}
 
당신의 관점에서 이 주제에 대한 의견을 제시하십시오.
다른 참여자의 의견이 있다면 그에 대해 반론이나 동의를 표명하십시오.
핵심 논점을 2-3개로 정리하여 제시하십시오."""
 
                response = self.client.messages.create(
                    model="claude-sonnet-4-20250514",
                    max_tokens=2048,
                    system=perspective["system_prompt"],
                    messages=[{"role": "user", "content": prompt}],
                )
 
                statement = response.content[0].text
                round_responses[perspective["name"]] = statement
                print(f"[{perspective['name']}] {statement[:100]}...")
 
            history.append(round_responses)
 
        # 최종 종합
        return self._synthesize_debate(topic, history)
 
    def _format_history(self, history: list[dict]) -> str:
        if not history:
            return ""
        parts = []
        for i, round_data in enumerate(history):
            parts.append(f"\n라운드 {i + 1}:")
            for name, statement in round_data.items():
                parts.append(f"  [{name}]: {statement}")
        return "\n".join(parts)
 
    def _synthesize_debate(self, topic: str, history: list) -> str:
        prompt = f"""다음 토론의 결과를 중립적 입장에서 종합하십시오.
 
주제: {topic}
토론 내용: {json.dumps(history, ensure_ascii=False)}
 
1. 각 관점의 핵심 주장을 요약하십시오.
2. 합의된 사항과 합의되지 않은 사항을 구분하십시오.
3. 근거가 가장 강한 결론을 제시하십시오."""
 
        response = self.client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=4096,
            messages=[{"role": "user", "content": prompt}],
        )
        return response.content[0].text

class Pipeline:
    def __init__(self, stages: list[Agent]):
        self.stages = stages
 
    def run(self, initial_input: str) -> str:
        """파이프라인을 순차적으로 실행합니다."""
        current = initial_input
 
        for i, agent in enumerate(self.stages):
            print(f"[파이프라인 {i + 1}/{len(self.stages)}] {agent.name}")
            current = agent.run(current)
            print(f"  출력 길이: {len(current)}자")
 
        return current
 
# 사용 예시: 문서 처리 파이프라인
pipeline = Pipeline([
    Agent("extractor", "텍스트에서 핵심 정보를 추출합니다.",
          "문서에서 핵심 사실, 수치, 인용을 추출하십시오."),
    Agent("analyzer", "추출된 정보를 분석합니다.",
          "추출된 정보의 패턴과 관계를 분석하십시오."),
    Agent("summarizer", "분석 결과를 요약합니다.",
          "분석 결과를 경영진용 요약문으로 작성하십시오."),
])
 
result = pipeline.run(document_text)

from dataclasses import dataclass, field
from datetime import datetime
from collections import defaultdict
 
@dataclass
class Message:
    sender: str
    receiver: str
    content: str
    message_type: str = "info"  # info, request, response, error
    timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
 
class MessageBus:
    def __init__(self):
        self._queues: dict[str, list[Message]] = defaultdict(list)
        self._history: list[Message] = []
 
    def send(self, message: Message):
        """메시지를 전송합니다."""
        self._queues[message.receiver].append(message)
        self._history.append(message)
        print(
            f"[MSG] {message.sender} -> {message.receiver}: "
            f"{message.content[:80]}..."
        )
 
    def receive(self, agent_name: str) -> list[Message]:
        """에이전트의 미읽은 메시지를 가져옵니다."""
        messages = self._queues[agent_name]
        self._queues[agent_name] = []
        return messages
 
    def get_history(self) -> list[Message]:
        return self._history

import threading
 
class SharedState:
    """에이전트 간 공유 상태를 관리합니다."""
    def __init__(self):
        self._state: dict = {}
        self._lock = threading.Lock()
        self._change_log: list[dict] = []
 
    def get(self, key: str, default=None):
        with self._lock:
            return self._state.get(key, default)
 
    def set(self, key: str, value, agent_name: str):
        with self._lock:
            old_value = self._state.get(key)
            self._state[key] = value
            self._change_log.append({
                "agent": agent_name,
                "key": key,
                "old_value": old_value,
                "new_value": value,
                "timestamp": datetime.now().isoformat(),
            })
 
    def get_changes_since(self, timestamp: str) -> list[dict]:
        """특정 시점 이후의 변경 사항을 반환합니다."""
        return [
            log for log in self._change_log
            if log["timestamp"] > timestamp
        ]

class ResilientSupervisor(Supervisor):
    def run_agent_safely(self, agent: Agent, task: str, context: str) -> str:
        """에이전트 실행을 안전하게 격리합니다."""
        try:
            result = agent.run(task, context)
            if not result or len(result.strip()) < 10:
                return f"[{agent.name}] 유효한 결과를 반환하지 못했습니다."
            return result
        except Exception as e:
            error_msg = f"[{agent.name}] 오류 발생: {str(e)}"
            print(error_msg)
            return error_msg