[Co Labor] RAG 맛보기! 챗봇을 보완해보자.

현재 챗봇은 GPT API만을 사용하고 있고 챗봇을 개선할 수 있는 방법은 다음과 같다.

1. Fine Tuning

2. RAG

• https://www.law.go.kr/LSW/main.html
• 국가법령정보센터 공동 활용 https://open.law.go.kr/LSO/lab/hangulAddr.do
• 위의 링크에서 데이터 수집 후 전처리
• 전처리된 데이터를 임베딩
• 임베딩 된 공간에서 사용자 질문 검색 후 유사도가 가장 높은 결과 반환
• Fine Tuning 혹은 Prompt에 결과와 함께 대답 요구 (CHAIN OF THOUGHT 기법 사용)
  • 참고로 CAHIN OF THOUGHT 기법은 풀이 과정을 대답에서 함께 내놓는 방법이다. 추론을 할 때 왜 이렇게 추론했는지 대답에 포함해주면 정확도가 더욱 올라간다.

여기서는 RAG를 자세히 다뤄보자.

RAG

1. 데이터 수집

• 국가법령정보센터 API (https://open.law.go.kr/LSO/lab/hangulAddr.do) → 데이터가 html로 넘어와 활용 불가

따라서 국가법령정보센터에서 직접 데이터 수집 후 전처리로 방향 결정

• https://www.law.go.kr/법령/외국인근로자의고용등에관한법률
• https://www.moel.go.kr/local/ansan/common/downloadFile.do?file_seq=21171050016&bbs_seq=63204&bbs_id=LOCAL5
• https://www.law.go.kr/법령/근로기준법
  → 외국인 노동자와 관련 노동법 판례 및 행정해석 모음

2. 데이터 전처리

# -*- coding: utf-8 -*-
import re
import os

# 원본 텍스트 데이터
data = """
전처리를 원하는 데이터
"""

# 정규식을 사용하여 "제숫자조" 및 "제숫자조의숫자"로 분리
clauses = re.split(r'(제\d+조(?:의\d+)?\s*\([^)]*\))', data.strip())

# 파일 저장 경로 설정
output_dir = 'output_clauses'
os.makedirs(output_dir, exist_ok=True)

# 각 조항을 파일로 저장
for i in range(1, len(clauses), 2):
    title = clauses[i].strip()
    content = clauses[i + 1].strip() if i + 1 < len(clauses) else ''

    # 파일명 생성
    file_name = title.split('(')[0].strip() + '.txt'

    # 파일 저장
    with open(os.path.join(output_dir, file_name), 'w', encoding='utf-8') as f:
        f.write(f"{title} {content}")

print("각 조항이 개별 텍스트 파일로 저장되었습니다.")

파일을 직접 읽어서 구현하는 방식이 더 좋아보이지만 일단 테스트 용도로 직접 넣는 방식을 사용했다.

처음에는 제3조, 제3조의 2 이런식으로 조항을 구분하는 정규식을 짰는데, 법률 본문에 제24조에 따르면… 이런식의 문구가 많이 등장했다. 따라서 제39조(사용증명서)를 표준으로 정규식을 clauses = re.split(r'(제\d+조(?:의\d+)?\s*\([^)]*\))', data.strip()) 이렇게 구성하였다.

3. 데이터 임베딩 및 저장

먼저 OpenAI API를 사용하여 텍스트 임베딩을 생성하고 임베딩 간 코사인 유사도를 계산하는 기능을 구현하였다.

embedding_openAI.py

import os
import openai
from numpy import dot
from numpy.linalg import norm
from dotenv import load_dotenv

# OpenAI API 설정
load_dotenv()
openai.api_key = os.getenv("OPENAI_API_KEY")

def get_embedding(text, engine="text-embedding-ada-002"):
    text = text.replace("\n", " ")
    return openai.Embedding.create(input=[text], model=engine)['data'][0]['embedding']

def cos_sim(A, B):
    return dot(A, B) / (norm(A) * norm(B))

 

def get_embedding(text, engine="text-embedding-ada-002"):
    text = text.replace("\n", " ")
    return openai.Embedding.create(input=[text], model=engine)['data'][0]['embedding']

해당 함수가 실질적으로 OpenAI API를 호출하여 텍스트 임베딩을 생성한다. cos_sim 함수는 두 벡터 간 코사인 유사도를 계산하는 함수이다.

 

다음은 텍스트 파일 데이터를 읽고 get_embedding을 실행한 후, 그 결과를 CSV 파일로 저장하는 기능을 구현했다.

save_embedding.py

import os
import pandas as pd
from embedding_openAI import get_embedding


def embed_text_files(folders):
    for folder in folders:
        data = []

        if not os.path.exists(folder):
            print(f"Warning: Folder not found - {folder}")
            continue

        for file_name in os.listdir(folder):
            if file_name.endswith('.txt'):
                file_path = os.path.join(folder, file_name)
                with open(file_path, 'r', encoding='utf-8') as f:
                    text = f.read()
                    embedding = get_embedding(text)
                    data.append({
                        'folder': folder,
                        'file_name': file_name,
                        'text': text,
                        'embedding': embedding
                    })

        if data:
            # 폴더 이름을 기반으로 출력 CSV 파일 이름을 생성
            output_csv = os.path.join(rag_dir, f"{os.path.basename(folder)}_embeddings.csv")
            df = pd.DataFrame(data)
            df.to_csv(output_csv, index=False)
            print(f"CSV file created: {output_csv}")
        else:
            print(f"No data to save in folder: {folder}. CSV file not created.")


# 프로젝트의 rag 디렉토리 경로
rag_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))

