현재 에이전트 AI (Agentic AI)를 배우고 있는 제가 아는 모든 사람들은 프레임워크를 '사용하는' 법을 배우고 있습니다. 그들은 .invoke()를 호출하고, 무언가 일어나면 다음 단계로 넘어갑니다.

도구 (Tool) 클래스가 정의될 때 무엇이 실행되는지 물어보세요. 왜 검증 (Validation)이 __init__ 대신 디스크립터 (Descriptor)에 위치해야 하는지 물어보세요. ParamSpec이 데코레이터 스택 (Decorator stacks)을 통해 어떻게 타입 안전성 (Type safety)을 유지하는지 물어보세요. 다들 멍한 표정만 지을 뿐입니다.

저는 실제로 이 내용들을 이해하고 싶었습니다. 그래서 실제 에이전트 시스템의 내부 구조를 반영하는 미니 Python 프레임워크를 구축했습니다 — 레지스트리 (Registry), 검증된 설정 (Validated config), 믹스인 (Mixins), 데코레이터 (Decorators), 세션 (Sessions), 스트리밍 (Streaming), TypedDicts, 프로토콜 (Protocols), 그리고 CLI까지 포함했습니다. 외부 의존성은 전혀 없습니다. 전체 과정은 소스 코드만으로 설명 가능합니다.

상세한 분석은 다음과 같습니다.

아키텍처 개요

agent_cli/
├── core/
│   ├── base.py        ← BaseTool: __init_subclass__, run(), stream()
...

하나씩 살펴보겠습니다.

1. __init_subclass__를 이용한 자동 등록 도구

모든 에이전트 프레임워크에서 마주하는 첫 번째 실제 설계 질문은 이것입니다: 런타임 (Runtime)은 어떤 도구들이 존재하는지 어떻게 알 수 있는가?

가장 단순한 접근 방식은 수동으로 관리하는 딕셔너리 (Dict)를 사용하는 것입니다. 문제는 명확합니다 — 도구 클래스를 추가하고 딕셔너리 항목을 넣는 것을 잊으면, 나중에 조용히 "도구를 찾을 수 없음" 오류가 발생합니다. 이는 실제로 흔히 발생하는 버그입니다.

해결책: 베이스 클래스 (Base class)에 __init_subclass__를 사용하는 것입니다. 이는 인스턴스화 (Instantiation) 시점이 아니라, 클래스 정의 시점에 실행됩니다.

