[Chat gpt - Slack연동4] - Slack API 사용하기

Slack API?

• 지피티와 대화를 주고 받을 슬랙 채널에서 메시지 내용을 서버로 전달하는 역할
• 지피티에게 받은 응답을 슬랙 채널 적재적소에 출력하는 역할

슬랙에는 원하는 API 기능을 가지는 챗봇의 형태로 슬랙 봇이라는 이름으로 존재한다.

슬랙 봇을 만드는 방법은 https://junhee742.tistory.com/24 여기 나와있다.

[Chat gpt - Slack연동2] - chat gpt API 구매, Slack legacy bot 생성

 

위와같은 방법으로 슬랙 봇을 만들게 되면, RTM(RealTimeMessage) 기능을 가지는 슬랙 봇을 사용할 수 있게 되는 것이다.

그리고, 만든 슬랙 봇을 챗 지피티를 사용할 채널에 초대하면 RTM API 를 호출하는 코드가 정상적으로 구동 됐을 때 실시간으로 등록된 메시지의 정보를 받고 응답을 출력할 수 있다.

---

기본적인 동작 방식은 아래 링크에 나와 있다.

https://junhee742.tistory.com/25

[Chat gpt - Slack연동1] - 프로젝트 시작 전

---

아래는 RTM API 를 통해 받아올 수 있는 추적된 메시지의 정보이다.

RTM API 는 실시간 메시지를 받아오는 API 로 실시간으로 등록 되는 메시지 한 건을 받아온다. 이전에 등록되어 있던 메시지의 정보는 불러올 수 없다.

그래서, 처음 채널에 메시지가 등록 됐을 때 나중에도 꺼내쓰기 위해 내장 DB 를 활용해 저장해 놓는 작업이 필요하다. 이후 포스팅에서 자세히 알아볼테지만, 대화형 AI 인 Chat GPT 를 효율적으로 사용하기 위해선 대화 맥락의 이해는 필수이다. 해당 기능 구현을 위해 실시간 추적된 메시지의 필요한 정보를 그때그때 DB에 저장하고 Chat GPT API 요청 시 이전 대화 내용을 불러온다.

추적된 정보 안에는 다양한 데이터가 존재한다. 필요한 데이터만 간단히 알아보자.

• Channel_id
  • 메시지가 등록된 슬랙 채널의 아이디 값
• bot_id
  • 글쓴이가 슬랙 봇인지, 봇이라면 봇의 아이디 값
• subtype
  • 추적된 메시지의 타입. 채널 홈에 달리는 글인지, 스레드에 달리는 글인지, 수정된 글인지, 삭제된 글인지 등등 판별 가능
• text
  • 등록된 내용
• user_id
  • 등록한 사람의 아이디(암호화 된 값으로 저장됨)
• message_ts
  • 메시지가 등록된 시간.
  • channel_id 와 조합해 메시지 고유 키 값으로 사용된다.
  • (channel_id)+(message_ts)
• thread_ts
  • 메시지가 등록된 스레드가 만들어진 시간
  • 같은 스레드에 등록되는 메시지들은 모두 같은 thread_ts 를 가진다.

이외에도 여러가지 값들이 있는데, 직접 진행해 보면서 더 필요한 값이 있으면 빼다 쓰면 된다.

지금은 아래 코드의 tracking_msg_data_list 만 어떤식으로 받아오는지만 보면 될 듯 하다.

from slack import RTMClient
from chatgpt import ChatGPT
import gpt_sql
import traceback
import api_token
import re
import log_config
import gpt_scheduler

# 로거 생성
logger = log_config.get_gpt_logger(name="gpt_log")


