TorchSystem: AI 학습 시스템을 구조적이고 유지보수하기 쉽게 만드는 도메인 주도 설계(DDD) 기반 프레임워크

9bow · 10월 5, 2025, 3:00오전

TorchSystem 소개

TorchSystem은 PyTorch 위에서 동작하는 도메인 주도 설계(DDD, Domain-Driven Design) 기반의 프레임워크로, 복잡한 인공지능(AI) 학습 시스템을 구조적이고 유지보수 가능한 방식으로 설계하기 위해 만들어졌습니다. 이 프레임워크는 모델의 정의와 학습 과정, 데이터 처리, 로깅 및 평가 등의 요소를 하나의 아키텍처적 틀 안에서 통합적으로 관리할 수 있도록 돕습니다.

안내: TorchSystem은 독립적인 오픈소스 프로젝트로, Meta 및 PyTorch로부터 어떠한 인정/후원/지원을 받은 바가 없습니다. TorchSystem이라는 이름은 기술적 설명의 편의상 사용한 것일뿐, PyTorch 또는 그 개발 주체와의 연관성을 의미하지 않습니다.

일반적으로 PyTorch 프로젝트는 모델, 옵티마이저, 손실 함수, 데이터로더, 로깅 및 체크포인트 관리 코드가 뒤섞여 있어서 시스템이 커질수록 유지보수가 어려워집니다. TorchSystem은 이러한 문제를 해결하기 위해 “모델 중심”이 아닌 “시스템 중심”의 접근법을 제시합니다. 모델을 하나의 독립적인 단위로 다루면서도, 그 주변에서 일어나는 서비스 로직이나 이벤트 흐름을 체계적으로 관리할 수 있도록 지원합니다.

TorchSystem의 근본적인 목표는 모델 로직과 학습 로직을 명확히 분리하고, 이벤트 중심으로 동작하는 모듈형 구조를 제공하며, 의존성 주입(Dependency Injection)을 통해 구성 요소 간 결합도를 최소화하는 것입니다. 이러한 구조를 통해 대규모 AI 시스템에서도 재사용성과 확장성을 확보할 수 있습니다.

TorchSystem의 철학과 설계 원칙

TorchSystem은 도메인 주도 설계(DDD)의 원칙을 딥러닝 시스템에 적용한 프레임워크입니다. 도메인 주도 설계란 시스템을 구성하는 요소를 기술적 관점이 아닌 “문제 영역(Domain)” 중심으로 정의하고, 각 영역을 독립적인 구조로 분리하여 관리하는 설계 방식입니다.

TorchSystem에서는 모델과 관련된 객체들을 하나의 “Aggregate(집합체)”로 묶고, 학습과 평가와 같은 동작을 “Service(서비스)” 계층으로 분리하며, 서비스의 실행 결과를 다른 모듈에 전달하기 위해 “Event(이벤트)”와 “Consumer(소비자)”를 사용합니다. 또한 모델을 초기화하거나 디바이스로 옮기고, 가중치를 복원하는 등의 과정을 자동화하기 위해 “Compiler(컴파일러)” 개념을 도입했습니다. 이러한 구성 요소들은 의존성 주입 시스템을 통해 느슨하게 연결되며, 결과적으로 대규모 시스템에서도 쉽게 테스트하고 확장할 수 있는 구조를 형성합니다.

TorchSystem의 설치 및 시작

TorchSystem은 파이썬 기반으로 개발되었기 때문에 추가적인 서버나 외부 인프라 없이도 바로 사용할 수 있습니다. 다음 명령어로 간단히 설치할 수 있습니다:

pip install torchsystem

설치가 완료되면 PyTorch 환경 위에서 바로 시스템을 설계할 수 있습니다. 예시 코드와 함께 주요 개념들에 대해서 살펴보겠습니다.

Aggregate: 모델을 하나의 논리 단위로 묶기

TorchSystem의 가장 핵심적인 개념은 “Aggregate”입니다. 일반적으로 딥러닝 모델은 신경망 구조만을 중심으로 구현되지만, 실제로 모델이 동작하려면 손실 함수, 옵티마이저, 메트릭, 학습 루프 등 다양한 요소가 함께 필요합니다. TorchSystem에서는 이러한 요소를 하나의 단위로 묶어 관리합니다.

