SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day44_Vector DB와 LLM을 결합한 RAG 아키텍처

RAG Retrieval-Augmented Generation 검색 증강 생성

답이있는 문서를 먼저 검색해 내용을 근거로 답한다. 

데이터로드 - 청킹 - 임베딩 - 백터db에 보관 - 사용자질문 임베딩 - 벡터db에서 유사조각 k개 - 근거로 LLM 으로 답변 생성.

https://standout.tistory.com/1876

RAG(Retrieval-Augmented Generation)와 LLM의 Hallucination(환각): 외부 문서를 검색한 후 검색 결과를 바탕으

 

 

 

 

 

RecursiveCharacterTextSplitter 청킹도구 

langchain은 RecursiveCharacterTextSplitter 로 청킹한다. Recursive 재귀적이라는 이름을 단락 - 문장 - 단어 순으로 자연스러운 경계를 우선해 자른다. 문장중간에서 뚝 끊기는 일이 적은 똑똑한 자르기.

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)

chunks = text_splitter.split_documents(documents)

https://standout.tistory.com/1903

RecursiveCharacterTextSplitter 청킹도구, 단락 - 문장 - 단어 순으로 자연스러운 경계를 우선해 자른다.

 

 

 

 

 

예시 프로젝트 rag_console_project를 분석해보자. 

 

 

common.py

import os, pathlib, dotenv

env를 읽고 변수설정, require_key 함수설정 및 api 설정

get_genai_client 클라이언트 만들어 객체 return

get_chat key와 객체로 langhain_google_gemai에서 chat 객체 생성.

get_embeddings 임베딩객체 만들어 리턴.

__main__ common파일을 직접 실행했을때만 상세 정보 txt 출력.

# -*- coding: utf-8 -*-
"""모든 콘솔 실습에서 함께 사용하는 공통 설정 모듈입니다."""

# 운영체제 환경변수를 읽기 위해 os 모듈을 가져옵니다.
import os
# 운영체제와 관계없이 경로를 안전하게 다루기 위해 pathlib 모듈을 가져옵니다.
import pathlib
# 프로젝트 루트의 .env 파일을 읽기 위해 load_dotenv 함수를 가져옵니다.
from dotenv import load_dotenv

# 현재 common.py 파일의 부모(code)의 부모를 프로젝트 최상위 폴더로 지정합니다.
ROOT = pathlib.Path(__file__).resolve().parent.parent
# 프로젝트의 실습 데이터가 저장될 data 폴더 경로를 생성합니다.
DATA = ROOT / "data"
# PDF 등 문서 파일이 저장될 data/docs 폴더 경로를 생성합니다.
DOCS = DATA / "docs"

# 프로젝트 루트에 있는 .env 파일의 환경변수를 현재 프로세스로 불러옵니다.
load_dotenv(ROOT / ".env")

# Gemini 채팅 모델명을 환경변수에서 읽고, 없으면 기본 모델명을 사용합니다.
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
# Gemini 임베딩 모델명을 환경변수에서 읽고, 없으면 기본 모델명을 사용합니다.
GEMINI_EMBED_MODEL = os.getenv("GEMINI_EMBED_MODEL", "models/gemini-embedding-001")
# OpenAI 채팅 모델명을 환경변수에서 읽고, 없으면 비교적 저비용 모델명을 사용합니다.
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")


def require_key(name: str) -> str:
    """필수 API 키가 없거나 예시 문자열이면 친절한 오류와 함께 실행을 종료합니다."""
    # 전달받은 이름에 해당하는 환경변수 값을 읽습니다.
    value = os.getenv(name)
    # 값이 비어 있거나 예시 문자열로 시작하면 실제 키가 입력되지 않은 것으로 판단합니다.
    if not value or value.startswith("여기에"):
        # 설정 방법을 보여 주는 SystemExit 예외를 발생시켜 현재 메뉴 실행만 중단합니다.
        raise SystemExit(
            f"[설정 필요] {name}가 .env 파일에 없습니다.\n"
            f"1) .env.example 파일을 .env 이름으로 복사합니다.\n"
            f"2) .env 파일에서 {name} 값을 실제 API 키로 변경합니다."
        )
    # 검증을 통과한 API 키 문자열을 호출한 쪽에 반환합니다.
    return value


