[MeMind 리팩토링] 1. SRP 원칙 적용하기

 

0. 리팩토링을 시작한 이유

본격적인 리팩토링을 시작하기 전에, 리팩토링을 해야겠다라고 생각한 계기를 간단히 얘기해보려고 한다.

조코딩 AI 해커톤에 참여하고 운이 좋아서 수상까지 하게 되었지만,
개발을 하고 운영을 하는 내내, 내 코드가 불안하다는 느낌을 지울 수가 없었다.

프론트에서 뭔가 에러가 나면, 기억에 의존해서 코드를 찾아가며 고치는 일이 잦았고
뭔가 구조를 가지고 단단하게 코드를 짜는 느낌이 들지 않았다.

그래서 객체지향 프로그래밍의 개발 원칙인 SOLID 원칙을 적용하는 것부터
테스트 코드, 시스템 아키텍처까지 천천히 그러나 꼼꼼하게 리팩토링 해보려고 한다.

 

포스팅 내에는 분량 상, 일부 파일의 코드만 올라갈 예정이고
만약 전체 디렉토리 및 코드를 보고 싶다면, 아래 깃허브를 참고해주시길!!

GitHub - Me-mind-hackerthon/memind-backend

 

그럼 본격적으로 시작해보자!!

 

비즈니스 로직을 포함하는 Service 디렉토리의 코드에 SRP 원칙 (단일 책임 원칙)을 적용해보려고 한다.

1. 기존 코드 분석

# service/talk.py
""" ChatGPT와의 API 통신을 통해 대화하는 로직 """

from fastapi import HTTPException, status
import openai
from datetime import datetime, date

from models import Conversation, Message

