문서를 찾아 읽고 답하는 RAG

 1# hallucination.py - RAG 없이 사내 규정 질문
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse
 4
 5load_dotenv()
 6llm = ChatBedrockConverse(
 7    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
 8    region_name="us-east-1",
 9    temperature=0,
10)
11
12def get_text(message):
13    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
14    content = message.content
15    if isinstance(content, list):
16        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
17    return content
18
19
20print(get_text(llm.invoke("삼성SDS 국내 출장 식비 한도는 하루 얼마인가요? 정확한 금액만 답해주세요.")))
21print(get_text(llm.invoke("삼성SDS 재택근무는 주 며칠까지 가능한가요? 규정 조항 번호와 함께 알려주세요.")))

 1# 연차 규정
 2
 3본 문서는 교육용 더미 데이터입니다.
 4
 5## 부여 기준
 6연차 휴가는 연 15일이 부여됩니다. 입사 1년 미만은 매월 1일씩 최대 11일까지 받을 수 있습니다.
 73년 이상 근속자는 매 2년마다 1일이 더해져 최대 25일까지 받을 수 있습니다.
 8
 9## 이월
10미사용 연차는 5일까지 다음 해로 이월할 수 있습니다. 이월한 연차는 다음 해 6월 말까지 먼저 사용해야 합니다.
11
12## 신청 절차
13연차는 사용 5영업일 전까지 인사 포털에서 연차 신청서(양식 HR-103)로 신청합니다.
14부서장 승인 후 확정되며 긴급한 경우 사후 신청은 다음 영업일 오전까지 마쳐야 합니다.

 1# 재택근무 규정
 2
 3본 문서는 교육용 더미 데이터입니다.
 4
 5## 운영 기준
 6재택근무는 주 2일까지 사용할 수 있습니다. 팀 필수 출근일(수요일)에는 재택근무를 사용할 수 없습니다.
 7신청은 전주 목요일까지 인사 포털에서 재택근무 신청서(양식 RW-201)로 제출합니다.
 8
 9## 보안 수칙
10재택근무 중 사내망 접속은 VPN 사용이 필수입니다.
11개인 PC를 사용할 경우 보안 점검 프로그램을 설치하고 보안 서약서를 제출해야 합니다.
12사내 자료는 승인 없이 개인 저장소나 외부 서비스에 올릴 수 없습니다.
13
14## 근태
15재택근무 시작과 종료 시 협업 메신저에 상태를 등록합니다. 근무 시간은 사무실 근무와 동일하게 적용됩니다.

 1# 출장 규정
 2
 3본 문서는 교육용 더미 데이터입니다.
 4
 5## 신청과 승인
 6출장은 출발 3영업일 전까지 인사 포털에서 출장 신청서(양식 TR-102)로 신청하고 부서장 승인을 받아야 합니다.
 7항공권과 숙박은 출발 전에 사내 출장 시스템에서 미리 예약해야 합니다.
 8
 9## 경비 기준
10국내 출장 식비 한도는 하루 5만원, 해외 출장 식비 한도는 하루 8만원입니다.
11숙박비는 국내 1박 12만원, 해외 1박 20만원 한도입니다.
12
13## 정산
14출장비는 복귀 후 7영업일 안에 영수증을 첨부해 정산해야 합니다.
15법인카드 사용 내역은 매월 말일까지 경비 시스템에 등록합니다.

 1# loaders.py - 문서 로더
 2from pathlib import Path
 3
 4from langchain_community.document_loaders import TextLoader, DirectoryLoader
 5
 6# 실행 위치와 무관하게 이 파일 옆의 docs 폴더를 사용합니다
 7DOCS_DIR = Path(__file__).parent / "docs"
 8
 9# 1) 파일 하나 로드
