[RAG] 문서 Loader부터 다국어 검색 챗봇 정리

1. 도입부 (Introduction)

대형 언어 모델(LLM)은 놀라운 비서 역할을 수행하지만, "학습하지 않은 내부 문서나 실시간 정보"에 대해 물어보면 엉뚱한 답변을 지어내는 환각(Hallucination) 현상을 보인다. 이를 해결하는 가장 근본적이고 실무적인 기법이 바로 RAG(Retrieval-Augmented Generation, 검색 증강 생성)다.

RAG는 쉽게 말해 "LLM에게 시험 치기 직전 관련 오픈북(참고 문서)을 제공하는 시스템"이다.

많은 개발자가 RAG 시스템을 구축할 때 대형 모델(LLM) 자체에만 집중하지만, 실무 RAG 성능의 90%는 "데이터 전처리(Data Preprocessing)와 임베딩(Embedding)" 단계에서 결정된다. 아무리 똑똑한 LLM이라도 엉망으로 쪼개지고 유실된 참고 문서를 받으면 정답을 맞출 수 없기 때문이다.

본 포스팅에서는 랭체인(LangChain) 생태계를 기반으로 다양한 문서를 정교하게 읽어 들이는 방법부터 시작하여, 문맥을 해치지 않고 텍스트를 나누는 전략, 다국어 임베딩의 성능 비교, 그리고 최종적으로 다국어 검색 루프와 대화형 Gradio 챗봇 UI를 붙이는 전체 파이프라인을 완전 정복해 본다.

[RAG] LCEL(랭체인 표현 언어)과 Runnable을 이용한 RAG 파이프라인

 

 

2. 다양한 문서 형식 처리하기 (Document Loaders)

RAG의 첫 단추는 로컬이나 웹에 파편화되어 존재하는 지식 베이스를 파이썬 메모리 상의 Document 객체로 적재하는 것이다. 랭체인은 다양한 소스에 맞춤형으로 대응할 수 있는 강력한 Loader 인터페이스를 제공한다.

(1) PDF 문서 처리 (PyPDFLoader)

PDF는 레이아웃이 고정되어 있어 텍스트 추출이 까다로운 포맷 중 하나다. PyPDFLoader를 사용하면 각 페이지별로 데이터를 깔끔하게 구분하여 로드할 수 있다.

from langchain_community.document_loaders import PyPDFLoader

# PDF 로더 초기화 및 로드 (pypdf 패키지 필요)
pdf_loader = PyPDFLoader("./data/transformer.pdf")
pdf_docs = pdf_loader.load()

print(f"PDF 로드 완료. 총 페이지 수: {len(pdf_docs)}")
# 첫 번째 페이지의 내용과 메타데이터(출처 및 페이지 정보) 출력
print(f"첫 페이지 내용 샘플: {pdf_docs[0].page_content[:150]}")
print(f"첫 페이지 메타데이터: {pdf_docs[0].metadata}")

 

(2) 웹 페이지 크롤링 및 파싱 (WebBaseLoader)

특정 블로그나 공식 가이드 문서를 RAG의 참고 자료로 쓰고자 할 때 편리하다. 내부적으로 BeautifulSoup을 사용하여 HTML 태그 중 텍스트 본문만 효율적으로 추려낸다.

from langchain_community.document_loaders import WebBaseLoader

# 크롤링할 대상 웹사이트 지정 (bs4 패키지 필요)
web_loader = WebBaseLoader("[https://python.langchain.com/](https://python.langchain.com/)")
web_docs = web_loader.load()

print(f"웹 문서 로드 완료. 메타데이터: {web_docs[0].metadata}")
print(f"본문 내용 샘플: {web_docs[0].page_content[:150]}")

 

(3) JSON 및 메타데이터 사용자 정의 매핑 (JSONLoader)

구조화된 채팅 로그나 API 결과물을 분석할 때 활용된다. jq 라이브러리의 쿼리 스펙을 적용해 특정 필드만 낚아채고, 작성자 정보(sender)나 시간 정보(timestamp)를 문서의 메타데이터 필드로 유기적으로 이식할 수 있다.

