2장: 멀티에이전트 팀 아키텍처 설계

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
 
# 워커 에이전트 정의
research_agent = create_react_agent(
    model="claude-sonnet-4-6",
    tools=[web_search, document_reader],
    name="researcher",
    prompt="최신 기술 동향을 조사하는 리서치 전문가입니다."
)
 
writer_agent = create_react_agent(
    model="claude-sonnet-4-6",
    tools=[text_editor, grammar_checker],
    name="writer",
    prompt="조사 결과를 정리하여 기술 문서를 작성하는 전문가입니다."
)
 
reviewer_agent = create_react_agent(
    model="claude-sonnet-4-6",
    tools=[fact_checker, style_validator],
    name="reviewer",
    prompt="작성된 문서의 정확성과 품질을 검증하는 전문가입니다."
)
 
# 감독자가 워커를 조율
supervisor = create_supervisor(
    agents=[research_agent, writer_agent, reviewer_agent],
    model="claude-opus-4-6",
    prompt="""당신은 기술 문서 제작팀의 관리자입니다.
    사용자 요청을 분석하여 적절한 전문가에게 작업을 할당하세요.
    리서치 → 작성 → 검토 순서를 따르되, 검토에서 문제가 발견되면
    해당 단계로 돌아가 수정을 요청하세요."""
)
 
app = supervisor.compile()

# 감독자의 동적 라우팅 로직 예시
supervisor_prompt = """
현재 상태를 분석하여 다음 행동을 결정하세요:
 
1. 리서치가 아직 안 되었다면 → researcher에게 위임
2. 리서치가 완료되었고 초안이 없다면 → writer에게 위임
3. 초안이 있고 검토가 안 되었다면 → reviewer에게 위임
4. 검토 결과 수정이 필요하면 → writer에게 피드백과 함께 재위임
5. 검토가 통과되었다면 → 최종 결과 반환
 
최대 재작업 횟수는 2회입니다. 2회 수정 후에도 미흡하면
현재 상태의 최선의 결과를 반환하세요.
"""

# 하위 팀 1: 리서치 팀
web_researcher = create_react_agent(
    model="claude-sonnet-4-6",
    tools=[web_search, url_fetcher],
    name="web_researcher",
    prompt="웹에서 최신 정보를 검색하고 수집합니다."
)
 
paper_analyst = create_react_agent(
    model="claude-sonnet-4-6",
    tools=[arxiv_search, pdf_reader],
    name="paper_analyst",
    prompt="학술 논문을 검색하고 핵심 내용을 분석합니다."
)
 
research_supervisor = create_supervisor(
    agents=[web_researcher, paper_analyst],
    model="claude-sonnet-4-6",
    prompt="리서치 팀을 관리합니다. 웹 조사와 논문 분석을 조율합니다."
)
 
# 하위 팀 2: 콘텐츠 제작 팀
outline_writer = create_react_agent(
    model="claude-sonnet-4-6",
    tools=[outline_generator],
    name="outline_writer",
    prompt="조사 결과를 바탕으로 문서 구조를 설계합니다."
)
 
content_writer = create_react_agent(
    model="claude-sonnet-4-6",
    tools=[text_editor, image_generator],
    name="content_writer",
    prompt="설계된 구조에 따라 본문을 작성합니다."
)
 
content_supervisor = create_supervisor(
    agents=[outline_writer, content_writer],
    model="claude-sonnet-4-6",
    prompt="콘텐츠 제작 팀을 관리합니다."
)
 
# 최상위 감독자: 팀 간 조율
top_supervisor = create_supervisor(
    agents=[research_supervisor, content_supervisor, reviewer_agent],
    model="claude-opus-4-6",
    prompt="""전체 문서 제작 프로세스를 총괄합니다.
    리서치 팀의 조사 결과를 콘텐츠 팀에 전달하고,
    최종 결과를 검토자에게 보내 품질을 확인합니다."""
)

from agents import Agent, handoff
 
billing_agent = Agent(
    name="billing",
    instructions="결제 관련 문의를 처리합니다.",
    tools=[query_payment, process_refund],
    handoffs=[
        handoff(
            agent="shipping",
            description="배송 관련 문의가 포함된 경우"
        ),
        handoff(
            agent="technical",
            description="기술적 문제가 발견된 경우"
        )
    ]
)
 
shipping_agent = Agent(
    name="shipping",
    instructions="배송 추적과 물류를 관리합니다.",
    tools=[track_package, update_delivery],
    handoffs=[
        handoff(
            agent="billing",
            description="환불이 필요한 배송 사고인 경우"
        )
    ]
)
 
technical_agent = Agent(
    name="technical",
    instructions="기술적 문제를 해결합니다.",
    tools=[check_system_status, run_diagnostics],
    handoffs=[
        handoff(
            agent="billing",
            description="기술 문제로 인한 보상이 필요한 경우"
        )
    ]
)

# 순환 방지 메커니즘
class HandoffTracker:
    def __init__(self, max_handoffs: int = 5):
        self.max_handoffs = max_handoffs
        self.history: list[str] = []
 
    def can_handoff(self, target: str) -> bool:
        if len(self.history) >= self.max_handoffs:
            return False
        # 같은 에이전트로 2번 이상 돌아가면 차단
        if self.history.count(target) >= 2:
            return False
        return True
 
    def record(self, target: str) -> None:
        self.history.append(target)

from autogen import GroupChat, GroupChatManager
 
analyst = AssistantAgent(
    name="Analyst",
    system_message="데이터를 분석하고 인사이트를 도출합니다."
)
 
engineer = AssistantAgent(
    name="Engineer",
    system_message="분석 결과를 기반으로 기술적 구현을 제안합니다."
)
 
critic = AssistantAgent(
    name="Critic",
    system_message="제안된 솔루션의 약점과 개선점을 지적합니다."
)
 
group_chat = GroupChat(
    agents=[analyst, engineer, critic],
    messages=[],
    max_round=10,
    speaker_selection_method="auto",  # LLM이 다음 발언자 결정
    allow_repeat_speaker=False        # 연속 발언 방지
)
 
manager = GroupChatManager(
    groupchat=group_chat,
    llm_config={"model": "claude-sonnet-4-6"}
)

def select_architecture(requirements: dict) -> str:
    if requirements["fixed_order"] and requirements["agent_count"] <= 5:
        return "pipeline"
    elif requirements["realtime_conversation"]:
        return "handoff"
    elif requirements["multi_perspective"]:
        return "group_chat"
    elif requirements["agent_count"] > 10:
        return "hierarchical"
    else:
        return "supervisor_worker"  # 기본값