10loader = TextLoader(str(DOCS_DIR / "leave_policy.md"), encoding="utf-8")
11docs = loader.load()
12print(f"문서 수: {len(docs)}")
13print(f"metadata: {docs[0].metadata}")
14print(f"본문 앞 80자: {docs[0].page_content[:80]}")
15
16# 2) 폴더 통째로 로드
17dir_loader = DirectoryLoader(
18    str(DOCS_DIR),
19    glob="**/*.md",
20    loader_cls=TextLoader,
21    loader_kwargs={"encoding": "utf-8"},
22)
23all_docs = dir_loader.load()
24print(f"\n전체 문서 수: {len(all_docs)}")
25for d in all_docs:
26    print(f"  - {d.metadata['source']} ({len(d.page_content)}자)")

 1# chunking.py - 청킹 설정 비교
 2import warnings
 3from pathlib import Path
 4
 5# langchain-community 지원 종료 안내 경고는 import 시점에 뜨므로 import 전에 꺼야 합니다
 6warnings.filterwarnings("ignore", message=".*langchain-community.*", category=DeprecationWarning)
 7
 8from langchain_community.document_loaders import TextLoader
 9from langchain_text_splitters import RecursiveCharacterTextSplitter
10
11# 실행 위치와 무관하게 이 파일 옆의 docs 폴더를 사용합니다
12DOCS_DIR = Path(__file__).parent / "docs"
13
14docs = TextLoader(str(DOCS_DIR / "leave_policy.md"), encoding="utf-8").load()
15
16for size, overlap in [(100, 0), (100, 50), (300, 50)]:
17    splitter = RecursiveCharacterTextSplitter(chunk_size=size, chunk_overlap=overlap)
18    chunks = splitter.split_documents(docs)
19    print(f"\n=== chunk_size={size}, overlap={overlap} -> {len(chunks)}개 청크 ===")
20    for i, c in enumerate(chunks):
21        print(f"[{i}] ({len(c.page_content)}자) {c.page_content[:40].replace(chr(10), ' ')}...")

 1# chunking_md.py - 마크다운 헤더 기준 분할
 2import langchain_community.document_loaders
 3from langchain_text_splitters import MarkdownHeaderTextSplitter
 4
 5docs = langchain_community.document_loaders.TextLoader("docs/leave_policy.md", encoding="utf-8").load()
 6
 7headers = [("#", "title"), ("##", "section")]
 8md_splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers)
 9md_chunks = md_splitter.split_text(docs[0].page_content)
10for c in md_chunks:
11    print(c.metadata, "->", c.page_content[:30])

 1# metadata.py - 메타데이터 부착 (뒤 실습에서 재사용)
 2from langchain_community.document_loaders import TextLoader
 3from langchain_text_splitters import RecursiveCharacterTextSplitter
 4
 5# 파일별 업무 메타데이터 (교육용 더미 데이터)
 6FILE_META = {
 7    "docs/leave_policy.md":       {"department": "인사팀", "updated_at": "2026-01-15"},
 8    "docs/remote_work_policy.md": {"department": "인사팀", "updated_at": "2026-03-02"},
 9    "docs/travel_policy.md":      {"department": "총무팀", "updated_at": "2025-11-20"},
10}
11
12
13def load_chunks():
14    """세 문서를 로드해 메타데이터를 붙이고 청크로 나눕니다."""
15    splitter = RecursiveCharacterTextSplitter(chunk_size=300, chunk_overlap=50)
16    all_chunks = []
17    for path, meta in FILE_META.items():
18        docs = TextLoader(path, encoding="utf-8").load()
19        for d in docs:
20            d.metadata.update(meta)      # source 위에 업무 정보 추가
21        all_chunks.extend(splitter.split_documents(docs))
22    return all_chunks
23
24
25if __name__ == "__main__":
26    chunks = load_chunks()
27    for c in chunks[:3]:
28        print(c.metadata)
29    print(f"\n총 청크 수: {len(chunks)}")

 1# embedding.py - 임베딩 확인
 2from dotenv import load_dotenv
 3from langchain_aws import BedrockEmbeddings
 4
 5load_dotenv()
 6embeddings = BedrockEmbeddings(
 7    model_id="amazon.titan-embed-text-v2:0",
 8    region_name="us-east-1",
 9)