예를 들어, 분류 모델을 정의할 때 다음과 같이 신경망, 손실 함수, 옵티마이저를 하나의 객체로 묶어 관리할 수 있습니다:

from torch import Tensor
from torch.nn import Module
from torch.optim import Optimizer
from torchsystem import Aggregate
from torchsystem.registry import gethash, getname

class Classifier(Aggregate):
    def __init__(self, nn: Module, criterion: Module, optimizer: Optimizer):
        super().__init__()
        self.epoch = 0
        self.nn = nn
        self.criterion = criterion
        self.optimizer = optimizer
        self.name = getname(nn)
        self.hash = gethash(nn)

    def fit(self, inputs: Tensor, targets: Tensor):
        self.optimizer.zero_grad()
        outputs = self.nn(inputs)
        loss = self.criterion(outputs, targets)
        loss.backward()
        self.optimizer.step()
        return loss

이처럼 Aggregate를 사용하면 모델 내부에서 학습과 평가 로직을 명확히 구분할 수 있으며, 각 구성 요소의 상태를 일관성 있게 관리할 수 있습니다. gethash()와 getname() 함수는 모델의 고유 식별자를 생성하여 로그 기록이나 체크포인트 파일명 관리에 활용됩니다.

Service: 상태 없는 학습 및 평가 계층

TorchSystem의 “Service”는 모델의 상태를 직접 변경하지 않고 학습, 평가, 예측 등의 연산을 수행하는 독립 계층입니다. Service는 상태를 가지지 않으며, 모델의 동작을 조율하고 결과를 이벤트로 외부에 전달합니다.

다음 예시 코드는 모델 학습을 수행하는 서비스의 형태를 보여줍니다:

from torchsystem.services import Service, Producer, event
from torchsystem.depends import Depends, Provider

provider = Provider()
producer = Producer()
service = Service(provider=provider)

@event
class Trained:
    model: Classifier
    results: dict[str, float]

@service.handler
def train(model: Classifier, loader, device: str = Depends(lambda: "cuda")):
    model.train()
    for inputs, targets in loader:
        loss = model.fit(inputs.to(device), targets.to(device))
    producer.dispatch(Trained(model, {"loss": loss}))

위 예시 코드에서 train() 함수는 단순히 학습 과정을 수행할 뿐, 저장이나 로깅과 같은 부수적 기능은 담당하지 않습니다. 대신 학습이 끝나면 Trained 이벤트를 발행하여 다른 구성 요소가 이를 처리하도록 합니다. 이 방식은 학습 로직과 인프라 로직을 완전히 분리해 코드의 명확성과 재사용성을 높입니다.

Consumer: 이벤트를 처리하는 후속 단계

서비스가 발행한 이벤트는 “Consumer”가 받아서 처리합니다. Consumer는 로그 기록, 체크포인트 저장, 메트릭 계산, 모델 버전 관리 등과 같은 후처리를 담당합니다. 다음은 Consumer를 사용하는 예시 코드입니다:

from torchsystem.services import Consumer
from torch import save
from os import makedirs

@consumer.handler
def persist_model(event: Trained):
    makedirs("weights", exist_ok=True)
    path = f"weights/{event.model.name}-{event.model.hash}.pth"
    save({
        "epoch": event.model.epoch,
        "state_dict": event.model.nn.state_dict()
    }, path)

이벤트 기반 구조 덕분에 학습과 저장 로직이 완전히 분리되어, 모델 구조를 변경하거나 로깅 방식을 바꾸더라도 학습 로직을 수정할 필요가 없습니다. 이는 대규모 ML 파이프라인에서 특히 중요한 설계상의 장점입니다.

Compiler: 모델 빌드와 초기화의 자동화

TorchSystem은 모델의 빌드, 초기화, 디바이스 이동, 가중치 복원, torch.compile 적용 등을 자동화하기 위한 컴파일러 개념을 제공합니다. 컴파일러는 일련의 단계를 파이프라인처럼 정의하고, 이를 순서대로 실행해 모델을 완성합니다.

from torchsystem.compiler import Compiler, compile

compiler = Compiler[Classifier]()

@compiler.step
def build_model(nn, criterion, optimizer):
    return Classifier(nn, criterion, optimizer)

