aegis-gov: 멀티 에이전트 태스크 그래프 및 서킷 브레이커를 위한 작은 Python 라이브러리

멀티 에이전트 LLM (Large Language Model) 시스템은 대부분의 튜토리얼이 그냥 지나치는 조정(coordination) 문제를 안고 있습니다. 몇 개의 asyncio.gather 호출이나 프롬프트(prompt) 리스트를 연결할 수는 있지만, 세 개 또는 네 개의 에이전트가 정의된 순서에 따라 서로에게 작업을 전달해야 하고, 하나의 호출이 실패했을 때 전체 시스템이 우아하게 성능 저하(degrade gracefully)를 일으켜야 하는 상황이 되면, 스캐폴딩(scaffolding)은 빠르게 늘어나고 제공자(provider)별 SDK 코드와 뒤엉키게 됩니다.

저는 이러한 조정용 스캐폴딩을 비즈니스 로직으로부터 분리하기 위해 aegis-gov를 작성했습니다. 이는 다음과 같은 기능을 제공하는 작은 Python 라이브러리(단 하나의 필수 의존성: requests)입니다:

• Anthropic, OpenAI, 그리고 Ollama를 위한 제공자 불가지론적 어댑터 프로토콜 (provider-agnostic adapter protocol)
• 의존성을 해결하고, 독립적인 태스크를 병렬로 실행하며, 상위 태스크가 실패할 때 건너뛰기 신호(skip signals)를 전달하는 DAG 태스크 스케줄러 (DAG task scheduler)
• 동시성을 제한하고 이미 에러를 반환하고 있는 엔드포인트로의 작업 전송을 중단하는 서킷 브레이커 풀 (circuit-breaker pool)

이 글에서는 각 구성 요소를 살펴보고, 실제 코드를 보여주며, 아직 구현되지 않은 부분에 대해서도 솔직하게 다룹니다.

구체적인 문제 상황

연구자(researcher), 작가(writer), 번역가(translator), 발행인(publisher)이라는 네 명의 에이전트가 있다고 가정해 봅시다. 작가는 연구자에게 의존합니다. 번역가와 발행인은 모두 작가에게 의존하지만 병렬로 실행될 수 있습니다. 만약 작가가 실패한다면 발행인은 전혀 실행되어서는 안 됩니다.

스케줄러가 없다면, 여러분은 매번 이를 수동으로 작성해야 합니다. 특히 실패 전파(failure-cascade) 로직은 새로운 의존성 엣지(dependency edge)가 추가될 때마다 늘어나는 중첩된 조건문 세트가 되기 쉽습니다. 그리고 만약 여러분의 LLM 제공자가 429 또는 5xx 에러를 연속해서 반환한다면, 프로세스를 종료할 때까지 루프가 엔드포인트를 계속해서 두들기는 것을 막을 방법이 없습니다.

aegis-gov는 두 개의 집중된 클래스인 TaskQueue와 AgentPool을 통해 이 두 가지 문제를 해결합니다.

설치

pip install aegis-gov                      # requests만 설치
pip install "aegis-gov[anthropic]"         # + Anthropic SDK
pip install "aegis-gov[openai]"            # + OpenAI SDK
...

Python 3.10+ 버전이 필요합니다.

어댑터 계층 (The adapter layer)

LLMAdapter 프로토콜은 두 개의 메서드를 가집니다: generate()는 문자열을 반환하고, stream()은 문자열 청크 (string chunks)를 생성(yield)합니다. 세 가지 구체적인 어댑터(concrete adapters) 모두 이 프로토콜을 준수합니다:

from aegis_gov import AnthropicAdapter, OpenAIAdapter, OllamaAdapter

# Anthropic
...

어댑터는 AgentConfig의 필드이므로, 단일 에이전트의 제공자(provider)를 전환하는 것은 한 줄의 코드 변경만으로 가능합니다. 오케스트레이션 (orchestration) 코드의 나머지 부분은 어떤 제공자가 사용되고 있는지 알 필요가 없습니다.

단일 에이전트 실행하기 (Running a single agent)

from aegis_gov import OpenMultiAgent, AgentConfig, AnthropicAdapter

oma = OpenMultiAgent()
...

DAG 태스크 스케줄러 (The DAG task scheduler)

TaskQueue는 Task 객체 리스트를 받으며, 생성 시점에 의존성 그래프 (dependency graph)의 순환(cycle) 여부를 검증합니다 (순환이 발견되면 CyclicDependencyError를 발생시킵니다). 또한 모든 의존성이 done 상태인 태스크들을 반환하는 ready() 메서드를 제공합니다.

