6. 에이전트 코드 리팩터링: 계층 분리와 이벤트 기반 루프

저번에 5편 글에서는 파이썬(Python)으로 병렬 도구 호출(Parallel Tool Use)을 어떻게 처리하는지 알아보았다.

 

하지만 loop.py 한 파일에 API 호출, 도구 정의, 루프 처리, 터미널 출력까지 다 때려박아 놔서 코드가 점점 복잡해지고 있었다..

이제 진짜 언리얼 엔진 연동이나 더 많은 도구들을 붙이려면 깔끔하게 리팩터링해야 한다.

 

그래서 이번에는 코드를 LLM 프로바이더, 도구 레지스트리, 이벤트 기반 루프, UI 진입점으로 역할을 나누고 완전히 쪼개보았다.

 

1. 변경된 폴더 구조

역할 분담에 맞춰 파일을 다음과 같이 정리했다.

 

agent/
├── llm/
│   ├── __init__.py
│   ├── base.py
│   └── claude.py
├── tools/
│   ├── __init__.py
│   ├── base.py
│   └── builtin/
│       ├── time_tool.py
│       └── math_tool.py
├── loop/
│   ├── agent_loop.py
│   └── events.py
ui/
└── backend/
    └── run.py

 

2. 1단계: LLM 프로바이더 분리 (agent/llm)

먼저 다양한 LLM(Claude, OpenAI, DeepSeek 등)을 유연하게 교체할 수 있도록 추상 클래스 LLMProvider를 만들었다.

 

# agent/llm/base.py
from abc import ABC, abstractmethod

class LLMProvider(ABC):
    @abstractmethod
    def call(self, messages: list, tools: list) -> dict:
        """messages + tools 를 보내고 응답 dict 를 반환한다."""
        raise NotImplementedError

 

그리고 이 추상 클래스를 상속받아 실제로 Anthropic Claude API를 호출하는 ClaudeProvider를 구현했다.

 

# agent/llm/claude.py
import requests
from agent.llm.base import LLMProvider

class ClaudeProvider(LLMProvider):
    def __init__(self, api_key: str, model: str = "claude-haiku-4-5", max_tokens: int = 500):
        self.api_key = api_key
        self.model = model
        self.max_tokens = max_tokens

    def call(self, messages: list, tools: list) -> dict:
        headers = {
            "x-api-key": self.api_key,
            "anthropic-version": "2023-06-01",
            "content-type": "application/json",
        }
        body = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "tools": tools,
            "messages": messages,
        }
        response = requests.post("https://api.anthropic.com/v1/messages", headers=headers, json=body, timeout=60)
        if response.status_code != 200:
            raise RuntimeError(f"Claude API {response.status_code}: {response.text}")
        return response.json()

 

3. 2단계: 도구(Tools) 모듈화 및 레지스트리 (agent/tools)

도구들도 클래스 구조로 묶어 모듈화했다. Tool 구조체에 이름, 설명, 입력 스키마, 그리고 실제 실행할 run 함수를 정의한다.

 

# agent/tools/base.py
from dataclasses import dataclass
from typing import Callable

@dataclass
class Tool:
    name: str
    description: str
    input_schema: dict
    run: Callable[..., str]

    def spec(self) -> dict:
        return {
            "name": self.name,
            "description": self.description,
            "input_schema": self.input_schema,
        }

 

이렇게 쪼개놓으면 개별 도구(예: time_tool.py, math_tool.py)를 각각 독립된 파일로 깔끔하게 관리할 수 있게 된다.

도구들을 모아서 LLM에 스펙을 넘기고 실행하는 기능은 레지스트리(__init__.py)에서 중앙 제어한다.

 

# agent/tools/__init__.py
from agent.tools.builtin.time_tool import get_time
from agent.tools.builtin.math_tool import add_numbers

_REGISTRY = {t.name: t for t in [get_time, add_numbers]}

def specs() -> list:
    return [t.spec() for t in _REGISTRY.values()]

def execute(name: str, tool_input: dict) -> tuple[str, bool]:
    tool = _REGISTRY.get(name)
    if tool is None:
        return f"에러: 알 수 없는 도구 '{name}'", True
    try:
        return str(tool.run(**tool_input)), False
    except Exception as error:
        return f"에러: {error}", True

 

4. 3단계: 이벤트 기반 루프 구현 (agent/loop)

이 리팩터링의 핵심 중 하나다. 기존에는 루프 코드 안에서 print를 직접 호출하여 출력을 처리했다.