# 모든 폴더 리스트
folders = [
    os.path.join(rag_dir, 'data', 'output_case'),
    os.path.join(rag_dir, 'data', 'output_employee'),
    os.path.join(rag_dir, 'data', 'output_foreign')
]

# 각 폴더의 데이터를 개별 CSV 파일로 저장
embed_text_files(folders)

print("Process completed.")
print(f"Rag directory: {rag_dir}")
for folder in folders:
    print(f"Checking folder: {folder}")
    if os.path.exists(folder):
        print(f"  Folder exists: {folder}")
        print(f"  Contents: {os.listdir(folder)}")
    else:
        print(f"  Folder does not exist: {folder}")

거의 다 파일 처리와 관련된 코드이고 실제 임베딩과 관련된 코드는 embedding = get_embedding(text) 이 부분이다.

4. 유사도 계산 및 결과 반환

데이터프레임에 저장된 문서들의 임베딩과 비교하여 가장 유사한 벡터를 찾아내는 기능을 구현하면 끝이다.

find_similar.py

import pandas as pd
import numpy as np
import ast
import os
from rag.embedding.embedding_openAI import get_embedding, cos_sim

def find_most_similar_documents(df, query, top_n=2):
    # 쿼리에 대한 임베딩 계산
    query_embedding = get_embedding(query)

    # 임베딩 간의 유사도 계산
    df['similarity'] = df['embedding'].apply(lambda x: cos_sim(query_embedding, np.array(ast.literal_eval(x))))

    # 유사도 순으로 정렬 후 상위 top_n 개의 문서 반환
    return df.sort_values('similarity', ascending=False).head(top_n)

# CSV 파일들의 경로 (루트 디렉토리 기준)
embedding_files = [
    'output_case_embeddings.csv',
    'output_employee_embeddings.csv',
    'output_foreign_embeddings.csv'
]

# CSV 파일들을 로드하여 하나의 DataFrame으로 결합
dfs = []
for file_name in embedding_files:
    file_path = os.path.join(os.getcwd(), file_name)
    if os.path.exists(file_path):
        dfs.append(pd.read_csv(file_path))
    else:
        print(f"Warning: File not found - {file_path}")

if not dfs:
    raise FileNotFoundError("No embedding CSV files found")

df_combined = pd.concat(dfs, ignore_index=True)

# 테스트 코드 (필요한 경우 주석 처리)
if __name__ == "__main__":
    # 질문에 대한 가장 유사한 문서 찾기
    query = "근로자의 권리에 대한 질문"
    most_similar_docs = find_most_similar_documents(df_combined, query)

    # 결과 출력
    print(most_similar_docs[['folder', 'file_name', 'similarity']])

해당 코드 또한 주석의 흐름대로 전에 만들어둔 함수를 사용하기만 한다.

 

이제 app.py 에서 만들어둔 임베딩을 사용해서 쿼리 파라미터로 날아온 메시지에 대해 가장 유사도가 높은 법률 2개만 출력하면 된다.

현재 임베딩 파일이 세 가지인데, 그들을 통틀어서 가장 유사도가 높은 벡터가 필요하므로 하나의 데이터프레임으로 결합하였다.

# 세 가지 임베딩 파일을 로드하여 하나의 DataFrame으로 결합
embedding_files = [
    'output_case_embeddings.csv',
    'output_employee_embeddings.csv',
    'output_foreign_embeddings.csv'
]

dfs = []
for file_name in embedding_files:
    file_path = os.path.join(os.getcwd(), file_name)
    if os.path.exists(file_path):
        logging.info(f"Loading CSV file from: {file_path}")
        dfs.append(pd.read_csv(file_path))
    else:
        logging.warning(f"CSV file not found: {file_path}")

if not dfs:
    logging.error("No CSV files were found. Please check the file paths and run save_embedding.py if necessary.")
    raise FileNotFoundError("No embedding CSV files found")

# 모든 데이터를 하나의 DataFrame으로 결합
df_combined = pd.concat(dfs, ignore_index=True)

유사도가 가장 높은 법률 명만 리턴하게끔 라우팅해주면 끝이다!

# 새로운 엔드포인트 추가: /arg-chat
@app.route('/arg-chat', methods=['GET'])
def arg_chat():
    query = request.args.get('message', '')
    if not query:
        return "검색어를 입력하세요", 400

    logging.info(f"Received search query: {query}")

    # 가장 유사한 문서 3개 찾기
    most_similar_docs = find_most_similar_documents(df_combined, query)

    # 파일 이름에서 ".txt"를 제거하고 "제XX조" 부분만 추출하여 텍스트로 반환
    response = "\n".join([row['file_name'].replace('.txt', '') for index, row in most_similar_docs.iterrows()])

    return response

개선 방안

현재 유사도를 판단하는 방식의 정확도가 굉장히 떨어진다. 그 이유는 아마 한국어에 대한 자연어 처리의 성능이 굉장히 떨어지고 데이터셋을 임베딩 시키는 그 문서의 길이가 너무 짧아서 유사도 판단이 제대로 되지 않아서 그런 것으로 판단된다.

따라서 sentence-transformers 라이브러리와 같은 사전 훈련된 다국어 모델을 사용하거나 법률 텍스트에 특화된 사전 훈련 모델이나 fine-tuning을 적용한 모델을 사용하면 더 나은 결과를 얻을 수 있을 것이다.