from langchain_community.document_loaders import JSONLoader

# 데이터 레코드에서 특정 키값을 추출하여 메타데이터에 추가하는 매퍼 함수 정의
def custom_metadata_func(record: dict, metadata: dict) -> dict:
    metadata["sender"] = record.get("sender")
    metadata["timestamp"] = record.get("timestamp")
    return metadata

# JSONLoader 설정 (jq 쿼리로 메시지 배열의 content를 파싱)
json_loader = JSONLoader(
    file_path="./data/kakao_chat.json",
    jq_schema=".messages[]",
    content_key="content",
    metadata_func=custom_metadata_func
)
json_docs = json_loader.load()

print(f"JSON 파싱 완료. 총 메시지 수: {len(json_docs)}")
print(f"첫 번째 메시지 데이터: {json_docs[0].page_content}")
print(f"매핑된 메타데이터: {json_docs[0].metadata}")

 

(4) CSV 행 기반 처리 (CSVLoader)

행 단위가 하나의 독립적인 지식(예: 상품 목록, 팀 정보)일 때는 CSVLoader가 적합하다. 한 행씩을 고유한 메타데이터와 함께 별도의 문서 인스턴스로 분할해 준다.

from langchain_community.document_loaders.csv_loader import CSVLoader

# CSV 로더 가동 (구분자 지정 가능)
csv_loader = CSVLoader(file_path="./data/kbo_teams_2023.csv")
csv_docs = csv_loader.load()

print(f"CSV 로드 완료. 총 데이터 행 수: {len(csv_docs)}")
print(f"첫 행 데이터 구조: {csv_docs[0].page_content}")

 

 

3. 효과적인 텍스트 분할 전략 (Text Splitters)

수백 페이지에 달하는 책 한 권을 통째로 임베딩 모델에 넣거나 LLM에게 프롬프트로 전달하면 모델의 제한 컨텍스트(Token Limit)를 가볍게 넘겨버리거나, 엉뚱한 맥락을 참고하는 문제가 생긴다.

따라서 데이터를 작은 '청크(Chunk, 덩어리)'로 쪼개야 한다. 이때 각 청크 간 맥락의 단절을 막기 위해 오버랩(Overlap, 일부 중첩) 영역을 두는 것이 기술적 핵심이다.

(1) 가장 표준적인 재귀적 캐릭터 분할기 (RecursiveCharacterTextSplitter)

줄바꿈 문단 단위(\n\n), 줄 단위(\n), 단어 단위( ), 글자 단위(``) 구분자 리스트를 우선순위 순서대로 재귀적으로 타진하면서 사용자가 정의한 chunk_size 한도 내에서 문맥을 보존하며 영리하게 분할을 감행하는 가장 범용적인 분할기다.

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 재귀적 텍스트 분할기 정의
recursive_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,      # 한 청크당 최대 글자 수 (또는 토큰 수)
    chunk_overlap=200,    # 이전 청크와 겹치게 할 글자 수 (맥락 유지 장치)
    length_function=len,  # 길이 측정 기준 함수
    separators=["\n\n", "\n", " ", ""] # 분할 기준 우선순위 배열
)

recursive_chunks = recursive_splitter.split_documents(pdf_docs)
print(f"재귀적 분할을 적용한 결과 총 청크 수: {len(recursive_chunks)}")

 

(2) 정규식을 활용한 정밀 분할기 (CharacterTextSplitter)

단순 줄바꿈 외에 마침표, 물음표 등 특정 패턴 문자를 기준으로 커스텀 분리를 원할 때 강력하다.

from langchain_text_splitters import CharacterTextSplitter

# 문장 종결 기호 정규식을 분할 기준으로 삼는 설정
regex_splitter = CharacterTextSplitter(
    chunk_size=150,
    chunk_overlap=0,
    separator=r"(?<=[.!?])\s+", # 마침표, 느낌표, 물음표 뒤 공백을 기준으로 분할
    is_separator_regex=True
)