def get_genai_client():
    """Google Gen AI의 원시 SDK 클라이언트를 생성하여 반환합니다."""
    # 함수가 실제 호출될 때만 google-genai 패키지를 가져와 선택적 의존성을 처리합니다.
    from google import genai
    # 검증된 Google API 키를 사용해 Gemini 클라이언트 객체를 생성합니다.
    return genai.Client(api_key=require_key("GOOGLE_API_KEY"))


def get_chat(provider: str = "gemini", temperature: float = 0.0):
    """provider 값에 맞는 LangChain 채팅 모델 객체를 생성하여 반환합니다."""
    # 사용자가 Gemini 공급자를 선택한 경우의 처리입니다.
    if provider == "gemini":
        # 모델 객체를 만들기 전에 Google API 키 존재 여부를 검증합니다.
        require_key("GOOGLE_API_KEY")
        # Gemini용 LangChain 채팅 모델 클래스를 지연 import합니다.
        from langchain_google_genai import ChatGoogleGenerativeAI
        # 공통 모델명과 temperature를 적용한 Gemini 채팅 객체를 반환합니다.
        return ChatGoogleGenerativeAI(model=GEMINI_MODEL, temperature=temperature)
    # 사용자가 OpenAI 공급자를 선택한 경우의 처리입니다.
    if provider == "openai":
        # 모델 객체를 만들기 전에 OpenAI API 키 존재 여부를 검증합니다.
        require_key("OPENAI_API_KEY")
        # OpenAI용 LangChain 채팅 모델 클래스를 지연 import합니다.
        from langchain_openai import ChatOpenAI
        # 공통 모델명과 temperature를 적용한 OpenAI 채팅 객체를 반환합니다.
        return ChatOpenAI(model=OPENAI_MODEL, temperature=temperature)
    # 지원하지 않는 공급자 문자열이 들어오면 명확한 ValueError를 발생시킵니다.
    raise ValueError(f"알 수 없는 provider: {provider}")


def get_embeddings(provider: str = "gemini"):
    """provider 값에 맞는 LangChain 임베딩 객체를 생성하여 반환합니다."""
    # Gemini 임베딩을 선택한 경우의 처리입니다.
    if provider == "gemini":
        # Google API 키가 설정되어 있는지 먼저 검증합니다.
        require_key("GOOGLE_API_KEY")
        # Gemini 임베딩 클래스를 실제 사용 시점에 가져옵니다.
        from langchain_google_genai import GoogleGenerativeAIEmbeddings
        # 저장 공간과 실습 속도를 고려해 출력 차원을 768로 제한한 객체를 반환합니다.
        return GoogleGenerativeAIEmbeddings(model=GEMINI_EMBED_MODEL, output_dimensionality=768)
    # OpenAI 임베딩을 선택한 경우의 처리입니다.
    if provider == "openai":
        # OpenAI API 키가 설정되어 있는지 먼저 검증합니다.
        require_key("OPENAI_API_KEY")
        # OpenAI 임베딩 클래스를 실제 사용 시점에 가져옵니다.
        from langchain_openai import OpenAIEmbeddings
        # 비용과 성능 균형이 좋은 text-embedding-3-small 모델 객체를 반환합니다.
        return OpenAIEmbeddings(model="text-embedding-3-small")
    # 지원하지 않는 공급자 문자열이면 오류를 발생시킵니다.
    raise ValueError(f"알 수 없는 provider: {provider}")


# 이 파일을 직접 실행했을 때만 아래 진단 정보를 출력합니다.
if __name__ == "__main__":
    # 계산된 프로젝트 루트 경로를 출력합니다.
    print("ROOT:", ROOT)
    # data 폴더 경로와 실제 존재 여부를 출력합니다.
    print("DATA:", DATA, "(존재:", DATA.exists(), ")")
    # 현재 선택된 Gemini 모델명을 출력합니다.
    print("GEMINI_MODEL:", GEMINI_MODEL)
    # API 키의 실제 값은 노출하지 않고 설정 여부만 True/False로 출력합니다.
    print("키 설정 상태 - GOOGLE_API_KEY:", bool(os.getenv("GOOGLE_API_KEY")),
          "/ OPENAI_API_KEY:", bool(os.getenv("OPENAI_API_KEY")))

 

 

 

 

document_service.py

import path,

langchain_communoty에서 csvloader, pypdfloader, textloader. 가져오기