하지만 앞으로 데스크톱 앱(Electron 등)이나 소켓으로 메시지를 보내야 할 수도 있으므로, 출력을 루프와 완전히 분리했다.

루프 내부에서는 print 대신 특정 사건이 발생할 때마다 on_event() 콜백 함수로 이벤트를 발행한다.

 

# agent/loop/events.py
from dataclasses import dataclass

@dataclass
class TurnStart:
    turn: int
    stop_reason: str | None

@dataclass
class AssistantText:
    text: str

@dataclass
class ToolCall:
    name: str
    input: dict

@dataclass
class ToolResult:
    name: str
    output: str
    is_error: bool

@dataclass
class FinalText:
    text: str

@dataclass
class Usage:
    input_tokens: int
    output_tokens: int

 

이벤트를 발행해 주는 에이전트 루프의 핵심 로직이다.

 

# agent/loop/agent_loop.py
from typing import Any, Callable
from agent.tools import specs, execute
from agent.loop import events

MAX_TURNS = 10

def run(question: str, provider, on_event: Callable) -> None:
    messages: list[dict[str, Any]] = [{"role": "user", "content": question}]
    total_in, total_out = 0, 0

    for turn in range(1, MAX_TURNS + 1):
        data = provider.call(messages, specs())
        
        usage = data.get("usage", {})
        total_in += usage.get("input_tokens", 0)
        total_out += usage.get("output_tokens", 0)

        stop_reason = data.get("stop_reason")
        on_event(events.TurnStart(turn, stop_reason))

        if stop_reason != "tool_use":
            for block in data.get("content", []):
                if block.get("type") == "text":
                    on_event(events.FinalText(block.get("text", "")))
            break

        for block in data.get("content", []):
            if block.get("type") == "text":
                on_event(events.AssistantText(block.get("text", "")))

        messages.append({"role": "assistant", "content": data["content"]})
        results: list[dict[str, Any]] = []

        for block in data["content"]:
            if block.get("type") != "tool_use":
                continue

            name = block["name"]
            tool_input = block.get("input", {})
            on_event(events.ToolCall(name, tool_input))

            output, is_error = execute(name, tool_input)
            on_event(events.ToolResult(name, output, is_error))

            result_block = {
                "type": "tool_result",
                "tool_use_id": block["id"],
                "content": output,
            }
            if is_error:
                result_block["is_error"] = True
            results.append(result_block)

        messages.append({"role": "user", "content": results})
    on_event(events.Usage(total_in, total_out))

 

5. 4단계: 백엔드 진입점 및 터미널 출력 (ui/backend)

마지막으로 루프를 실행하고 이벤트를 받아 실제로 화면에 출력해 주는 진입점(run.py)이다.

 

# ui/backend/run.py
import sys
from pathlib import Path
from dotenv import load_dotenv
from agent.llm import get_provider
from agent.loop.agent_loop import run
from agent.loop import events

load_dotenv(Path(__file__).resolve().parents[2] / "agent" / ".env")

def cli_print(event) -> None:
    """이벤트를 받아서 터미널에 출력한다. (나중에 Electron과 연동할 땐 이 함수만 JSON 출력으로 변경)"""
    if isinstance(event, events.TurnStart):
        print(f"\n[턴 {event.turn}] stop_reason={event.stop_reason}")
    elif isinstance(event, events.AssistantText):
        print(f"  (Claude) {event.text}")
    elif isinstance(event, events.ToolCall):
        print(f"  → 도구 호출: {event.name}({event.input})")
    elif isinstance(event, events.ToolResult):
        print(f"  ← 결과: {event.output}")
    elif isinstance(event, events.FinalText):
        print(event.text)
    elif isinstance(event, events.Usage):
        print(f"\n=== 누적 토큰 : 입력 {event.input_tokens} / 출력 {event.output_tokens} ===")

def main() -> None:
    if len(sys.argv) < 2:
        print('사용법 : python -m ui.backend.run "질문"')
        sys.exit(1)

    provider = get_provider("claude")
    run(" ".join(sys.argv[1:]), provider, on_event=cli_print)

if __name__ == "__main__":
    main()

 

6. 느낀 점

설계 분리가 끝나니까 결합도가 낮아져서 확실히 코드가 단정해진 느낌이다.

만약 OpenAI 모델로 바꾸고 싶다면 agent/llm 계층에 프로바이더만 추가하면 되고,

CLI 대신 웹이나 GUI 앱 화면에 에이전트를 올리고 싶다면 run.py의 cli_print 콜백 내부만 갱신해 주면 된다.