[BricksAssistant] 디스코드 AI 챗봇 구축기 - (3) chatGPT 연동하기

📌 Intro

안녕하세요!! Brickstudy 김민준입니다.

 

BricksAssitant 지난 블로그 글을 통해 디스코드 챗봇 설정은 완료했고, 이번엔 해당 챗봇을 통해 chatGPT와 대화할 수 있는 기능을 구현합니다.

 

GPT 연동의 핵심은 이전 대화내용을 기반으로 chatGPT에 요청을 보내 이전 히스토리를 기반으로 한 질문으로 더 정확한 답변을 제공하고자하는 목표로 합니다.(+프로젝트에 특화된 GPT 제작!!)

 

현재는 위의 목표를 달성하기 위해 간단히 프롬프팅을 활용해 기능을 구현하지만, 추후 RAG를 사용하여 좀 더 고도화된 GPT 제작을 진행할 예정입니다.

 

🗂️ Table of Contents

1. AI 챗봇 요구 사항

2. 시스템 구성

3. 코드 아키텍처

4. GPT 프롬프팅

5. 챗봇 동작 확인

 

💡 Background

이전 블로그 글

[BricksAssistant] 디스코드 AI 챗봇 구축기 - (1) 기획 [링크]

[BricksAssistant] 디스코드 AI 챗봇 구축기 - (2) 디스코드 연동하기 [링크]

 

Project Github [링크]

 

 

1. AI 챗봇 요구 사항

A. 이전 대화 내용을 기반으로 AI와 대화가 필요

• 질문에 대한 히스토리를 디스코드 대화상에 기록
  → "디스코드 쓰레드 기능 활용"
• 질문 하나에 대한 답변이 아닌 이전 대화 내용을 기반으로 GPT에 요청하여 더 정확도 높은 답변 구성
  → "이전 대화내용 DB 적재 후 다음 대화에 사용"

 

B. 확장 가능한 형태로 기능 개발 필요

• 챗봇의 경우 사용자의 피드백을 받아 개선 및 기능 추가(Claude, RAG)가 계속될 예정이라 확장 가능한 형태의 코드 구성 필요
  → "Python 개발 시 Hexagonal Architecture + 추상화 계층 사용" 

 

2. 시스템 구성

A. Chatbot 시스템 구성

• 이전 대화 내용 History 저장을 위해 DocumentDB 사용
  • 사용자 피드백에 따라 시스템 구성이 달라질 수 있어 정해진 스키마보단 유연하게 대처할 수 있는 NoSQL 선택
  • 여러 NoSQL 시스템 중 AWS DocumentDB는 AWS에서 관리되어 데이터베이스 인프라를 직접 관리할 필요X
  • 프리티어를 제공하기 때문에 비용 절약
• ChatGPT의 경우 "GPT-4o mini" 모델 사용
  • 챗봇의 특성 상 높은 수준의 문제 해결 질의 보단 빠르고 가벼운 작업을 위한 모델이 적합할 것으로 판단
  • GPT-4o mini 모델은 소형 모델 중 가장 발전된 모델로 챗봇에 사용하기에 가장 적합할 것으로 선택
    • GPT 모델의 경우 추가적인 리서치와 챗봇의 발전 방향에 따라 변경될 수 있음.
  • https://platform.openai.com/docs/models/gpt-4o-mini

BrickAssitant 챗봇 시스템 구성

 

 

3. 코드 아키텍처

A. Project Tree

.
├── README.md
├── app.py									# bot 실행파일
├── requirements.txt						# 라이브러리 설치
├── src
│   ├── __init__.py
│   ├── adapter							# adapter 폴더(infra, application 연결)
│   │   ├── __init__.py
│   │   ├── bot.py							# bot adapter 코드
│   │   ├── database.py						# database adapter 코드
│   │   └── gpt.py							# gpt dapter 코드
│   ├── application						# application 폴더
│   │   ├── __init_.py
│   │   ├── command.py						# 내부 service와 adapter 코드
│   │   ├── entity.py
│   │   └── service							# 내부 Domain service 코드
│   │       ├── __init__.py
│   │       └── request_answer.py				# GPT응답 요청 처리
│   └── infra							# infra 폴더
│       ├── __init__.py
│       ├── database						# Database infra 폴더
│       │   ├── __init__.py
│       │   ├── abs_database.py
│       │   └── dynamodb.py
│       └── gpt								# gpt infra 폴더
│           ├── __init__.py
│           ├── abs_gpt.py
│           └── chatgpt.py
└── tests								# 테스트 코드 폴더
    ├── application
    │   ├── __init__.py
    │   └── service
    │       ├── __init__.py
    │       └── test_request_answer.py
    └── infra
        ├── __init__.py
        ├── database
        │   └── test_dynamodb.py
        └── gpt
            └── test_chatgpt.py

 

B. 세부 사항

 

• Hexagonal Architecture
  • infra, application, adapter 3개의 폴더로 구성
  • infra : 비즈니스 도메인 영역을 처리하는 외부 요소 정의 (ex. database, gpt)
  • application : 비즈니스 도메인 영역 처리 (ex, command, service)
    • command : 외부 요소를 받아 service 영역에 전달해주는 역할
    • service : 도메인 영역을 처리하는 코드
  • adapter : infra, application 영역 연결
    • 주의!! application에서 외부 요소를 불러올 때, infra영역 코드를 불러오는 것이 아닌 adapter을 통해서 불러와야 함
    • 이를 위해 팩토리 패턴 활용(application은 팩토리에서 어떤 종류의 데이터베이스를 썼는지 알 필요 없음!!)
      • 파이썬의 경우 별도의 인터페이스가 없기 때문에 팩토리 패턴과 아래에 있는 추상화 계층 활용

