ClawTeam: HKUDS가 공개한 오픈소스 에이전트 스웜 인텔리전스 프레임워크 구현 가이드

이 요약은 AI가 원문을 분석해 생성했습니다. 정확한 내용은 원문 기준으로 확인하세요.

핵심 요약

멀티 에이전트 시스템의 복잡성을 해결하기 위해 HKUDS는 ClawTeam 프레임워크를 개발했다. 이 아키텍처는 리더 에이전트가 목표를 분해하고, 여러 워커 에이전트가 이를 자율적으로 수행하며, 공유 태스크 보드와 메시징 시스템을 통해 협업한다. 본 튜토리얼은 로컬 인프라 없이 Colab 환경에서 OpenAI API를 사용하여 ClawTeam의 핵심 메커니즘을 파이썬으로 구현하는 방법을 제시한다. 이를 통해 RAG나 복잡한 분석 업무에서 에이전트 간의 유기적인 조율과 결과 합성이 가능해진다.

배경

Python 프로그래밍 기초, OpenAI API 사용 경험, 멀티 스레딩 및 동시성 제어에 대한 기본 이해

대상 독자

멀티 에이전트 시스템 및 LLM 오케스트레이션을 구현하려는 개발자

의미 / 영향

ClawTeam 아키텍처는 복잡한 작업을 구조화된 하위 태스크로 자동 분해하고 관리함으로써, 인간의 개입을 최소화하면서도 고품질의 결과물을 도출하는 차세대 AI 워크플로우의 표준을 제시한다.

섹션별 상세

ClawTeam 아키텍처의 핵심 구성 요소인 TaskBoard, InboxSystem, TeamRegistry를 구축한다. 모든 시스템은 멀티 스레딩 환경에서 안전하게 작동하도록 threading.Lock을 사용하여 동시성 제어를 구현하며, 태스크의 생명주기와 에이전트 간 통신을 중앙에서 관리한다.

python

class TaskBoard:
    def __init__(self):
        self._tasks: dict[str, Task] = {}
        self._lock = threading.Lock()

    def create(self, subject: str, description: str = "", owner: str = "", blocked_by: list = None) -> Task:
        task = Task(subject=subject, description=description, owner=owner, blocked_by=blocked_by or [])
        if task.blocked_by:
            task.status = TaskStatus.BLOCKED
        with self._lock: self._tasks[task.id] = task
        return task

    def _resolve_dependencies(self, completed_id: str):
        for t in self._tasks.values():
            if completed_id in t.blocked_by:
                t.blocked_by.remove(completed_id)
                if not t.blocked_by and t.status == TaskStatus.BLOCKED:
                    t.status = TaskStatus.PENDING

태스크의 생성과 의존성 해결을 담당하는 TaskBoard 클래스 구현 예시

리더 에이전트는 사용자의 고수준 목표를 수신하여 3~5개의 구체적인 하위 태스크로 분해한다. 이때 각 태스크 간의 의존성(blocked_by)을 정의하고 적합한 전문 워커 에이전트의 역할을 할당하는 구조화된 JSON 데이터를 생성하여 스웜의 전략적 방향을 설정한다.

워커 에이전트는 gpt-4o-mini 모델을 기반으로 자율 실행 루프를 수행한다. task_list, task_update, inbox_send 등의 도구를 사용하여 자신의 할당량을 확인하고 상태를 업데이트하며, 의존성이 있는 태스크는 선행 작업이 완료될 때까지 대기 상태를 유지하다가 자동으로 활성화된다.

python

def run(self, user_message: str, max_iterations: int = 6) -> str:
    self.conversation_history.append({"role": "user", "content": user_message})
    for iteration in range(max_iterations):
        response = client.chat.completions.create(
            model=MODEL,
            messages=[{"role": "system", "content": self._build_system_prompt()}, *self.conversation_history],
            tools=SWARM_TOOLS,
            tool_choice="auto",
            temperature=0.4,
        )
        // ...(중략)
        if not msg.tool_calls: return msg.content or "(No response)"
        for tc in msg.tool_calls:
            result = self._handle_tool_call(tc.function.name, json.loads(tc.function.arguments))
            self.conversation_history.append({"role": "tool", "tool_call_id": tc.id, "content": result})

OpenAI 도구 호출을 활용한 자율 에이전트의 실행 루프 구현

시스템은 여러 라운드에 걸쳐 에이전트들을 실행하며, 모든 태스크가 완료되면 리더 에이전트가 각 워커의 결과물과 메시지를 수집한다. 리더는 수집된 정보를 종합하여 핵심 인사이트와 권장 사항이 포함된 최종 보고서를 작성하여 사용자에게 전달한다.

AI 헤지펀드, 연구 스웜, 엔지니어링 팀 등 다양한 도메인별 템플릿을 제공하여 프레임워크의 범용성을 입증한다. 사용자는 단일 명령으로 복잡한 멀티 에이전트 워크플로우를 시작하고 Rich 라이브러리로 구현된 대시보드를 통해 실시간 진행 상황을 모니터링할 수 있다.

실무 Takeaway

복잡한 LLM 애플리케이션 설계 시 단일 에이전트 대신 리더-워커 구조의 스웜 아키텍처를 도입하여 작업의 정확도와 병렬 처리 효율을 높일 수 있다.
TaskBoard와 같은 중앙 집중식 상태 관리 시스템을 구축하여 에이전트 간의 작업 중복을 방지하고 의존성 있는 워크플로우를 자동화할 수 있다.
OpenAI의 function-calling을 활용해 에이전트가 스스로 태스크 상태를 변경하고 메시지를 주고받게 함으로써 자율적인 협업 환경을 조성할 수 있다.

언급된 리소스

튜토리얼ClawTeam Swarm Intelligence Tutorial Notebook

class TaskBoard: def __init__(self): self._tasks: dict[str, Task] = {} self._lock = threading.Lock() def create(self, subject: str, description: str = "", owner: str = "", blocked_by: list = None) -> Task: task = Task(subject=subject, description=description, owner=owner, blocked_by=blocked_by or []) if task.blocked_by: task.status = TaskStatus.BLOCKED with self._lock: self._tasks[task.id] = task return task def _resolve_dependencies(self, completed_id: str): for t in self._tasks.values(): if completed_id in t.blocked_by: t.blocked_by.remove(completed_id) if not t.blocked_by and t.status == TaskStatus.BLOCKED: t.status = TaskStatus.PENDING

def run(self, user_message: str, max_iterations: int = 6) -> str: self.conversation_history.append({"role": "user", "content": user_message}) for iteration in range(max_iterations): response = client.chat.completions.create( model=MODEL, messages=[{"role": "system", "content": self._build_system_prompt()}, *self.conversation_history], tools=SWARM_TOOLS, tool_choice="auto", temperature=0.4, ) // ...(중략) if not msg.tool_calls: return msg.content or "(No response)" for tc in msg.tool_calls: result = self._handle_tool_call(tc.function.name, json.loads(tc.function.arguments)) self.conversation_history.append({"role": "tool", "tool_call_id": tc.id, "content": result})

ClawTeam: HKUDS가 공개한 오픈소스 에이전트 스웜 인텔리전스 프레임워크 구현 가이드

핵심 요약

배경

대상 독자

의미 / 영향

섹션별 상세

실무 Takeaway

언급된 리소스

ClawTeam: HKUDS가 공개한 오픈소스 에이전트 스웜 인텔리전스 프레임워크 구현 가이드

핵심 요약

배경

대상 독자

의미 / 영향

섹션별 상세

실무 Takeaway

언급된 리소스

관련 토론

댓글

관련 피드

관련 토론

댓글

관련 피드