sentence_chunks = regex_splitter.split_documents(json_docs)
print(f"정규식 문장 분할 적용 결과 총 청크 수: {len(sentence_chunks)}")

 

(3) 혁신적인 의미론적 분할기 (SemanticChunker)

단순히 글자 수나 문장 기호로 자르는 전통적인 분할기와 달리, 문단 간의 임베딩 유사도를 실시간 분석하여 "말의 화제나 주제가 변하는 지점(Semantic Shift)"을 포착하여 지능적으로 청크 경계선을 긋는 가장 현대적인 방식이다.

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai.embeddings import OpenAIEmbeddings

# 임베딩 값의 급격한 변화 지점(임계치)을 감지하여 텍스트를 나누는 의미 기반 분할기
semantic_splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    breakpoint_threshold_type="gradient" # 경사 하강 변화량이 임계치를 초과하면 분할
)

# 다소 연산 리소스가 소모되므로 첫 페이지만 샘플로 확인
semantic_chunks = semantic_splitter.split_documents(pdf_docs[:1])
print(f"의미론적(Semantic) 지능형 분할 결과 생성된 청크 수: {len(semantic_chunks)}")

 

 

4. 임베딩 모델 선택 및 비교 (Embeddings)

임베딩(Embedding)은 컴퓨터가 이해할 수 없었던 비정형 자연어 청크를 고차원의 수치 벡터(Dense Vector)로 변환하는 핵심 단계다. 임베딩의 표현 한계가 곧 검색 정밀도로 연결된다.

성공적인 상용 RAG 시스템을 안착시키기 위해 현재 시장을 지배하는 3대 임베딩 파이프라인의 특징을 정확하게 조율하여 선택해야 한다.

임베딩 제공 모델	장점	단점	권장 사용 상황
OpenAI Embeddings (text-embedding-3-small)	- 압도적인 다국어 매핑 능력 - 긴 입력 토큰 한도 - 튜닝 불필요한 일관성	- 외부 API 호출로 인한 통신 비용 - 민감 보안 데이터의 외부 유출 우려	성능 최우선, 빠른 엔터프라이즈 프로토타이핑
Hugging Face (BAAI/bge-m3)	- 완전 무료 오픈소스 오픈 이식 - 다국어 및 한-영 매핑 우수 - 로컬 독립 구동 가능	- 로컬 인프라(GPU) 자원 직접 소모 - 초기 호스팅 개발 공수 증가	금융/의료 분야 폐쇄망, 자체 인프라 보유 팀
Ollama Embeddings (nomic-embed-text)	- 네트워크 없이 노트북 단독 실행 - 가볍고 민첩한 추론 속도	- 상대적인 다국어(한국어) 이해 능력 저하 - 제한적인 차원 표현력	개인 개발 환경, 모바일 오프라인 엣지 디바이스

 

 

5. 다국어 RAG 시스템 구축 실습 (Multilingual RAG Engine)

사용자가 영어로 *"Explain what Computer Vision researches"*라고 질문했을 때, 한글로 적힌 내부 문서("컴퓨터 비전은 디지털 이미지나 비디오를 이해하는 방법을 연구합니다.")를 정확히 찾아내어 한글 혹은 영어로 지연 없이 대답하는 똑똑한 검색 시스템을 구축해 보자.

우리는 검색 쿼리 언어를 감지하여 내부 지식 베이스의 타겟 언어로 실시간 변환하는 "언어 감지 및 자동번역 통합형 RAG"를 구현할 것이다.

[사용자 영어 질문] ──> [langdetect 언어 감지] ──> [DeepL API 자동 한글 번역]
                                                               │
                                                               ▼
[최종 영어/한글 응답] <── [gpt-4o-mini 생성] <── [ChromaDB 로컬 문서 검색]

 

(1) 실전 다국어 RAG 통합 소스 코드