@compiler.step
def compile_model(classifier):
    return compile(classifier)

classifier = compiler.compile(nn, criterion, optimizer)

이러한 접근법은 모델 초기화 과정을 체계적으로 관리할 수 있게 하며, 각 단계를 명확히 분리해 유지보수를 쉽게 합니다.

Dependency Injection: 동적인 자원 주입

TorchSystem은 의존성 주입 시스템을 내장하고 있어, 서비스나 컴파일러가 필요한 리소스를 자동으로 주입받을 수 있습니다. 이를 통해 디바이스 설정이나 데이터 경로, 외부 리소스 접근을 코드 수정 없이 유연하게 변경할 수 있습니다.

def device() -> str:
    return 'cuda' if torch.cuda.is_available() else 'cpu'

@service.handler
def evaluate(model: Classifier, loader, device: str = Depends(device)):
    ...

이러한 방식은 테스트 환경과 실제 배포 환경을 쉽게 전환할 수 있게 해주며, 코드의 하드코딩을 최소화할 수 있게 해줍니다.

TorchSystem의 실제 활용 예시

하나의 예시로, TorchSystem을 사용해 간단한 MLP 모델을 훈련하는 전체 스크립트를 살펴볼 수 있습니다.

from torch import cuda
from torch.nn import CrossEntropyLoss
from torch.optim import SGD
from torch.utils.data import DataLoader
from torchsystem import registry

def device() -> str:
    return 'cuda' if cuda.is_available() else 'cpu'

if __name__ == '__main__':
    from src import training, checkpoints
    from src.compilation import compiler
    from model import MLP
    from dataset import Digits

    registry.register(MLP)
    training.provider.override(training.device, device)
    training.producer.register(checkpoints.consumer)

    nn = MLP(input_size=784, hidden_size=256, output_size=10, dropout=0.5)
    criterion = CrossEntropyLoss()
    optimizer = SGD(nn.parameters(), lr=0.001)
    classifier = compiler.compile(nn, criterion, optimizer)

    datasets = {
        'train': Digits(train=True, normalize=True),
        'eval': Digits(train=False, normalize=True)
    }

    loaders = {
        'train': DataLoader(datasets['train'], batch_size=256, shuffle=True),
        'eval': DataLoader(datasets['eval'], batch_size=256, shuffle=False)
    }

    for epoch in range(5):
        training.train(classifier, loaders['train'])
        training.evaluate(classifier, loaders['eval'])

위 스크립트는 기존의 체크포인트를 불러오고, 학습을 수행하며, 평가 후 자동으로 모델을 저장합니다. TorchSystem은 이 모든 과정을 서비스, 이벤트, 컴파일러, 소비자 등의 계층 구조를 통해 관리합니다.

마무리

TorchSystem은 단순히 PyTorch 위에서 모델을 학습하기 위한 도구가 아니라, AI 시스템 전체를 아키텍처적으로 설계할 수 있는 프레임워크입니다. 모델과 서비스, 이벤트, 소비자, 그리고 컴파일러가 유기적으로 연결된 구조를 통해, 복잡한 학습 시스템을 안정적이고 확장 가능한 방식으로 구축할 수 있습니다.

PyTorch의 유연성과 도메인 주도 설계의 견고함을 결합한 TorchSystem은, 이제 개발자가 단순히 “모델을 학습시키는 사람”이 아니라, “AI 시스템을 설계하는 엔지니어”로 성장할 수 있도록 돕는 강력한 도구입니다.

라이선스

TorchSystem 공식 문서

TorchSystem 프로젝트 GitHub 저장소

이 글은 GPT 모델로 정리한 글을 바탕으로 한 것으로, 원문의 내용 또는 의도와 다르게 정리된 내용이 있을 수 있습니다. 관심있는 내용이시라면 원문도 함께 참고해주세요! 읽으시면서 어색하거나 잘못된 내용을 발견하시면 덧글로 알려주시기를 부탁드립니다.

파이토치 한국 사용자 모임이 정리한 이 글이 유용하셨나요? 회원으로 가입하시면 주요 글들을 이메일로 보내드립니다! (기본은 Weekly지만 Daily로 변경도 가능합니다.)

아래쪽에 좋아요를 눌러주시면 새로운 소식들을 정리하고 공유하는데 힘이 됩니다~