10
11v = embeddings.embed_query("오늘은 며칠인가요?")
12print(f"벡터 차원: {len(v)}")
13print(f"앞 5개 값: {[round(x, 4) for x in v[:]]}")

 1# similarity.py - 코사인 유사도 체험
 2import math
 3from dotenv import load_dotenv
 4from langchain_aws import BedrockEmbeddings
 5
 6load_dotenv()
 7embeddings = BedrockEmbeddings(
 8    model_id="amazon.titan-embed-text-v2:0",
 9    region_name="us-east-1",
10)
11
12
13def cosine(a, b):
14    dot = sum(x * y for x, y in zip(a, b))
15    na = math.sqrt(sum(x * x for x in a))
16    nb = math.sqrt(sum(x * x for x in b))
17    return dot / (na * nb)
18
19
20pairs = [
21    ("연차 며칠 받을 수 있어?", "월차 휴가 일수 알려줘"),   # 같은 의미, 다른 단어
22    ("연차 며칠 받을 수 있어?", "VPN 설정 방법"),           # 무관한 주제
23    ("Python", "파이썬"),                                    # 다른 언어, 같은 의미
24    ("출장비 정산", "경비 처리 기한"),                       # 비슷한 업무 영역
25]
26
27for a, b in pairs:
28    va, vb = embeddings.embed_query(a), embeddings.embed_query(b)
29    print(f"'{a}' vs '{b}' -> {cosine(va, vb):.3f}")

 1# vectorstore.py - 벡터 저장소 구축과 디스크 저장
 2from dotenv import load_dotenv
 3from langchain_aws import BedrockEmbeddings
 4from langchain_chroma import Chroma
 5from metadata import load_chunks
 6
 7load_dotenv()
 8embeddings = BedrockEmbeddings(
 9    model_id="amazon.titan-embed-text-v2:0",
10    region_name="us-east-1",
11)
12
13# Chroma(
14#     collection_name="sds_policies",
15#     embedding_function=embeddings,
16#     persist_directory="./chroma_db",
17# ).delete_collection()
18
19
20db = Chroma.from_documents(
21    documents=load_chunks(), 
22    embedding=embeddings,
23    collection_name="sds_policies",
24    persist_directory="./chroma_db",   # 이 폴더에 저장됩니다 (6일차까지 사용)
25)
26print(f"적재 완료: {db._collection.count()}개 청크")
27
28# 의미 검색 확인 (점수 포함)
29for q in ["월차 며칠 받을 수 있어?", "집에서 일해도 되나요?", "법카 등록 기한"]:
30    hits = db.similarity_search_with_score(q, k=2)
31    print(f"\n질문: {q}")
32    for doc, score in hits:
33        print(f"  [{doc.metadata['source']}] (거리 {score:.3f}) {doc.page_content[:40]}...")
34
35# 메타데이터 필터 검색
36hits = db.similarity_search("경비 정산 규정", k=2, filter={"department": "총무팀"})
37print(f"\n총무팀 문서로 한정 검색: {[h.metadata['source'] for h in hits]}")

 1# vectorstore.py - 벡터 저장소 구축과 디스크 저장
 2from dotenv import load_dotenv
 3from langchain_aws import BedrockEmbeddings
 4from langchain_chroma import Chroma
 5from metadata import load_chunks
 6
 7load_dotenv()
 8embeddings = BedrockEmbeddings(
 9    model_id="amazon.titan-embed-text-v2:0",
10    region_name="us-east-1",
11)
12
13# Chroma(
14#     collection_name="sds_policies",
15#     embedding_function=embeddings,
16#     persist_directory="./chroma_db",
17# ).delete_collection()
18
19db = Chroma.from_documents(
20    documents=load_chunks(),
21    embedding=embeddings,
22    collection_name="sds_policies",
23    persist_directory="./chroma_db",   # 이 폴더에 저장됩니다 (6일차까지 사용)
24)
25print(f"적재 완료: {db._collection.count()}개 청크")
26
27# 의미 검색 확인 (점수 포함)
28for q in ["월차 며칠 받을 수 있어?", "집에서 일해도 되나요?", "법카 등록 기한"]:
29    hits = db.similarity_search_with_score(q, k=2)
30    print(f"\n질문: {q}")
31    for doc, score in hits:
32        print(f"  [{doc.metadata['source']}] (거리 {score:.3f}) {doc.page_content[:40]}...")
33
34# 메타데이터 필터 검색
35hits = db.similarity_search("경비 정산 규정", k=2, filter={"department": "총무팀"})
36print(f"\n총무팀 문서로 한정 검색: {[h.metadata['source'] for h in hits]}")

 1# retriever.py - Retriever 생성과 k값 조정
 2from dotenv import load_dotenv
 3from langchain_aws import BedrockEmbeddings
 4from langchain_chroma import Chroma
 5
 6load_dotenv()
 7embeddings = BedrockEmbeddings(
 8    model_id="amazon.titan-embed-text-v2:0",
 9    region_name="us-east-1",
10)
11db = Chroma(
12    collection_name="sds_policies",
13    embedding_function=embeddings,
14    persist_directory="./chroma_db",
15)
16
17retriever = db.as_retriever(search_kwargs={"k": 3})
18docs = retriever.invoke("연차 신청은 어떻게 하나요?")
19for d in docs:
20    print(f"[{d.metadata['source']}] {d.page_content[:50]}...")
21
22# k값에 따른 차이 관찰
23for k in [1, 3, 5]:
24    r = db.as_retriever(search_kwargs={"k": k})
25    hits = r.invoke("연차 이월 규정")
26    print(f"k={k}: {[h.metadata['source'] for h in hits]}")
27
28# MMR - 넓은 질문에서 여러 문서를 고르게
29mmr = db.as_retriever(search_type="mmr", search_kwargs={"k": 3, "fetch_k": 6})
30print("MMR:", [h.metadata["source"] for h in mmr.invoke("회사 규정 전반")])

 1# rag_chain.py - 기본 RAG 체인
 2from dotenv import load_dotenv
 3from langchain_aws import ChatBedrockConverse, BedrockEmbeddings
 4from langchain_chroma import Chroma
 5from langchain_core.prompts import ChatPromptTemplate
 6from langchain_core.output_parsers import StrOutputParser
 7from langchain_core.runnables import RunnablePassthrough
 8
 9load_dotenv()
