이미지 생성 Slack Bot 만들기

조홍철·2024년 3월 17일

개발 과정

이미지 생성 기술을 사용중, 미드저니와 니지저니와 같이 이미지 생성을 비설치로 도와주는 서비스들을 참고하여 자체적으로 slack bot을 이용해 만들어보면 재미있을 것 같아 약 5일 동안 진행한 프로젝트 입니다.

첫 목표는 slack bot을 이용해 prompt를 그대로 전달하면 이미지를 생성하는 것을 목표로하고, 이 후 한글을 영어로 입력하고 특정 checkpoint모델을 선정하여 default prompt를 관리하여 유저 입력 prompt가 더 checkpoint의 특성을 더 잘가지고 갈 수 있도록 후처리하는 작업을 진행하였습니다.

대략적으로 아래 4개의 story로 나눠보고, 그 안의 여러가지 task를 나눠 관리해보았습니다.

1. slack bot을 만들고, local에서 fastAPI를 이용한 서버를 동작시켜 특정 기능을 수행하는 bot 만들기 
2. 이미지 생성 서버를 local에 별도로 두고(성능, 비용 문제로 인해) local API서버와 이미지 생성 서버의 통신을 통해 이미지 저장 및 관리 
3. api 서버를 aws EC2를 이용해 배포해보기
4. 추가 기능인 한글 -> 영어 번역, 특정 prompt 소거, 에러 핸들링

첫 개발시에 글또의 또봇 코드를 보면서 많이 참고할 수 있었습니다. 감사합니다 :)

Slack Bot 사용 (어떻게 Slack Bot은 내가 짠 로직을 수행시킬까?)

slack bot을 만들고, 간단하게 동작하도록하는 자료는 다른 분들이 상세하게 설명해주신게 많아 넘어가고자 힙니다.

다만, 저는 처음에 slack bot에서 slash command 혹은 특정 명령을 수행하는 동작 과정에 흥미가 갔었습니다. 분명 bot 자체는 slack server에서 관리할텐데 어떻게 내부 로직은 사용자가 만든 서버에서 작동시키도록 할 수 있는지가 궁금했습니다.

물론 그 중에서도 slack bot 통신 방법 중 websocket모드가 있는데 특정 url을 사용하지 않고도 slack ui에서 사용자가 발생시킨 action을 slack에서 서버에서 api서버를 찾아 보내준다는 것이 놀라웠습니다.

가설 중 하나는, 서버가 켜짐과 동시에 여러 init과정 중에 api서버가 가지고 있는 통신으로 slack server와 통신이 이뤄지고 slack에서는 해당 public ip를 관리하여 통신하는 방식이 아닐까 했습니다.

이에 따라, 리서치를 진행하고 결과적으로 GPT가 정리한 내용은 다음과 같습니다

Slack AsyncApp 통신
AsyncApp 인스턴스 생성: AsyncApp 인스턴스는 Slack 이벤트와 상호작용을 위한 주요 객체입니다. 
이 인스턴스는 Slack Bot 토큰을 사용하여 Slack API와의 인증 및 상호작용을 처리합니다.

AsyncSocketModeHandler 인스턴스 생성: AsyncSocketModeHandler는 AsyncApp 인스턴스와 APP_TOKEN을 인자로 받습니다. 
이 핸들러는 Socket Mode를 사용하여 Slack과의 연결을 관리하고, Slack으로부터 실시간 이벤트를 받습니다. 
여기서 APP_TOKEN은 Socket Mode에서 사용되는 특별한 슬랙 앱 토큰(Slack App-Level Token)으로, 이 토큰은 WebSocket 연결을 위한 인증에 사용됩니다.

연결 및 이벤트 리스닝 시작: AsyncSocketModeHandler 인스턴스에 의해 Slack과의 WebSocket 연결이 초기화되고, 이 연결을 통해 실시간으로 이벤트를 수신하기 시작합니다. 
이때, 서버는 Slack으로부터 오는 데이터를 실시간으로 처리할 수 있게 됩니다.

이벤트 처리: Slack으로부터 오는 이벤트는 AsyncApp에 의해 처리됩니다. 
개발자는 다양한 이벤트 타입에 대해 콜백 함수를 정의할 수 있으며, 이러한 콜백 함수는 특정 이벤트가 발생했을 때 실행됩니다.

이를 바탕으로 실제 코드를 도식화 하면 다음 과정을 통해 slack event에 대한 listening이 시작된다고 이해했습니다.

그렇다면 여러 서버에서 init하게 되면 어떻게 동작할까 궁금하여 시도해봤을 때, websocket방식은 여러 서버에서 init이 발생했을 때 제일 마지막에 init된 서버를 기준으로 통신하게 됩니다

slack handler

SlackService라는 class를 만들어, 처음 fastapi가 실행될 때 미들웨어에 주입하는 방식으로 의존성 주입을 사용했습니다

@app.middleware
async def inject_service_middleware(req: BoltRequest, next: Callable) -> None:
    req.context["service"] = SlackService()
    await next()
    return

