서론
3월부터 다시 시작한 프로젝트는 생성형 AI를 활용한 프로젝트이기에 python과 AI기술과 관련된 공부를 시작하게 되었다. 역시 시간이 촉박하기에, pre-trained 되어있는 모델을 사용하기로 하였으며 결과적으로 ddsp, diff 두 개의 모델을 사용하기로 하였다.
ddsp - https://github.com/yxlllc/DDSP-SVC/blob/master/ko_README.md
diff - https://github.com/prophesier/diff-svc
프로젝트의 구조는 프론트엔드 서버 - 백엔드 서버 - AI 서버의 구조로 이뤄지게 되었으며, 그 중에서 AI 서버 부분을 맡게 되었다.
본론
1. 파이썬 프레임 워크 선택
서버를 구축하기 위해 맨 처음으로 고민한것은, 대표적인 프레임워크인 Django, Flask 둘 중 무엇을 사용할 지 고민했다.
1.1. Flask 특징
-
경량 프레임워크
Flask는 마이크로 프레임워크로 분류되며, 간단하고 유연한 구조를 가지고 있습니다. 필요한 기능만 선택하여 추가할 수 있어, 프로젝트의 초기 구조를 단순하게 유지하고 싶을 때 좋습니다. -
빠른 개발
Flask는 러닝 커브가 낮아 빠르게 개발을 시작할 수 있습니다. 작은 규모의 프로젝트나 프로토타이핑에 적합합니다. -
확장성
필요에 따라 다양한 확장 프로그램을 추가하여 기능을 확장할 수 있습니다. 대규모 애플리케이션에도 적용 가능하지만, 그 과정에서 필요한 구조와 기능을 직접 설계해야 할 수도 있습니다.
1.2. Django 특징
-
Full Stack 프레임워크
Django는 "배터리 포함(batteries included)" 접근 방식을 취하고 있어, 인증, 세션, 템플릿과 같은 기능이 기본적으로 포함되어 있습니다. 복잡한 웹 애플리케이션을 구축할 때 시간을 절약할 수 있습니다. -
보안
Django는 보안 기능이 잘 구축되어 있어, CSRF, SQL 인젝션 등의 웹 공격으로부터 애플리케이션을 보호하는 데 도움이 됩니다. -
확장 가능
대규모 애플리케이션에 적합하며, 많은 트래픽과 데이터를 처리할 수 있는 강력한 ORM(Object-Relational Mapping)을 제공합니다.
1.3. 결론
Flask의 경량 프레임워크, 빠른 개발의 장점과 현재 개발하고 있는 웹 어플리케이션은 비로그인 서비스이기에 보안기능은 우선순위가 낮다고 생각하여, Flask를 선택하게 되었다.
2. 비동기 적용을 위한 Celery
생성형 AI는 결과를 출력하는데 오랜 시간이 걸립니다. Flask만 사용한다면 해당 결과 출력까지 Session을 연결해야되기 때문에 Http의 stateless 설계 원칙에 위배됩니다. 그렇기에 Celery를 적용시켜 백엔드 서버에서는 요청만 보내고, task id를 활용해 해당 작업을 관리하는 방식으로 stateless 방식으로 구현하고자 했습니다.
Celery는 Python으로 작성된 비동기 작업 큐/태스크 큐 시스템으로, 실시간 처리와 배치 처리를 관리하는 데 사용됩니다. 이 프레임워크는 분산 시스템에서 작업을 실행할 수 있도록 도와주며, RabbitMQ, Redis 등의 메시지 브로커와 함께 사용되어 작업 메시지를 교환합니다.
2.1. 설치
Celery를 pip를 사용하여 설치해줍니다
pip install celery2.2. Celery 인스턴스 생성
celery 애플리케이션 인스턴스를 생성하려면, 다음과 같이 Python 모듈에 Celery를 설정합니다. 예를 들어, redis를 메시지 브로커로 사용하는 경우의 설정입니다. 저는 celery_config.py를 따로 작성하여 인스턴스 생성 부분을 따로 작성했습니다.
from celery import Celery
def make_celery(app_name=__name__):
app = Celery(app_name,
broker='redis://:password@redis server url',
backend='redis://:password@redis server url')
return app
celery = make_celery()
celery.autodiscover_tasks(['tasks'], force=True)
밑에 있는 autodiscover_tasks는 __init__.py 와 같은 디렉토리 안에있는 작업 목록을 자동으로 찾아줍니다.
2.3. 작업 정의
이제 Celery를 사용하여 비동기 작업을 정의할 수 있습니다. 아래 예제는 간단한 덧셈 작업을 정의한 것입니다.
from celery_config import celery
@celery.task(bind=True, name='tasks.diff')
def add(x, y):
return x + y
2.4. Celery 워커 실행
작업을 처리하기 위해 Celery 워커를 실행해야 합니다. 터미널에서 아래 명령어를 사용하여 워커를 시작할 수 있습니다. 이 명령어는 Celery 애플리케이션이 정의된 모듈이 위치한 디렉토리에서 실행되어야 합니다:
celery -A celery_config worker --loglevel=info2.5. 작업 실행
작업이 큐에 추가되고, 설정한 Redis 브로커를 통해 이를 Celery 워커가 가져가 처리합니다.
from tasks import add
@app.route("/api3/svc", methods=["POST"])
def request_svc():
message = "Request received"
task = add.delay(4,4)
return jsonify({"message": message, "message_queue_id": task.id}), 200
작업의 결과는 발급받은 task_id와 result 변수를 통해 접근할 수 있습니다.
@app.route('/api3/status/<task_id>')
def task_status(task_id):
task = celery.AsyncResult(task_id)
if task.state == 'PENDING':
return jsonify({"message": "작업 진행중"}), 200
elif task.state != 'FAILURE':
# task.info가 사전인지 확인
if isinstance(task.info, dict):
status = task.info.get('status', '')
result = task.info.get('result', None)
else:
# task.info가 사전이 아니면, 직접 문자열로 처리
status = task.info or '' # task.info가 None이 아니면 사용, 그렇지 않으면 빈 문자열
result = None
response = {
'state': task.state,
'status': status,
'result': result # 'result'를 여기에 추가
}
else:
# 작업 실패
return jsonify({"message": "작업 실패"}), 404
return jsonify(response)
Celery와 Redis를 함께 사용함으로써, Python 애플리케이션에서 비동기 작업 처리를 효율적으로 구현할 수 있습니다.
결론
최초에는 celery 없이 구현을 했었습니다. 하지만, AI 작업이 시간이 오래걸려 Session 연결을 지속하는 것은 적절치 않다고 판단하고, 비동기 작업을 처리하기 위해 검색하던 와중 celery를 찾게 되었습니다!
SSAFY에 와서 정말 다양한 기술스택을 경험하고, 필요한 것을 찾아쓸 수 있는 능력이 생기는 것 같아 뿌듯헙니다~
새로 진행하는 프로젝트도 성공적으로 끝마치길 기원하며 이번 포스팅 마치도록 하겠습니다.