10
11llm = ChatBedrockConverse(
12    model="us.anthropic.claude-sonnet-4-5-20250929-v1:0",
13    region_name="us-east-1",
14    temperature=0,
15)
16embeddings = BedrockEmbeddings(
17    model_id="amazon.titan-embed-text-v2:0",
18    region_name="us-east-1",
19)
20
21db = Chroma(
22    collection_name="sds_policies",
23    embedding_function=embeddings,
24    persist_directory="./chroma_db",
25)
26retriever = db.as_retriever(search_kwargs={"k": 3})
27
28prompt = ChatPromptTemplate.from_messages([
29    ("system",
30     "너는 삼성SDS 사내 정책 안내 어시스턴트야. 아래 사내 문서만 참고해 답해. "
31     "문서에 없는 내용은 '해당 내용은 사내 문서에서 찾을 수 없습니다'라고 답해. 추측하지 마. "
32     "답변 끝에 참고한 문서를 '(출처: 파일명)' 형식으로 붙여.\n\n"
33     "--- 사내 문서 ---\n{context}"),
34    ("human", "{question}"),
35])
36
37
38def format_docs(docs):
39    return "\n\n---\n\n".join(
40        f"[{d.metadata.get('source', '?')}] {d.page_content}" for d in docs
41    )
42
43
44rag_chain = (
45    {"context": retriever | format_docs, "question": RunnablePassthrough()}
46    | prompt
47    | llm
48    | StrOutputParser()
49)
50
51if __name__ == "__main__":
52    print(rag_chain.invoke("회사에서 교육을 보냈는데 환경이 x같아? 누구에게 항의해야해?"))

 1# verify.py - RAG 유무 비교 검증
 2from rag_chain import rag_chain, llm
 3
 4
 5def get_text(message):
 6    """ChatBedrockConverse는 content를 블록 리스트로 주기도 하므로 텍스트만 모아 반환합니다."""
 7    content = message.content
 8    if isinstance(content, list):
 9        return "".join(block.get("text", "") for block in content if isinstance(block, dict))