langchain_core.documents 에서 document 가져오기

langchain_text_aplitters 에서 MarkdownHeaderTextAplitter, RecursiveCharacterTextSplitter 가져오기

common에서 필요함수 가져오기

 - CSVLoader CSV 파일을 읽어 각 행(row)을 Document 객체로 변환하는 문서 로더

 -  PyPDFLoader PDF 파일의 텍스트를 추출하여 페이지별 Document 객체로 변환하는 문서 로더

 -  TextLoader TXT 등 일반 텍스트 파일을 읽어 Document 객체로 변환하는 문서 로더

 -  Document 문서 내용(page_content)과 메타데이터(metadata)를 함께 저장하는 LangChain의 기본 문서 객체

 -  MarkdownHeaderTextSplitter Markdown의 헤더(#, ##, ### 등)를 기준으로 문서를 의미 단위로 분할하는 텍스트 분할기

 -  RecursiveCharacterTextSplitter 문맥을 최대한 유지하면서 지정한 크기와 겹침(overlap)을 기준으로 텍스트를 재귀적으로 분할하는 텍스트 분할기

ensure_file() 파일존재여부 확인, load_pdf() PyPDFLoader로 pdf 로드해 return하기

split_documents()  RecursiveCharacterTextSplitter를 사용해 청킹해 return하기

load_csv() , load_text() 마찬가지 로드

load_markdown() 마크다운 매칭해 가져와 aplitter하기

load_all_sources pdf csv text markdown 모두 합쳐 return하기

# 문서 로드와 청킹 기능을 한곳에 모은 서비스 모듈입니다.
from pathlib import Path
from langchain_community.document_loaders import CSVLoader, PyPDFLoader, TextLoader
from langchain_core.documents import Document
from langchain_text_splitters import MarkdownHeaderTextSplitter, RecursiveCharacterTextSplitter
from common import DATA, DOCS


def ensure_file(path: Path) -> Path:
    """실습 파일이 실제로 존재하는지 검사한 뒤 경로를 반환합니다."""
    if not path.exists():
        raise FileNotFoundError(f"실습 파일을 찾을 수 없습니다: {path}")
    return path


def load_pdf(file_name: str = "환불교환정책.pdf") -> list[Document]:
    """PDF를 페이지 단위 Document 목록으로 읽습니다."""
    pdf_path = ensure_file(DOCS / file_name)
    loader = PyPDFLoader(str(pdf_path))
    return loader.load()


def split_documents(docs: list[Document], chunk_size: int = 500, chunk_overlap: int = 50) -> list[Document]:
    """Document 목록을 지정한 크기와 겹침으로 청킹합니다."""
    if chunk_size <= 0:
        raise ValueError("chunk_size는 1 이상이어야 합니다.")
    if chunk_overlap < 0 or chunk_overlap >= chunk_size:
        raise ValueError("chunk_overlap은 0 이상이고 chunk_size보다 작아야 합니다.")
    splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=chunk_overlap)
    return splitter.split_documents(docs)


def load_csv(file_name: str = "faq.csv") -> list[Document]:
    """CSV의 각 행을 하나의 Document로 읽습니다."""
    csv_path = ensure_file(DATA / file_name)
    return CSVLoader(file_path=str(csv_path), encoding="utf-8-sig").load()


def load_text(file_name: str = "notice.txt") -> list[Document]:
    """일반 텍스트 파일 전체를 하나의 Document로 읽습니다."""
    text_path = ensure_file(DATA / file_name)
    return TextLoader(str(text_path), encoding="utf-8").load()


def load_markdown(file_name: str = "policy.md") -> list[Document]:
    """마크다운을 1단계와 2단계 헤더 기준으로 분리합니다."""
    markdown_path = ensure_file(DATA / file_name)
    markdown_text = markdown_path.read_text(encoding="utf-8")
    splitter = MarkdownHeaderTextSplitter(headers_to_split_on=[("#", "대분류"), ("##", "소분류")])
    return splitter.split_text(markdown_text)


def load_all_sources() -> list[Document]:
    """PDF, CSV, TXT, Markdown 결과를 하나의 Document 목록으로 합칩니다."""
    return load_pdf() + load_csv() + load_text() + load_markdown()

 

 

 

llm_service.py

