우리 회사 플랫폼의 제품군은 크게 Label, Curate, Model, 그리고 Apps로 구성되어 있다.
고객이 업로드한 이미지나 비디오에서 Curate를 통해 데이터를 선별하고, Label에서 라벨을 붙이고, Model에서 AI 모델을 학습시켜 플랫폼 사용자는 나만의 Vision AI 모델을 손쉽게 API 형태로 사용할 수 있다.
그중 Apps는 고객이 플랫폼을 더 쉽고 유연하게 활용할 수 있도록 돕는 자동화 도구들을 제공한다.
주요 목적은 다음과 같다:
기존에 다른 라벨링 툴에서 작업한 데이터를 Superb Platform 에 업로드
(ex. YOLO, COCO, Labelme 등 다양한 어노테이션 포맷 → Superb 형식으로 자동 변환 후 플랫폼에 업로드)
이미지에서 사람 얼굴을 감지하고 자동으로 비식별화(blur)
OCR로 텍스트 영역을 자동 감지하고 바운딩 박스를 생성
즉, 고객이 “이사”오는 과정에서 겪는 여러 번거로운 과정을 자동화함으로써 onboarding을 부드럽게 만든다.
Apps는 어떻게 실행되는가?
Apps는 여러 기능들을 개별 app 형태로 제공하는 Superb Platform의 구성 요소이며, 각 app은 독립적인 컨테이너로 실행되어 Kubernetes 상에서 운영된다. 실행 흐름은 다음과 같다:
사용자가 웹 프론트를 통해 데이터를 업로드하면, 해당 파일은 S3에 저장된다.
지정된 app이 실행되어 S3의 입력 데이터를 내려받아 처리한다.
처리 결과는 다시 S3에 업로드되거나, 다운로드 링크 또는 플랫폼 내 리소스로 제공된다.
이 모든 과정은 Apps의 백엔드 서버와의 통신을 통해 상태 및 진행 상황이 관리된다.
개별 app은 다양한 개발자가 만들지만, 공통된 실행 흐름과 시스템 환경을 모두 이해해야만 동작하도록 만들고 싶지는 않았다.
AppWrapper의 목적은 앱 실행에 필요한 공통 작업들을 추상화하여, 개발자가 비즈니스 로직에만 집중할 수 있게 하는 것이었다.
하지만 이 실행 흐름은 단순해 보이는 것과 달리, 환경 설정, presigned URL 요청, 오류 처리, 결과 업로드 등을 앱마다 수작업으로 구현하면 유지보수와 품질 관리가 매우 어려워진다.
AppWrapper의 역할
Apps의 실행 흐름에는 파일 다운로드, 결과 업로드, 상태 보고 등 반복되는 작업이 필수적으로 포함된다. 이런 공통 작업을 자동으로 처리해주는 유틸리티가 바로 AppWrapper다.
AppWrapper는 클래스 데코레이터 방식으로 동작하며, 다음과 같은 실행 흐름을 관리한다:
앱 실행 전 (Pre-Processing)
앱 실행에 필요한 입력 파일 목록과 파일 저장 경로, 작업 ID 등의 실행 정보를 시스템으로부터 전달받는다.
전달받은 정보를 기반으로, 사용자가 웹에서 업로드한 파일들을 S3에서 가져와 Pod의 로컬 디렉터리에 저장한다. (정확히는 emptyDir 형태)
만약 실행 준비 중 오류가 발생하면, 해당 작업은 즉시 실패 처리되고 상태가 백엔드로 보고된다.
앱 실행 중
개발자가 작성한 process() 함수를 호출하여 실제 비즈니스 로직을 수행한다.
예를 들어 다음과 같이 간단히 작성할 수 있다:
@AppWrapper()
def process():
# 예: YOLO 라벨 포맷을 Superb 포맷으로 변환
return {
"type": "download",
"file_path": "/tmp/converted.zip"
}
process()
실행 후 (Post-Processing)
결과의 타입이 download인 경우 → presigned PUT URL을 통해 파일 업로드 후 웹 프론트에서 사용자가 결과 파일 download 가능
결과의 타입이 link인 경우 → URL 그대로 반환
상태 업데이트 및 로그 저장 (S3에 로그 업로드 포함)
결과 포맷은 단순하지만 명확하다
이 통일된 구조 덕분에, 어떤 앱이든 동일한 방식으로 결과를 처리할 수 있다.
Case 1: type이 link인 경우
{
"type": "link",
"url": "특정url",
"label": "Go to Project"
}
app의 결과물이 hyperlink인 경우
url key가 필요 (ex) platform의 특정 url)
label key는 화면에서 보여줄 내용 (ex) 아래 사진에 가장 오른쪽 “Go to Project “버튼 )
Case 2: type이 download인 경우
app의 결과물이 파일인경우
file_path key가 필요 (Pod 내부의 파일 경로)
AppWrapper의 도입 효과
항목
개선 전
개선 후
코드 중복
presigned 요청, S3 처리 반복
데코레이터로 추상화
예외 처리
앱마다 오류 처리 제각각
실패 시 상태 일괄 업데이트
로그 수집
개발자가 직접 구현
자동 S3 업로드
개발 속도
새로운 앱 추가 시 부가작업 다수
비지니스 로직만 구현하면 끝
이 구조 덕분에 앱 개발자는 “파일을 어떻게 받을지, 어디에 저장할지, 상태를 어떻게 보고할지” 같은 운영상의 디테일을 몰라도 앱 로직만 작성하면 된다.
실제 AppWrapper 데코레이터 내부는 어떻게 구현되어 있을까?
AppWrapper는 클래스지만, __call__ 메서드를 구현함으로써 함수 데코레이터처럼 동작한다. 로컬 모드와 운영 모드가 명확히 분기되어 있어 개발/운영 환경에서 모두 활용 가능하고, 예외 발생 시 즉시 로그 저장 & 상태 보고를 통해 시스템 안정성 확보한다.
class AppWrapper:
....
def __call__(self, func):
def inner(*args, **kwargs):
if self.LOCAL_MODE:
try:
result = func(*args, **kwargs)
except Exception as e:
logger.error(e)
sys.exit(1)
else:
try:
result = func(*args, **kwargs) # 비지니스 로직 함수
except Exception as e:
logger.error(traceback.format_exc())
.....
if task_data["Task"]["status"] != Status.Canceled.value:
self.send_pod_log_to_s3() # pod 로그 s3로 전송
self.update_status(Status.Failed.value, str(e))
try:
# 반환된 결과 포맷 유효성 검사
self.validate_result_format(result)
# 파일이 포함된 경우 업로드
if result["type"] == "download":
self.upload_file_to_s3(result)
# 결과를 backend에 전송
data = {"task_id": self.task_id, "result": result}
self.send_pod_log_to_s3()
requests.post(
f"{self.TOAD_HOST}/output/",
data=json.dumps(data),
)
# 성공 상태 업데이트
self.update_status(Status.Complete.value, "App completed")
except Exception as e:
error_log = f"Post app failed: {e}"
self.update_status(Status.Failed.value, error_log)
return inner
AppWrapper는 어떻게 활용되고 있을까?
AppWrapper는 내부 앱 개발자들이 공통으로 사용할 수 있도록 구조화되어 있으며,
별도의 설치 없이 앱 코드에서 쉽게 가져다 사용할 수 있도록 PyPI에 패키징되어 배포되어 있다.
$ pip install AppWrapper
PyPI에 공개(링크)되어 있어 누구나 설치는 가능하지만, 소스 코드는 private GitHub repository에서 관리되고 있어 내부 사용자만 직접 수정하거나 검토할 수 있다.
앱 개발자는 별도의 설정 없이 @AppWrapper 데코레이터만 붙이면, 파일 다운로드, 결과 업로드, 상태 보고, 로그 저장 등 반복되는 실행 흐름을 자동으로 처리할 수 있다.
개선 포인트와 앞으로의 방향
AppWrapper는 현재까지 수십 개의 앱에서 안정적으로 사용되며, 개별 app들의 실행 흐름에 많은 역할을 하고 있다. 하지만 실제 운영하면서 다음과 같은 개선 가능성도 보였다:
1. 단일 클래스가 너무 많은 책임을 가짐
환경 초기화, 파일 다운로드/업로드, 상태 전송, 로그 처리, 예외 핸들링 등 너무 많은 역할이 한 클래스에 몰려 있다.
SRP(Single Responsibility Principle)를 따르는 구조로 리팩토링할 필요가 있다. (예: S3Client, StatusManager, ResultValidator 등으로 분리)
2. 결과 포맷의 유효성 검사는 코드로만 정의됨
result는 dict 형식에 "type", "url" 혹은 "file_path"가 필요하지만, 이를 Pydantic 등으로 명시하지 않았다. Pydantic 기반의 result schema 도입을 통해 유효성 검사를 개선할 수 있겠다.
/* Tokenizer state */
struct tok_state {
/* Input state; buf <= cur <= inp <= end */
/* NB an entire line is held in the buffer */
char *buf; /* Input buffer, or NULL; malloc'ed if fp != NULL */
char *cur; /* Next character in buffer */
char *inp; /* End of data in buffer */
const char *end; /* End of input buffer if buf != NULL */
const char *start; /* Start of current token if not NULL */
int done; /* E_OK normally, E_EOF at EOF, otherwise error code */
/* NB If done != E_OK, cur must be == inp!!! */
FILE *fp; /* Rest of input; NULL if tokenizing a string */
int tabsize; /* Tab spacing */
int indent; /* Current indentation index */
int indstack[MAXINDENT]; /* Stack of indents */
int atbol; /* Nonzero if at begin of new line */
int pendin; /* Pending indents (if > 0) or dedents (if < 0) */
const char *prompt, *nextprompt; /* For interactive prompting */
int lineno; /* Current line number */
int first_lineno; /* First line of a single line or multi line string
expression (cf. issue 16806) */
int level; /* () [] {} Parentheses nesting level */
/* Used to allow free continuations inside them */
char parenstack[MAXLEVEL];
int parenlinenostack[MAXLEVEL];
PyObject *filename;
/* Stuff for checking on different tab sizes */
int altindstack[MAXINDENT]; /* Stack of alternate indents */
/* Stuff for PEP 0263 */
enum decoding_state decoding_state;
int decoding_erred; /* whether erred in decoding */
int read_coding_spec; /* whether 'coding:...' has been read */
char *encoding; /* Source encoding. */
int cont_line; /* whether we are in a continuation line. */
const char* line_start; /* pointer to start of current line */
const char* multi_line_start; /* pointer to start of first line of
a single line or multi line string
expression (cf. issue 16806) */
PyObject *decoding_readline; /* open(...).readline */
PyObject *decoding_buffer;
const char* enc; /* Encoding for the current str. */
char* str;
char* input; /* Tokenizer's newline translated copy of the string. */
int type_comments; /* Whether to look for type comments */
/* async/await related fields (still needed depending on feature_version) */
int async_hacks; /* =1 if async/await aren't always keywords */
int async_def; /* =1 if tokens are inside an 'async def' body. */
int async_def_indent; /* Indentation level of the outermost 'async def'. */
int async_def_nl; /* =1 if the outermost 'async def' had at least one
NEWLINE token after it. */
};
(2) 토크나이저 상태는 커서의 현재 위치 같은 정보를 저장 (Parser/tokenizer.h)
/* Tokenizer state */
struct tok_state {
/* Input state; buf <= cur <= inp <= end */
/* NB an entire line is held in the buffer */
char *buf; /* Input buffer, or NULL; malloc'ed if fp != NULL */
char *cur; /* Next character in buffer */
char *inp; /* End of data in buffer */
const char *end; /* End of input buffer if buf != NULL */
const char *start; /* Start of current token if not NULL */
int done; /* E_OK normally, E_EOF at EOF, otherwise error code */
/* NB If done != E_OK, cur must be == inp!!! */
FILE *fp; /* Rest of input; NULL if tokenizing a string */
int tabsize; /* Tab spacing */
int indent; /* Current indentation index */
int indstack[MAXINDENT]; /* Stack of indents */
int atbol; /* Nonzero if at begin of new line */
int pendin; /* Pending indents (if > 0) or dedents (if < 0) */
const char *prompt, *nextprompt; /* For interactive prompting */
int lineno; /* Current line number */
int first_lineno; /* First line of a single line or multi line string
expression (cf. issue 16806) */
int level; /* () [] {} Parentheses nesting level */
/* Used to allow free continuations inside them */
char parenstack[MAXLEVEL];
int parenlinenostack[MAXLEVEL];
PyObject *filename;
/* Stuff for checking on different tab sizes */
int altindstack[MAXINDENT]; /* Stack of alternate indents */
/* Stuff for PEP 0263 */
enum decoding_state decoding_state;
int decoding_erred; /* whether erred in decoding */
int read_coding_spec; /* whether 'coding:...' has been read */
char *encoding; /* Source encoding. */
int cont_line; /* whether we are in a continuation line. */
const char* line_start; /* pointer to start of current line */
const char* multi_line_start; /* pointer to start of first line of
a single line or multi line string
expression (cf. issue 16806) */
PyObject *decoding_readline; /* open(...).readline */
PyObject *decoding_buffer;
const char* enc; /* Encoding for the current str. */
char* str;
char* input; /* Tokenizer's newline translated copy of the string. */
int type_comments; /* Whether to look for type comments */
/* async/await related fields (still needed depending on feature_version) */
int async_hacks; /* =1 if async/await aren't always keywords */
int async_def; /* =1 if tokens are inside an 'async def' body. */
int async_def_indent; /* Indentation level of the outermost 'async def'. */
int async_def_nl; /* =1 if the outermost 'async def' had at least one
NEWLINE token after it. */
};
파서-토크나이저는 tok_get()으로 다음 토큰을 얻고 그 아이디를 파서로 전달
// Parser/tokenizer.c L1174
/* Get next token, after space stripping etc. */
static int
tok_get(struct tok_state *tok, const char **p_start, const char **p_end)
{
int c;
int blankline, nonascii;
*p_start = *p_end = NULL;
nextline:
tok->start = NULL;
blankline = 0;
/* Get indentation level */
if (tok->atbol) {
...
return PyToken_OneChar(c);
}
파서는 파서 생성기 오토마타(DFA)로 CST에 노드를 추가
640줄이 넘는 tok_get()은 CPython 코드 중에서도 손꼽히게 복잡한 부분중 하나. 루프에서 토크나이저와 파서를 호출하는 과정은 아래와 같음
CST→AST로 변환하려면 PyParser_ParseFileObject()에서 반환된 CST의 루트인 node가 필요
typedef struct _node {
short n_type;
char *n_str;
int n_lineno;
int n_col_offset;
int n_nchildren;
struct _node *n_child;
int n_end_lineno;
int n_end_col_offset;
} node;
CST는 구문, 토큰 아이디, 심벌을 모두 포함하기 때문에 컴파일러가 사용하기에는 적합하지 않음
AST를 살펴보기에 앞서 파서 단계의 결과를 확인하는 방법이 있는데, parser 모듈은 C함수의 파이썬 API를 제공
>>> import parser
st<stdin>:1: DeprecationWarning: The parser module is deprecated and will be removed in future versions of Python
>>> st = parser.expr('a+1')
>>>
>>>
>>> pprint(parser.st2list(st))
[258,
[332,
[306,
[310,
[311,
[312,
[313,
[316,
[317,
[318,
[319,
[320,
[321, [322, [323, [324, [325, [1, 'a']]]]]],
[14, '+'],
[321, [322, [323, [324, [325, [2, '1']]]]]]]]]]]]]]]]],
[4, ''],
[0, '']]
Parser 모듈의 출력은 숫자형식으로 make regen-grammar 단계에서 Include/token.h파일에 저장된 코튼과 심벌의 번호와 같음
좀더 보기 쉽게 symbol과 token 모듈의 모든 번호로 딕셔너리를 만든 후 parser.st2list()의 출력을 토큰과 심벌의 이름으로 재귀적으로 바꾸면 아래와 같음
import symbol
import token
import parser
from pprint import pprint
def lex(expression):
symbols = {v: k for k, v in symbol.__dict__.items() if isinstance(v, int)}
tokens = {v: k for k, v in token.__dict__.items() if isinstance(v, int)}
lexicon = {**symbols, **tokens}
st = parser.expr(expression)
st_list = parser.st2list(st)
def replace(l: list):
r = []
for i in l:
if isinstance(i, list):
r.append(replace(i))
else:
if i in lexicon:
r.append(lexicon[i])
else:
r.append(i)
return r
return replace(st_list)
pprint(lex("a + 1"))
CST는 코드 파일의 텍스트를 있는 그대로 표현하는 구조로, 텍스트로부터 토큰을 추출하여 토큰 종류만 구분해 둔 상태에 불과
CST로 기본적인 문법 구조는 알 수 있지만 함수, 스코프, 루프 같은 파이썬 언어 사양에 대한 의미를 결정할 수 없음
코드를 컴파일 하기 전 CST를 실제 파이썬 언어구조와 의미 요소를 표현하는 고수준 구조인 AST로 변환해야 함
예를 들어 AST에서 이항 연산은 표현식의 한 종류인 BinOp로 표현. 해당 표현식은 세가지 요소로 이루어짐
left: 왼쪽 항
op: 연산자(+, -, * 등)
right: 오른쪽 항
다음은 a + 1 에 대한 AST
AST는 CPython 파싱 과정 중에 생성하지만 표준 라이브러리 ast모듈을 사용해서 파이썬 코드에서 AST를 생성할 수도 있음
6.3.1 AST 연관된 소스 파일 목록
Include/python-ast.h: Parser/asdl_c.py 로 생성한 AST 노드 타입 선언
parser/Python.asdl: 도메인 특화 언어인 ASDL(abstract syntax description language) 5로 작성된 ast 노드 타입들과 프로퍼티 목록
Python/ast.c: AST 구현
6.3.2 인스타비즈로 AST 시각화하기
$ pip install instaviz
AST와 컴파일된 코드를 웹 인터페이스로 시각화 하는 파이썬 패키지
❯ python
Python 3.9.20 (main, Sep 6 2024, 19:03:56)
[Clang 15.0.0 (clang-1500.3.9.4)] on darwin
Type "help", "copyright", "credits" or "license" for more information.
>>> import instaviz
>>> def example():
... a = 1
... b = a + 1
... return b
...
>>>
>>> instaviz.show(example)
트리의 각 노드의 타입은 AST 노드 클래스
ast 모듈에서 찾을 수 있는 노드 클래스들은 모두 _ast.AST를 상속
CST와 달리, AST의 노드들은 특정 프로퍼티들을 통해 자식 노드와 연결됨
b = a + 1 이 선언된 줄과 연결된 assign 노드를 클릭하면 아래와 같음
Assign 노드는 2개의 프로퍼티를 가짐
targets는 값이 할당될 이름의 목록. 언패킹을 통해 한 번에 여러 이름에 값을 할당할 수 있기 때문에 목록이 필요
value는 이름에 할당할 값. 이 경우에 BinOp 표현식 a + 1 이 할당됨
BinOp노드는 세 개의 프로퍼티를 가짐
left: 왼쪽 항
op: 연산자, 이 경우에는 더 하기를 뜻하는 Add 노드(+)
right: 오른쪽 항
6.3.3 AST 컴파일
C에서 AST를 컴파일하는 것을 매우 복잡한 작업으로 Python/ast.c 모듈은 5000줄이 넘는 코드로 이루어져 있음
AST의 공개 API는 CST와 파일 이름, 컴파일러 플래그 ,메모리 저장 영역을 인자로 받음
반환 타입은 파이썬 모듈을 표현하는 mod_ty 타입으로 Include/Python-ast.h 에서 정의
mod_ty는 아래 4가지 모듈 타입 중 하나를 담는 컨테이너 구조체
Module
Interactive
Expression
FunctionType
모듈 타입은 Parser/Python.asdl에서 정의하는데 문장, 표현식, 연산자, 컴프리헨션 타입들도 찾을 수 있음
AST가 생성하는 클래스들과 표준 라이브러리 ast 모듈의 클래스들은 Parser/Python.asdl에서 정의하는 타입들
AST는 ALMOSTEQUAL 토큰이 비교 연산자인 AlE라는 것을 아직 알 수 없어서, 토큰을 연산자로 인식할 수 있게 AST관련 C 코드를 수정
Python/ast.c의 ast_for_comp_op()로 이동해서 switch문을 찾아보면, _cmpop 열거형 값 중 하나를 반환 함
static cmpop_ty
ast_for_comp_op(struct compiling *c, const node *n)
{
/* comp_op: '<'|'>'|'=='|'>='|'<='|'!='|'in'|'not' 'in'|'is'
|'is' 'not'
*/
REQ(n, comp_op);
if (NCH(n) == 1)
{
n = CHILD(n, 0);
switch (TYPE(n))
{
case LESS:
return Lt;
case GREATER:
return Gt;
case ALMOSTEQUAL: // 추가된 내용
return AlE; // 추가된 내용
case EQEQUAL: /* == */
return Eq;
case LESSEQUAL:
return LtE;
case GREATEREQUAL:
return GtE;
case NOTEQUAL:
return NotEq;
case NAME:
if (strcmp(STR(n), "in") == 0)
return In;
if (strcmp(STR(n), "is") == 0)
return Is;
/* fall through */
이제 토크나이저와 AST모두 코드를 파싱할 수 있지만 컴파일러는 아직 이 연산자를 실행하는 방법을 모름
AST로 거의 같음 연산자를 아래와 같이 확인해볼 수 있음
import ast
m = ast.parse('1 ~= 2')
m.body[0].value.ops[0]
<_ast.AlE object at 0x111111>