class ConversationHandler:
    def __init__(self, session, nickname):
        self.session = session
        self.nickname = nickname
        # OpenAI GPT-3.5 Turbo API 인증 설정
        openai.api_key = os.environ["GPT_APIKEY"]

    def __parsing_date(self, date):
        date_object = datetime.strptime(date, "%Y-%m")

        return date_object.year, date_object.month

    def get_all_messages(self, conversation_id):
        chat_history = []
        message_object = select(Message).where(Message.conversation_id == conversation_id)
        messages = self.session.exec(message_object).all()

        for m in messages:
            if(m.is_from_user):
                role = "user"
            elif(not m.is_from_user):
                role = "assistant"

            chat_history.append({"role": role, "content": m.message})

        return chat_history

    def __get_all_full_messages(self, conversation_id):
        message_object = select(Message).where(Message.conversation_id == conversation_id)
        messages = self.session.exec(message_object).all()

        return messages

    def get_conversation_by_month(self, date):
        year, month = self.__parsing_date(date)
        conversation_object = select(Conversation).where(Conversation.nickname == self.nickname).where(Conversation.year == year).where(Conversation.month == month)
        conversation_list = self.session.exec(conversation_object).all()

        if(not conversation_list):
            raise HTTPException(
                status_code = status.HTTP_404_NOT_FOUND, detail = "해당 월에 작성된 일기가 없습니다"
            )

        return {
            "conversation_list": conversation_list
        }

    def create_message(self, conversation_id, order, message, is_from_user):
        message_object = Message(
            conversation_id = conversation_id,
            order = order,
            is_from_user = is_from_user,
            message = message
        )

        self.session.add(message_object)
        self.session.commit()

        return

    def __get_conversation_id_by_date(self, date):
        date_object = datetime.strptime(date, "%Y-%m-%d")

        conversation_object = select(Conversation).where(Conversation.nickname == self.nickname).where(Conversation.year == date_object.year).where(Conversation.month == date_object.month).where(Conversation.day == date_object.day)
        conversation_object = self.session.exec(conversation_object).first()

        if(not conversation_object):
            return 0

        return conversation_object.conversation_id

    def start_conversation(self, date):
        conversation_id = self.__get_conversation_id_by_date(date)

        if(conversation_id):
            chat_history = self.__get_all_full_messages(conversation_id)
        else:
            conversation_id = uuid4().hex
            conversation = Conversation(
                nickname = self.nickname,
                conversation_id = conversation_id
            )

            self.session.add(conversation)
            self.session.commit()

            messages = [
                {"role": "system", "content": "너는 친절한 심리상담가야. 사용자에게 오늘 하루는 어땠는지 물어보고 사용자가 응답하면 더 자세히 물어봐주고 위로해주는 상담가의 역할을 해줘. 사용자에게 보내는 너의 첫 메세지는 '안녕하세요! 오늘 하루는 어땠나요?'로 고정이야. 사용자의 응답에 적절하게 반응해주고 항상 더 자세히 질문해줘야 해. 그리고 2번 이상 응답을 받으면, '충분히 이야기를 나눈 것 같네요. 오늘 하루를 평가한다면 몇 점을 주시겠어요?'라는 말로 대화를 마무리해줘"},
                {"role": "user", "content": "안녕"}
            ]

            # OpenAI GPT-3.5 Turbo 모델에 대화를 요청합니다.
            response = openai.ChatCompletion.create(
                model="gpt-3.5-turbo",
                messages=messages
            )

            self.create_message(conversation_id, 0, response['choices'][0]['message']['content'], False)

        chat_history = self.__get_all_full_messages(conversation_id)

        if(len(chat_history) > 13):
            is_enough = True
        else:
            is_enough = False

        return {
            "conversation_id": conversation_id,
            "chat_history": chat_history,
            "is_enough": is_enough
        }

    def answer_conversation(self, user_answer, conversation_id):
        messages = [
            {"role": "system", "content": "너는 친절한 심리상담가야. 사용자의 응답에 대해서 더 자세히 물어봐주고 위로해주는 상담가의 역할을 해줘. 단, 위로보다는 더 자세히 물어봐주는 경우가 더 많아야 해. 그리고 2줄 이상 말하지마"},
            {"role": "user", "content": "안녕"}
        ]

        # 이전 대화 내역이 있으면 채팅에 추가합니다.
        chat_history = self.get_all_messages(conversation_id = conversation_id)
        messages.extend(chat_history)

        # 사용자 입력을 채팅에 추가합니다.
        user_message = {"role": "user", "content": user_answer}
        messages.append(user_message)

        # message object 추가
        order = len(messages)
        self.create_message(conversation_id, order, user_answer, True)
    
        # OpenAI GPT-3.5 Turbo 모델에 대화를 요청합니다.
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages=messages
        )

        # 모델의 답변을 가져옵니다.
        assistant_response = response['choices'][0]['message']['content']

        self.create_message(conversation_id, order + 1, assistant_response, False)

        if(order > 13):
            is_enough = True
        else:
            is_enough = False

        # 챗봇의 답변을 사용자 메시지와 함께 반환합니다.
        return {
            "message": assistant_response,
            "is_enough": is_enough
        }

 

기존 코드의 문제점은 다음과 같다.
(문제점이 매우매우 많지만 이번 포스팅에선 SRP 원칙과 관련된 문제점만 언급한다!!)

 

1. 한 클래스 내에 너무 많은 메서드 (책임)이 공존한다.

이번 포스팅의 핵심이다.

ConversationHandler라는 한 클래스 내에 모든 메서드가 몰빵되어 있다.
즉, 너무 많은 책임이 한 클래스 내에 공존한다.

SRP 원칙에서는 원칙을 위반할 경우,
"한 메서드에서 생긴 문제가 여러 메서드로 퍼져나가 유지보수가 어렵고, 코드 가독성이 떨어진다"고 말한다.
즉, 코드 응집도가 저하되고 결합도가 높아지는 좋지 않은 코드라는 것이다.

실제로 나는 해커톤 내내, 유지보수의 어려움을 굉장히 많이 느꼈다.

대화와 관련된 API가 모두 ConversationHandler 인스턴스를 생성하고 있었기 때문에
어떤 메서드에 문제가 생기든 하나의 클래스 내의 메서드들을 찾아가며 수정해야 했고,
한 메서드를 고치면, 다른 메서드에 연결되어 문제가 생기는 일이 잦았다.

뭔가 코드가 안정되지 못하고 불안정하다라는 느낌을 많이 받았던 것 같다.

 

2. 메서드마다 어떤 리턴 타입을 가지고 있는지 명시되어 있지 않다.

코드 가독성에 대한 문제다.