import langchain_core.documents의 document가져오기

common에서 필요함수 get_chat가져오기

format_context() for문을 돌려 파일을 읽어 없으면 대체문구를 사용한다. pdf일때 존재할수있는 page 번호를 메타데이터로 get.

페이지 번호가 있으면 사용자가 읽는 실질적인 페이지 1부터.

각 근거 사이를 빈줄 두개로 구분해 프롬프트 문자열로 return.

answer_with_context() result가 없으면 안내문구, 검색결과를 context에 할당, prompt 설정 get_chat() model.invoke(), getattr() content 를 읽거나 response을 읽어 문자열로 반환

# 검색된 문서 근거를 OpenAI 또는 Gemini에 전달하는 RAG 응답 서비스입니다.
# LangChain의 표준 Document 타입을 타입 힌트에 사용하기 위해 가져옵니다.
from langchain_core.documents import Document
# 제공된 common.py의 공통 채팅 모델 생성 함수를 가져옵니다.
from common import get_chat


def format_context(results: list[tuple[Document, float]]) -> str:
    """검색 결과를 LLM 프롬프트에 넣기 쉬운 문자열로 변환합니다."""
    # 문서별 근거 문자열을 저장할 빈 리스트를 생성합니다.
    sections: list[str] = []
    # 검색 순위 번호와 Document, 유사도 점수를 순서대로 처리합니다.
    for number, (doc, score) in enumerate(results, start=1):
        # metadata에서 원본 파일 경로를 읽고 없으면 대체 문구를 사용합니다.
        source = doc.metadata.get("source", "알 수 없는 출처")
        # PDF 문서일 때 존재할 수 있는 0부터 시작하는 페이지 번호를 읽습니다.
        page = doc.metadata.get("page")
        # 정수 페이지 번호가 있으면 사용자가 읽는 1부터 시작하는 번호로 변환합니다.
        page_text = f", page={page + 1}" if isinstance(page, int) else ""
        # 근거 번호, 출처, 점수, 본문을 하나의 문자열로 묶어 리스트에 추가합니다.
        sections.append(
            f"[근거 {number}] source={source}{page_text}, score={score:.4f}\n"
            f"{doc.page_content}"
        )
    # 각 근거 사이를 빈 줄 두 개로 구분하여 하나의 프롬프트 문자열로 반환합니다.
    return "\n\n".join(sections)


def answer_with_context(
    provider: str,
    question: str,
    results: list[tuple[Document, float]],
) -> str:
    """선택한 LLM에 근거 제한 프롬프트를 전달하고 텍스트 답변을 반환합니다."""
    # 검색 결과가 없으면 API를 호출하지 않고 안내 문장을 즉시 반환합니다.
    if not results:
        return "검색된 근거가 없어 답변을 생성할 수 없습니다."
    # 검색 결과를 모델 입력용 근거 문자열로 변환합니다.
    context = format_context(results)
    # 환각을 줄이기 위해 제공된 근거만 사용하도록 명시한 프롬프트를 작성합니다.
    prompt = f"""당신은 사내 문서 기반 질의응답 도우미입니다.
아래 [검색 근거]에 포함된 사실만 사용하여 한국어로 답하세요.
근거에 없는 내용은 추측하지 말고 '제공된 문서에서 확인할 수 없습니다'라고 답하세요.
답변 마지막에는 사용한 근거 번호를 표시하세요.

[사용자 질문]
{question}

[검색 근거]
{context}
"""
    # common.py를 통해 OpenAI 또는 Gemini LangChain 채팅 모델을 생성합니다.
    model = get_chat(provider=provider, temperature=0.0)
    # 완성한 프롬프트를 모델에 전달하여 응답 객체를 받습니다.
    response = model.invoke(prompt)
    # LangChain 메시지 객체라면 content를 읽고, 아니면 응답 자체를 사용합니다.
    content = getattr(response, "content", response)
    # 콘솔에서 안전하게 출력할 수 있도록 최종 결과를 문자열로 변환해 반환합니다.
    return str(content)

 

 

 

 

menu_actions.py

common에서 data, docs를 가져온다. 

features에서 service를 가져온다.

pause() 사용자의 enter키 기다리기 안내문구.

print_documents() 도큐먼트 보여주기

run_pdf_load() pdf 보여주기

run_chunking() pdf를 로드한 뒤 split_documents 수행해 pritn