import os
import deepl
from langdetect import detect
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from langchain.prompts import ChatPromptTemplate

# 1. 시뮬레이션용 소량 한글 지식 베이스 정의 및 벡터 DB 초기화
ko_knowledge_base = [
    "인공지능은 컴퓨터 과학의 한 분야로, 인간의 학습 및 추론 능력을 모델링합니다.",
    "머신러닝은 인공지능의 하위 연구 분야로, 데이터로부터 패턴을 기계가 스스로 학습하게 만듭니다.",
    "딥러닝은 머신러닝의 일종으로, 인간의 뇌 신경망을 본뜬 다층 인공 신경망을 활용합니다.",
    "자연어 처리는 컴퓨터가 인간의 언어를 분석하고, 이해하며, 고품질로 생성하도록 돕는 인공지능 기술입니다.",
    "컴퓨터 비전은 인공지능 분야 중 하나로, 컴퓨터가 디지털 이미지나 비디오를 정밀하게 분석하고 이해하게 합니다."
]

# 임베딩 및 Chroma Vector Store 셋업
embeddings_engine = OpenAIEmbeddings(model="text-embedding-3-small")
vector_db = Chroma.from_texts(texts=ko_knowledge_base, embedding=embeddings_engine)

# 2. DeepL 번역 미들웨어 연동 설정
# 환경 변수에 DEEPL_API_KEY가 적재되어 있어야 작동하며, 없으면 기본값 우회 처리됩니다.
DEEPL_KEY = os.getenv("DEEPL_API_KEY", "")
translator = deepl.Translator(DEEPL_KEY) if DEEPL_KEY else None

def translate_query_middleware(user_input: str, target_lang: str = "KO") -> tuple[str, str]:
    """
    입력된 질문의 원래 언어를 자동 감지하여 대상 타겟 언어로 실시간 교차 번역해 주는 수문장입니다.
    """
    if not translator:
        return user_input, "KO" # 번역 엔진 미세팅 시 입력값 그대로 우회
    
    try:
        # 사용자의 질문 언어 분석
        detected_lang = detect(user_input).upper()
        
        # 질문이 타겟 언어(한국어)와 다르면 강제 변환 처리
        if detected_lang != target_lang and detected_lang != "KO":
            translation_result = translator.translate_text(user_input, target_lang=target_lang)
            return str(translation_result), detected_lang
        
        return user_input, detected_lang
    except Exception as error:
        print(f"[번역 예외] {error}")
        return user_input, "KO"

# 3. RAG 수칙 프롬프트 템플릿 및 LLM 기동
rag_prompt_template = """Answer the question based only on the following context.
Do not make up any facts or use external knowledge.
If you don't know the answer, reply with "잘 모르겠습니다.".

[Context]
{context}

[Question] 
{question}

[Answer]
"""

prompt_interface = ChatPromptTemplate.from_template(rag_prompt_template)
llm_brain = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 문서 리스트 병합 헬퍼
def format_matching_docs(docs):
    return "\n\n".join([doc.page_content for doc in docs])

# 4. 엔터프라이즈 다국어 RAG 체인 호출 프로세스
def execute_multilingual_rag(user_question: str) -> str:
    # STEP 1: 입력 질문 다국어 처리 수문장 가동
    korean_translated_query, original_language = translate_query_middleware(user_question, target_lang="KO")
    print(f"\n[원문 언어] {original_language} -> [한글 번역 검색어] {korean_translated_query}")
    
    # STEP 2: ChromaDB 로컬 수집기 연동
    retriever_engine = vector_db.as_retriever(search_kwargs={"k": 2})
    matched_chunks = retriever_engine.invoke(korean_translated_query)
    context_material = format_matching_docs(matched_chunks)
    
    # STEP 3: LCEL 파이프라인 수동 바인딩 가동
    rag_chain = prompt_interface | llm_brain | StrOutputParser()
    korean_answer = rag_chain.invoke({
        "context": context_material,
        "question": korean_translated_query
    })
    
    # STEP 4: 질문한 원문 언어가 영어(EN)일 경우, 답변 또한 영어로 역번역 제공
    if original_language.startswith("EN") and translator:
        print("[영어 감지] 답변을 다시 원어인 영어로 번역하여 역출송합니다.")
        english_answer = translator.translate_text(korean_answer, target_lang="EN-US")
        return str(english_answer)
        
    return korean_answer

