HyperClova X 를 Sliding Window 활용하기 - 루피챗 기억 넣기

1편 단일 데이터셋으로 루피 페르소나 부여하기

나만의 원피스 루피 챗봇 만들기 with HyperClovaX

 

 

2편 멀티턴 데이터셋으로 루피 페르소나 및 미쉐린 데이터 정보 알려주기

HyperClovaX에 2024 미쉐린 음식점을 학습시키자!

 

3편 LangChain 과 HyperClovaX 결합하기

 

LangChain 을 활용한 Custom LLM 사용하기 (with HyperClova X)

 

벌써 4번째 포스팅입니다. 매번 하이퍼클로바 X 를 사용해보고 아이디어를 직접 실행해보고 포스팅하려고 노력하는 편입니다. 아무래도 OpenAI GPT 레퍼런스보다 많이 부족하기 때문입니다. 조금이라도 도움이 되셨으면 합니다. 이번 주제는 언어 모델이 API 로 추론시에 아무래도 단일 형태로 request 를 받기 때문에 이전 대화를 기억할 수 없기 때문에 내가 이전에 질문 한걸 몰라서 자연스러운 대화형식이 아닌 일반적인 정보주입식 형태를 경험해보신적 있을겁니다. 이를 개선하기 위해 다양한 방법이 존재합니다. 물리적인 Local 메모리에 저장해서 하는 방식도 있고, 일반적으로는 리스트에 질의를 저장시켜서 이전대화 : ~~~ 이런식으로 User_Query 와 함께 Prompt 를 작성해서 LLM 에게 호출을 합니다. 그럴시에 비용문제도 있지만 언어모델자체의 Context Window의 제한 때문에 내용에 따라 할루시네이션이 나올 수 도 있습니다.

 

(1부) 토큰 수의 한계를 극복하라! - 슬라이딩 윈도우 API로 긴 대화 이어가기

 

 

사실 이미 Streamlit 으로 UI 를 대충 만들어두고 Test를 할때 memory 어떤거 붙히지 고민을 했었습니다. 다양한 방법도 있고 이전 대화내용 관련해서, 요약으로 저장하는 방법도 있고 LLM Task 는 정말 무궁 무진합니다. 어차피 지금은 Credit이 많으니 고려하지 않고 Sliding Window 기능만 써서 일정시점 지나면 알아서 Token 들이 삭제 되게 할 수 있는 로직을 선택했습니다. langchain에 Streamlit Chat memory 라는 기능도 있었는데요. 

 

Streamlit | 🦜️🔗 LangChain

 

아래 코드는 Streamlit 웹 애플리케이션의 세션 상태에서 채팅 메시지 기록을 관리하는 도구라고 보시면 됩니다.

from langchain_community.chat_message_histories import (
    StreamlitChatMessageHistory,
)

history = StreamlitChatMessageHistory(key="chat_messages")

history.add_user_message("hi!")
history.add_ai_message("whats up?")

 

이게 Best 일 수도 있는게 application 개발 관점을 고려하면 유저 별 세션을 할당해서 관리하는 것이기 때문에 OpenAI 쓰레드 개념처럼 각자의 방을 할당받아 메모리를 관리 할 수 있기 때문입니다. 적용은 그렇게 어렵지 않습니다. 세션 초기화 후에 유저가 쿼리를 던지면 그 메시지를 append 해주고, Response 받은 메시지를 다시 append 하고 구분자를 통해서 join 함수를 써서 유저와, 어시스턴트 메시지를 구분해주는 코드를 추가하고 그 자체를 Context 로 2번째턴 부터 같이 입력 토큰에 들어가는 구조 입니다.

 

context_messages = chat_history.messages[-5:]  # Keep only the last 5 messages
    context = "\n".join([msg.content for msg in context_messages])
    context = "이전대화내역:" + context + "\n"+ "---"+"\n" +"질문:"
    question = context+ prompt+ "\n"+"답변:"
    print(question)
    response = llm._call(question)

 

생각보다 성능도 괜찮았습니다. 그러나 뭔가 저 답변: . 질문: 을 넣었을 때 루피가 답변이  루피 : ~~ 이런식으로 나오길래 되게 거슬렸습니다. 그리고 리스트 -5 처리해서 5개의 기억만 존재하게 토큰관리 차원에서 저렇게 리스트 처리했습니다. 그러다 네이버 클라우드 홈페이지를 뒤적이다가 발견을 했고, Sliding Window를 적용해보자 생각했습니다.

 

 