run_keyword_search() 키워드를 검색해 spli_documnets 검색결과 출력

run_chunk_tuning() pdf 로드해 키워드 입력받아 청킹 설정 정의해 for문으로 결과계싼

split_documents() 한뒤 각 평균을 계산, 정렬해 출력

run_various_loaders 각 파일들 확인

ask_question() 질문받기. 각 docs를 합쳐 청킹, search_documents해 질문과 result return

run_torch_search() result에서 유사도가 높은 순서대로 출력

run_llm_rag() for문을 돌려 근거번호, 내용을 출력하고 answer_width_context() 후 answer return

run_path_check() 경로확인

# 콘솔 메뉴에서 호출할 각 실행 기능을 정의합니다.
# 공통 모듈에서 계산한 데이터 폴더 경로를 가져옵니다.
from common import DATA, DOCS
# 문서 로드와 청킹에 필요한 서비스 함수를 가져옵니다.
from features.document_service import (
    load_all_sources,
    load_csv,
    load_markdown,
    load_pdf,
    load_text,
    split_documents,
)
# 검색 근거를 LLM에 전달하는 함수를 가져옵니다.
from features.llm_service import answer_with_context
# PyTorch 기반 로컬 검색 함수를 가져옵니다.
from features.local_vector_service import search_documents


def pause() -> None:
    """결과를 읽은 뒤 Enter를 누르면 메뉴로 돌아가게 합니다."""
    # 콘솔 결과가 바로 사라지지 않도록 사용자의 Enter 입력을 기다립니다.
    input("\nEnter 키를 누르면 메뉴로 돌아갑니다...")


def print_documents(title: str, docs, limit: int = 3) -> None:
    """Document의 본문과 metadata를 일정 개수만 보기 좋게 출력합니다."""
    # 출력 결과의 제목을 표시합니다.
    print(f"\n=== {title} ===")
    # 전체 Document 개수를 출력합니다.
    print(f"Document 개수: {len(docs)}")
    # 지정한 최대 개수만큼 Document를 순서대로 출력합니다.
    for index, doc in enumerate(docs[:limit], start=1):
        # 여러 줄 본문을 한 줄로 바꾸고 최대 250자까지만 미리보기로 만듭니다.
        preview = doc.page_content.replace("\n", " ")[:250]
        # 현재 문서의 번호와 본문 일부를 출력합니다.
        print(f"\n[{index}] page_content: {preview}")
        # 현재 문서의 출처와 페이지 등의 metadata를 출력합니다.
        print(f"    metadata: {doc.metadata}")


def run_pdf_load() -> None:
    """PDF 페이지별 로드 결과를 확인합니다."""
    # 기본 PDF를 페이지 단위 Document 목록으로 읽습니다.
    docs = load_pdf()
    # 페이지 본문 일부와 metadata를 출력합니다.
    print_documents("PDF 페이지 로드 결과", docs)
    # 사용자가 결과를 확인할 시간을 제공합니다.
    pause()


def run_chunking() -> None:
    """PDF를 기본 설정으로 청킹하고 metadata 보존을 확인합니다."""
    # 원본 PDF를 먼저 로드합니다.
    docs = load_pdf()
    # HTML 예제의 기본값인 500자와 50자 겹침으로 문서를 나눕니다.
    chunks = split_documents(docs, chunk_size=500, chunk_overlap=50)
    # 최대 5개 청크의 본문과 metadata를 출력합니다.
    print_documents("PDF 청킹 결과", chunks, limit=5)
    # 사용자가 결과를 확인할 시간을 제공합니다.
    pause()


def run_keyword_search() -> None:
    """동일 단어 포함 여부만 검사하는 단순 키워드 검색을 실행합니다."""
    # 빈 입력이면 HTML 예제와 같은 '환불'을 기본 키워드로 사용합니다.
    keyword = input("검색할 키워드 [기본값: 환불]: ").strip() or "환불"
    # PDF를 로드한 뒤 기본 설정으로 청킹합니다.
    chunks = split_documents(load_pdf())
    # 사용자가 입력한 문자열이 본문에 그대로 포함된 청크만 선택합니다.
    hits = [chunk for chunk in chunks if keyword in chunk.page_content]
    # 검색된 청크를 최대 10개까지 출력합니다.
    print_documents(f"'{keyword}' 키워드 검색 결과", hits, limit=10)
    # 사용자가 결과를 확인할 시간을 제공합니다.
    pause()