미들웨어 의존성 주입 이유 또한 GPT가 정리한 내용으로 보면 다음과 같습니다

middleware 의존성 주입 이유
서비스의 재사용성: lackService와 같은 서비스 인스턴스를 요청 컨텍스트에 주입함으로써, 
애플리케이션의 다른 부분에서 해당 인스턴스를 쉽게 재사용할 수 있습니다. 
이는 코드 중복을 줄이고, 서비스 로직의 일관성을 유지하는 데 도움이 됩니다.

의존성 관리: 의존성 주입은 의존성 관리의 한 형태로, 각 요청이 처리될 때 필요한 의존성(이 경우 SlackService)을 자동으로 주입함으로써, 
컴포넌트 간의 결합도를 낮추고, 유지 보수성 및 테스트 용이성을 향상시킵니다.

중앙 집중화된 설정: 모든 요청에 대해 공통된 설정이나 초기화 로직(예: 로깅, 서비스 인스턴스 생성, 요청 검증 등)을 실행하고 싶을 때 미들웨어를 사용하면, 이러한 로직을 중앙 집중적으로 관리할 수 있습니다. 
이는 애플리케이션의 구조를 깔끔하게 유지하는 데 도움이 됩니다.

성능 최적화: 특정 서비스 인스턴스를 요청 컨텍스트에 한 번만 생성하고, 
이후의 요청 처리 과정에서 재사용함으로써, 불필요한 객체 생성 오버헤드를 줄이고 애플리케이션의 성능을 최적화할 수 있습니다.

이미지 생성 (Comfy UI와 어떻게 통신해서 이미지를 가져올까?)

이미지 생성에는 Comfy UI를 사용했습니다. comfy ui를 사용해보면서 comfy ui자체에서도 이미지를 생성해주는 과정과 생성된 이미지를 가져와서 보여주는 방식이 있을 것이라고 생각하여 네트워크 통신을 보게 되었습니다.

queue를 작동시켰을 때, prompt라는 객체가 생기는 것을 볼 수 있는데 이 객체가 전체 comfy ui의 workflow를 담고있다는 것을 알 수 있습니다. 따라서 사용자가 해당 object안에 내용만 잘 채워넣는다면 ui없이도 이미지를 생성하는 API를 만들 수 있다고 판단했습니다. 이에 따라 리서치를 시작해보고 다음 코드를 참고하여 web socket통신을 통해 구현할 수 있음을 알게되었습니다
참고 코드 : https://github.com/itsKaynine/comfy-ui-client

이에 따라 이미지 생성쪽은 websocket 방식을 채택하여 구현하였습니다. 로컬 이미지 생성 서버와 연결 뒤에 listening하여 특정 이베트에 따라 핸들링하였습니다.

특히 이미지를 얻는 부분에 있어서는 들어오는 객체의

message = await websocket.recv()  
message = message.replace('null', 'None')
message_dict = eval(message)

if message_dict['type'] == 'executed' ~

인 부분을 이용하여 처리하였고, 이미지를 불러오는 것 또한 comfy ui에서 네트워크 통신을 참고하여 다음과 같이 가져오도록 했습니다

filename = message_dict['data']['output']['images'][0]['filename']
subfolder = message_dict['data']['output']['images'][0]['subfolder']
output_type = message_dict['data']['output']['images'][0]['type']

image_url = await ImageGenerateRouter.get_image_url_from_generator_server(
	filename,
    subfolder,
    output_type
 )
 
 async with httpx.AsyncClient() as client:
 	response = await client.get(image_url)

결과적으로 slack event handler와 websocket handler

@app.on_event("startup")
	async def startup():
    # lifespan - 인스턴스 생성시 실행 함수
    # 슬랙 소켓 모드 실행 - handler 설정
    loop = asyncio.get_event_loop()
    if not loop.is_running():
    	asyncio.run(message_handler())
    else:
    	loop.create_task(message_handler())
    await slack_handler.connect_async()

1차 결과물

<이미지 생성 중 알림>

<이미지 업로드>

추가 기능 개발

사용성을 위해서 한글로 프롬프트를 입력할 수 있도록 번역 기능을 추가하였고, 이는 google translate api를 사용하였습니다. 또한 nsfw라는 negative prompt를 넣어도 부적절한 이미지가 생성되어 특정 언어를 필터링하는 기능도 만들게 되었습니다.

입력된 prompt 혹은 문장을 split하여 하나씩 검사하는 방식보다는 set을 통해서 forbidden keyword를 추출하여 이를 제거하는 방식을 채택하였습니다

split_text = re.split(',\s*|\s+', translated_text)
identified_forbidden_words = set(forbidden_words).intersection(split_text)

filtered_text = translated_text or word in identified_forbidden_words:
filtered_text = filtered_text.replace(word, "")

또한 checkpoint에 따라 유저가 신경쓰지 않아도 이미지 퀄리티를 올릴 수 있도록 default prompt를 구성하고 env파일을 통해 관리하였습니다