# 지속적으로 슬랙 메세지 트래킹
@RTMClient.run_on(event="message")
def chatgpt_bot(**payload):
    try:
        # 추적한 메세지 정보 세팅
        web_client = payload["web_client"]
        data_list = tracking_msg_data_list(payload["data"])
        if data_list is None or data_list["channel_valid"] == "unnecessary message":
            return
        elif data_list["channel_valid"] == "different channel":
            web_client.chat_postMessage(channel=data_list["channel_id"],
                                        text="#all-askanythings에서 질문해 주세요:laughing:",
                                        thread_ts=data_list["ts"])
            logger.info(
                f"chatgptbot| different channel | channel_id : {data_list['channel_id']} | user : {data_list['user']}")
            return
        elif data_list["channel_valid"] == "message_changed":
            web_client.chat_postMessage(channel=data_list["channel_id"],
                                        text="편집한 질문은 인식할 수 없어요. 새롭게 질문해 주세요:laughing:",
                                        thread_ts=data_list["ts"])
            return
        logger.info(data_list)

        # DB 접속
        conn = gpt_sql.connect_db(YOUR_DB_FILE_PATH)
        cur = conn.cursor()

        # 새롭게 추가된 메세지 insert
        gpt_sql.insert_data(cur, data_list)
        conn.commit()

        # 스레드 대화 내용 select
        slack_msgs = gpt_sql.select_slack_msg(cur, data_list)

        # DB 연결 종료
        conn.close()

        # list dictionary 형식으로 정제
        req_msg = request_msg_list(slack_msgs)

        # GPT 슬랙봇 멘션이 아닐 경우 ChatGPT 동작 x
        if data_list["tag_code"] == YOUR_SLACK_BOT_TAG_CODE:
            logger.info(
                f"chatgptbot| {data_list['thread_ts']} | Message receive (USER {data_list['user']}) : {data_list['text']}")
            # 받아온 리스트를 ChatGPT에 전달하고 ChatGPT의 답변 저장
            response = ChatGPT(req_msg)
            # 슬랙에 메세지 전달
            web_client.chat_postMessage(channel=data_list["channel_id"], text=response, thread_ts=data_list["ts"])
            # 요청-응답 메시지 로깅
            logger.info(f"chatgptbot| {data_list['thread_ts']} | Message Send (GPT) : {response}")

    except BaseException as e:
        logger.error(f"chatgptbot| An Exception has occur during tracking message!")
        logger.error(traceback.format_exc())
        payload["web_client"].chat_postMessage(channel=YOUR_SLACK_GPT_CHANNEL,
                                               text="오류가 발생했습니다. 관리자에게 문의해 주세요:exclamation:",
                                               ts=payload["data"]["ts"])


def tracking_msg_data_list(data):
    data_list = {}
    data_list["subtype"] = data.get("subtype", "")
    data_list["channel_id"] = data["channel"]
    data_list["ts"] = data["ts"]
    data_list["user"] = data.get("user", "")
    data_list["channel_valid"] = ""
    tag_code = re.search("<[^>]*>", data.get("text", ""))
    if tag_code is not None:
        data_list["tag_code"] = tag_code.group()
    else:
        data_list["tag_code"] = ""
    # 스레드 제목 읽어올 경우 return, 채널 내에서 gpt 에게 질문 하지 않은 경우 return, 다른 채널에서 질문할 경우 return
    if data_list["subtype"] == "message_replied":
        return
    elif data_list["subtype"] == "" and data_list["tag_code"] == "":
        return
    elif data_list["subtype"] == "bot_message" and ":laughing:" in data["text"]:
        return
    validator = channel_validator(data_list["channel_id"], data_list["tag_code"])
    if validator == "different channel":
        data_list["channel_valid"] = "different channel"
        return data_list
    elif validator == "unnecessary message":
        data_list["channel_valid"] = "unnecessary message"
        return data_list
    elif data_list["subtype"] == "message_changed":
        tag_code = re.search("<[^>]*>", data["message"]["text"])
        if tag_code is not None and tag_code.group() == api_token.spectra_gpt_tag_code:
            data_list["channel_valid"] = "message_changed"
        else:
            data_list["channel_valid"] = "unnecessary message"
        return data_list
    data_list["bot_id"] = data.get("bot_id", "")
    data_list["thread_ts"] = data.get("thread_ts", data_list["ts"])
    data_list["block_id"] = data["blocks"][0]["block_id"]
    data_list["role"] = data["blocks"][0]["elements"][0]["elements"][0]["type"]
    data_list["text"] = data["text"].replace(data_list["tag_code"], "")
    return data_list


def channel_validator(channel_id, tag_code):
    if channel_id != YOUR_SLACK_GPT_CHANNEL:
        if tag_code == YOUR_SLACK_BOT_TAG_CODE:
            return "different channel"
        elif tag_code != YOUR_SLACK_BOT_TAG_CODE:
            return "unnecessary message"


def request_msg_list(slack_msgs):
    messages = []
    for slackMsg in slack_msgs:
        message = {}
        if slackMsg[1] == "text":
            message["role"] = "assistant"
        elif slackMsg[1] == "user":
            message["role"] = "user"
        message["content"] = slackMsg[2]
        messages.append(message)
    return messages


if __name__ == "__main__":
    try:
    	# DB 파일 처음 만들때만 아래 주석 해제하고 사용
        # gpt_sql.create_table()
        # 지난 데이터 삭제 스케줄러 시작
        gpt_scheduler.start_scheduler()
        logger.info(f"main| delete_old_data schedule job start!")

        # RTM 클라이언트 호출 및 시작
        rtm_client = RTMClient(token=YOUR_SLACK_BOT_TOKEN)
        logger.info(f"main| GPT start!")
        rtm_client.start()
    except Exception as err:
        logger.error(f"main| An Exception has occur during Bot connect!")
        logger.error(traceback.format_exc())