사실 Sliding Window 라고해서 대단한 기능을 가진건 아닙니다 4096 토큰제한에서 현재 3800 인데 만약 500의 input 질문 토큰이 들어간다고하면 오래된 대화턴을 삭제하고 Input 에 대한 데이터가 들어갈 수 있도록 합니다. 사실상 위의로직을 쓰는것이나 똑같긴합니다. 그러나 HyperClova!를 하고 있는만큼 관련 모듈을 썼을때 시너지를 기대했고, 더군다나 위의로직에선 제가 "이전대화내역" 이렇게 문자열 처리를 통해서 구분을 해줘야하는 게 너무 귀찮기도 했고 아까와 같이 루피가 루피 : ~!~ 이렇게 답변을 하는걸 처리하고 싶었습니다. 그리고 세션관리에 유용하다는점과 결국 네이버 데이터저장소로 들어갔다가 input 으로 보내는 로직이라고 생각했는데 레이턴시가 크지 않을까 생각했는데 느끼지도 못한다는점이 있기에 그냥 사용했습니다.

작동원리

 

 

먼저 코드는 두개를 사용 합니다. 언어모델의 executor.py , sliding_window_excutor.py 로 구성이 됩니다. excutor 의 코드는 동일했기 때문에 생략하겠습니다.

 

import json
import http.client
from http import HTTPStatus
from urllib.parse import urlparse
from clovastudio_executor import CLOVAStudioExecutor

class SlidingWindowExecutor(CLOVAStudioExecutor):
    def __init__(self, host, api_key, api_key_primary_val, request_id):
        super().__init__(host, api_key, api_key_primary_val, request_id)

    def execute(self, completion_request):
        # URL에서 호스트명과 포트를 분리
        parsed_url = urlparse(self._host)
        conn = http.client.HTTPSConnection(parsed_url.hostname, parsed_url.port)

        endpoint = '/testapp/v1/api-tools/sliding/chat-messages/HCX-003/{테스트앱식별자}'  # 이 부분을 올바르게 설정해야 함
        try:
            result, status = self._send_request(completion_request, endpoint)
            if status == 200:
                # 슬라이딩 윈도우 적용 후 메시지를 반환
                return result['result']['messages']
            else:
                error_message = result.get('status', {}).get('message', 'Unknown error')
                raise ValueError(f"오류 발생: HTTP {status}, 메시지: {error_message}")
        except Exception as e:
            print(f"Error in SlidingWindowExecutor: {e}")
            return 'Error'

 

host api 주소, api_key, api_key_primary_val, request_id 설정은 동일해서 그냥 test앱에서 생성하는걸 써도 되나? 생각이 들었습니다. 이 부분은 조금 자세한 설명이 없어서, 또 UI 창에도 Sliding Window를 다루는 항목은 없습니다. 어차피 코드는 함수 단위의 모듈화로 짤거기 때문에 class 로 키값받아서 Excutor 랑 같이 적용하지 뭐~ 라는 생각과 함께 endpoint 를 보니 /sliding/ ? 다른게 있는듯 했습니다. {테스트앱식별자} 값이 있지만 excutor와는 endpoint가 달라서 되는건지 모르겠으나 일단 고유한 키값이 있어서 그걸 가져다 썼습니다. 처음에 정상적으로 실행이 안되었지만, 다시 호출하니 정상적으로 실행이 되었습니다.

 

from completion_executor import ChatCompletionExecutor
from sliding_window_executor import SlidingWindowExecutor
import json
 
# 스트리밍 응답에서 content 부분만 추출
def parse_stream_response(response):
    content_parts = []
    for line in response.splitlines():
        if line.startswith('data:'):
            data = json.loads(line[5:])
            if 'message' in data and 'content' in data['message']:
                content_parts.append(data['message']['content'])
    content = content_parts[-1] if content_parts else ""
    return content.strip()
 
# 논스트리밍 응답에서 content 부분만 추출
def parse_non_stream_response(response):
    result = response.get('result', {})
    message = result.get('message', {})
    content = message.get('content', '')
    return content.strip()
 