10    return content
11
12
13questions = [
14    "국내 출장 식비 한도는 하루 얼마인가요?",   # 문서에 답 있음
15    "재택근무는 주 며칠까지 가능한가요?",       # 문서에 답 있음
16    "육아휴직은 얼마나 쓸 수 있나요?",          # 문서에 답 없음 -> 모른다고 해야 정답
17]
18
19for q in questions:
20    print("=" * 60)
21    print(f"질문: {q}")
22    print(f"\n[RAG 없음] {get_text(llm.invoke(q))[:100]}")
23    print(f"\n[RAG 적용] {rag_chain.invoke(q)}")

 1# verify_sources.py - 근거 문서를 함께 반환
 2from langchain_core.runnables import RunnableParallel
 3from rag_chain import rag_chain, retriever
 4
 5rag_with_sources = RunnableParallel(
 6    answer=rag_chain,
 7    sources=retriever | (lambda docs: sorted({d.metadata["source"] for d in docs})),
 8)
 9result = rag_with_sources.invoke("물류플랫폼팀 박도윤입니다. 해외 출장 숙박비 한도가 얼마인가요?")
10print(result["answer"])
11print("근거 문서:", result["sources"])

 1# failures.py - 검색 실패 재현
 2from rag_chain import db
 3
 4fail_queries = [
 5    "출장 신청서 TR-102 어디서 내?",   # 유형 1: 고유 기호
 6    "3년차 연차 가산 규정",            # 유형 2: 청크 경계에 걸릴 수 있음
 7    "그거 신청 어떻게 해?",            # 유형 4: 모호한 질문
 8]
 9for q in fail_queries:
10    hits = db.similarity_search_with_score(q, k=2)
11    print(f"\n질문: {q}")
12    for doc, score in hits:
13        print(f"  (거리 {score:.3f}) [{doc.metadata['source']}] {doc.page_content[:40]}")

 1# hybrid.py - 하이브리드 검색 (BM25 구성은 사전 검증된 완성 코드)
 2from dotenv import load_dotenv
 3from langchain_community.retrievers import BM25Retriever
 4from langchain_classic.retrievers import EnsembleRetriever
 5from kiwipiepy import Kiwi
 6from metadata import load_chunks
 7from rag_chain import db
 8
 9load_dotenv()