from aegis_gov import OpenMultiAgent, Task
from aegis_gov import AgentConfig, AnthropicAdapter

...

translate와 publish는 모두 draft에만 의존하므로, draft가 완료되면 병렬로 실행됩니다.

연쇄 실패 (Cascade failure)

stop_on_failure=False (기본값)인 경우, 실패한 태스크에 직접적 또는 간접적으로 의존하는 태스크들만 건너뜁니다. 독립적인 브랜치(branches)는 계속 진행됩니다:

from aegis_gov import TaskQueue, Task

q = TaskQueue([
...

stop_on_failure=True로 설정하면 첫 번째 실패 발생 시 전체 큐(queue)가 중단됩니다.

서킷 브레이커 (The circuit breaker)

AgentPool은 threading.Semaphore를 래핑(wrap)하여 동시에 실행되는 에이전트의 수를 제한하며, 연속적인 실패를 추적하여 서킷을 개방(open)합니다:

from aegis_gov import AgentPool, OpenMultiAgent

pool = AgentPool(
...

상태 머신(state machine)에는 세 가지 상태가 있습니다: CLOSED (정상), OPEN (새로운 작업을 거부하며 CircuitOpenError를 발생시킴), 그리고 HALF_OPEN (recovery_timeout_s가 경과한 후 하나의 프로브(probe) 호출을 보냄). 성공적인 프로브는 회로를 CLOSED 상태로 되돌리며, 또 다른 실패가 발생하면 회로를 다시 엽니다.

도구 레지스트리 (The tool registry)

에이전트에게 호출 가능한 도구(callable tools)를 부여할 수 있습니다:

from aegis_gov import ToolRegistry

registry = ToolRegistry()
...

내장된 도구들은 얇은 래퍼(thin wrappers)입니다. 이들은 프로덕션(production) 환경에서 사용하기 위해 강화(hardened)되지 않았으므로, 직접 구현한 코드로 교체할 스텁(stub)으로 취급하십시오.

수행하지 않는 작업

범위(scope)에 대해 솔직하게 밝히는 것이 중요합니다:

• 비동기 런타임(async runtime) 미지원. 모든 실행은 threading.Semaphore 및 ThreadPoolExecutor를 사용하는 동기(synchronous) 방식입니다. asyncio 네이티브 에이전트가 필요하다면, 이 라이브러리는 아직 적합하지 않습니다.
• 재시도 로직(retry logic) 미내장. 서킷 브레이커(circuit breaker)는 연쇄적인 장애(cascading failures)로부터 보호하지만, 지수 백오프(backoff)를 사용하여 실패한 호출을 자동으로 재시도하지는 않습니다. 재시도는 작업 로직이나 어댑터(adapter)에서 직접 처리하십시오.
• 오케스트레이터(orchestrator)를 통한 스트리밍 미지원. 개별 어댑터는 stream()을 지원하지만, run_agent() 및 run_tasks()는 반환하기 전에 전체 응답을 수집합니다.
• 지속성(persistence) 미지원. 작업 상태(task state)는 프로세스가 지속되는 동안 메모리에 머뭅니다. 체크포인트(checkpoint)나 재개(resume) 기능은 없습니다.
• 최소한의 내장 도구. shell, http_get 등은 샌드박싱(sandboxing), 속도 제한(rate limiting), 또는 오류 강화(error enrichment) 기능이 없는 얇은 래퍼입니다.
• 알파(Alpha) 상태. 공개 API는 1.0 버전 이전에 변경될 수 있습니다. 이에 따라 적절히 취급하십시오.

로드맵 (Roadmap)

대략적인 우선순위 순서대로 제가 다음에 작업할 계획인 영역들입니다:

1. 비동기 지원 (asyncio.Semaphore, async def generate())
2. 풀(pool) 레벨에서의 지수 백오프(exponential backoff)를 포함한 설정 가능한 재시도
3. run_agent()에서의 스트리밍 패스스루(streaming pass-through)
4. 더 나은 관측성 훅(observability hooks) — 작업 상태 전환 시의 콜백(callbacks)
5. 내장 도구 구현의 견고화

기여(Contributions)와 이슈 보고(issue reports)를 환영합니다. 테스트 스위트는 pytest를 사용합니다. 개발용 추가 패키지는 pyproject.toml을 참조하십시오.

링크 (Links)

링크 (Links)

• GitHub: https://github.com/aegis-os/aegis-gov
• PyPI: https://pypi.org/project/aegis-gov/
• Changelog: https://github.com/aegis-os/aegis-gov/blob/main/CHANGELOG.md
• 설치 (Install): pip install "aegis-gov[all]"