Fastapi, Celery를 이용한 LLM Worker 만들기(feat. Redis)

서론

로컬 LLM 모델을 이용한 챗봇 서비스를 만들고 있다.

웹서비스 개발은 처음으로 하고 있어서, copilot과 ChatGPT, 구글링 등 여러군데에서 적당한 개발 방식을 찾아서 적용해 보고있다.

여러시도 중에 백앤드로 Fastapi를 사용해서 별도의 Worker서버에 Celery를 통한 요청으로 LLM response를 받는 방식으로

서비스를 구현하려고 하고있어 그 예제를 정리하여 올린다.

구성도

백앤드에서 redis를 브로커로 이용하여 worker서버에 작업 요청과 결과를받아오는 구성이다.

필요 패키지

• fastapi , uvicorn, celery, redis

pip install fastapi celery redis "uvicorn[standard]"

Redis 사용

Docker를 이용하여 Redis를 설치하고 사용하여도 되지만, 이번글에서는 무료로 제공되는 Redis cloud를 이용하겠다.

Redis : https://redis.com

(위에 사이트에서 회원 가입후 사용 자세한 예제는 다른 포스팅 글을 참조..)

 

Fastapi 서비스

- Celery의 백엔드와 브로커 URL을 동일하게 하여 생성(무료 Redis Cloud는 1개만 이용 가능하여 동일하게 생성)

#Fastapi 모듈
from fastapi import FastAPI

#Celery 요청 및 결과 수신을 위한 모듈
from celery import Celery
from celery.result import AsyncResult

REDIS_URL= #Redis Cloud에서 확인하여 입력
REDIS_PASSWORD= #Redis Cloud에서 확인하여 입력

def make_celery(app_name=__name__):

    connection_url = f"redis://:{REDIS_PASSWORD}@{REDIS_URL}"

    backend = connection_url+"/0"
    broker = connection_url+"/0"
    
    #backend = broker = f"redis://{REDIS_URL}"
    celery = Celery(app_name, backend=backend, broker=broker)
    
    return celery


celery_app = make_celery('Knowslog')

- send_msg : 입력 받은 메시지를 Broker로 전달

- receive_msg : Broker로 부터 요청한 Task의 결과를 수신

app = FastAPI()
@app.post("/send")
async def send_msg(msg:str):
    task = celery_app.send_task('generate_text_task', args=[msg])
    
    return {"taskid": task.id}

@app.get("/receive/{task_id}")
async def receive_msg(task_id:str):
    task = celery_app.AsyncResult(task_id)
    if task.ready():
        return {"result": task.get()}
    else:
        return {"result": "Not ready yet!"}

Worker 구성

- LLAMA 2 챗 모델 생성

from langchain import PromptTemplate, LLMChain
from langchain.llms import LlamaCpp

class ChatModel:
    
    template = """{question}"""
    prompt = PromptTemplate(template=template, input_variables=["question"])

    def __init__(self):
        
        self.llm = LlamaCpp(
        model_path="/Volumes/external/Models/LLAMACPP/llama-2-7b-chat.ggmlv3.q4_1.bin",
        callback_manager=None,
        verbose=False,
        )
        self.llm_chain = LLMChain(prompt=self.prompt, llm=self.llm)
    
    def chat(self,input:str):
        return self.llm_chain.run(input)

- Worker 구현

  - generate_text_task를 shared_task로 지정하였다. shared로 하지 않으면 백엔드에서 워커로 요청시 서버명을 지정해서 task를 요청하여야 한다.

from celery import Celery,shared_task

REDIS_URL= #Redis Cloud 확인
REDIS_PASSWORD= #Redis Cloud 확인

def make_celery(app_name=__name__):
    connection_url = f"redis://:{REDIS_PASSWORD}@{REDIS_URL}"
    backend = connection_url+"/0"
    broker = connection_url+"/0"
    
    celery = Celery(__name__, backend=backend, broker=broker)

    return celery

celery = make_celery('Knowslog')
chatbot = ChatModel()
@shared_task(name='generate_text_task')
def generate_text_task(prompt):
    print("Generating text...")
    response = chatbot.chat(prompt)
    print(response)
    return response

def main():
    celery.worker_main(argv=['worker', '--loglevel=info'])

if __name__ == '__main__':
    main()

worker구동 하면 콘솔메시지에 아래와 같이 task가 생성된 것을 확인할 수 있다.

task가 나오지 않는다면 뭔가 잘못된거...생성된 task로 백엔드 서버에서 요청하여야한다 

 

서비스 확인

 

Fastapi Doc(http://localhost:8000/docs) 서비스 확인

Send 메시지

  - 메시지를 작성해서 요청하고 결과로 회신된 task id를 확인 해당 id로 결과를 가져온다.

Worker 메시지 작성

  - Task가 수신되면 해당 Worker에서 해당 Task 수행  

  - Worker Msg 작성 완료

Receive 메시지

  - 메시지가 완성되면 task id를 인자 값으로 하여 작성된 메시지를 수신한다.

  (셜록 홈즈라고 자기를 소개한다..)

 

결론

간단한 예제를 통한 서비스 요청 및 수신까지 작성해 보았다. 해당 예제를 토대로 서비스를 구현해보고 있다.

해당 방식으로 구현시 Worker의 성능을 업그레이하거나 병렬로 자원을 늘려서 서버관리가 가능할 것으로 보인다.

초짜 개발자로 실제 현업에서 구현되고 있는 서비스의 방식을 알지는 못하지만 해당 방식으로 서비스를 구현하고 업그레이드 하려고 한다.