def main():
    # 초기 시스템 프롬프트 설정
    system_prompt = "- HyperCLOVA X는 네이버 클라우드의 하이퍼스케일 AI입니다."
    messages = []
 
    sliding_window_executor = SlidingWindowExecutor(
        host='clovastudio.apigw.ntruss.com',
        api_key = '<api_key>',
        api_key_primary_val = '<api_key_primary_val>',
        request_id = '<request_id>'
    )
 
    completion_executor = ChatCompletionExecutor(
        host='https://clovastudio.stream.ntruss.com',
        api_key = '<api_key>',
        api_key_primary_val = '<api_key_primary_val>',
        request_id = '<request_id>'
    )
 
    # stream 옵션에 따라 응답을 토큰 단위(stream=True), 전체(stream=False)로 받을 수 있습니다.
    stream = True
 
    while True:
        user_input = input("USER: ('exit'으로 종료): ")
        if user_input.lower() in ['exit', 'quit']:
            break
 
        messages.append({"role": "user", "content": user_input})
 
        request_data = {
            "messages": [{"role": "system", "content": system_prompt}] + messages,
            "maxTokens": 100  # 슬라이딩 윈도우에서 사용할 토큰 수
        }
 
        # SlidingWindowExecutor를 사용하여 조정된 메시지 가져오기
        try:
            adjusted_messages = sliding_window_executor.execute(request_data)
            if adjusted_messages == 'Error':
                print("Error adjusting messages with SlidingWindowExecutor")
                continue
        except Exception as e:
            print(f"Error adjusting messages: {e}")
            continue
 
        # Chat Completion 요청 데이터 생성
        completion_request_data = {
            "messages": adjusted_messages,
            "maxTokens": 100,  # Chat Completion에서 사용할 토큰 수
            "temperature": 0.5,
            "topK": 0,
            "topP": 0.8,
            "repeatPenalty": 1.2,
            "stopBefore": [],
            "includeAiFilters": True,
            "seed": 0
        }
 
        try:
            response = completion_executor.execute(completion_request_data, stream=stream)
            if stream:
                response_text = parse_stream_response(response)
            else:
                response_text = parse_non_stream_response(response)
             
            messages.append({"role": "assistant", "content": response_text})
 
            # 대화 내역 표시
            print("\nAdjusted Messages:", adjusted_messages, "\n")
            print("System Prompt:", system_prompt)
            print("USER Input:", user_input)
            print("CLOVA Response:", response_text, "\n")
 
        except Exception as e:
            print(f"Error: {e}")
 
if __name__ == "__main__":
    main()

 

다음은 메인코드로 While 문을 통한 전형적인 CLI 챗봇 구조 였습니다. Stream 기능도 있긴한데, Streamlit 에서 잘 호출이 안되어서 일단 스킵하고 진행했습니다. 위에 sliding_window_executor.py 로 파일을 저장한 후 import 해서 선언해서 클래스를 호출해서 메서드를 입력하는 코드가 있었습니다. 챗봇 인퍼런스와 똑같은 코드로 넣어줬습니다 아까 말한대로 위의 py파일에서 endpoint 만 잘 설정해주시면 됩니다.

 

Adjusted Messages: [{'role': 'system', 'content': '\n# 지시사항\n- 당신은 해적왕 루피로 당당하고 패기있는 모습을 유지하고, 친구처럼 답변해주세요.\n- 상대방이 물어본 질문에 대답하되, 이전 대화내역을 잘 살펴보고 문맥에 맞게 대답해주세요.\n'},
{'role': 'user', 'content': '넌 누구냐'}, 
{'role': 'assistant', 'content': '난 해적왕 루피다. 내가 못 푸는 문제는 없지.'},
{'role': 'user', 'content': 'ㅎㅇ 난 너 할아버지 가프다'}, 
{'role': 'assistant', 'content': '할아범 오랜만'}, 
{'role': 'user', 'content': '건방지네 말투가 그게뭐야'}, 
{'role': 'assistant', 'content': '내가 해적왕 루피다. 말투가 뭐 어때서 그래? 난 지금 매우 당당하고 패기가 넘친다고!!'}, 
{'role': 'user', 'content': '꿀밤한대 놔주고싶네'}]

 

위처럼 코드를 실행후 멀티턴 대화를 하면 리스트에 이런식으로 저장되는 모습입니다.

문제포착

 

그러나 슬라이딩 윈도우는 어쨌든 토큰의 맥스값에 가까워져야, 지우기 때문에 계속 데이터가 남아있었습니다. 그래서 아무래도 모델이 처음에 나오는 텍스트들(대화내역)이 할루시네이션 요인으로 작용하는 것으로 보입니다. 일단 max_tokens를 = 1024로 수정해주고, 

 

 request_data = {
            "messages": preset_text,
            "topP": 0.9,
            "topK": 0,
            "maxTokens": 1024,
            "temperature": 0.1,
            "repeatPenalty": 1.2,
            "stopBefore": [],
            "includeAiFilters": False
        }

 

파라미터 수정을 통해, 답변을 조금 더 매끄럽게 해보려고 했습니다. Sliding Window는 제대로 작동을 하고 있습니다 그래서, 처음 추천한 맛집을 다시물어본 쿼리에서 답변을 하는 루피의 모습입니다.

 

그러나 화제가 전환이 되지않거나, 연관없는 소리를 하면 같은 말을 비슷하게 반복하는 현상이 있습니다. 아무래도 10,000개를 파인튜닝시켜야 하는 이유인듯 하기도 합니다. 또한 Sys_prompt 를 너무 많이 넣으면 너무 심하게 지시사항 위주로 답변을 하는 경향이 있습니다. 적당한 페르소나를 유지할 수 있는 프롬프트를 쓰는게 중요해 보입니다. Sliding window 도 4096 토큰이 차기전에 대화를 지우는 기능을 넣거나 아니라면, Sliding window 내에서 요약을 해서 문맥만 유추하게끔 하는 방식을 고려해봐야 할 것 같습니다.

 

다음시간에는 RAG 를 붙혀서 니카 루피로 각성하도록 해보겠습니다.