LLM 컨텍스트 창을 초과하지 않고 자율 서브 에이전트를 조율하는 방법
Source: Dev.to
우리는 모두 “단일 LLM 장벽(monolithic LLM wall)”에 부딪혔습니다.
매우 강력한 AI 에이전트를 설계하고, 다양한 도구들을 장착한 뒤 복잡하고 다단계 작업—예를 들어 데이터 분석, 웹 조사, 코드 검증이 포함된 포괄적인 기술 논문 작성—을 부여합니다. 처음엔 완벽히 작동합니다. 하지만 단계가 쌓일수록 컨텍스트 윈도우가 가득 차고, 에이전트는 “주의력 흐트러짐(attention drift)”을 겪기 시작합니다. 원래 지시를 잊어버리고, 도구 출력이 환각되며, 결국 수백만 토큰과 API 예산을 소모하면서 통제 불능 상태에 빠집니다.
문제는 LLM의 추론 능력이 아니라 아키텍처에 있습니다. 복합적인 다도메인 문제를 단일 에이전트의 컨텍스트 윈도우 안에서 해결하려는 시도는, 마치 하나의 거대한 main() 함수 안에 전체 엔터프라이즈 애플리케이션을 작성하는 현대 소프트웨어와 같은 비효율적인 접근입니다.
현실 세계의 복잡성을 다룰 수 있는 AI 시스템을 구축하려면 단일 에이전트에서 계층적 다중 에이전트 오케스트레이션으로 전환해야 합니다.
복잡한 목표를 격리된, 전문화된 서브 에이전트들로 분해하고—각각이 자체 제한된 컨텍스트와 자원 예산 내에서 동작하도록—함으로써, 무한히 확장 가능한 회복력 있고 자체 개선 가능한 AI 시스템을 만들 수 있습니다.
이 글에서는 다중 에이전트 오케스트레이션의 아키텍처 패턴을 깊이 파고들고, 에이전트 수명 주기를 관리하는 방법을 탐구하며, 서브 에이전트를 생성·감독하는 프로덕션 급 파이썬 코드를 작성해 보겠습니다.
(여기에 소개된 개념과 코드는 제 전자책 Hermes Agent, The Self-Evolving AI Workforce 에서 발췌되었습니다.)
1. 핵심 개념: 계층적 분해와 감독 제어
다중 에이전트 오케스트레이션은 단순히 설계상의 편의가 아니라 아키텍처적 필수조건입니다. 이 접근법의 이론적 토대는 두 가지 기둥, **작업 분해(task decomposition)**와 감독 제어(supervisory control) 위에 세워집니다. 이 두 요소가 결합돼 단일 에이전트를 확장 가능하고 회복력 있는 전문화된 작업자들의 계층 구조로 변모시킵니다.
마스터 목수 비유
맞춤형 캐비닛을 만드는 마스터 목수를 떠올려 보세요. 마스터는 모든 도브테일을 직접 자르거나, 모든 표면을 샌딩하거나, 모든 힌지를 직접 설치하지 않습니다. 대신 프로젝트를 조인트 작업, 마감 작업, 하드웨어 설치와 같은 뚜렷한 하위 작업으로 나눕니다.
각 하위 작업마다 적절한 도구와 전문성을 갖춘 견습공에게 맡깁니다. 마스터는 그들의 진행 상황을 모니터링하고, 품질을 검사하며, 개별 결과물을 최종 제품에 통합합니다. 견습공이 난관에 봉착하면 마스터가 개입해 지도하거나 자원을 재배정합니다.
이 시나리오에서 부모 에이전트는 마스터 목수이고, 서브 에이전트는 견습공입니다. 각 견습공은 자신만의 집중된 도구 세트와 독립적인 **반복 예산(iteration budget)**을 가지고 작업합니다.
+------------------+
| Parent Agent | self.limit:
raise TimeoutError("Iteration budget exceeded!")
class AIAgent:
def __init__(self, **kwargs):
self.config = kwargs
self.session_id = kwargs.get("session_id")
self.budget = IterationBudget(kwargs.get("max_iterations", 50))
async def run_conversation(self, prompt: str) -> Dict[str, Any]:
# Simulate agent execution and tool calling
await asyncio.sleep(1)
self.budget.consume(5) # Simulate consuming 5 iterations of reasoning
return {
"status": "success",
"output": f"Processed prompt: '{prompt}' using model {self.config.get('model')}",
"iterations_used": self.budget.used
}
class SessionDB:
def __init__(self, db_path: Path):
self.db_path = db_path
self.db_path.mkdir(parents=True, exist_ok=True)
self.sessions_file = self.db_path / "sessions.json"
if not self.sessions_file.exists():
self.sessions_file.write_text("{}")
def ensure_tables(self):
# In a real SQL database, this would execute CREATE TABLE statements
pass
def upsert_session(self, session_id: str, metadata: Dict[str, Any]):
data = json.loads(self.sessions_file.read_text())
data[session_id] = metadata
self.sessions_file.write_text(json.dumps(data, indent=4))
print(f"💾 Session '{session_id}' persisted to database.")
def get_hermes_home() -> Path:
home = Path.home() / ".hermes"
home.mkdir(exist_ok=True)
return home
# Setup Logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("MultiAgentOrchestrator")
# ---------------------------------------------------------------------------
# Step 1: Parent Agent Supervisor Configuration
# ---------------------------------------------------------------------------
parent_config = {
"base_url": "https://api.openai.com/v1",
"api_key": "sk-mock-key",
"model": "gpt-4o",
"provider": "openai",
"api_mode": "chat",
"max_iterations": 90, # Parent gets a generous budget
"tool_delay": 1.0, # Rate-limiting safety delay
"enabled_toolsets": ["filesystem", "web", "terminal", "code_execution"],
"save_trajectories": True,
"session_id": "supervisor_session_101",
}
# Initialize Parent Agent
parent_agent = AIAgent(
base_url=parent_config["base_url"],
api_key=parent_config["api_key"],
model=parent_config["model"],
provider=parent_config["provider"],
api_mode=parent_config["api_mode"],
max_iterations=parent_config["max_iterations"],
tool_delay=parent_config["tool_delay"],
enabled_toolsets=parent_config["enabled_toolsets"],
save_trajectories=parent_config["save_trajectories"],
session_id=parent_config["session_id"],
)
logger.info(f"Supervisor Agent Initialized. Model: {parent_config['model']} | Session: {parent_config['session_id']}")
# ---------------------------------------------------------------------------
# Step 2: Initialize Persistent Session Storage
# ---------------------------------------------------------------------------
hermes_home = get_hermes_home()
session_db = SessionDB(db_path=hermes_home / "sessions")
session_db.ensure_tables()
# Register parent session in DB
session_db.upsert_session(
session_id=parent_config["session_id"],
metadata={
"role": "supervisor",
"model": parent_config["model"],
"max_iterations": parent_config["max_iterations"],
"status": "active"
}
)
# ---------------------------------------------------------------------------
# Step 3: Sub-Agent Spawner Configuration & Lifecycle Management
# ---------------------------------------------------------------------------
SUB_AGENT_MODEL = "gpt-4-mini" # Using a faster, cheaper model for sub-agents
SUB_AGENT_MAX_ITERATIONS = 50 # Capped iteration budget for safety
def build_sub_agent_config(task_slug: str, specialized_tools: List[str]) -> dict:
"""
Generates a tailored configuration for a specialized sub-agent.
"""
sub_session_id = f"{parent_config['session_id']}_sub_{task_slug}"
return {
"base_url": parent_config["base_url"],
"api_key": parent_config["api_key"],
"model": SUB_AGENT_MODEL,
"provider": parent_config["provider"],
"api_mode": "chat",
"max_iterations": SUB_AGENT_MAX_ITERATIONS,
"tool_delay": 0.5,
"enabled_toolsets": specialized_tools, # Restrict tools to only what is needed!
"save_trajectories": True,
"session_id": sub_session_id,
}
async def orchestrate_sub_task(task_name: str, prompt: str, tools: List[str]) -> Dict[str, Any]:
"""
Spawns, executes, tracks, and terminates a sub-agent.
"""
logger.info(f"🚀 Spawning sub-agent for task: [{task_name}]")
# Generate configuration
sub_config = build_sub_agent_config(task_name, tools)
# Persist sub-agent creation t