def run_chunk_tuning() -> None:
    """여러 chunk_size와 overlap 조합을 비교 표로 출력합니다."""
    # 비교 대상이 되는 원본 PDF를 로드합니다.
    docs = load_pdf()
    # 키워드 포함 청크 수를 비교할 기준 단어를 입력받습니다.
    keyword = input("비교 기준 키워드 [기본값: 환불]: ").strip() or "환불"
    # HTML 예제에 제시된 세 가지 청킹 설정을 정의합니다.
    configs = [(300, 0), (500, 50), (1000, 100)]
    # 비교 표의 헤더를 정렬하여 출력합니다.
    print(f"\n{'설정(size, overlap)':<24}{'청크 수':>10}{'평균 길이':>12}{'키워드 포함':>14}")
    # 표 헤더와 데이터 행을 구분하는 선을 출력합니다.
    print("-" * 60)
    # 각 설정을 차례대로 적용해 결과를 계산합니다.
    for size, overlap in configs:
        # 현재 설정으로 원본 문서를 청킹합니다.
        chunks = split_documents(docs, chunk_size=size, chunk_overlap=overlap)
        # 생성된 전체 청크 수를 계산합니다.
        count = len(chunks)
        # 청크가 있을 때 각 본문 길이의 평균을 계산합니다.
        average = sum(len(chunk.page_content) for chunk in chunks) / count if count else 0.0
        # 기준 키워드를 포함한 청크의 수를 계산합니다.
        hits = sum(1 for chunk in chunks if keyword in chunk.page_content)
        # 현재 설정과 계산 결과를 한 행으로 정렬해 출력합니다.
        print(f"({size}, {overlap}){count:>20}{average:>12.1f}{hits:>14}")
    # 사용자가 결과를 확인할 시간을 제공합니다.
    pause()


def run_various_loaders() -> None:
    """CSV, TXT, Markdown가 같은 Document 구조로 변환되는지 확인합니다."""
    # CSV의 한 행이 하나의 Document로 변환된 결과를 출력합니다.
    print_documents("CSVLoader 결과", load_csv())
    # 텍스트 파일 전체가 Document로 변환된 결과를 출력합니다.
    print_documents("TextLoader 결과", load_text())
    # 마크다운 헤더별로 분리된 Document 결과를 출력합니다.
    print_documents("MarkdownHeaderTextSplitter 결과", load_markdown())
    # 사용자가 결과를 확인할 시간을 제공합니다.
    pause()


def ask_question() -> tuple[str, list]:
    """전체 문서를 청킹하고 PyTorch 검색을 수행해 질문과 검색 결과를 반환합니다."""
    # 사용자에게 문서 기반 질문을 입력받습니다.
    question = input("문서에 질문할 내용을 입력하세요: ").strip()
    # 빈 질문은 검색할 수 없으므로 ValueError를 발생시킵니다.
    if not question:
        raise ValueError("질문은 비워 둘 수 없습니다.")
    # PDF, CSV, TXT, Markdown 문서를 하나의 목록으로 합칩니다.
    all_docs = load_all_sources()
    # 모든 문서를 동일한 청킹 파이프라인으로 나눕니다.
    chunks = split_documents(all_docs, chunk_size=500, chunk_overlap=50)
    # PyTorch 코사인 유사도를 이용해 상위 3개 근거를 검색합니다.
    results = search_documents(question, chunks, top_k=3)
    # 질문과 검색 결과를 호출한 메뉴 함수에 반환합니다.
    return question, results


def run_torch_search() -> None:
    """API 없이 PyTorch 코사인 유사도 검색 결과만 출력합니다."""
    # 질문을 입력받고 로컬 벡터 검색을 수행합니다.
    _question, results = ask_question()
    # 검색 결과 영역의 제목을 출력합니다.
    print("\n=== PyTorch 로컬 벡터 검색 결과 ===")
    # 유사도가 높은 순서대로 검색 결과를 출력합니다.
    for index, (doc, score) in enumerate(results, start=1):
        # 현재 문서의 순위와 유사도 점수를 출력합니다.
        print(f"\n[{index}] 유사도: {score:.4f}")
        # 현재 문서의 출처 metadata를 출력합니다.
        print("출처:", doc.metadata)
        # 현재 문서 본문을 최대 500자까지 출력합니다.
        print(doc.page_content[:500])
    # 사용자가 결과를 확인할 시간을 제공합니다.
    pause()