# [검증 데모 1] 외부 불량 정보 유도에 대한 방어벽 (환각 방지 검증)
print(execute_multilingual_rag("Who is the CEO of Apple?"))

# [검증 데모 2] 영어 질문을 통해 한글 지식을 매핑하여 정확한 영어 결과를 뽑아내기
print(execute_multilingual_rag("What is Computer Vision?"))

 

 

6. Gradio 기반 실시간 챗봇 인터페이스 구축

RAG 파이프라인의 완성은 사용자가 쉽고 편하게 접근할 수 있는 대화형 웹 인터페이스를 제공하는 것이다. 파이썬 환경에서 가장 빠르고 직관적으로 UI 앱을 가동시킬 수 있는 Gradio 라이브러리를 사용해 챗봇을 완성해 보자.

import gradio as gr
from langchain_core.messages import HumanMessage, AIMessage

# RAG 결과 도출 및 이전 대화 기록 흐름 제어 함수
def chat_with_rag_engine(message, history):
    # 랭체인이 문맥 히스토리를 이해할 수 있도록 구조적 포맷으로 누적 가공
    history_langchain_format = []
    for human_text, ai_text in history:
        history_langchain_format.append(HumanMessage(content=human_text))
        history_langchain_format.append(AIMessage(content=ai_text))
    
    # 1단계: 이번 질문에 대해 다국어 번역 결합형 RAG 체인을 수행하여 답안 확보
    rag_context_answer = execute_multilingual_rag(message)
    
    # 2단계: 과거 히스토리에 현재 참고 컨텍스트와 유저 발화를 최종적으로 덧붙입니다.
    history_langchain_format.append(AIMessage(content=f"[RAG 검색 참고 결과] {rag_context_answer}"))
    history_langchain_format.append(HumanMessage(content=message))
    
    # 3단계: 전체 문맥 대화 이력을 LLM에게 다시 전달하여 연속성 있는 챗봇 응답 완성
    final_output = llm_brain.invoke(history_langchain_format)
    return final_output.content

# Gradio 인터페이스 커스텀 선언 및 기동
chatbot_ui = gr.ChatInterface(
    fn=chat_with_rag_engine,
    title="🏢 엔터프라이즈 다국어 RAG 지식 챗봇",
    description="로컬 지식 문서 데이터를 지능적으로 탐색하고 실시간 다국어 번역을 연동하여 가장 안전한 답변을 내놓는 챗봇입니다."
)

if __name__ == "__main__":
    # 로컬 서버 호스트 포트 7860으로 웹 대시보드 구동
    chatbot_ui.launch(share=False)

 

7. 결론 및 마무리 (Conclusion)

본 가이드를 통해 우리는 문서 로드(Load) -> 분할(Split) -> 임베딩(Embed) -> 벡터 DB 저장(Store) -> 실시간 번역 결합 검색(Retrieve) -> 답변 생성(Generate)이라는 고도화된 RAG 시스템의 라이프사이클을 완벽하게 완수했다.

단순히 질문을 던지는 프롬프팅에서 한 걸음 더 나아가, 똑똑한 데이터 쪼개기와 번역 미들웨어를 거치면 어떤 다국어 환경의 엔터프라이즈 요구 사항에도 부합하는 고성능 지식 챗봇을 서비스할 수 있다.

이러한 지능형 데이터 처리 기법들은 점차 정밀한 사내 자동화 인프라로 자리 잡아 비즈니스의 지식 격차를 메워줄 강력한 핵심 무기가 될 것이다.