10kiwi = Kiwi()
11
12
13def korean_tokenizer(text: str) -> list[str]:
14    """한국어 형태소 단위 토크나이저 (사전 검증본). 명사, 어간, 숫자, 외래어만 남깁니다."""
15    keep = {"NNG", "NNP", "NNB", "NR", "NP", "VV", "VA", "SL", "SN"}
16    return [t.form for t in kiwi.tokenize(text) if t.tag in keep]
17
18
19# 1) BM25 검색기 - 벡터 DB가 아니라 청크 리스트에서 직접 만듭니다
20bm25 = BM25Retriever.from_documents(load_chunks(), preprocess_func=korean_tokenizer)
21bm25.k = 3
22
23# 2) 벡터 검색기
24vector = db.as_retriever(search_kwargs={"k": 3})
25
26# 3) 하이브리드 - 가중치 조합 [BM25 비중, 벡터 비중]
27hybrid = EnsembleRetriever(retrievers=[bm25, vector], weights=[0.3, 0.7])
28
29queries = [
30    "월차 며칠 받아?",        # 의미형 질문 -> 벡터가 유리
31    "TR-102 양식 규정",       # 키워드형 질문 -> BM25가 유리
32]
33for q in queries:
34    print("=" * 55)
35    print(f"질문: {q}")
36    for name, r in [("BM25만", bm25), ("벡터만", vector), ("하이브리드", hybrid)]:
37        hits = r.invoke(q)
38        srcs = [f'{h.metadata["source"].split("/")[-1]} | {h.page_content[:14].replace(chr(10), " ")}' for h in hits[:3]]
39        print(f"  {name}: {srcs}")
40
41        # hybrid_weights.py - 가중치 조정 실험 (hybrid.py에 이어서)
42for w in [0.2, 0.5, 0.8]:
43    h = EnsembleRetriever(retrievers=[bm25, vector], weights=[w, 1 - w])
44    top = h.invoke("월차 며칠 받아?")[0].metadata["source"].split("/")[-1]
45    print(f"BM25 {w:.1f} / 벡터 {1 - w:.1f} -> 1위: {top}")

 1[
 2  {
 3    "question": "연차 이월은 며칠까지 가능한가요?",
 4    "expected_source": "docs/leave_policy.md",
 5    "type": "fact"
 6  },
 7  {
 8    "question": "월차 얼마나 받을 수 있어?",
 9    "expected_source": "docs/leave_policy.md",
10    "type": "paraphrase"
11  },
12  {
13    "question": "집에서 일하려면 어떻게 신청해?",
14    "expected_source": "docs/remote_work_policy.md",
15    "type": "paraphrase"
16  },
17  {
18    "question": "해외 출장 식비 한도는 하루 얼마인가요?",
19    "expected_source": "docs/travel_policy.md",
20    "type": "fact"
21  },
22  {
23    "question": "육아휴직은 얼마나 쓸 수 있나요?",
24    "expected_source": null,
25    "type": "no_answer"
26  }
27]

 1# eval.py - 평가 세트 자동 채점
 2import json
 3from rag_chain import rag_chain, retriever
 4
 5PASS_THRESHOLD = 4   # 2일차 성공 기준: 5문항 중 4문항 이상 정답
 6
 7with open("eval_set.json", encoding="utf-8") as f:
 8    eval_set = json.load(f)
 9
10passed = 0
11for item in eval_set:
12    hits = retriever.invoke(item["question"])
13    sources = sorted({h.metadata["source"] for h in hits})
14
15    if item["expected_source"] is None:
16        # no_answer 유형: 체인이 '찾을 수 없다'고 거절하는지로 채점
17        answer = rag_chain.invoke(item["question"])
18        ok = "찾을 수 없" in answer
19    else:
20        # 기대 출처가 검색 결과 상위 k개 안에 있는지로 채점
21        ok = item["expected_source"] in sources
22
23    passed += ok
24    mark = "통과" if ok else "실패"
25    print(f"[{mark}] ({item['type']}) {item['question']}")
26    print(f"       기대: {item['expected_source']} / 실제 검색: {sources}")
27
28print(f"\n정답 {passed}/{len(eval_set)}문항")
29if passed >= PASS_THRESHOLD:
30    print(f"[성공 기준 통과] {PASS_THRESHOLD}문항 이상 정답입니다.")
31else:
32    print(f"[성공 기준 미달] {PASS_THRESHOLD}문항 이상 필요합니다. {PASS_THRESHOLD - passed}문항 더 맞혀야 합니다.")

 1# rerank_demo.py - 재랭킹 전후 순위 비교
 2from sentence_transformers import CrossEncoder
 3from hybrid import hybrid
 4
 5# 다국어 재랭킹 모델 (한국어 지원, 최초 실행 시 다운로드)
 6reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")
 7
 8query = "연차 이월 규정"
 9candidates = [d.page_content for d in hybrid.invoke(query)]  # 1차 검색 후보
10
11# 재랭킹 전: 하이브리드 검색이 돌려준 순서 그대로
12print("=== 재랭킹 전 (하이브리드 순위) ===")
13for i, c in enumerate(candidates, 1):
14    print(f"{i}위  {c[:50].replace(chr(10), ' ')}...")
15
16scores = reranker.predict([(query, c) for c in candidates])
17ranked = sorted(zip(scores, candidates), reverse=True)
18print("\n=== 재랭킹 후 (상위 3개) ===")
19for i, (s, c) in enumerate(ranked[:3], 1):
20    before = candidates.index(c) + 1
21    print(f"{i}위 (전 {before}위)  {s:.3f}  {c[:50].replace(chr(10), ' ')}...")