class BaseTool(LoggingMixin, RetryMixin, MetricsMixin, ABC):
    def __init_subclass__(
        cls,
...

도구는 그저 자신을 선언하기만 하면 됩니다:

class SearchTool(BaseTool, tool_name="search", streamable=True):
    def execute(self, context: ToolContext) -> str:
        ...

Python이 해당 클래스를 파싱하는 순간 — SearchTool은 레지스트리에 등록됩니다. 수동 목록은 필요 없습니다. tool_name=을 잊어버리면 나중에 혼란스러운 실패가 발생하는 대신 즉시 TypeError를 받게 됩니다.

ToolRegistry는 평면적인 dict[str, type]를 저장하며, 두 클래스가 동일한 이름을 사용하려고 하면 DuplicateToolError를 발생시킵니다.

@classmethod
def register(cls, name: str, tool_cls: type[Any]) -> None:
    existing = cls._tools.get(name)
...

이 테스트는 등록이 오직 임포트 (import)를 통해 발생함을 증명합니다:

def test_builtin_tools_register_automatically(self) -> None:
    # agent_cli.tools를 임포트하면 세 클래스 모두에서 __init_subclass__가 트리거됩니다
    self.assertEqual(("search", "summarize", "translate"), ToolRegistry.names())

2. 데이터 디스크립터 (Data Descriptors)를 통한 설정 검증

재시도 루프 (retry loops) 내부로 조용히 전파되는 잘못된 설정 값은 디버깅하기 매우 어렵습니다. 해결책은 실행 시점 (execution time)이 아니라 할당 시점 (assignment time)에 검증하는 것입니다.

Python의 데이터 디스크립터 (data descriptors)는 __set__을 가로챕니다:

class ValidatedField(Generic[T]):
    def __set_name__(self, owner, name):
        self.public_name = name
...

구체적인 예시:

class IdentifierField(NonEmptyString):
    _pattern = re.compile(r"^[a-z][a-z0-9_-]*$")

...

ToolConfig는 이를 클래스 수준의 선언으로 사용합니다:

class ToolConfig:
    __slots__ = ("_retries", "_streaming_enabled", "_timeout", "_tool_name")

...

ToolConfig(tool_name="Bad Name", retries=9)는 생성 시 두 개의 ValueError를 발생시킵니다. 테스트는 이를 명시적으로 다룹니다:

def test_invalid_config_fails_fast(self) -> None:
    with self.assertRaises(ValueError):
        ToolConfig(tool_name="Bad Name", retries=1, timeout=10.0)
...

3. MRO를 통한 믹스인 (Mixins) 조합

class BaseTool(LoggingMixin, RetryMixin, MetricsMixin, ABC):

세 개의 믹스인 (mixins)이 있으며, 각 믹스인은 하나의 관심사 (concern)를 담당합니다. 모두 __slots__ = ()를 사용하여 인스턴스 상태 (instance state) 없이 순수 동작 (pure behavior)만을 가집니다:

LoggingMixin — 도구 이름을 접두사로 붙이는 범위가 지정된 self.log(message)를 제공합니다.

RetryMixin — self.with_retries(operation)는 모든 호출 가능한 객체 (callable)를 래핑 (wrap)합니다:

def with_retries(self, operation: Callable[[], R]) -> R:
    last_error = None
    for attempt in range(1, self.config.max_attempts + 1):
...

MetricsMixin — 실행 횟수, 실패, 총 소요 시간을 추적하는 슬롯화된 (slotted) ToolMetrics를 기반으로 하는 self.record_metric() 및 self.metrics를 제공합니다.

런타임에 전체 체인을 검사할 수 있습니다:

$ python main.py describe search
MRO: SearchTool -> BaseTool -> LoggingMixin -> RetryMixin -> MetricsMixin -> ABC -> object

4. 타입 시그니처를 보존하는 ParamSpec 데코레이터 (decorators)

단순한(Naive) 데코레이터는 타입 시그니처를 지워버립니다. mypy는 실제 타입 대신 Callable[..., Any]를 보게 됩니다. ParamSpec은 이 문제를 해결합니다:

P = ParamSpec("P")
R = TypeVar("R")

...

BaseTool.run()은 두 데코레이터를 모두 쌓습니다:

@log_execution
@measure_time
def run(self, raw_input: str, *, stream: bool = False, session_id: str = "standalone") -> ToolOutput:
...

두 래퍼(wrapper)를 거친 후에도 mypy는 여전히 run()의 전체 시그니처를 인식합니다. 이는 프로덕션 프레임워크가 사용자 대상 메서드 위에 계측(instrumentation) 레이어를 쌓을 때 사용하는 패턴입니다.

5. ExecutionSession: 컨텍스트 매니저 (context manager) 생명주기

모든 도구 실행은 세션(session) 내부에서 발생합니다:

with ExecutionSession() as session:
    session.add_resource(f"{tool.name}-runtime")
    result = tool.run(raw_input, stream=args.stream, session_id=session.session_id)

__enter__는 시작 시간을 기록하고 세션 시작을 로그로 남깁니다. __exit__는 성공 여부나 예외 발생 여부와 관계없이 모든 경우에 cleanup()을 호출하며, 추적된 리소스를 역순으로 해제하고 총 소요 시간을 기록합니다. __exit__는 False를 반환하여 예외가 정상적으로 전파되도록 합니다.

각 세션은 ToolContext로 흘러 들어가 ToolOutput을 통해 출력되는 고유한 session_id (session-{uuid4().hex[:10]})를 생성하므로, 모든 결과는 추적 가능합니다.

6. 제너레이터 스트리밍 (Generator streaming) + TypedDict 계약 (contracts)

두 실행 경로 모두 동일한 ToolOutput을 생성합니다:

class ToolOutput(TypedDict):
    tool: str
    content: str
...

스트리밍은 실제 제너레이터(generator)를 사용합니다:

def stream(self, context: ToolContext) -> Iterator[str]:
    for token in self.execute(context).split(" "):
        yield f"{token} "

BaseTool.run()은 경로에 관계없이 finally 블록에서 제너레이터 출력을 수집하고 메트릭(metrics)을 기록합니다:

try:
    if stream:
        tokens = self.with_retries(lambda: list(self.stream(context)))
...

finally 블록은 실패 시를 포함하여 메트릭(metrics)이 항상 기록되도록 보장합니다. 이는 프로덕션 관측성 (production observability) 측면에서 매우 중요합니다.

7. Protocol을 이용한 구조적 타이핑 (Structural typing)

레지스트리는 BaseTool이 아닌 ToolProtocol을 반환합니다:

@runtime_checkable
class ToolProtocol(Protocol):
    @property
...

CLI는 클래스가 아닌 계약 (contract)에 의존합니다. BaseTool을 상속받지 않더라도 ToolProtocol을 구현한 제3자 도구(third-party tool)라면 완전히 호환됩니다. 이것이 바로 프로덕션 프레임워크가 확장을 위해 개방된 상태를 유지하는 정확한 방식입니다.

Repo: https://github.com/Sajid0875/agentic-systems-bootcamp/tree/main/Week-01-Python-Agentic-Systems/Session%201/project_agent_cli%20

이 프로젝트는 16주간 진행되는 공개 빌드 시리즈의 1주 차입니다. Python → NLP → 임베딩 (embeddings) → LLM 내부 구조 (LLM internals) → RAG → 멀티 에이전트 (multi-agent) → MCP → 배포 (deployment) 순으로 진행됩니다. 매주 동일한 코드베이스를 확장하며 빌드와 포스팅이 이어집니다.