def run_llm_rag(provider: str) -> None:
    """PyTorch 검색 결과를 선택한 LLM에 전달하여 근거 기반 답변을 생성합니다."""
    # 질문을 입력받고 상위 문서 근거를 검색합니다.
    question, results = ask_question()
    # LLM 호출 전에 실제 전달될 검색 근거를 출력합니다.
    print("\n=== 검색된 근거 ===")
    # 검색 순위에 따라 각 근거를 출력합니다.
    for index, (doc, score) in enumerate(results, start=1):
        # 근거 번호, 유사도, metadata를 출력합니다.
        print(f"[{index}] score={score:.4f}, metadata={doc.metadata}")
        # 근거 본문 일부를 출력합니다.
        print(doc.page_content[:250], "\n")
    # 선택한 공급자의 LLM에 질문과 검색 근거를 전달합니다.
    answer = answer_with_context(provider, question, results)
    # 어떤 모델 공급자의 결과인지 제목으로 표시합니다.
    print(f"\n=== {provider.upper()} 근거 기반 답변 ===")
    # 최종 생성 답변을 출력합니다.
    print(answer)
    # 사용자가 결과를 확인할 시간을 제공합니다.
    pause()


def run_path_check() -> None:
    """공통 모듈이 계산한 데이터 경로와 실습 파일 존재 여부를 점검합니다."""
    # common.py가 계산한 data 폴더의 절대 경로를 출력합니다.
    print("\nDATA 경로:", DATA)
    # common.py가 계산한 docs 폴더의 절대 경로를 출력합니다.
    print("DOCS 경로:", DOCS)
    # 반드시 필요한 네 개의 실습 파일 경로를 순서대로 검사합니다.
    for path in [
        DOCS / "환불교환정책.pdf",
        DATA / "faq.csv",
        DATA / "notice.txt",
        DATA / "policy.md",
    ]:
        # 각 파일이 존재하는지 사람이 읽기 쉬운 문자열로 출력합니다.
        print(f"- {path.name}: {'존재' if path.exists() else '없음'}")
    # 사용자가 결과를 확인할 시간을 제공합니다.
    pause()

 

 

 

local_vector_service.py

import hashlib, re, torch, langchain_core.documents에서 document 가져오기

tokenize() 토큰 추출 한글, 영문, 숫자로 이루어진 2글자 이상. lower()

token_index 토큰인덱스. 

embed_text() torch.zeros를 만들어 l2정규화

search_documents()  embed한뒤 torch.stack document를 쌓아 torch.mv 유사도 점수 계산, max가져오기 value, indices들을 return하기 langchain이 지원하는 invoke 즉 질문인메딩, 유사도계산, top k 선택, document 반환등을 직접 구현한다.

# PyTorch만으로 간단한 로컬 문서 벡터와 코사인 유사도 검색을 구현합니다.
import hashlib
import re
import torch
from langchain_core.documents import Document

VECTOR_SIZE = 512


def tokenize(text: str) -> list[str]:
    """한글, 영문, 숫자로 이루어진 2글자 이상의 토큰을 추출합니다."""
    return re.findall(r"[가-힣A-Za-z0-9]{2,}", text.lower())


def token_index(token: str) -> int:
    """문자열 토큰을 항상 같은 벡터 위치로 변환합니다."""
    digest = hashlib.sha256(token.encode("utf-8")).digest()
    return int.from_bytes(digest[:4], byteorder="big") % VECTOR_SIZE


def embed_text(text: str) -> torch.Tensor:
    """해시 기반 단어 빈도 벡터를 만들고 L2 정규화합니다."""
    vector = torch.zeros(VECTOR_SIZE, dtype=torch.float32)
    for token in tokenize(text):
        vector[token_index(token)] += 1.0
    norm = torch.linalg.vector_norm(vector)
    if norm > 0:
        vector = vector / norm
    return vector