메서드가 어떤 결과 타입을 리턴하는지를 명시해놓지 않아서,
확장을 하거나 수정을 할 때, 머릿 속에서 코드를 한번씩 돌려봐야했다.

 

2. 기존 코드에 SRP 원칙 적용하기

# service/talk.py
""" ChatGPT와의 API 통신을 통해 대화하는 로직 """

from uuid import uuid4
import os
from sqlmodel import select
import openai
from datetime import datetime, date
from typing import Dict, List, Any

from models import Conversation, Message
from exceptions import ConversationNotFound, NoSuchConversationIdError

class DateParser:
    @staticmethod
    def parsing_date(date) -> datetime:
        """ date 정보를 year과 month로 파싱하는 함수 """
        return datetime.strptime(date, "%Y-%m-%d")

class EnoughJudge:
    @staticmethod
    def is_enough(conversation_lenght) -> bool:
        return conversation_lenght > 13

class MonthlyConversationLoader:
    def __init__(self, session) -> None:
        self.session = session

    def _get_conversation_list(self, date_object, nickname) -> List[Conversation]:
        conversation_object = select(Conversation).where(
            Conversation.nickname == nickname,
            Conversation.year == date_object.year,
            Conversation.month == date_object.month
        )

        return self.session.exec(conversation_object).all()

    async def get_conversation_by_month(self, date, nickname) -> Dict[str, List[Any]]:
        """ 해당 월에 나눈 대화 목록을 리턴하는 함수 """
        date_object = DateParser.parsing_date(date)
        conversation_list = self._get_conversation_list(date_object, nickname)

        if(not conversation_list):
            raise ConversationNotFound

        return {
            "conversation_list": conversation_list
        }

class FullMessageLoader:
    def __init__(self, session) -> None:
        self.session = session

    def get_all_full_messages(self, conversation_id) -> List[Message]:
        """ 해당 conversation에서 나누었던 message들을 모두 리턴하는 함수 """
        try:
            message_object = select(Message).where(Message.conversation_id == conversation_id)
            messages = self.session.exec(message_object).all()
        except Exception as e:
            raise NoSuchConversationIdError

        return messages

class MessageSender:
    def __init__(self) -> None:
        openai.api_key(os.environ["GPT_APIKEY"])
        self.premessage = [
                {"role": "system", "content": "너는 친절한 심리상담가야. 사용자에게 오늘 하루는 어땠는지 물어보고 사용자가 응답하면 더 자세히 물어봐주고 위로해주는 상담가의 역할을 해줘. 사용자에게 보내는 너의 첫 메세지는 '안녕하세요! 오늘 하루는 어땠나요?'로 고정이야. 사용자의 응답에 적절하게 반응해주고 항상 더 자세히 질문해줘야 해. 그리고 2번 이상 응답을 받으면, '충분히 이야기를 나눈 것 같네요. 오늘 하루를 평가한다면 몇 점을 주시겠어요?'라는 말로 대화를 마무리해줘"},
                {"role": "user", "content": "안녕"}
            ]

    async def send_message_to_chatgpt(self, messages = None) -> str:
        # OpenAI GPT-3.5 Turbo 모델에 대화를 요청합니다.
        response = openai.ChatCompletion.create(
            model="gpt-3.5-turbo",
            messages = self.premessage.extend(messages)
        )

        return response["choices"][0]["message"]["content"]
        
class ConversationStarter:
    def __init__(self, session) -> None:
        self.session = session

    def _get_conversation_id_by_date(self, date, nickname) -> str:
        """ 해당 일자에 conversation_id를 리턴하는 함수 """
        date_object = DateParser.parsing_date(date)

        conversation_object = select(Conversation).where(Conversation.nickname == nickname).where(Conversation.year == date_object.year).where(Conversation.month == date_object.month).where(Conversation.day == date_object.day)
        conversation_object = self.session.exec(conversation_object).first()

        return conversation_object.conversation_id

    # conversation 생성, message load, 길이 check
    async def start_conversation(self, date, nickname) -> Dict[str, Any]:
        """ conversation이 있으면, 대화 내용을 리턴하고, 없으면 새로 conversation을 생성하는 함수 """
        conversation_id = self._get_conversation_id_by_date(date, nickname)

        # conversation이 존재하지 않는 경우, 새로 conversation 생성
        if(not conversation_id):
            conversation_id = uuid4().hex
            conversation = Conversation(
                nickname = nickname,
                conversation_id = conversation_id
            )

            self.session.add(conversation)
            self.session.commit()

            response = MessageSender().send_message_to_chatgpt()

            MessageCreator().create_message(conversation_id, 0, False, response)

        chat_history = FullMessageLoader().get_all_full_messages(conversation_id)

        return {
            "conversation_id": conversation_id,
            "chat_history": chat_history,
            "is_enough": EnoughJudge.is_enough(len(chat_history))
        }