▼ 팩토리 패턴(src/adapter/database.py)

from src.infra.database.dynamodb import Dynamodb


class DatabaseFactory:
    @staticmethod
    def create_database_gpt(db_type: str = "dynamodb"):
        if db_type == "dynamodb":
            return Dynamodb()
        else:
            raise ValueError(f"Unknown database type: {db_type}")

 

• 추상화 계층
  • 시스템 확장성을 고려하여 시스템 유지 보수성을 향상하기 위해 사용
  • 장점
    • 복잡성 감소
    • 유지보수성 증가
    • 코드 재사용성 향상
    • 협업 용이성 증가
    • 독립적인 개발 및 테스트 가능

 

▼ 추상화 클래스 정의(src/infra/database/abs_database)

from abc import ABC, abstractmethod
from src.application.entity import GPTConversationInfo


class AbstractDatabaseGPT(ABC):
    @abstractmethod
    def insert_item(self, item: GPTConversationInfo):
        """데이터베이스에 GPT 요청 응답 데이터 삽입하는 추상화 메서드"""
        pass

    @abstractmethod
    def get_item(self, thread_id: str):
        """데이터베이스에 특정 thread_id 기준 응답을 가져오는 추상화 매서드"""
        pass

    @abstractmethod
    def delete_item(self, thread_id: str, request_time: str):
        """데이터베이스에 특정 thread_id 기준 응답을 삭제하는 추상화 매서드"""
        pass

 

 

▼ 추상화 클래스 상속받아 코드 구현(src/infra/database/dynamodb.py)

import os
import boto3
from boto3.dynamodb.conditions import Key

from src.infra.database.abs_database import AbstractDatabaseGPT


class Dynamodb(AbstractDatabaseGPT):
    def __init__(self) -> None:
        dynamodb = boto3.resource('dynamodb',
                                  aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
                                  aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
                                  region_name=os.getenv("AWS_DEFAULT_REGION"))
        self.table = dynamodb.Table("brickstudy-gpt")

    def insert_item(self, item: dict):
        if item:
            self.table.put_item(Item=item)

    def get_item(self, thread_id: str):
        try:
            response = self.table.query(
                KeyConditionExpression=Key('thread_id').eq(thread_id)
                )
            return response["Items"]
        except Exception:
            return []

    def delete_item(self, thread_id: str, request_time: str):
        self.table.delete_item(Key={
            'thread_id': thread_id,
            "request_time": request_time
            })

 

 

4. GPT 프롬프팅

A. 기본 role 지정

• brickAssitant라는 이름 지정
• 이전 History를 assitant Role로 전달해줌을 지정
• 마지막에 GPT의 답변 내용을 영어로 요약해주도록 설정
  • 영어로 요약하는 이유는 토큰을 조금이라도 줄이기 위해(?)

self.model = "gpt-4o-mini"
self.messages = [
    {
        "role": "system",
        "content": "Your name is brickAssistant. Respond in Korean. "
        + "The role of assistant's message is the content of your previous conversation with me. "
        + "At the end of your response, summarize the key points in bullet form. "
        + "Provide the summary in English and indicate it as [Summary] followed by the bullet points."
    }
]

 

B. 이전 대화 내용을 Assitant로 전달

# src/application/service/request_answer.py
class GPTRequestService:
    def request_answer(self, client, db, info: GPTConversationInfo):
        # find history
        history = []
        if info.thread_id:
            response = db.get_item(info.thread_id)
            for idx, r in enumerate(reversed(response)):
                if idx == 5: break  # 최근 5개만 저장

                history_q = r["question"]
                hisotry_a = r["record"]
                content = f"user: {history_q}, brickAssistant: {hisotry_a}"
                history.append(content)

        # request gpt
        response = client.request_gpt(info.question, history)
        all_answer = response.choices[0].message.content

-------------
# src/infra/gpt/chatgpt.py

	def request_gpt(self, msg: str, history: list = []):
        try:
        	# history를 message에 추가
            for old_msg in history:
                self.messages.append(
                    {"role": "assistant", "content": old_msg}
                )

            self.messages.append(
                {"role": "user", "content": msg}
            )
            response = self.client.chat.completions.create(
                model=self.model,
                messages=self.messages
            )
            return response

 

5. 챗봇 동작 확인

A. 서버 실행

• 클라우드 가상 서버에 배포

B. 디스코드 실행 후 채널에 챗봇추가

• 기존에 채널이 있는 경우 챗봇이 해당 채널에 있는지 확인

 

C. "!GPT "로 질문하기

• !GPT과 함께 질문하면 챗봇이 인식하여 답변을 위해 쓰레드 생성

 

C. 동일한 쓰레드에 질문 시 이전 답변을 기억하여 답변

 

정상 동작됨을 확인!!

실제 모임 내에서 운영하여 피드백을 통해 챗봇을 점차 개선해 나갈 예정입니다!!