def search_documents(query: str, docs: list[Document], top_k: int = 3) -> list[tuple[Document, float]]:
    """질문과 각 문서 벡터의 코사인 유사도를 계산해 상위 문서를 반환합니다."""
    if not docs:
        return []
    query_vector = embed_text(query)
    document_matrix = torch.stack([embed_text(doc.page_content) for doc in docs])
    scores = torch.mv(document_matrix, query_vector)
    actual_k = min(max(top_k, 1), len(docs))
    values, indices = torch.topk(scores, k=actual_k)
    return [(docs[index], float(score)) for score, index in zip(values.tolist(), indices.tolist())]

 

main.py

import menu를 가져온다. 

print_menu() 메뉴를 보여준다 각메뉴에 맞게 actions한다.

# 콘솔 앱의 시작점이며 숫자 메뉴를 반복 실행합니다.
from features.menu_actions import (
    run_chunk_tuning,
    run_chunking,
    run_keyword_search,
    run_llm_rag,
    run_path_check,
    run_pdf_load,
    run_torch_search,
    run_various_loaders,
)


def print_menu() -> None:
    """사용자가 실행할 수 있는 실습 기능만 메뉴로 출력합니다."""
    # 이전 실행 결과와 새 메뉴를 시각적으로 구분하기 위해 빈 줄과 구분선을 출력합니다.
    print("\n" + "=" * 64)
    # 콘솔 프로그램의 제목을 출력합니다.
    print("RAG 문서 로드·청킹·검색 콘솔 실습")
    # 제목 아래쪽 구분선을 출력합니다.
    print("=" * 64)
    # 아래 메뉴는 HTML 설명을 보여 주지 않고 실제 실행 기능만 제공합니다.
    print("1. 공통 경로 및 실습 파일 확인")
    print("2. PDF 페이지 단위 로드")
    print("3. PDF 청킹 및 metadata 확인")
    print("4. 청크 키워드 검색")
    print("5. chunk_size / chunk_overlap 비교")
    print("6. CSV · TXT · Markdown 로더 확인")
    print("7. PyTorch 로컬 벡터 검색")
    print("8. OpenAI API 근거 기반 답변")
    print("9. Gemini API 근거 기반 답변")
    print("0. 프로그램 종료")


def main() -> None:
    """메뉴 번호를 입력받아 해당 함수를 호출하고 오류를 사용자에게 안내합니다."""
    # 메뉴 문자열과 실행 함수를 딕셔너리로 연결하여 긴 조건문을 줄입니다.
    actions = {
        "1": run_path_check,
        "2": run_pdf_load,
        "3": run_chunking,
        "4": run_keyword_search,
        "5": run_chunk_tuning,
        "6": run_various_loaders,
        "7": run_torch_search,
        "8": lambda: run_llm_rag("openai"),
        "9": lambda: run_llm_rag("gemini"),
    }
    # 사용자가 종료 메뉴를 선택할 때까지 메뉴 입력을 반복합니다.
    while True:
        # 현재 선택 가능한 메뉴를 출력합니다.
        print_menu()
        # 사용자 입력 앞뒤 공백을 제거하여 메뉴 번호를 얻습니다.
        choice = input("메뉴 번호를 선택하세요: ").strip()
        # 0번을 선택하면 반복문을 종료합니다.
        if choice == "0":
            # 프로그램 종료 메시지를 출력합니다.
            print("프로그램을 종료합니다.")
            # while 반복문을 빠져나갑니다.
            break
        # 입력한 번호와 연결된 실행 함수를 찾습니다.
        action = actions.get(choice)
        # 등록되지 않은 메뉴 번호이면 안내 후 다시 입력받습니다.
        if action is None:
            print("0~9 사이의 올바른 메뉴 번호를 입력하세요.")
            continue
        try:
            # 선택한 기능을 실행합니다.
            action()
        except (FileNotFoundError, ValueError, SystemExit) as error:
            # 파일 누락, 잘못된 입력, API 키 누락처럼 예상 가능한 오류를 출력합니다.
            print(f"\n[실행 오류] {error}")
        except Exception as error:
            # 그 밖의 예외가 발생해도 앱 전체가 종료되지 않도록 오류 정보를 출력합니다.
            print(f"\n[예상하지 못한 오류] {type(error).__name__}: {error}")


# 이 파일을 직접 실행한 경우에만 main 함수를 호출합니다.
if __name__ == "__main__":
    # 콘솔 메뉴 프로그램을 시작합니다.
    main()