class MessageGetter:
    def __init__(self, session) -> None:
        self.session = session

    async def classify_writer(self, conversation_id) -> List[Dict[str, str]]:
        """ message들을 화자에 따라서 분류하여 리턴하는 함수 """
        chat_history = []
        messages = FullMessageLoader().get_all_full_messages(conversation_id)
    
        for m in messages:
            if(m.is_from_user):
                role = "user"
            elif(not m.is_from_user):
                role = "assistant"

            chat_history.append({"role": role, "content": m.message})

        return chat_history

class MessageCreator:
    def __init__(self, session) -> None:
        self.session = session

    async def create_message(self, conversation_id, order, is_from_user, message) -> None:
        """ 채팅 이력을 저장하는 함수 """
        message_object = Message(
            conversation_id = conversation_id,
            order = order,
            is_from_user = is_from_user,
            message = message
        )

        try:
            self.session.add(message_object)
            self.session.commit()
        except Exception as e:
            raise NoSuchConversationIdError

class MessageRespondent:
    def __init__(self, session) -> None:
        self.session = session

    async def answer_conversation(self, user_answer, conversation_id) -> Dict[str, Any]:
        """ 사용자의 응답에 대한 AI의 응답을 리턴하는 함수 """
        # 이전 대화 내역이 있으면 채팅에 추가합니다.
        messages = FullMessageLoader().get_all_full_messages(conversation_id)

        # 사용자 입력을 채팅에 추가합니다.
        messages.append({"role": "user", "content": user_answer})
        order = len(messages)
        await MessageCreator().create_message(conversation_id, order, user_answer, True)
    
        # OpenAI GPT-3.5 Turbo 모델에 대화를 요청합니다.
        response = await MessageSender().send_message_to_chatgpt(messages)

        await MessageCreator().create_message(conversation_id, order + 1, False, response)

        # 챗봇의 답변을 사용자 메시지와 함께 반환합니다.
        return {
            "message": response,
            "is_enough": EnoughJudge.is_enough(order)
        }

 

기존 코드에 SRP 원칙을 적용하여, 여러 클래스로 분리해보았다.

분리하는 기준은 참고한 레퍼런스마다 의견이 달라서
나는 "수정했을 때, 연쇄적으로 파급효과가 있는가?"를 기준으로 분리했던 것 같다.

메서드마다 어떤 리턴 타입을 가지고 있는지를 명시했고,
최대한 단일 클래스에는 하나의 메서드만을 배치하여 응집도를 높이되
서로 강한 결합을 가지고 있는 메서드들 중, 상호 배타적이지 않은 (결이 같은) 메서드들 또한
한 클래스 내에 배치하였다.

날짜 문자열을 파싱하여 Date 객체를 리턴하는 DataParser 클래스나
나눈 대화의 길이를 체크하여 충분한 대화를 나눴는지 여부를 리턴하는 EnoughJudge 클래스는
굳이 인스턴스를 생성할 필요가 없는 유틸리티 클래스에 해당한다고 판단하여
staticmethod로 선언해주었다.

이렇게 수정하고 보니, 추상화해서 인터페이스로 빼야할 부분이나
추후 설명할 LSP 원칙을 적용할 부분이 조금 더 명확하게 보이는 것 같다.

 

다음 포스팅에서는 SOLID 원칙 중, OCP 원칙을 적용하여
본 코드에 대한 리팩토링을 계속 진행할 예정이다.

오류나 첨언은 언제든지 댓글로 남겨주시길!!