Study/LangChain
1.1. create_retrieval_chain 완전 가이드
01-1_create_retrieval_chain_완전_가이드.ipynb
0.16MB
01-1_create_retrieval_chain_완전_가이드.md
0.04MB
1. create_retrieval_chain 개요
- `create_retrieval_chain`은 LangChain이 제공하는 RAG 파이프라인 구축 헬퍼 함수이다. 검색(Retrieval)과 생성(Generation)을 하나의 체인으로 결합하여, 사용자 질의에 대해 문서를 검색하고 LLM이 답변을 생성하는 전체 흐름을 단 몇 줄의 코드로 구성할 수 있다.
1.1. 왜 create_retrieval_chain이 필요한가
| 문제 | create_retrieval_chain의 해결 방식 |
| RAG 파이프라인 구성이 복잡 | 검색과 생성을 하나의 함수로 결합 |
| 검색 결과를 프롬프트에 수동 삽입 필요 | 자동으로 검색 결과를 context에 주입 |
| 출처 추적이 어려움 | context 키로 검색된 문서를 자동 반환 |
| 체인 연결 로직 반복 작성 | 표준화된 입출력 인터페이스 제공 |
| 레거시 API(RetrievalQA) 지원 종료 | 공식 권장 대체 API |
1.2. RetrievalQA vs create_retrieval_chain
- LangChain의 RAG 체인은 세대별로 발전해 왔다. `RetrievalQA`는 더 이상 권장되지 않으며, `create_retrieval_chain`이 공식 대체 API이다.
| 기준 | RetrievalQA (레거시) | create_retrieval_chain (현재) |
| 상태 | Deprecated (사용 중단 권고) | 공식 권장 |
| 입력 키 | query | input |
| 출력 키 | result, source_documents | answer, context |
| 내부 구조 | 블랙박스 (커스텀 어려움) | 모듈화 (문서 체인 분리) |
| LCEL 호환 | 부분적 | 완전 호환 |
| 스트리밍 | 제한적 | 완전 지원 |
| 프롬프트 제어 | chain_type_kwargs로 간접 전달 | 직접 커스텀 프롬프트 사용 |
from langchain_openai import ChatOpenAI, OpenAIEmbeddings from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate
# LLM 초기화 - 질의에 대한 응답을 생성하는 언어 모델 llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 임베딩 → 벡터 스토어 → retriever 구성
# retriever: 사용자 질의와 유사한 문서를 벡터 DB에서 검색하는 검색기
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = Chroma(collection_name="example", embedding_function=embeddings)
retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
# [레거시] RetrievalQA - 더 이상 사용하지 말 것
# RetrievalQA.from_chain_type(): LLM과 retriever를 결합하여 질의응답 체인을 생성 (단일 블랙박스 구조)
from langchain.chains import RetrievalQA
qa = RetrievalQA.from_chain_type(llm=llm, retriever=retriever)
result = qa.invoke({"query": "질문"})
# result["result"], result["source_documents"]
# [현재 권장] create_retrieval_chain
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
# 프롬프트 정의 - LLM에게 검색된 문서(context)를 기반으로 답변하도록 지시
prompt = ChatPromptTemplate.from_template(
"다음 문맥을 참고하여 질문에 답하세요:\n\n{context}\n\n질문: {input}"
)
# create_stuff_documents_chain(): 검색된 문서들을 하나의 프롬프트에 합쳐서 LLM에 전달하는 체인 생성
combine_docs_chain = create_stuff_documents_chain(llm, prompt)
# create_retrieval_chain(): retriever와 문서 결합 체인을 연결하여 전체 RAG 파이프라인 구성
chain = create_retrieval_chain(retriever, combine_docs_chain)
result = chain.invoke({"input": "질문"})
# result["answer"], result["context"]
- 핵심 포인트: 신규 프로젝트에서는 반드시 `create_retrieval_chain`을 사용한다. 기존 `RetrievalQA` 코드가 있다면 마이그레이션을 권장한다.
1.3. create_retrieval_chain의 아키텍처
[사용자 질의]
│
├──→ [Retriever] ──→ [검색된 문서들 (context)]
│ │
│ ↓
└──→ [combine_docs_chain] ←── [context + input]
│
↓
[LLM 생성]
│
↓
[최종 응답 반환]
{"input": "질문",
"context": [문서들],
"answer": "답변"}1단계: 사용자 질의 수신
사용자가 자연어 질문(input)을 입력하면, create_retrieval_chain은 이 질의를 두 갈래로 동시에 전달한다. 하나는
Retriever로, 다른 하나는 combine_docs_chain으로 향한다. 원본 질의가 그대로 보존되어 최종 응답에도 포함된다는 점이
핵심이다.
2단계: Retriever — 관련 문서 검색
Retriever는 사용자 질의를 받아 벡터 데이터베이스(Vector Store)에서 의미적으로 유사한 문서를 검색한다. 내부적으로는
질의를 임베딩 벡터로 변환한 뒤, 사전에 인덱싱된 문서 벡터들과 유사도 비교(cosine similarity 등)를 수행한다. 검색 결과는
관련도가 높은 상위 k개의 문서(Document 객체 리스트)로 반환되며, 이것이 곧 context가 된다.
3단계: combine_docs_chain — 문서와 질의 결합
검색된 문서들(context)과 원본 사용자 질의(input)가 combine_docs_chain에 합류한다. 이 체인은 일반적으로
create_stuff_documents_chain으로 구성되며, 다음 작업을 수행한다:
- 검색된 문서들의 page_content를 하나의 텍스트로 병합(stuff)
- PromptTemplate에 context와 input을 삽입하여 완성된 프롬프트를 생성
- 이 프롬프트를 LLM에 전달
4단계: LLM 생성
완성된 프롬프트를 받은 LLM(예: ChatOpenAI, ChatUpstage 등)이 검색된 문서의 내용을 근거로 답변을 생성한다. LLM은 자신의
사전 학습 지식이 아닌, context로 제공된 문서를 기반으로 응답하므로 환각(hallucination)을 줄이고 사실에 기반한 답변을
만들어낸다.
5단계: 최종 응답 반환
create_retrieval_chain은 단순한 텍스트가 아닌 구조화된 딕셔너리를 반환한다:
┌─────────┬─────────────────────────────────────────┐
│ 키 │ 설명 │
├─────────┼─────────────────────────────────────────┤
│ input │ 사용자가 입력한 원본 질문 │
├─────────┼─────────────────────────────────────────┤
│ context │ Retriever가 검색한 Document 객체 리스트 │
├─────────┼─────────────────────────────────────────┤
│ answer │ LLM이 생성한 최종 답변 문자열 │
└─────────┴─────────────────────────────────────────┘
이 구조 덕분에 답변의 근거가 된 원본 문서를 추적할 수 있고, 디버깅이나 평가(Evaluation) 시 어떤 문서가 참조되었는지
확인할 수 있다.
- `create_retrieval_chain`은 내부적으로 두 단계로 동작한다:
1) 검색 단계: `retriever`가 사용자 질의(`input`)를 받아 관련 문서를 검색하여 `context` 키에 저장한다.
2) 생성 단계: `combine_docs_chain`이 `input`과 `context`를 받아 LLM으로 최종 답변(`answer`)을 생성한다.
2. 핵심 구성 요소 상세
2.1. create_retrieval_chain 함수 시그니처
# [함수 시그니처 참고 - 실행 코드가 아닙니다]
from langchain.chains import create_retrieval_chain
create_retrieval_chain(
retriever, # 검색기: BaseRetriever 또는 Runnable
combine_docs_chain, # 문서 결합 체인: Runnable (보통 create_stuff_documents_chain)
) → Runnable
# ──────────────────────────────────────────────
# 입력 스키마 (invoke 시 전달하는 딕셔너리)
# ──────────────────────────────────────────────
# {"input": str}
#
# ──────────────────────────────────────────────
# 출력 스키마 (invoke 결과로 반환되는 딕셔너리)
# ──────────────────────────────────────────────
# {
# "input": str, # 사용자 원본 질문 (그대로 전달)
# "context": List[Document], # retriever가 검색한 문서 리스트
# "answer": str # LLM이 생성한 최종 답변
# } 1. 매개변수 타입 상세
매개변수: retriever
타입: BaseRetriever | Runnable[dict, List[Document]]
설명: 질의를 받아 Document 리스트를 반환하는 객체. VectorStore.as_retriever()로 생성하거나, 커스텀 Runnable을 전달할 수 있다.
────────────────────────────────────────
매개변수: combine_docs_chain
타입: Runnable[dict, str]
설명: {"context": List[Document], "input": str}을 입력받아 str(답변)을 반환하는 체인. 일반적으로 create_stuff_documents_chain()으로 생성한다.
2. 반환값 타입
# 반환되는 Runnable의 타입 시그니처
Runnable[ dict[str, str], dict[str, Any] ]
# ↑ 입력 ↑ 출력
# {"input": "질문"} {"input": ..., "context": ..., "answer": ...}
반환값은 Runnable 객체이므로, .invoke(), .stream(), .batch() 등 LangChain Expression Language(LCEL)의 모든 실행 메서드를 지원한다.
3. 호출 방식 정리
# [참고용 예시 - 실행 코드가 아닙니다]
# 기본 호출
result = chain.invoke({"input": "질문 내용"})
# 스트리밍 호출 (토큰 단위 출력)
for chunk in chain.stream({"input": "질문 내용"}):
print(chunk)
# 배치 호출 (여러 질문 동시 처리)
results = chain.batch([
{"input": "첫 번째 질문"},
{"input": "두 번째 질문"},
])
# 비동기 호출
result = await chain.ainvoke({"input": "질문 내용"})
4. 내부 동작 흐름 (의사 코드)
# [내부 동작 의사 코드 - 실행 코드가 아닙니다]
def create_retrieval_chain(retriever, combine_docs_chain):
def chain(input_dict):
# Step 1: 원본 질의 보존
user_input = input_dict["input"]
# Step 2: retriever로 관련 문서 검색
context = retriever.invoke(user_input)
# Step 3: combine_docs_chain에 context + input 전달
answer = combine_docs_chain.invoke({
"input": user_input,
"context": context,
})
# Step 4: 구조화된 결과 반환
return {
"input": user_input,
"context": context,
"answer": answer,
}
return chain # Runnable 객체로 반환
5. combine_docs_chain이 기대하는 입력 키
combine_docs_chain(보통 create_stuff_documents_chain)은 내부적으로 다음 두 키를 사용한다:
┌─────────┬────────────────┬─────────────────────────────────────┐
│ 키 │ 타입 │ 공급 주체 │
├─────────┼────────────────┼─────────────────────────────────────┤
│ input │ str │ 사용자 입력에서 그대로 전달 │
├─────────┼────────────────┼─────────────────────────────────────┤
│ context │ List[Document] │ retriever의 검색 결과에서 자동 주입 │
└─────────┴────────────────┴─────────────────────────────────────┘
따라서 create_stuff_documents_chain의 PromptTemplate에는 반드시 {context}와 {input} 변수가 포함되어야 한다.
2.1.1. 파라미터 상세
| 파라미터 | 타입 | 설명 |
| retriever | BaseRetriever 또는 Runnable | 사용자 질의를 받아 관련 문서를 검색하는 객체. db.as_retriever()로 생성하거나 커스텀 Runnable을 사용 |
| combine_docs_chain | Runnable | 검색된 문서와 질의를 결합하여 LLM에 전달하는 체인. 보통 create_stuff_documents_chain()으로 생성 |
2.1.2. 반환값 구조
| 키 | 타입 | 설명 |
| input | str | 원본 사용자 질의 (그대로 전달) |
| context | List[Document] | retriever가 검색한 문서 리스트 |
| answer | str | LLM이 생성한 최종 답변 |
2.2. create_stuff_documents_chain 상세
- `create_stuff_documents_chain`은 검색된 문서들을 하나의 프롬프트에 "채워넣는(stuff)" 방식으로 결합하는 체인이다.
# [함수 시그니처 참고 - 실행 코드가 아닙니다]
from langchain.chains.combine_documents import create_stuff_documents_chain
create_stuff_documents_chain(
llm, # LLM 모델
prompt, # ChatPromptTemplate (반드시 {context} 변수 포함)
output_parser=None, # 출력 파서 (기본: StrOutputParser)
document_variable_name="context", # 문서가 삽입될 변수명
document_separator="\n\n", # 문서 간 구분자
) → Runnable
# ──────────────────────────────────────────────
# 입력 스키마
# ──────────────────────────────────────────────
# {
# "context": List[Document], # 검색된 문서 리스트
# "input": str, # 사용자 질문 (프롬프트에 {input}이 있을 경우)
# ... # 프롬프트에 정의된 기타 변수
# }
#
# ──────────────────────────────────────────────
# 출력 스키마
# ──────────────────────────────────────────────
# str (기본 StrOutputParser 사용 시)
# Any (커스텀 output_parser 사용 시 해당 파서의 반환 타입)내부 동작 흐름
[입력]
{"context": [Doc1, Doc2, Doc3], "input": "사용자 질문"}
│
▼
┌─────────────────────────────────────────────┐
│ Step 1. 문서 텍스트 추출 및 병합 │
│ │
│ Doc1.page_content = "세금 신고 기한은..." │
│ Doc2.page_content = "부가가치세란..." │
│ Doc3.page_content = "종합소득세 계산..." │
│ │
│ ↓ document_separator로 결합 │
│ │
│ context_str = """ │
│ 세금 신고 기한은... │
│ │
│ 부가가치세란... │
│ │
│ 종합소득세 계산... │
│ """ │
└─────────────────────────┬───────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ Step 2. 프롬프트 템플릿에 변수 삽입 │
│ │
│ prompt.format( │
│ context = context_str, ← 병합된 문서 │
│ input = "사용자 질문", ← 원본 질의 │
│ ) │
│ │
│ 결과 예시: │
│ ┌─────────────────────────────────────┐ │
│ │ System: 다음 문서를 참고하여 │ │
│ │ 질문에 답하세요. │ │
│ │ │ │
│ │ 세금 신고 기한은... │ │
│ │ │ │
│ │ 부가가치세란... │ │
│ │ │ │
│ │ 종합소득세 계산... │ │
│ │ │ │
│ │ Human: 사용자 질문 │ │
│ └─────────────────────────────────────┘ │
└─────────────────────────┬───────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ Step 3. LLM 호출 │
│ │
│ llm.invoke(완성된_프롬프트) │
│ → AIMessage(content="답변 텍스트...") │
└─────────────────────────┬───────────────────┘
│
▼
┌─────────────────────────────────────────────┐
│ Step 4. output_parser 적용 │
│ │
│ StrOutputParser().parse(AIMessage) │
│ → "답변 텍스트..." (str) │
└─────────────────────────┬───────────────────┘
│
▼
[출력: str]
2.2.1. 프롬프트 요구사항
- `create_stuff_documents_chain`에 전달하는 프롬프트는 반드시 `{context}` 변수를 포함해야 한다. 이 변수에 검색된 문서의 `page_content`가 자동으로 삽입된다.
from langchain_core.prompts import ChatPromptTemplate
# 기본 프롬프트 템플릿
prompt = ChatPromptTemplate.from_template(
"""다음 컨텍스트를 기반으로 질문에 답변하세요.
컨텍스트:
{context}
질문: {input}
답변:"""
)
- 주의: `{context}`는 `create_stuff_documents_chain`이 내부적으로 문서들의 `page_content`를 결합하여 삽입하는 예약 변수명이다. `document_variable_name` 파라미터로 변경할 수 있지만, 관례적으로 `context`를 사용한다.
2.3. 문서 결합 전략 비교
- `create_stuff_documents_chain`은 "Stuff" 전략을 구현한다. LangChain은 다양한 문서 결합 전략을 제공한다.
| 전략 | 설명 | LLM 호출수 | 장점 | 단점 | 적합한 경우 |
| Stuff | 모든 문서를 하나의 프롬프트에 삽입 | 1회 | 구현 간단, 빠른 속도 | 토큰 제한 초과 가능 | 문서 합계가 컨텍스트 내 (가장 보편적) |
| Map-Reduce | 각 문서별 처리 후 결과 종합 | N+1회 | 대량 문서 처리 가능 | 속도 느림, 비용 높음 | 대량 문서 요약 |
| Refine | 순차적으로 답변 정제 | N회 | 정밀한 답변 | 순서 의존적, 느림 | 장문 종합 분석 |
| Map-Rerank | 각 문서별 답변+점수 후 최고 선택 | N회 | 최적 답변 선택 | 비용 높음 | 단일 최적 답변 필요 |
- 실무 권장: 대부분의 RAG 시나리오에서는 Stuff 전략이 충분하다. GPT-4o의 128K 토큰 컨텍스트를 감안하면, 일반적인 검색 결과(3~5개 청크, 각 1000~2000자)는 Stuff 방식으로 처리 가능하다. 문서 합계가 토큰 제한을 초과하는 경우에만 Map-Reduce를 고려한다.
3. 기본 사용법: 단계별 구축
3.1. 전체 파이프라인 흐름
[1. 환경 설정] → [2. 문서 로딩] → [3. 청킹] → [4. 임베딩 + 벡터 DB]
↓
[7. 질의 및 응답] ← [6. RAG 체인 구성] ← [5. 검색기 생성]
3.2. Step 1: 환경 설정
import os
from dotenv import load_dotenv, find_dotenv
load_dotenv(find_dotenv(), override=True)
assert os.environ.get("OPENAI_API_KEY", "").startswith("sk-"), \
"OPENAI_API_KEY가 설정되지 않았습니다."
3.3. Step 2: 문서 로딩
# ──────────────────────────────────────────────
# 1. 문서 로더 임포트
# ────────────────────────────────────────────── # LangChain 커뮤니티 패키지에서 .docx 파일을 읽기 위한 로더를 가져온다.
# 내부적으로 docx2txt 라이브러리를 사용하여 Word 문서의 텍스트를 추출한다.
from langchain_community.document_loaders import Docx2txtLoader
# ──────────────────────────────────────────────
# 2. 문서 로딩
# ──────────────────────────────────────────────
# Docx2txtLoader에 .docx 파일 경로를 전달하여 로더 인스턴스를 생성한다.
loader = Docx2txtLoader("../../tax_docs/tax_with_markdown.docx")
# .load()를 호출하면 파일 내용을 읽어 Document 객체 리스트로 반환한다.
# 각 Document 객체는 page_content(텍스트)와 metadata(출처 등)를 포함한다.
# Docx2txtLoader는 문서 전체를 하나의 Document로 반환한다. (리스트 길이 = 1)
documents = loader.load()
# ──────────────────────────────────────────────
# 3. 로딩 결과 확인
# ──────────────────────────────────────────────
print(f"로딩된 문서 수: {len(documents)}") # Document 객체 개수
print(f"첫 문서 길이: {len(documents[0].page_content)}자") # 첫 번째 문서의 텍스트 길이(글자 수)
3.4. Step 3: 청킹
● # ──────────────────────────────────────────────
# 1. 텍스트 분할기 임포트
# ────────────────────────────────────────────── # RecursiveCharacterTextSplitter는 여러 구분자를 우선순위 순으로 시도하며
# 문서를 재귀적으로 분할하는 LangChain의 대표적인 텍스트 분할기이다.
from langchain_text_splitters import RecursiveCharacterTextSplitter
# ──────────────────────────────────────────────
# 2. 분할기 설정
# ──────────────────────────────────────────────
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, # 각 청크의 최대 길이 (글자 수 기준)
chunk_overlap=150, # 인접 청크 간 겹치는 글자 수 (문맥 유지를 위한 중복 구간)
separators=[ # 분할 시 시도할 구분자 (우선순위 높은 순)
"\n\n", # 1순위: 빈 줄 (문단 경계)
"\n", # 2순위: 줄바꿈 (행 경계)
". ", # 3순위: 마침표+공백 (문장 경계)
" ", # 4순위: 공백 (단어 경계)
"" # 5순위: 글자 단위 (최후 수단)
]
)
# ──────────────────────────────────────────────
# 3. 문서 분할 실행
# ──────────────────────────────────────────────
# 앞서 로딩한 Document 리스트를 청크 단위로 분할한다.
# 각 청크는 원본 Document의 metadata를 그대로 상속받는다.
chunks = splitter.split_documents(documents)
# ──────────────────────────────────────────────
# 4. 분할 결과 확인
# ──────────────────────────────────────────────
print(f"청크 수: {len(chunks)}") # 생성된 총 청크 개수
print(f"첫 청크 미리보기: {chunks[0].page_content[:100]}...") # 첫 번째 청크의 앞 100자 미리보기
3.5. Step 4: 임베딩 + 벡터 DB 저장
# ────────────────────────────────────────────── # 1. 임베딩 모델 및 벡터 스토어 임포트
# 1. 임베딩 모델 및 벡터 스토어 임포트
# ────────────────────────────────────────────── # OpenAI의 임베딩 모델을 사용하여 텍스트를 벡터로 변환한다.
from langchain_openai import OpenAIEmbeddings # Chroma는 오픈소스 벡터 데이터베이스로, 로컬 환경에서 별도 서버 없이 사용 가능하다.
from langchain_community.vectorstores import Chroma
# ──────────────────────────────────────────────
# 2. 임베딩 모델 설정
# ──────────────────────────────────────────────
# text-embedding-3-small: OpenAI의 경량 임베딩 모델 (1536차원)
# 텍스트를 고정 길이의 숫자 벡터로 변환하여 의미적 유사도 비교를 가능하게 한다.
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
# ──────────────────────────────────────────────
# 3. 벡터 DB 생성 및 문서 저장
# ──────────────────────────────────────────────
# from_documents()는 다음 과정을 한 번에 수행하는 클래스 메서드이다:
# 1) 각 청크의 page_content를 임베딩 모델로 벡터 변환
# 2) 벡터와 원본 텍스트를 Chroma DB에 저장
# 3) 저장이 완료된 Chroma 인스턴스를 반환
db = Chroma.from_documents(
documents=chunks, # 앞서 분할한 청크 리스트 (List[Document])
embedding=embedding, # 벡터 변환에 사용할 임베딩 모델
collection_name="tax_retrieval_chain" # Chroma 내부 컬렉션 이름 (데이터 그룹 구분용)
)
# ──────────────────────────────────────────────
# 4. 저장 결과 확인
# ──────────────────────────────────────────────
# _collection.count()로 실제 저장된 벡터(문서) 수를 조회한다.
print(f"벡터 DB 저장 완료: {db._collection.count()}개 문서")
3.6. Step 5: 검색기(Retriever) 생성
# ──────────────────────────────────────────────
# 1. Retriever 생성
# ──────────────────────────────────────────────
# as_retriever()는 Chroma 벡터 스토어를 BaseRetriever 인터페이스로 변환한다.
# 이를 통해 create_retrieval_chain 등 LangChain 체인에 직접 연결할 수 있다.
retriever = db.as_retriever(
search_type="similarity", # 코사인 유사도 기반 검색 (기본값)
# 다른 옵션: "mmr" (최대 한계 관련성 - 다양성 확보)
# "similarity_score_threshold" (유사도 임계값 필터링)
search_kwargs={"k": 4} # 검색 결과로 반환할 상위 문서 수 (상위 4개)
)
# ──────────────────────────────────────────────
# 2. 검색 테스트
# ──────────────────────────────────────────────
# invoke()에 질의 문자열을 전달하면 내부적으로 다음 과정이 수행된다:
# 1) 질의를 임베딩 벡터로 변환
# 2) 벡터 DB에서 코사인 유사도가 높은 상위 k개 문서를 검색
# 3) List[Document] 형태로 반환
test_docs = retriever.invoke("소득세 계산 방법")
# ──────────────────────────────────────────────
# 3. 검색 결과 출력
# ──────────────────────────────────────────────
print(f"검색 결과: {len(test_docs)}개 문서") # 반환된 문서 수 (최대 k=4)
for i, doc in enumerate(test_docs):
print(f" [{i+1}] {doc.page_content[:80]}...") # 각 문서의 앞 80자를 미리보기로 출력
3.7. Step 6: RAG 체인 구성 (create_retrieval_chain)
# ──────────────────────────────────────────────
# 1. 필요 모듈 임포트 # 1. 필요 모듈 임포트
# ────────────────────────────────────────────── from langchain_openai import ChatOpenAI # OpenAI 채팅 모델
from langchain_core.prompts import ChatPromptTemplate # 프롬프트 템플릿 from langchain.chains import create_retrieval_chain # 검색 + 생성 통합 체인
from langchain.chains.combine_documents import create_stuff_documents_chain # 문서 결합 체인
# ──────────────────────────────────────────────
# 2. LLM 초기화
# ──────────────────────────────────────────────
# gpt-4o-mini: 비용 대비 성능이 우수한 경량 모델
# temperature=0: 무작위성을 제거하여 일관된 답변을 생성 (RAG에 권장)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ──────────────────────────────────────────────
# 3. 프롬프트 정의
# ──────────────────────────────────────────────
# {context}: create_stuff_documents_chain이 검색된 문서를 병합하여 삽입하는 위치
# {input} : 사용자의 원본 질문이 삽입되는 위치
# 답변할 수 없는 경우의 지시문을 포함하여 환각(hallucination)을 방지한다.
prompt = ChatPromptTemplate.from_template(
"""다음 컨텍스트를 기반으로 질문에 답변하세요.
컨텍스트에 답변이 없으면 "제공된 문서에서 해당 정보를 찾을 수 없습니다."라고 답변하세요.
컨텍스트:
{context}
질문: {input}
답변:"""
)
# ──────────────────────────────────────────────
# 4. 문서 결합 체인 생성
# ──────────────────────────────────────────────
# create_stuff_documents_chain은 다음 역할을 수행한다:
# 1) 검색된 Document 리스트의 page_content를 하나의 문자열로 병합
# 2) 병합된 텍스트를 프롬프트의 {context}에 삽입
# 3) 완성된 프롬프트를 LLM에 전달하여 답변 생성
combine_docs_chain = create_stuff_documents_chain(llm, prompt)
# ──────────────────────────────────────────────
# 5. RAG 체인 생성 (검색 + 생성 통합)
# ──────────────────────────────────────────────
# create_retrieval_chain은 retriever와 combine_docs_chain을 하나의 파이프라인으로 연결한다.
# - 입력: {"input": str}
# - 내부: retriever로 문서 검색 → combine_docs_chain으로 답변 생성
# - 출력: {"input": str, "context": List[Document], "answer": str}
rag_chain = create_retrieval_chain(retriever, combine_docs_chain)
3.8. Step 7: 질의 및 응답
# ──────────────────────────────────────────────
# 1. RAG 체인 실행 # 1. RAG 체인 실행
# ────────────────────────────────────────────── # invoke()에 {"input": 질문}을 전달하면 내부적으로 다음 과정이 수행된다:
# 1) retriever가 "근로소득세 계산 방법은?"을 벡터로 변환 → 유사 문서 4개 검색 # 2) combine_docs_chain이 검색된 문서를 프롬프트에 삽입 → LLM 호출
# 3) 결과를 {"input", "context", "answer"} 딕셔너리로 반환
result = rag_chain.invoke({"input": "근로소득세 계산 방법은?"})
# ──────────────────────────────────────────────
# 2. 답변 출력
# ──────────────────────────────────────────────
# result["answer"]: LLM이 검색된 문서를 기반으로 생성한 최종 답변 (str)
print("=== 답변 ===")
print(result["answer"])
# ──────────────────────────────────────────────
# 3. 참조 문서 출력
# ──────────────────────────────────────────────
# result["context"]: retriever가 검색한 Document 리스트 (List[Document])
# 답변의 근거가 된 원본 문서를 확인하여 신뢰성을 검증할 수 있다.
print("\n=== 참조 문서 ===")
for i, doc in enumerate(result["context"]):
# metadata에서 문서 출처(파일 경로) 추출, 없으면 'N/A' 표시
print(f" [{i+1}] 출처: {doc.metadata.get('source', 'N/A')}")
# 각 문서의 앞 100자를 미리보기로 출력하여 어떤 내용이 참조되었는지 확인
print(f" 내용: {doc.page_content[:100]}...")
3.8.1. 반환값 상세 분석
# result 딕셔너리의 키 확인
print("반환된 키:", list(result.keys()))
# → ['input', 'context', 'answer']
# 각 키의 타입과 내용
print(f"input 타입: {type(result['input']).__name__}") # str
print(f"context 타입: {type(result['context']).__name__}") # list
print(f"answer 타입: {type(result['answer']).__name__}") # str
print(f"context 문서 수: {len(result['context'])}") # 4 (k=4)
4. 프롬프트 커스터마이징
4.1. 기본 프롬프트 구조
- `create_stuff_documents_chain`에 전달하는 프롬프트는 최소한 `{context}`와 `{input}` 두 변수를 포함해야 한다.
# 최소 프롬프트
minimal_prompt = ChatPromptTemplate.from_template(
"""Context: {context}
Question: {input}
Answer:"""
)
4.2. 시스템 메시지를 포함한 프롬프트
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
system_prompt = """당신은 세법 전문 AI 어시스턴트입니다.
다음 규칙을 따르세요:
1. 제공된 컨텍스트에 기반하여 정확하게 답변합니다.
2. 컨텍스트에 없는 내용은 추측하지 않습니다.
3. 관련 법률 조항이 있으면 명시합니다.
4. 답변은 구조화하여 가독성 있게 작성합니다.
컨텍스트:
{context}"""
prompt_with_system = ChatPromptTemplate.from_messages([
("system", system_prompt),
("human", "{input}"),
])
combine_chain = create_stuff_documents_chain(llm, prompt_with_system)
rag_chain = create_retrieval_chain(retriever, combine_chain)
result = rag_chain.invoke({"input": "종합소득세 세율 구간을 알려주세요"})
print(result["answer"])
4.3. 추가 변수를 포함한 프롬프트
- 프롬프트에 `{context}`와 `{input}` 외에 추가 변수를 포함할 수 있다.
prompt_with_lang = ChatPromptTemplate.from_template(
"""당신은 세법 전문가입니다. {language}로 답변하세요.
컨텍스트:
{context}
질문: {input}
답변:"""
)
combine_chain = create_stuff_documents_chain(llm, prompt_with_lang)
rag_chain = create_retrieval_chain(retriever, combine_chain)
# 추가 변수는 invoke 시 함께 전달
result = rag_chain.invoke({
"input": "소득세율은?",
"language": "한국어"
})
print(result["answer"])
4.4. LangChain Hub 프롬프트 활용
- 검증된 프롬프트를 LangChain Hub에서 가져와 사용할 수 있다.
# ──────────────────────────────────────────────
# 0. 필요 모듈 임포트 (이전 셀에서 이미 정의된 경우 생략 가능)
# ────────────────────────────────────────────── from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_community.document_loaders import Docx2txtLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
# ──────────────────────────────────────────────
# 1. 문서 로딩
# ──────────────────────────────────────────────
# .docx 파일을 읽어 Document 객체 리스트로 변환한다.
loader = Docx2txtLoader("../../tax_docs/tax_with_markdown.docx")
documents = loader.load()
# ──────────────────────────────────────────────
# 2. 문서 분할
# ──────────────────────────────────────────────
# 긴 문서를 1000자 단위 청크로 분할하고, 150자씩 겹쳐 문맥 단절을 방지한다.
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=150,
separators=["\n\n", "\n", ". ", " ", ""]
)
chunks = splitter.split_documents(documents)
# ──────────────────────────────────────────────
# 3. 벡터 DB 생성
# ──────────────────────────────────────────────
# 각 청크를 임베딩 벡터로 변환하여 Chroma에 저장한다.
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
db = Chroma.from_documents(
documents=chunks,
embedding=embedding,
collection_name="tax_retrieval_chain"
)
# ──────────────────────────────────────────────
# 4. Retriever 생성
# ──────────────────────────────────────────────
# 코사인 유사도 기반으로 상위 4개 문서를 검색하는 retriever를 생성한다.
retriever = db.as_retriever(
search_type="similarity",
search_kwargs={"k": 4}
)
# ──────────────────────────────────────────────
# 5. LLM 초기화
# ──────────────────────────────────────────────
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# ──────────────────────────────────────────────
# 6. LangChain Hub에서 검증된 RAG 프롬프트 가져오기
# ──────────────────────────────────────────────
# LangChain Hub는 커뮤니티가 공유하는 프롬프트 저장소이다.
# "langchain-ai/retrieval-qa-chat"은 LangChain 공식 팀이 제공하는
# RAG 전용 프롬프트로, 직접 작성하는 것보다 검증된 품질을 기대할 수 있다.
# 내부적으로 {context}와 {input} 변수를 포함하고 있어
# create_stuff_documents_chain과 바로 호환된다.
from langchain import hub
hub_prompt = hub.pull("langchain-ai/retrieval-qa-chat")
# ──────────────────────────────────────────────
# 7. RAG 체인 구성
# ──────────────────────────────────────────────
# Hub에서 가져온 프롬프트를 사용하여 문서 결합 체인을 생성한다.
# 이전 셀에서 직접 작성한 프롬프트 대신 Hub 프롬프트를 사용한다는 점만 다르다.
combine_chain = create_stuff_documents_chain(llm, hub_prompt)
# retriever(검색) + combine_chain(생성)을 하나의 RAG 파이프라인으로 통합한다.
rag_chain = create_retrieval_chain(retriever, combine_chain)
# ──────────────────────────────────────────────
# 8. RAG 체인 실행 및 결과 출력
# ──────────────────────────────────────────────
# invoke() 호출 시: 질의 → 문서 검색 → 프롬프트 완성 → LLM 답변 생성
result = rag_chain.invoke({"input": "근로소득 공제 방법은?"})
# result["answer"]: 검색된 문서를 근거로 생성된 최종 답변
print(result["answer"])
5. 검색기(Retriever) 옵션
5.1. 검색 방식별 Retriever 설정
# ────────────────────────────────────────────── # 1. 기본 유사도 검색 (가장 보편적)
# ────────────────────────────────────────────── # 질의 벡터와 문서 벡터 간 코사인 유사도가 가장 높은 상위 k개를 반환한다.
# 가장 단순하고 빠른 검색 방식으로, 대부분의 RAG 파이프라인에서 기본값으로 사용된다. # 단점: 유사한 내용의 문서가 중복 반환될 수 있다.
retriever_similarity = db.as_retriever(
search_type="similarity", # 코사인 유사도 기반 검색
search_kwargs={"k": 4} # 상위 4개 문서 반환
)
# ──────────────────────────────────────────────
# 2. MMR 검색 (Maximal Marginal Relevance)
# ──────────────────────────────────────────────
# 유사도와 다양성을 동시에 고려하는 검색 방식이다.
# 먼저 fetch_k개의 후보 문서를 유사도 기준으로 가져온 뒤,
# 후보 중에서 서로 내용이 겹치지 않도록 k개를 최종 선별한다.
# 검색 결과의 중복을 줄이고 다양한 관점의 문서를 확보할 때 유용하다.
retriever_mmr = db.as_retriever(
search_type="mmr", # MMR 알고리즘 적용
search_kwargs={
"k": 4, # 최종 반환 문서 수
"fetch_k": 20, # 1차로 가져올 후보 문서 수 (k보다 충분히 커야 다양성 확보 가능)
"lambda_mult": 0.5 # 유사도-다양성 균형 조절 파라미터
# 0에 가까울수록 → 다양성 극대화 (서로 다른 내용 우선)
# 1에 가까울수록 → 유사도 극대화 (similarity와 동일하게 동작)
# 0.5 → 유사도와 다양성의 균형점
}
)
# ──────────────────────────────────────────────
# 3. 유사도 점수 임계값 검색
# ──────────────────────────────────────────────
# 유사도 점수가 지정한 임계값(threshold) 이상인 문서만 반환한다.
# 관련성이 낮은 문서가 포함되는 것을 방지할 수 있다.
# 주의: 임계값을 넘는 문서가 없으면 빈 리스트가 반환될 수 있고,
# 임계값이 너무 낮으면 k개까지 모두 반환되어 similarity와 유사하게 동작한다.
retriever_threshold = db.as_retriever(
search_type="similarity_score_threshold",
search_kwargs={
"score_threshold": 0.5, # 최소 유사도 점수 (0~1 범위, 높을수록 엄격)
"k": 4 # 임계값을 넘는 문서 중 최대 반환 수
}
)
---
검색 방식 비교 요약
┌──────────────────────────┬──────────────┬──────────────┬──────────────────────┐
│ 방식 │ 속도 │ 다양성 │ 특징 │
├──────────────────────────┼──────────────┼──────────────┼──────────────────────┤
│ similarity │ 가장 빠름 │ 낮음 │ 항상 k개 반환 │
│ mmr │ 보통 │ 높음 │ 중복 제거에 효과적 │
│ similarity_score_threshold│ 빠름 │ 낮음 │ 관련 없는 문서 필터링 │
└──────────────────────────┴──────────────┴──────────────┴──────────────────────┘
5.1.1. 검색 방식 선택 기준
| 검색 방식 | 적합한 상황 | k 권장값 |
| similarity | 일반 Q&A, 빠른 응답 | 3~5 |
| mmr | 다양한 관점 필요, 중복 결과 문제 | 3~5 (fetch_k: 15~30) |
| similarity_score_threshold | 정밀도 중시, 무관련 결과 배제 | 3~5 (threshold: 0.8~0.9) |
5.2. 각 Retriever로 RAG 체인 구성 비교
# ──────────────────────────────────────────────────────────────
# 각 Retriever로 RAG 체인 구성 비교
# ────────────────────────────────────────────────────────────── # create_retrieval_chain의 핵심 장점은 retriever와 combine_docs_chain이
# 독립적인 컴포넌트로 분리되어 있다는 점이다.
# 이 구조 덕분에 combine_docs_chain(LLM + 프롬프트)은 그대로 유지한 채
# retriever만 교체하여 검색 전략에 따른 결과 차이를 비교할 수 있다.
#
# 아래 코드는 동일한 질문("소득세 계산 방법은?")을 두 가지 검색 방식으로 실행하고,
# 각 방식이 생성한 답변의 길이와 내용을 비교한다.
# - similarity : 유사도가 높은 문서를 그대로 반환 → 정확도 중심
# - mmr : 유사도와 다양성을 동시에 고려 → 폭넓은 정보 제공
#
# 동일한 combine_docs_chain을 재사용하므로 LLM과 프롬프트는 통제 변인이 되고,
# retriever의 검색 전략만 조작 변인이 되어 공정한 비교가 가능하다.
# ──────────────────────────────────────────────────────────────
for name, ret in [("similarity", retriever_similarity),
("mmr", retriever_mmr)]:
# retriever만 교체하여 새로운 RAG 체인을 구성한다.
# combine_chain(LLM + 프롬프트)은 동일하게 재사용된다.
chain = create_retrieval_chain(ret, combine_chain)
# 동일한 질문으로 체인을 실행하여 검색 전략별 결과를 비교한다.
result = chain.invoke({"input": "소득세 계산 방법은?"})
# 답변 길이와 미리보기를 출력하여 검색 전략에 따른 차이를 확인한다.
# similarity는 유사한 문서가 중복될 수 있어 답변이 좁고 깊을 수 있고,
# mmr은 다양한 문서를 참조하므로 답변이 넓고 포괄적일 수 있다.
print(f"\n[{name}] 답변 길이: {len(result['answer'])}자")
print(f" 답변 미리보기: {result['answer'][:100]}...")for name, ret in [(...), (...)] 동작 원리
핵심: 튜플 언패킹(Tuple Unpacking)을 활용한 반복문
for name, ret in [("similarity", retriever_similarity),
("mmr", retriever_mmr)]:
이 코드는 리스트 안에 담긴 튜플을 순회하면서, 각 튜플의 요소를 두 변수에 자동으로 분리 대입하는 구문이다.
---
단계별 동작
반복 대상 리스트:
┌─────────────────────────────────────────────────┐
│ [ │
│ ("similarity", retriever_similarity), ← 1회차│
│ ("mmr", retriever_mmr), ← 2회차│
│ ] │
└─────────────────────────────────────────────────┘
1회차 반복:
("similarity", retriever_similarity) → name = "similarity"
ret = retriever_similarity
2회차 반복:
("mmr", retriever_mmr) → name = "mmr"
ret = retriever_mmr
---
풀어쓴 동등 코드
위 코드는 아래와 완전히 동일하게 동작한다.
# 1회차: similarity 검색
name = "similarity"
ret = retriever_similarity
chain = create_retrieval_chain(ret, combine_chain)
result = chain.invoke({"input": "소득세 계산 방법은?"})
print(f"\n[{name}] 답변 길이: {len(result['answer'])}자")
print(f" 답변 미리보기: {result['answer'][:100]}...")
# 2회차: mmr 검색
name = "mmr"
ret = retriever_mmr
chain = create_retrieval_chain(ret, combine_chain)
result = chain.invoke({"input": "소득세 계산 방법은?"})
print(f"\n[{name}] 답변 길이: {len(result['answer'])}자")
print(f" 답변 미리보기: {result['answer'][:100]}...")
---
이 패턴을 사용하는 이유
반복문 사용 시 풀어쓴 코드
───────────────────── ─────────────────────
코드 중복 없음 동일 코드가 2번 반복
retriever 추가 시 retriever 추가 시
→ 리스트에 튜플 1개만 추가 → 코드 블록 전체를 복사·붙여넣기
비교 대상(retriever)이 늘어나도 리스트에 튜플만 추가하면 된다.
# threshold 검색을 추가하고 싶다면 → 튜플 1개만 추가
for name, ret in [("similarity", retriever_similarity),
("mmr", retriever_mmr),
("threshold", retriever_threshold)]: # ← 이 한 줄만 추가
chain = create_retrieval_chain(ret, combine_chain)
result = chain.invoke({"input": "소득세 계산 방법은?"})
print(f"\n[{name}] 답변 길이: {len(result['answer'])}자")
print(f" 답변 미리보기: {result['answer'][:100]}...")
요약: name에는 출력용 라벨(문자열)이, ret에는 실제 retriever 객체가 대입되어, 동일한 체인 로직을 검색 전략별로 반복 실행하는 구조다.
5.3. 메타데이터 필터링
# ──────────────────────────────────────────────────────────────
# 메타데이터 필터링을 활용한 조건부 검색
# ────────────────────────────────────────────────────────────── # 벡터 DB에 여러 출처의 문서가 저장되어 있을 때,
# 특정 출처(source)의 문서만 대상으로 검색을 제한할 수 있다. # filter 파라미터는 Document의 metadata 딕셔너리와 매칭되어
# 조건에 부합하는 문서만 검색 후보에 포함시킨다.
#
# 활용 예시:
# - 여러 법률 문서 중 특정 법률만 검색하고 싶을 때
# - 연도별 문서 중 최신 문서만 대상으로 검색하고 싶을 때
# - 부서별 문서 중 특정 부서 문서만 참조하고 싶을 때
# ──────────────────────────────────────────────────────────────
retriever_filtered = db.as_retriever(
search_type="similarity",
search_kwargs={
"k": 4,
# filter: metadata 필드를 기준으로 검색 범위를 제한한다.
# Docx2txtLoader가 자동으로 부여한 metadata의 "source" 키와 매칭하여
# 해당 파일에서 생성된 청크만 검색 대상에 포함시킨다.
"filter": {"source": "../../tax_docs/tax_with_markdown.docx"}
}
)
# ──────────────────────────────────────────────────────────────
# 필터링된 retriever로 RAG 체인 구성 및 실행
# ──────────────────────────────────────────────────────────────
# combine_chain(LLM + 프롬프트)은 동일하게 재사용하고,
# retriever만 필터링 버전으로 교체하여 검색 범위를 제한한다.
chain = create_retrieval_chain(retriever_filtered, combine_chain)
result = chain.invoke({"input": "세율 구간은?"})
# 필터링된 문서만을 근거로 생성된 답변을 출력한다.
print(result["answer"])
---
filter가 적용되는 위치
사용자 질의: "세율 구간은?"
│
▼
┌─────────────────────────────────────────┐
│ Chroma 벡터 DB │
│ │
│ [청크A] source: tax_with_markdown.docx │ ✅ 필터 통과 → 유사도 비교 대상
│ [청크B] source: tax_with_markdown.docx │ ✅ 필터 통과 → 유사도 비교 대상
│ [청크C] source: other_document.pdf │ ❌ 필터 제외 → 검색 대상에서 제외
│ [청크D] source: tax_with_markdown.docx │ ✅ 필터 통과 → 유사도 비교 대상
│ │
└─────────────────────────────────────────┘
│
▼ 필터 통과한 문서 중 유사도 상위 k개 선택
│
[검색 결과: 최대 4개 문서]
필터는 유사도 계산 이전 단계에서 적용된다. 즉, 조건에 맞지 않는 문서는 유사도 비교 자체를
수행하지 않으므로 검색 정확도와 효율이 모두 향상된다.
6. 스트리밍 응답
6.1. 기본 스트리밍
- 실시간으로 답변을 출력하려면 `stream()` 메서드를 사용한다.
# 스트리밍 출력
for chunk in rag_chain.stream({"input": "근로소득세 계산 과정을 상세히 설명해주세요"}):
# 'answer' 키가 있는 청크만 출력 (생성 단계 결과)
if "answer" in chunk:
print(chunk["answer"], end="", flush=True)
print() # 줄바꿈
6.2. 스트리밍 시 context 함께 수집
# ────────────────────────────────────────────────────────────── # 스트리밍 실행: 토큰 단위 실시간 출력 + 참조 문서 수집
# ────────────────────────────────────────────────────────────── # invoke()는 전체 답변이 완성될 때까지 기다린 뒤 한 번에 반환하지만,
# stream()은 LLM이 토큰을 생성할 때마다 즉시 chunk 단위로 전달한다. # 사용자에게 실시간으로 답변이 출력되는 효과를 줄 수 있어
# 긴 답변에서의 체감 대기 시간을 크게 줄여준다.
#
# stream()이 반환하는 각 chunk는 딕셔너리이며,
# 매 chunk마다 모든 키가 포함되는 것이 아니라
# 해당 시점에 생성된 데이터의 키만 포함된다:
# - {"context": List[Document]} → 검색 완료 시점에 1회 전달
# - {"answer": str} → LLM이 토큰을 생성할 때마다 반복 전달
# ──────────────────────────────────────────────────────────────
# 스트리밍 결과를 누적할 변수 초기화
context_docs = [] # 검색된 참조 문서를 저장할 리스트
answer_text = "" # 토큰 단위로 전달되는 답변을 이어붙일 문자열
for chunk in rag_chain.stream({"input": "소득세 신고 기한은?"}):
# context 키는 retriever의 검색이 완료된 시점에 한 번만 전달된다.
# 이 시점에서 참조 문서 리스트를 저장해둔다.
if "context" in chunk:
context_docs = chunk["context"]
# answer 키는 LLM이 토큰을 생성할 때마다 반복적으로 전달된다.
# 각 chunk["answer"]에는 토큰 1개(또는 소수의 토큰)가 담겨 있다.
if "answer" in chunk:
answer_text += chunk["answer"] # 전체 답변 누적
print(chunk["answer"], end="", flush=True) # 토큰을 즉시 출력
# end="" : 줄바꿈 없이 이어서 출력
# flush=True : 버퍼를 즉시 비워 실시간 표시
# 스트리밍 완료 후 참조 문서 수를 출력한다.
print(f"\n\n참조 문서: {len(context_docs)}개")
---
스트리밍 chunk 흐름 시각화
시간 →
─────────────────────────────────────────────────────────
chunk 1: {"context": [Doc1, Doc2, Doc3, Doc4]} ← 검색 결과 (1회)
chunk 2: {"answer": "소득세"} ← 토큰 스트리밍 시작
chunk 3: {"answer": " 신고"}
chunk 4: {"answer": " 기한은"}
chunk 5: {"answer": " 매년"}
chunk 6: {"answer": " 5월"}
chunk 7: {"answer": " 1일부터"}
... ...
chunk N: {"answer": "입니다."} ← 토큰 스트리밍 종료
─────────────────────────────────────────────────────────
화면 출력 (실시간):
소득세 신고 기한은 매년 5월 1일부터 ... 입니다.
↑ 한 글자씩 나타나는 효과
참조 문서: 4개
invoke() vs stream() 비교
┌────────────┬──────────────────────────┬───────────────────────────┐
│ │ invoke() │ stream() │
├────────────┼──────────────────────────┼───────────────────────────┤
│ 반환 형태 │ 완성된 딕셔너리 1개 │ chunk 딕셔너리의 반복자 │
│ 응답 시점 │ 전체 생성 완료 후 │ 토큰 생성 즉시 │
│ 체감 대기 │ 긴 답변일수록 오래 대기 │ 첫 토큰부터 바로 출력 │
│ 용도 │ 후처리·평가용 │ 사용자 대면 인터페이스 │
└────────────┴──────────────────────────┴───────────────────────────┘
7. 대화 이력을 포함한 RAG (Conversational RAG)
7.1. 대화형 RAG의 필요성
- 단일 질의-응답이 아닌, 이전 대화를 기억하는 RAG 체인이 필요한 경우가 있다.
| 시나리오 | 예시 |
| 후속 질문 | "그러면 법인세는?" (이전 질문: 소득세 세율) |
| 대명사 참조 | "그것의 계산 방법은?" (이전 답변의 특정 개념) |
| 맥락 연속 | "더 자세히 설명해줘" (이전 답변의 확장) |
7.2. create_history_aware_retriever
- 대화 이력을 고려하여 검색 질의를 재구성하는 retriever를 생성한다.
# ──────────────────────────────────────────────────────────────
# 대화형 RAG 체인 구성 (Conversational RAG)
# ────────────────────────────────────────────────────────────── # 기본 create_retrieval_chain은 매 질문을 독립적으로 처리한다.
# 그러나 실제 대화에서는 "그것의 세율은?" 같은 대명사 참조나 # "좀 더 자세히 알려줘" 같은 맥락 의존적 질문이 빈번하다.
#
# 이 문제를 해결하기 위해 retriever 앞에 "질의 재구성" 단계를 추가한다.
# LLM이 대화 이력을 참고하여 모호한 질문을 독립적인 검색 질의로 변환한 뒤,
# 재구성된 질의로 문서를 검색하는 구조이다.
#
# 전체 흐름:
# chat_history + input → 질의 재구성 → 문서 검색 → 답변 생성
# ──────────────────────────────────────────────────────────────
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# ──────────────────────────────────────────────────────────────
# 1. 질의 재구성 프롬프트 (Contextualize Prompt)
# ──────────────────────────────────────────────────────────────
# 대화 이력을 참고하여 맥락 의존적 질문을 독립적인 질문으로 변환한다.
#
# 예시:
# chat_history: [("human", "근로소득세란?"), ("ai", "근로소득세는...")]
# input: "세율은 어떻게 돼?"
# → 재구성: "근로소득세의 세율은 어떻게 되나요?"
#
# MessagesPlaceholder("chat_history")는 대화 이력 메시지 리스트가
# 삽입될 위치를 지정한다.
contextualize_prompt = ChatPromptTemplate.from_messages([
("system", "대화 이력과 최신 질문을 고려하여, 검색에 사용할 독립적인 질문으로 재구성하세요. "
"대화 이력이 없으면 질문을 그대로 반환하세요."),
MessagesPlaceholder("chat_history"), # 이전 대화 내역이 삽입되는 위치
("human", "{input}"), # 현재 사용자 질문
])
# ──────────────────────────────────────────────────────────────
# 2. 대화 이력 인식 Retriever 생성
# ──────────────────────────────────────────────────────────────
# create_history_aware_retriever는 기존 retriever를 감싸서
# 검색 전에 질의 재구성 단계를 추가한다.
#
# 내부 동작:
# 1) chat_history가 비어 있으면 → input을 그대로 retriever에 전달
# 2) chat_history가 있으면 → LLM이 질의를 재구성 → 재구성된 질의로 검색
#
# 즉, LLM 호출이 최대 2회 발생한다: 질의 재구성 1회 + 답변 생성 1회
history_aware_retriever = create_history_aware_retriever(
llm, # 질의 재구성에 사용할 LLM
retriever, # 실제 검색을 수행할 기존 retriever
contextualize_prompt # 질의 재구성 프롬프트
)
# ──────────────────────────────────────────────────────────────
# 3. 답변 생성 프롬프트
# ──────────────────────────────────────────────────────────────
# 검색된 문서(context)와 대화 이력(chat_history)을 모두 참고하여
# 답변을 생성하는 프롬프트이다.
# 대화 이력을 포함함으로써 이전 답변과 일관된 톤과 맥락을 유지한다.
answer_prompt = ChatPromptTemplate.from_messages([
("system", "당신은 세법 전문 AI입니다. 컨텍스트를 기반으로 답변하세요.\n\n{context}"),
MessagesPlaceholder("chat_history"), # 이전 대화 내역 (맥락 유지)
("human", "{input}"), # 현재 사용자 질문
])
# ──────────────────────────────────────────────────────────────
# 4. 대화형 RAG 체인 구성
# ──────────────────────────────────────────────────────────────
# combine_chain: 검색된 문서 + 대화 이력을 기반으로 답변을 생성
combine_chain = create_stuff_documents_chain(llm, answer_prompt)
# conversational_rag: history_aware_retriever + combine_chain 통합
# 입력: {"input": str, "chat_history": List[BaseMessage]}
# 출력: {"input": str, "chat_history": ..., "context": List[Document], "answer": str}
conversational_rag = create_retrieval_chain(history_aware_retriever, combine_chain)
---
기본 RAG vs 대화형 RAG 구조 비교
기본 RAG (create_retrieval_chain)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
{"input": "세율은?"}
│
├──→ retriever ──→ 문서 검색
│ │
└──→ combine_chain ←──┘
│
"답변" (LLM 1회 호출)
대화형 RAG (history_aware_retriever 적용)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
{"input": "세율은?", "chat_history": [...]}
│
▼
┌──────────────────────────────────┐
│ history_aware_retriever │
│ │
│ chat_history + input │
│ │ │
│ ▼ │
│ LLM: 질의 재구성 (1회차 호출) │
│ "세율은?" → "근로소득세 세율은?" │
│ │ │
│ ▼ │
│ retriever ──→ 문서 검색 │
└──────────────┬───────────────────┘
│
▼
┌──────────────────────────────────┐
│ combine_chain │
│ context + chat_history + input │
│ │ │
│ ▼ │
│ LLM: 답변 생성 (2회차 호출) │
└──────────────┬───────────────────┘
│
▼
{"input": ..., "context": ..., "answer": ...}
7.3. 대화 이력을 사용한 질의
# ──────────────────────────────────────────────────────────────
# 대화형 RAG 실행: 멀티턴 대화 시뮬레이션
# ────────────────────────────────────────────────────────────── # 대화형 RAG의 핵심은 이전 대화 맥락을 유지하면서
# 후속 질문의 의미를 정확히 해석하는 것이다. # 아래 코드는 2턴 대화를 통해 이 과정을 시연한다.
# ──────────────────────────────────────────────────────────────
from langchain_core.messages import HumanMessage, AIMessage
# ──────────────────────────────────────────────────────────────
# 대화 이력 초기화
# ──────────────────────────────────────────────────────────────
# chat_history는 HumanMessage와 AIMessage 객체를 번갈아 저장하는 리스트이다.
# 빈 리스트로 시작하며, 매 턴마다 질문-답변 쌍을 추가한다.
chat_history = []
# ──────────────────────────────────────────────────────────────
# 1턴: 첫 번째 질문 (독립적 질문)
# ──────────────────────────────────────────────────────────────
# chat_history가 비어 있으므로 history_aware_retriever는
# 질의 재구성 없이 원본 질문을 그대로 retriever에 전달한다.
result1 = conversational_rag.invoke({
"input": "소득세 세율을 알려주세요",
"chat_history": chat_history # 빈 리스트 → 재구성 생략
})
print("Q1:", "소득세 세율을 알려주세요")
print("A1:", result1["answer"][:200])
# ──────────────────────────────────────────────────────────────
# 대화 이력 업데이트
# ──────────────────────────────────────────────────────────────
# 1턴의 질문과 답변을 각각 HumanMessage, AIMessage로 감싸서
# chat_history에 추가한다. 이 이력이 다음 턴에서 맥락으로 활용된다.
#
# chat_history 상태:
# [HumanMessage("소득세 세율을 알려주세요"),
# AIMessage("소득세 세율은...")]
chat_history.extend([
HumanMessage(content="소득세 세율을 알려주세요"),
AIMessage(content=result1["answer"])
])
# ──────────────────────────────────────────────────────────────
# 2턴: 후속 질문 (맥락 의존적 질문)
# ──────────────────────────────────────────────────────────────
# "그러면 과세표준 계산은 어떻게 하나요?"는 앞선 대화의 맥락(소득세)을
# 전제로 한 질문이다. 이 질문만으로는 검색 키워드가 부족하다.
#
# history_aware_retriever 내부 동작:
# 1) chat_history가 비어있지 않으므로 질의 재구성 실행
# 2) LLM이 대화 이력을 참고하여 질의를 변환:
# "그러면 과세표준 계산은 어떻게 하나요?"
# → "소득세 과세표준 계산 방법은?" (독립적 질문으로 재구성)
# 3) 재구성된 질의로 retriever 검색 실행
result2 = conversational_rag.invoke({
"input": "그러면 과세표준 계산은 어떻게 하나요?",
"chat_history": chat_history # 1턴 이력 포함 → 재구성 실행
})
print("\nQ2:", "그러면 과세표준 계산은 어떻게 하나요?")
print("A2:", result2["answer"][:200])
---
2턴 대화의 내부 흐름
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
1턴 (chat_history = [])
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
input: "소득세 세율을 알려주세요"
│
│ chat_history가 비어 있음 → 재구성 생략
│
├──→ retriever("소득세 세율을 알려주세요")
│ → [Doc1, Doc2, Doc3, Doc4]
│
└──→ combine_chain → LLM → "소득세 세율은..."
chat_history 업데이트:
[Human("소득세 세율을 알려주세요"), AI("소득세 세율은...")]
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
2턴 (chat_history = [Human(...), AI(...)])
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
input: "그러면 과세표준 계산은 어떻게 하나요?"
│
│ chat_history가 존재 → 질의 재구성 실행
│
▼
LLM 재구성: "소득세 과세표준 계산 방법은?" ← 맥락 반영
│
├──→ retriever("소득세 과세표준 계산 방법은?")
│ → [Doc5, Doc6, Doc7, Doc8] ← 더 정확한 검색 결과
│
└──→ combine_chain → LLM → "과세표준은..."
- 핵심 포인트: `create_history_aware_retriever`는 대화 이력을 분석하여 "그러면 과세표준 계산은?"을 "소득세 과세표준 계산 방법은?"으로 재구성한 후 검색한다. 이를 통해 대명사 참조나 맥락 의존적 질문도 정확하게 처리할 수 있다.
8. 고급 활용: 커스텀 체인 조합
8.1. 출력 파서와 함께 사용
- 구조화된 출력이 필요할 때 출력 파서를 결합한다.
# ══════════════════════════════════════════════════════════════
# 고급 활용: 커스텀 체인 조합
# ══════════════════════════════════════════════════════════════
# ────────────────────────────────────────────────────────────── # 8.1 출력 파서와 함께 사용
# ──────────────────────────────────────────────────────────────
# 기본 RAG 체인은 LLM의 답변을 단순 문자열(str)로 반환한다.
# 그러나 실무에서는 답변·신뢰도·출처 등을 구조화된 형태로 받아야
# 후속 처리(UI 렌더링, DB 저장, API 응답 등)가 용이하다.
#
# JsonOutputParser와 Pydantic 모델을 결합하면
# LLM의 자유 텍스트 출력을 검증된 JSON 구조로 변환할 수 있다.
# ──────────────────────────────────────────────────────────────
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
# ──────────────────────────────────────────────────────────────
# 1. 출력 스키마 정의 (Pydantic 모델)
# ──────────────────────────────────────────────────────────────
# LLM이 반환해야 할 JSON의 구조를 Pydantic 모델로 정의한다.
# Field의 description은 LLM에게 각 필드가 무엇인지 알려주는 역할을 한다.
class TaxAnswer(BaseModel):
answer: str = Field(description="질문에 대한 답변")
confidence: str = Field(description="답변 신뢰도 (높음/중간/낮음)")
sources: list = Field(description="참조한 법률 조항 리스트")
# ──────────────────────────────────────────────────────────────
# 2. JSON 출력 파서 생성
# ──────────────────────────────────────────────────────────────
# JsonOutputParser는 두 가지 역할을 수행한다:
# 1) get_format_instructions(): LLM에게 전달할 JSON 형식 지시문 생성
# 2) parse(): LLM 출력 문자열을 파싱하여 Python 딕셔너리로 변환
parser = JsonOutputParser(pydantic_object=TaxAnswer)
# ──────────────────────────────────────────────────────────────
# 3. 구조화된 출력용 프롬프트 정의
# ──────────────────────────────────────────────────────────────
# {format_instructions}: 파서가 생성한 JSON 형식 지시문이 삽입될 위치
# {context}: 검색된 문서가 삽입될 위치
# {input}: 사용자 질문이 삽입될 위치
structured_prompt = ChatPromptTemplate.from_template(
"""다음 컨텍스트를 기반으로 질문에 답변하세요.
{format_instructions}
컨텍스트:
{context}
질문: {input}"""
)
# ──────────────────────────────────────────────────────────────
# 4. format_instructions를 프롬프트에 사전 주입
# ──────────────────────────────────────────────────────────────
# .partial()은 프롬프트의 특정 변수를 미리 고정하는 메서드이다.
# format_instructions는 매 호출마다 동일하므로 사전에 주입해두면
# invoke() 시 {context}와 {input}만 전달하면 된다.
#
# 주입되는 내용 예시:
# "The output should be formatted as a JSON instance that conforms to
# the JSON schema below: {"answer": str, "confidence": str, "sources": list}"
structured_prompt = structured_prompt.partial(
format_instructions=parser.get_format_instructions()
)
# ──────────────────────────────────────────────────────────────
# 5. 구조화된 RAG 체인 구성
# ──────────────────────────────────────────────────────────────
# output_parser=parser를 전달하면 create_stuff_documents_chain의
# 내부 파이프라인이 다음과 같이 변경된다:
# 기본: 문서 병합 → 프롬프트 → LLM → StrOutputParser → str
# 변경: 문서 병합 → 프롬프트 → LLM → JsonOutputParser → dict
combine_chain = create_stuff_documents_chain(
llm, structured_prompt, output_parser=parser
)
structured_rag = create_retrieval_chain(retriever, combine_chain)
# ──────────────────────────────────────────────────────────────
# 6. 체인 실행 및 결과 확인
# ──────────────────────────────────────────────────────────────
result = structured_rag.invoke({"input": "소득세 세율 구간은?"})
# result["answer"]는 이제 str이 아닌 dict(또는 Pydantic 객체)이다.
# 예시 출력:
# {
# "answer": "소득세 세율 구간은 과세표준에 따라...",
# "confidence": "높음",
# "sources": ["소득세법 제55조", "소득세법 시행령 제100조"]
# }
print(result["answer"])
---
기본 체인 vs 구조화된 체인 출력 비교
기본 RAG 체인 (StrOutputParser)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
result["answer"] → str
"소득세 세율은 과세표준 구간에 따라
1,200만원 이하 6%, 4,600만원 이하 15%..."
구조화된 RAG 체인 (JsonOutputParser)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
result["answer"] → dict
{
"answer": "소득세 세율은 과세표준 구간에 따라...",
"confidence": "높음",
"sources": ["소득세법 제55조"]
}
프롬프트에 주입되는 format_instructions 내용
parser.get_format_instructions() 가 생성하는 지시문:
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
"The output should be formatted as a JSON instance
that conforms to the JSON schema below.
Here is the output schema:
{
"answer": {"description": "질문에 대한 답변", "type": "string"},
"confidence": {"description": "답변 신뢰도 (높음/중간/낮음)", "type": "string"},
"sources": {"description": "참조한 법률 조항 리스트", "type": "array"}
}"
↓ .partial()로 프롬프트에 사전 주입
┌────────────────────────────────────────────┐
│ 다음 컨텍스트를 기반으로 질문에 답변하세요. │
│ │
│ The output should be formatted as ... │ ← 여기에 삽입됨
│ │
│ 컨텍스트: │
│ {context} │
│ │
│ 질문: {input} │
└────────────────────────────────────────────┘
8.2. 후처리 체인 추가
- RAG 체인의 출력을 후처리하여 요약이나 검증을 추가할 수 있다.
# ══════════════════════════════════════════════════════════════
# 후처리 체인 추가
# ══════════════════════════════════════════════════════════════ # RAG 체인의 출력은 {"input", "context", "answer"} 딕셔너리이다.
# 이 원시 출력을 그대로 사용자에게 보여주기보다, # 후처리 단계를 추가하여 답변 포맷팅, 출처 정리, 요약, 검증 등을
# 수행하는 것이 실무에서 일반적이다.
#
# LCEL(LangChain Expression Language)의 파이프 연산자(|)를 사용하면
# 기존 체인 뒤에 후처리 함수를 자연스럽게 연결할 수 있다.
# ══════════════════════════════════════════════════════════════
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
# ──────────────────────────────────────────────────────────────
# 1. 후처리 함수 정의
# ──────────────────────────────────────────────────────────────
# rag_chain의 출력 딕셔너리를 받아 사용자 친화적인 문자열로 변환한다.
# 이 함수는 LLM을 호출하지 않으므로 추가 비용이 발생하지 않는다.
def format_answer(result):
"""RAG 결과를 포맷팅하는 후처리 함수"""
# result["answer"]: LLM이 생성한 답변 문자열
answer = result["answer"]
# result["context"]: 검색된 Document 리스트
# 각 Document의 metadata에서 출처(source) 정보를 추출한다.
sources = [doc.metadata.get("source", "N/A") for doc in result["context"]]
# 동일 파일에서 여러 청크가 검색될 수 있으므로 중복 출처를 제거한다.
unique_sources = list(set(sources))
# 마크다운 형식으로 답변, 출처, 문서 수를 정리하여 반환한다.
# chr(10)은 줄바꿈 문자('\n')로, f-string 내부에서 사용하기 위한 표현이다.
formatted = f"""## 답변
{answer}
## 참조 출처
{chr(10).join(f'- {s}' for s in unique_sources)}
## 참조 문서 수: {len(result['context'])}개
"""
return formatted
# ──────────────────────────────────────────────────────────────
# 2. 후처리를 포함한 체인 구성
# ──────────────────────────────────────────────────────────────
# LCEL 파이프 연산자(|)로 rag_chain 뒤에 후처리 함수를 연결한다.
# RunnableLambda는 일반 Python 함수를 Runnable 인터페이스로 감싸서
# LCEL 체인에 결합할 수 있게 해주는 래퍼이다.
#
# 데이터 흐름:
# {"input": "질문"}
# → rag_chain → {"input": ..., "context": [...], "answer": "..."}
# → RunnableLambda(format_answer) → 포맷팅된 문자열 (str)
formatted_chain = rag_chain | RunnableLambda(format_answer)
# ──────────────────────────────────────────────────────────────
# 3. 체인 실행 및 결과 출력
# ──────────────────────────────────────────────────────────────
# 최종 결과는 format_answer가 반환한 포맷팅된 문자열이다.
result = formatted_chain.invoke({"input": "근로소득세 계산 방법은?"})
print(result)
---
파이프라인 전체 흐름
formatted_chain = rag_chain | RunnableLambda(format_answer)
{"input": "근로소득세 계산 방법은?"}
│
▼
┌──────────────────────────────────────────────┐
│ rag_chain (create_retrieval_chain) │
│ │
│ retriever → 문서 검색 → combine_chain → LLM │
│ │
│ 출력: │
│ { │
│ "input": "근로소득세 계산 방법은?", │
│ "context": [Doc1, Doc2, Doc3, Doc4], │
│ "answer": "근로소득세는 총급여에서..." │
│ } │
└──────────────────┬───────────────────────────┘
│ 파이프(|) 연산자로 전달
▼
┌──────────────────────────────────────────────┐
│ RunnableLambda(format_answer) │
│ │
│ 1. result["answer"] → 답변 추출 │
│ 2. result["context"] → 출처 추출 + 중복 제거 │
│ 3. 마크다운 형식으로 조합 │
│ │
│ 출력: │
│ "## 답변 │
│ 근로소득세는 총급여에서... │
│ │
│ ## 참조 출처 │
│ - ../../tax_docs/tax_with_markdown.docx │
│ │
│ ## 참조 문서 수: 4개" │
└──────────────────┬───────────────────────────┘
│
▼
print(result) → 포맷팅된 문자열 출력
RunnableLambda vs 일반 함수 호출
# ──────────────────────────────────────────────
# 이 두 코드는 결과는 동일하지만 구조적 차이가 있다.
# [참고용 비교 - 실행 코드가 아닙니다]
# ──────────────────────────────────────────────
# 방법 1: RunnableLambda로 체인에 포함 (권장)
formatted_chain = rag_chain | RunnableLambda(format_answer)
result = formatted_chain.invoke({"input": "질문"})
# 방법 2: 함수를 별도로 호출
raw_result = rag_chain.invoke({"input": "질문"})
result = format_answer(raw_result)
# ──────────────────────────────────────────────
# 방법 1의 장점:
# - 하나의 Runnable 체인으로 통합 → stream(), batch(), ainvoke() 모두 지원
# - LangSmith 추적 시 후처리 단계도 트레이스에 포함
# - 다른 체인과 추가 조합이 가능 (| 로 계속 연결)
# ──────────────────────────────────────────────
8.3. 다중 검색기 결합 (Ensemble Retriever)
- 여러 검색기의 결과를 결합하여 검색 품질을 높인다.
# ══════════════════════════════════════════════════════════════
# 다중 검색기 결합 (Ensemble Retriever)
# ══════════════════════════════════════════════════════════════ # 단일 검색 방식은 각각 고유한 약점을 가진다:
# - 벡터 검색: 의미적 유사도는 높지만, 고유명사·법률 조항 번호 등 # 정확한 키워드 매칭에 약하다.
# - BM25 검색: 키워드 매칭은 정확하지만, 동의어·유사 표현 등
# 의미적 유사성을 이해하지 못한다.
#
# EnsembleRetriever는 서로 다른 검색 방식의 결과를 가중 합산하여
# 각 방식의 단점을 보완하는 하이브리드 검색을 구현한다.
# 내부적으로 Reciprocal Rank Fusion(RRF) 알고리즘을 사용하여
# 각 검색기의 순위 결과를 통합한다.
# ══════════════════════════════════════════════════════════════
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
# ──────────────────────────────────────────────────────────────
# 1. BM25 키워드 검색기 생성
# ──────────────────────────────────────────────────────────────
# BM25(Best Matching 25)는 TF-IDF를 개선한 전통적인 키워드 기반 검색 알고리즘이다.
# 임베딩 벡터를 사용하지 않으며, 단어의 출현 빈도와 문서 길이를 고려하여 관련도를 계산한다.
# 장점: "제55조" 같은 정확한 키워드 매칭에 강하다.
# 단점: "세금" ↔ "조세" 같은 의미적 유사성을 인식하지 못한다.
bm25_retriever = BM25Retriever.from_documents(chunks, k=4) # 청크를 직접 인덱싱
# ──────────────────────────────────────────────────────────────
# 2. 벡터 유사도 검색기 생성
# ──────────────────────────────────────────────────────────────
# 앞서 생성한 Chroma DB에서 코사인 유사도 기반 검색기를 생성한다.
# 장점: "세금 계산" ↔ "조세 산출" 같은 의미적 유사성을 이해한다.
# 단점: "제55조"처럼 특정 키워드를 정확히 매칭해야 하는 경우 부정확할 수 있다.
vector_retriever = db.as_retriever(search_kwargs={"k": 4})
# ──────────────────────────────────────────────────────────────
# 3. 앙상블 검색기 구성 (하이브리드 검색)
# ──────────────────────────────────────────────────────────────
# 두 검색기의 결과를 weights 비율에 따라 통합한다.
# 내부적으로 Reciprocal Rank Fusion(RRF) 알고리즘으로 순위를 재정렬한다.
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.4, 0.6] # BM25 40% + 벡터 60%
# 의미 검색에 더 높은 가중치를 부여
# 합계는 반드시 1.0이어야 한다.
)
# ──────────────────────────────────────────────────────────────
# 4. 앙상블 검색기로 RAG 체인 구성 및 실행
# ──────────────────────────────────────────────────────────────
# create_retrieval_chain의 첫 번째 인자만 ensemble_retriever로 교체하면
# 나머지 구조(combine_chain, 프롬프트, LLM)는 동일하게 재사용된다.
hybrid_chain = create_retrieval_chain(ensemble_retriever, combine_chain)
# "제55조 세율 규정"은 법률 조항 번호(키워드)와 의미(세율 규정)가 결합된 질의이다.
# BM25는 "제55조"를 정확히 매칭하고, 벡터 검색은 "세율 규정"의 의미를 파악한다.
# 두 결과가 통합되어 단일 검색보다 높은 품질의 검색 결과를 기대할 수 있다.
result = hybrid_chain.invoke({"input": "제55조 세율 규정은?"})
print(result["answer"])
---
앙상블 검색 내부 동작
질의: "제55조 세율 규정은?"
│
├──────────────────────┬──────────────────────┐
▼ ▼ │
┌───────────────┐ ┌───────────────┐ │
│ BM25 검색기 │ │ 벡터 검색기 │ │
│ (키워드 매칭) │ │ (의미 유사도) │ │
└───────┬───────┘ └───────┬───────┘ │
│ │ │
▼ ▼ │
순위 문서 순위 문서 │
1위 청크A ("제55조") 1위 청크D ("세율 구간") │
2위 청크B ("제55조") 2위 청크A ("제55조") │
3위 청크E ("제56조") 3위 청크F ("과세표준") │
4위 청크G ("제54조") 4위 청크B ("제55조") │
│ │ │
└─────────┬──────────┘ │
▼ │
┌──────────────────────────┐ │
│ Reciprocal Rank Fusion │ │
│ │ │
│ 가중 점수 계산: │ │
│ 청크A: 0.4×(1위) + 0.6×(2위) = 높음 │ │
│ 청크B: 0.4×(2위) + 0.6×(4위) = 중간 │ │
│ 청크D: 0.4×(없음) + 0.6×(1위) = 중간 │ │
│ 청크F: 0.4×(없음) + 0.6×(3위) = 낮음 │ │
│ │ │
│ 최종 순위: │ │
│ 1위 청크A ← 양쪽 모두 상위권 │
│ 2위 청크D │ │
│ 3위 청크B │ │
│ 4위 청크F │ │
└──────────────────────────┘ │
│ │
▼ │
[상위 k개 문서 반환] │
검색 방식별 강점 비교
질의 유형 │ BM25 │ 벡터 │ 앙상블
─────────────────────────────┼───────────┼───────────┼───────────
"제55조" (정확한 키워드) │ ★★★ │ ★☆☆ │ ★★★
"세금 줄이는 방법" (의미 검색) │ ★☆☆ │ ★★★ │ ★★★
"제55조 세율 규정" (복합 질의) │ ★★☆ │ ★★☆ │ ★★★
- 실무 팁: 법률 문서, 기술 문서처럼 정확한 키워드 매칭이 중요한 도메인에서는 BM25와 벡터 검색을 결합한 하이브리드 검색이 단일 벡터 검색보다 우수한 성능을 보인다.
9. 디버깅과 성능 최적화
9.1. 체인 실행 과정 확인
# ══════════════════════════════════════════════════════════════
# 디버깅과 성능 최적화
# ══════════════════════════════════════════════════════════════ # RAG 체인은 여러 컴포넌트(retriever, 프롬프트, LLM, 파서)가
# 파이프라인으로 연결되어 있어, 문제 발생 시 어느 단계에서 # 오류가 발생했는지 파악하기 어려울 수 있다.
# LangChain의 verbose 모드를 활용하면 각 단계의 입출력을
# 콘솔에 실시간으로 출력하여 디버깅을 용이하게 할 수 있다.
# ══════════════════════════════════════════════════════════════
# ──────────────────────────────────────────────────────────────
# 9.1 체인 실행 과정 확인 (verbose 모드)
# ──────────────────────────────────────────────────────────────
# set_verbose(True)를 설정하면 LangChain의 모든 체인 및 컴포넌트가
# 실행될 때마다 다음 정보를 콘솔에 자동으로 출력한다:
# - 각 체인/컴포넌트의 시작과 종료
# - 각 단계에 전달된 입력값
# - 각 단계가 반환한 출력값
# - LLM에 전달된 최종 프롬프트 전문
from langchain.globals import set_verbose
# verbose 모드 활성화
set_verbose(True)
# 체인 실행 — 각 단계의 입출력이 콘솔에 상세히 출력된다.
# 출력되는 정보:
# 1) Retriever 단계: 검색 질의 → 반환된 문서 목록
# 2) StuffDocumentsChain 단계: 병합된 context + 완성된 프롬프트
# 3) LLM 단계: 모델에 전달된 메시지 → 생성된 응답
result = rag_chain.invoke({"input": "소득세 계산법은?"})
# 디버깅이 완료되면 반드시 verbose를 끈다.
# 운영 환경에서 verbose를 켜두면 로그 과다로 성능이 저하될 수 있다.
set_verbose(False)
---
verbose 출력 예시
set_verbose(True) 활성화 시 콘솔 출력 예시
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
> Entering new RetrievalChain chain...
[Retriever]
Query: "소득세 계산법은?"
Retrieved 4 documents:
- Document(page_content="소득세는 과세표준에...", metadata={...})
- Document(page_content="근로소득금액에서...", metadata={...})
- Document(page_content="종합소득세 계산...", metadata={...})
- Document(page_content="세율 구간별로...", metadata={...})
[StuffDocumentsChain]
Input:
context: "소득세는 과세표준에...\n\n근로소득금액에서..."
input: "소득세 계산법은?"
[ChatOpenAI]
Prompt:
System: 다음 컨텍스트를 기반으로 질문에 답변하세요...
Human: 소득세 계산법은?
Response:
"소득세 계산은 다음과 같은 단계로 이루어집니다..."
> Finished RetrievalChain chain.
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
LangChain 디버깅 도구 비교
┌──────────────────┬──────────────┬──────────────────────────────────┐
│ 도구 │ 설정 방법 │ 용도 │
├──────────────────┼──────────────┼──────────────────────────────────┤
│ set_verbose │ 코드 1줄 │ 각 단계의 입출력 텍스트 확인 │
│ set_debug │ 코드 1줄 │ verbose보다 더 상세한 내부 동작 │
│ LangSmith │ 환경변수 설정 │ 웹 UI 기반 트레이싱, 평가, 모니터링│
└──────────────────┴──────────────┴──────────────────────────────────┘
# [참고용 예시 - 실행 코드가 아닙니다]
# debug 모드: verbose보다 더 상세 (원시 API 호출, 토큰 수 등)
from langchain.globals import set_debug
set_debug(True)
# LangSmith: 웹 대시보드에서 시각적으로 트레이싱
# 환경변수 설정만으로 자동 연동
# LANGCHAIN_TRACING_V2=true
# LANGCHAIN_API_KEY=your_api_key
9.2. 검색 품질 확인
# ──────────────────────────────────────────────────────────────
# 9.2 검색 품질 확인
# ────────────────────────────────────────────────────────────── # RAG 시스템의 답변 품질은 검색 단계의 품질에 직접적으로 좌우된다.
# 아무리 뛰어난 LLM이라도 관련 없는 문서가 검색되면 정확한 답변을 생성할 수 없다. # 따라서 검색 결과에 기대한 핵심 키워드가 포함되어 있는지 정량적으로
# 평가하는 것이 RAG 파이프라인 최적화의 첫 단계이다.
# ──────────────────────────────────────────────────────────────
def evaluate_retrieval(chain, question, expected_keywords):
"""검색 품질을 평가하는 유틸리티 함수
Args:
chain: 평가할 RAG 체인 (create_retrieval_chain으로 생성된 Runnable)
question: 테스트 질문 문자열
expected_keywords: 검색 결과에 포함되어야 할 핵심 키워드 리스트
Returns:
result: RAG 체인의 전체 출력 딕셔너리 (후속 분석용)
"""
# RAG 체인 실행: 검색 + 답변 생성
result = chain.invoke({"input": question})
# ── 검색된 문서 평가 ──────────────────────────────────
# 검색된 모든 문서의 page_content를 하나의 문자열로 병합하여
# 키워드 존재 여부를 일괄 검사할 수 있도록 한다.
context_text = " ".join([doc.page_content for doc in result["context"]])
# 기대 키워드 중 검색 결과에 포함된 키워드와 누락된 키워드를 분리한다.
found = [kw for kw in expected_keywords if kw in context_text] # 적중 키워드
missed = [kw for kw in expected_keywords if kw not in context_text] # 미발견 키워드
# ── 평가 결과 출력 ──────────────────────────────────
print(f"질문: {question}")
print(f"검색 문서 수: {len(result['context'])}")
print(f"키워드 적중: {len(found)}/{len(expected_keywords)}") # 적중률 확인
if missed:
# 미발견 키워드가 있으면 출력 — 검색 전략 개선의 단서가 된다.
# 예: 특정 키워드가 자주 누락되면 chunk_size 조정이나
# 검색 방식 변경(similarity → mmr, 앙상블 등)을 검토한다.
print(f"미발견 키워드: {missed}")
print(f"답변 길이: {len(result['answer'])}자")
return result
# ──────────────────────────────────────────────────────────────
# 평가 실행
# ──────────────────────────────────────────────────────────────
# "근로소득세 계산"이라는 질문에 대해
# 검색 결과에 "근로소득", "세율", "과세표준" 키워드가 포함되는지 확인한다.
# 3개 모두 적중하면 검색 품질이 양호한 것으로 판단할 수 있다.
evaluate_retrieval(
rag_chain,
"근로소득세 계산",
["근로소득", "세율", "과세표준"]
)
---
출력 예시 및 해석
실행 결과 예시
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
질문: 근로소득세 계산
검색 문서 수: 4
키워드 적중: 3/3 ← 모든 키워드가 검색 결과에 포함됨
답변 길이: 342자
키워드 누락이 발생한 경우
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
질문: 근로소득세 계산
검색 문서 수: 4
키워드 적중: 2/3 ← 1개 키워드 누락
미발견 키워드: ['과세표준'] ← 이 키워드가 포함된 청크가 검색되지 않음
답변 길이: 215자
평가 결과에 따른 개선 방향
┌─────────────────────────┬──────────────────────────────────────────┐
│ 증상 │ 개선 방향 │
├─────────────────────────┼──────────────────────────────────────────┤
│ 키워드 적중률 낮음 │ chunk_size 축소 또는 chunk_overlap 증가 │
│ │ → 관련 내용이 하나의 청크에 온전히 포함 │
├─────────────────────────┼──────────────────────────────────────────┤
│ 특정 키워드만 반복 누락 │ BM25 앙상블 검색기 도입 │
│ │ → 키워드 매칭 능력 보완 │
├─────────────────────────┼──────────────────────────────────────────┤
│ 검색 문서는 적절하나 │ 프롬프트 개선 또는 LLM 모델 업그레이드 │
│ 답변 품질이 낮음 │ → 생성 단계의 문제 │
├─────────────────────────┼──────────────────────────────────────────┤
│ 유사한 문서만 중복 검색 │ MMR 검색으로 전환 │
│ │ → 검색 결과의 다양성 확보 │
└─────────────────────────┴──────────────────────────────────────────┘
9.3. 성능 최적화 체크리스트
| 항목 | 설명 | 권장값 |
| chunk_size | 검색 정밀도와 문맥 보존의 균형 | 500~1500 |
| chunk_overlap | 문맥 연속성 보장 | chunk_size의 10~15% |
| k (검색 문서 수) | 컨텍스트 크기와 비용의 균형 | 3~5 |
| 임베딩 모델 | 품질과 비용의 균형 | text-embedding-3-small (프로토타입), text-embedding-3-large (프로덕션) |
| LLM 모델 | 답변 품질과 비용의 균형 | gpt-4o-mini (일반), gpt-4o (고품질) |
| temperature | 답변의 일관성 | 0 (사실 기반), 0.3~0.7 (창의적) |
10. 실무 고급 프로젝트: 세법 Q&A 시스템
10.1. 요구사항 명세
- 프로젝트명: 세법 문서 기반 지능형 Q&A 시스템
- 기능 요구사항:
| ID | 요구사항 | 우선순위 |
| FR-01 | 세법 문서(DOCX)를 로딩하여 벡터 DB에 인덱싱 | 필수 |
| FR-02 | 사용자 질의에 대해 문서 기반 답변 생성 | 필수 |
| FR-03 | 답변과 함께 참조 문서 출처 반환 | 필수 |
| FR-04 | 대화 이력을 기반으로 후속 질의 처리 | 필수 |
| FR-05 | 답변에 신뢰도 표시 (검색 문서 적합성 기반) | 선택 |
| FR-06 | 키워드 사전 기반 질의 전처리 | 선택 |
| FR-07 | 하이브리드 검색 (벡터 + 키워드) | 선택 |
- 비기능 요구사항:
| ID | 요구사항 |
| NFR-01 | 응답 시간 5초 이내 |
| NFR-02 | 스트리밍 응답 지원 |
| NFR-03 | 검색 결과가 없을 때 적절한 안내 메시지 |
10.2. 시스템 설계
[사용자 질의]
│
├─→ [키워드 사전 변환] (FR-06)
│ │
│ ↓
├─→ [대화 이력 분석] (FR-04)
│ │
│ ↓
├─→ [하이브리드 검색] (FR-07)
│ ├─ BM25 키워드 검색
│ └─ 벡터 유사도 검색
│ │
│ ↓
└─→ [답변 생성 + 신뢰도] (FR-02, FR-05)
│
↓
[포맷된 응답 반환] (FR-03)
10.3. 구현
import os
from dotenv import load_dotenv, find_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import Docx2txtLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
# ============================================================
# 1. 환경 설정
# ============================================================
load_dotenv(find_dotenv(), override=True)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
# ============================================================
# 2. 문서 로딩 및 인덱싱 (FR-01)
# ============================================================
loader = Docx2txtLoader("../../tax_docs/tax_with_markdown.docx")
documents = loader.load()
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, chunk_overlap=150
)
chunks = splitter.split_documents(documents)
db = Chroma.from_documents(
documents=chunks,
embedding=embedding,
collection_name="tax_qa_system"
)
print(f"인덱싱 완료: {len(chunks)}개 청크")
# ============================================================
# 3. 하이브리드 검색기 구성 (FR-07)
# ============================================================
bm25_retriever = BM25Retriever.from_documents(chunks, k=3)
vector_retriever = db.as_retriever(search_kwargs={"k": 3})
hybrid_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.3, 0.7]
)
# ============================================================
# 4. 키워드 사전 기반 질의 전처리 (FR-06)
# ============================================================
KEYWORD_DICT = {
"직장인": "근로소득자",
"월급쟁이": "근로소득자",
"세금": "소득세",
"연봉": "총급여액",
"연말정산": "근로소득 연말정산",
}
def preprocess_query(query):
"""키워드 사전을 기반으로 질의를 변환"""
for colloquial, formal in KEYWORD_DICT.items():
query = query.replace(colloquial, formal)
return query
# ============================================================
# 5. 대화형 RAG 체인 구성 (FR-02, FR-04)
# ============================================================
# 질의 재구성 프롬프트
contextualize_prompt = ChatPromptTemplate.from_messages([
("system", "대화 이력과 최신 질문을 고려하여 독립적인 검색 질문으로 재구성하세요."),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
])
history_aware_retriever = create_history_aware_retriever(
llm, hybrid_retriever, contextualize_prompt
)
# 답변 생성 프롬프트 (FR-05: 신뢰도 포함)
answer_prompt = ChatPromptTemplate.from_messages([
("system", """당신은 세법 전문 AI 어시스턴트입니다.
규칙:
1. 컨텍스트에 기반하여 정확하게 답변합니다.
2. 컨텍스트에 정보가 부족하면 "제공된 문서에서 충분한 정보를 찾을 수 없습니다."라고 안내합니다.
3. 관련 법률 조항이 있으면 명시합니다.
4. 답변 마지막에 [신뢰도: 높음/중간/낮음]을 표시합니다.
- 높음: 컨텍스트에 직접적인 답변이 있음
- 중간: 컨텍스트에서 유추 가능
- 낮음: 컨텍스트 관련성이 떨어짐
컨텍스트:
{context}"""),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
])
combine_chain = create_stuff_documents_chain(llm, answer_prompt)
conversational_rag = create_retrieval_chain(history_aware_retriever, combine_chain)
# ============================================================
# 6. Q&A 시스템 실행 (FR-02, FR-03, FR-04)
# ============================================================
class TaxQASystem:
"""세법 Q&A 시스템 클래스"""
def __init__(self, chain):
self.chain = chain
self.chat_history = []
def ask(self, question):
"""질문을 처리하고 답변을 반환"""
# 키워드 전처리
processed = preprocess_query(question)
# RAG 체인 실행
result = self.chain.invoke({
"input": processed,
"chat_history": self.chat_history
})
# 대화 이력 업데이트
self.chat_history.extend([
HumanMessage(content=question),
AIMessage(content=result["answer"])
])
# 포맷된 결과 반환
return {
"question": question,
"processed_query": processed,
"answer": result["answer"],
"sources": [
{
"content": doc.page_content[:100] + "...",
"source": doc.metadata.get("source", "N/A")
}
for doc in result["context"]
],
"num_sources": len(result["context"])
}
def stream_ask(self, question):
"""스트리밍 방식으로 답변 반환 (NFR-02)"""
processed = preprocess_query(question)
full_answer = ""
for chunk in self.chain.stream({
"input": processed,
"chat_history": self.chat_history
}):
if "answer" in chunk:
full_answer += chunk["answer"]
print(chunk["answer"], end="", flush=True)
print()
self.chat_history.extend([
HumanMessage(content=question),
AIMessage(content=full_answer)
])
return full_answer
def reset_history(self):
"""대화 이력 초기화"""
self.chat_history = []
# ============================================================
# 7. 시스템 테스트
# ============================================================
qa = TaxQASystem(conversational_rag)
# 테스트 1: 기본 질의
print("=" * 60)
print("[테스트 1] 기본 질의")
print("=" * 60)
result = qa.ask("직장인의 세금 계산 방법은?")
print(f"원본 질문: {result['question']}")
print(f"변환 질문: {result['processed_query']}")
print(f"답변:\n{result['answer']}")
print(f"참조 문서: {result['num_sources']}개")
# 테스트 2: 후속 질의 (대화 이력 활용)
print("\n" + "=" * 60)
print("[테스트 2] 후속 질의")
print("=" * 60)
result2 = qa.ask("과세표준 구간은 어떻게 되나요?")
print(f"답변:\n{result2['answer']}")
# 테스트 3: 스트리밍 출력
print("\n" + "=" * 60)
print("[테스트 3] 스트리밍 출력")
print("=" * 60)
qa.reset_history()
qa.stream_ask("소득세 신고 기한을 알려주세요")
```
10.4 시스템 평가
```python
# 평가용 질문-기대 키워드 세트
test_cases = [
{
"question": "근로소득세 계산 방법은?",
"expected_keywords": ["근로소득", "세율", "과세표준"]
},
{
"question": "종합소득세 세율 구간은?",
"expected_keywords": ["세율", "과세표준", "종합소득"]
},
]
qa_eval = TaxQASystem(conversational_rag)
for tc in test_cases:
result = qa_eval.ask(tc["question"])
context_text = " ".join([s["content"] for s in result["sources"]])
found = [kw for kw in tc["expected_keywords"] if kw in context_text or kw in result["answer"]]
score = len(found) / len(tc["expected_keywords"])
print(f"\nQ: {tc['question']}")
print(f" 키워드 적중률: {score:.0%} ({len(found)}/{len(tc['expected_keywords'])})")
print(f" 답변 길이: {len(result['answer'])}자")
qa_eval.reset_history()
11. 정리
11.1. 핵심 요약
| 항목 | 설명 |
| create_retrieval_chain | 검색(Retriever) + 생성(LLM)을 결합한 RAG 체인 생성 헬퍼 |
| create_stuff_documents_chain | 검색된 문서를 프롬프트에 삽입하여 LLM에 전달하는 체인 |
| create_history_aware_retriever | 대화 이력을 고려한 질의 재구성 retriever |
| 입력 | {"input": "질문"} |
| 출력 | {"input": "질문", "context": [문서들], "answer": "답변"} |
| 프롬프트 필수 변수 | {context} (문서 삽입), {input} (질의) |
11.2. 사용 패턴 결정 트리
RAG 체인 구성이 필요한가?
├─ YES → 대화 이력이 필요한가?
│ ├─ YES → create_history_aware_retriever + create_retrieval_chain
│ └─ NO → create_retrieval_chain (기본)
└─ NO → 단순 LLM 호출 (prompt | llm | parser)
11.3. 마이그레이션 가이드 (RetrievalQA → create_retrieval_chain)
● # ══════════════════════════════════════════════════════════════
# 프로젝트: 세법 문서 기반 지능형 Q&A 시스템
# ══════════════════════════════════════════════════════════════ # 이 시스템은 지금까지 학습한 개별 컴포넌트들을 하나의 통합 파이프라인으로
# 조립한 종합 프로젝트이다. 각 기능 요구사항(FR)과 비기능 요구사항(NFR)에
# 대응하는 컴포넌트가 아키텍처 내에서 어떤 위치에 배치되는지 확인할 수 있다.
#
# [사용자 질의]
# │
# ├─→ [키워드 사전 변환] (FR-06)
# │ │
# │ ↓
# ├─→ [대화 이력 분석] (FR-04)
# │ │
# │ ↓
# ├─→ [하이브리드 검색] (FR-07)
# │ ├─ BM25 키워드 검색
# │ └─ 벡터 유사도 검색
# │ │
# │ ↓
# └─→ [답변 생성 + 신뢰도] (FR-02, FR-05)
# │
# ↓
# [포맷된 응답 반환] (FR-03)
# ══════════════════════════════════════════════════════════════
import os
from dotenv import load_dotenv, find_dotenv
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import Docx2txtLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain.chains.history_aware_retriever import create_history_aware_retriever
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
# ============================================================
# 1. 환경 설정
# ============================================================
# .env 파일에서 OPENAI_API_KEY 등 환경변수를 로드한다.
# find_dotenv()은 현재 디렉토리부터 상위로 올라가며 .env 파일을 자동 탐색한다.
# override=True로 설정하면 이미 존재하는 환경변수도 .env 값으로 덮어쓴다.
load_dotenv(find_dotenv(), override=True)
# LLM: temperature=0으로 설정하여 일관된 답변을 보장한다.
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# 임베딩: 텍스트를 1536차원 벡터로 변환하는 경량 모델
embedding = OpenAIEmbeddings(model="text-embedding-3-small")
# ============================================================
# 2. 문서 로딩 및 인덱싱 (FR-01)
# ============================================================
# FR-01: 세법 문서를 로딩하고 검색 가능한 형태로 인덱싱한다.
# 이 단계는 시스템 초기화 시 1회만 실행되며,
# 문서 로딩 → 청크 분할 → 벡터 변환 → DB 저장의 순서로 진행된다.
# ────────────────────────────────────────────────────────────
# .docx 파일을 읽어 Document 객체로 변환
loader = Docx2txtLoader("../../tax_docs/tax_with_markdown.docx")
documents = loader.load()
# 긴 문서를 1000자 단위로 분할 (150자 중복으로 문맥 유지)
splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, chunk_overlap=150
)
chunks = splitter.split_documents(documents)
# 각 청크를 임베딩 벡터로 변환하여 Chroma에 저장
db = Chroma.from_documents(
documents=chunks,
embedding=embedding,
collection_name="tax_qa_system"
)
print(f"인덱싱 완료: {len(chunks)}개 청크")
# ============================================================
# 3. 하이브리드 검색기 구성 (FR-07)
# ============================================================
# FR-07: BM25 키워드 검색과 벡터 유사도 검색을 결합하여
# 정확한 키워드 매칭과 의미적 유사도 검색을 동시에 수행한다.
#
# 가중치 배분 근거:
# - BM25(0.3): "제55조" 같은 법률 조항 번호의 정확한 매칭 담당
# - 벡터(0.7): "세금 줄이는 방법" 같은 의미적 검색 담당
# - 세법 질의는 의미 검색의 비중이 더 크므로 벡터에 높은 가중치 부여
# ────────────────────────────────────────────────────────────
# BM25: 단어 출현 빈도 기반의 키워드 검색기
bm25_retriever = BM25Retriever.from_documents(chunks, k=3)
# 벡터: 코사인 유사도 기반의 의미 검색기
vector_retriever = db.as_retriever(search_kwargs={"k": 3})
# 앙상블: 두 검색기의 결과를 RRF 알고리즘으로 통합
hybrid_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.3, 0.7]
)
# ============================================================
# 4. 키워드 사전 기반 질의 전처리 (FR-06)
# ============================================================
# FR-06: 사용자의 구어체/일상 표현을 세법 전문 용어로 변환한다.
# 사용자가 "직장인 세금"이라고 입력해도 내부적으로는
# "근로소득자 소득세"로 변환되어 검색 정확도가 향상된다.
#
# 이 전처리는 LLM 호출 없이 단순 문자열 치환으로 수행되므로
# 추가 비용이나 지연 시간이 발생하지 않는다.
# ────────────────────────────────────────────────────────────
KEYWORD_DICT = {
"직장인": "근로소득자",
"월급쟁이": "근로소득자",
"세금": "소득세",
"연봉": "총급여액",
"연말정산": "근로소득 연말정산",
}
def preprocess_query(query):
"""키워드 사전을 기반으로 구어체 표현을 전문 용어로 변환"""
for colloquial, formal in KEYWORD_DICT.items():
query = query.replace(colloquial, formal)
return query
# ============================================================
# 5. 대화형 RAG 체인 구성 (FR-02, FR-04)
# ============================================================
# FR-02: 검색된 문서를 기반으로 정확한 답변을 생성한다.
# FR-04: 대화 이력을 분석하여 맥락 의존적 후속 질문을 처리한다.
#
# 이 체인은 두 단계로 구성된다:
# 1단계: history_aware_retriever — 질의 재구성 + 하이브리드 검색
# 2단계: combine_chain — 검색 결과 + 대화 이력 기반 답변 생성
# ────────────────────────────────────────────────────────────
# ── 5-1. 질의 재구성 프롬프트 ────────────────────────────
# 대화 이력이 있을 때 "그건 어떻게 되나요?" 같은 모호한 질문을
# "근로소득세 과세표준 구간은?" 같은 독립적 질문으로 재구성한다.
contextualize_prompt = ChatPromptTemplate.from_messages([
("system", "대화 이력과 최신 질문을 고려하여 독립적인 검색 질문으로 재구성하세요."),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
])
# 질의 재구성 + 하이브리드 검색을 하나의 retriever로 통합
history_aware_retriever = create_history_aware_retriever(
llm, hybrid_retriever, contextualize_prompt
)
# ── 5-2. 답변 생성 프롬프트 (FR-05: 신뢰도 포함) ───────
# FR-05: 답변 말미에 신뢰도를 표시하여 사용자가
# 답변의 근거 수준을 판단할 수 있도록 한다.
# 시스템 프롬프트에 구체적인 신뢰도 판단 기준을 명시하여
# LLM이 일관된 기준으로 신뢰도를 평가하도록 유도한다.
answer_prompt = ChatPromptTemplate.from_messages([
("system", """당신은 세법 전문 AI 어시스턴트입니다.
규칙:
1. 컨텍스트에 기반하여 정확하게 답변합니다.
2. 컨텍스트에 정보가 부족하면 "제공된 문서에서 충분한 정보를 찾을 수 없습니다."라고 안내합니다.
3. 관련 법률 조항이 있으면 명시합니다.
4. 답변 마지막에 [신뢰도: 높음/중간/낮음]을 표시합니다.
- 높음: 컨텍스트에 직접적인 답변이 있음
- 중간: 컨텍스트에서 유추 가능
- 낮음: 컨텍스트 관련성이 떨어짐
컨텍스트:
{context}"""),
MessagesPlaceholder("chat_history"), # 이전 대화 내역 (맥락 유지)
("human", "{input}"),
])
# ── 5-3. 최종 RAG 체인 조립 ──────────────────────────────
# combine_chain: 검색 문서 + 대화 이력 → LLM → 답변(신뢰도 포함)
combine_chain = create_stuff_documents_chain(llm, answer_prompt)
# conversational_rag: history_aware_retriever + combine_chain
# 입력: {"input": str, "chat_history": List[BaseMessage]}
# 출력: {"input": str, "chat_history": ..., "context": List[Document], "answer": str}
conversational_rag = create_retrieval_chain(history_aware_retriever, combine_chain)
# ============================================================
# 6. Q&A 시스템 실행 (FR-02, FR-03, FR-04)
# ============================================================
# TaxQASystem 클래스는 지금까지 구성한 모든 컴포넌트를 하나의
# 사용자 인터페이스로 캡슐화한다.
#
# 주요 기능:
# - ask(): 질의 전처리 → RAG 실행 → 이력 관리 → 포맷된 결과 반환
# - stream_ask(): 실시간 스트리밍 출력 (NFR-02: 응답 지연 최소화)
# - reset_history(): 새로운 대화 세션 시작
#
# 내부 처리 흐름:
# 사용자 질문
# → preprocess_query() : "직장인 세금" → "근로소득자 소득세"
# → conversational_rag : 질의 재구성 → 하이브리드 검색 → 답변 생성
# → chat_history 업데이트 : 후속 질문 대비
# → 포맷된 딕셔너리 반환 : 질문, 변환 질문, 답변, 출처 정보
# ────────────────────────────────────────────────────────────
class TaxQASystem:
"""세법 Q&A 시스템 클래스
대화 이력을 내부적으로 관리하면서,
질의 전처리 → RAG 실행 → 결과 포맷팅까지의 전 과정을 캡슐화한다.
"""
def __init__(self, chain):
self.chain = chain
self.chat_history = [] # HumanMessage/AIMessage 리스트
def ask(self, question):
"""질문을 처리하고 구조화된 답변을 반환
처리 순서:
1. 키워드 사전 기반 전처리 (FR-06)
2. RAG 체인 실행 (검색 + 생성)
3. 대화 이력 업데이트 (FR-04)
4. 포맷된 결과 딕셔너리 반환 (FR-03)
"""
# 1. 키워드 전처리: 구어체 → 전문 용어
processed = preprocess_query(question)
# 2. RAG 체인 실행: 전처리된 질의로 검색 및 답변 생성
result = self.chain.invoke({
"input": processed,
"chat_history": self.chat_history
})
# 3. 대화 이력 업데이트: 원본 질문과 답변을 저장하여 후속 질문에 활용
# 원본 질문(question)을 저장하는 이유: 대화 이력은 LLM이 읽으므로
# 자연어 원문이 맥락 파악에 더 적합하다.
self.chat_history.extend([
HumanMessage(content=question),
AIMessage(content=result["answer"])
])
# 4. 포맷된 결과 반환 (FR-03)
# 원본 질문, 변환된 질문, 답변, 출처 정보를 구조화하여 반환한다.
# 출처 정보에는 각 문서의 앞 100자 미리보기와 파일 경로를 포함한다.
return {
"question": question, # 사용자 원본 질문
"processed_query": processed, # 키워드 변환된 질문
"answer": result["answer"], # LLM 생성 답변 (신뢰도 포함)
"sources": [ # 참조 문서 목록
{
"content": doc.page_content[:100] + "...",
"source": doc.metadata.get("source", "N/A")
}
for doc in result["context"]
],
"num_sources": len(result["context"]) # 참조 문서 수
}
def stream_ask(self, question):
"""스트리밍 방식으로 답변을 실시간 출력 (NFR-02)
토큰이 생성될 때마다 즉시 화면에 출력하여
사용자의 체감 대기 시간을 최소화한다.
"""
# 키워드 전처리
processed = preprocess_query(question)
# 스트리밍 실행: 토큰 단위로 chunk를 수신하며 즉시 출력
full_answer = ""
for chunk in self.chain.stream({
"input": processed,
"chat_history": self.chat_history
}):
if "answer" in chunk:
full_answer += chunk["answer"]
print(chunk["answer"], end="", flush=True)
print() # 스트리밍 완료 후 줄바꿈
# 대화 이력 업데이트 (스트리밍 완료 후 전체 답변으로 저장)
self.chat_history.extend([
HumanMessage(content=question),
AIMessage(content=full_answer)
])
return full_answer
def reset_history(self):
"""대화 이력을 초기화하여 새로운 대화 세션을 시작"""
self.chat_history = []
# ============================================================
# 7. 시스템 테스트
# ============================================================
# 3가지 시나리오로 시스템의 주요 기능을 검증한다:
# 테스트 1: 키워드 전처리 + 기본 Q&A (FR-01, FR-02, FR-06)
# 테스트 2: 대화 이력 기반 후속 질의 (FR-04)
# 테스트 3: 스트리밍 출력 (NFR-02)
# ────────────────────────────────────────────────────────────
qa = TaxQASystem(conversational_rag)
# ── 테스트 1: 기본 질의 ──────────────────────────────────
# "직장인"→"근로소득자", "세금"→"소득세"로 변환되는지 확인한다.
# 키워드 전처리 덕분에 전문 용어로 검색이 수행되어
# 더 정확한 문서가 검색될 것으로 기대한다.
print("=" * 60)
print("[테스트 1] 기본 질의")
print("=" * 60)
result = qa.ask("직장인의 세금 계산 방법은?")
print(f"원본 질문: {result['question']}") # "직장인의 세금 계산 방법은?"
print(f"변환 질문: {result['processed_query']}") # "근로소득자의 소득세 계산 방법은?"
print(f"답변:\n{result['answer']}")
print(f"참조 문서: {result['num_sources']}개")
# ── 테스트 2: 후속 질의 (대화 이력 활용) ──────────────────
# 테스트 1의 대화 이력이 남아 있으므로,
# "과세표준 구간"이라는 질문이 앞선 "소득세" 맥락과 연결된다.
# history_aware_retriever가 질의를 재구성:
# "과세표준 구간은?" → "소득세 과세표준 구간은?"
print("\n" + "=" * 60)
print("[테스트 2] 후속 질의")
print("=" * 60)
result2 = qa.ask("과세표준 구간은 어떻게 되나요?")
print(f"답변:\n{result2['answer']}")
# ── 테스트 3: 스트리밍 출력 ───────────────────────────────
# 대화 이력을 초기화한 뒤 새로운 세션으로 스트리밍 테스트를 수행한다.
# 토큰이 생성되는 즉시 화면에 출력되어 실시간 타이핑 효과를 확인한다.
print("\n" + "=" * 60)
print("[테스트 3] 스트리밍 출력")
print("=" * 60)
qa.reset_history()
qa.stream_ask("소득세 신고 기한을 알려주세요")
---
# ============================================================
# 10.4 시스템 평가
# ============================================================
# 시스템의 검색 품질을 정량적으로 측정한다.
# 사전 정의된 질문-키워드 쌍을 사용하여
# 검색 결과와 답변에 핵심 키워드가 포함되는지 확인한다.
#
# 평가 방식:
# - 각 테스트 케이스에 대해 기대 키워드의 적중률을 계산
# - 검색 결과(context)뿐 아니라 답변(answer)에서도 키워드를 탐색
# - 적중률이 낮은 케이스는 검색 전략 개선의 단서가 된다
# ────────────────────────────────────────────────────────────
# 평가용 질문-기대 키워드 세트
# 각 테스트 케이스는 질문과 해당 질문의 답변에 반드시 포함되어야 할
# 핵심 키워드를 쌍으로 정의한다.
test_cases = [
{
"question": "근로소득세 계산 방법은?",
"expected_keywords": ["근로소득", "세율", "과세표준"]
},
{
"question": "종합소득세 세율 구간은?",
"expected_keywords": ["세율", "과세표준", "종합소득"]
},
]
# 평가 전용 인스턴스 생성 (테스트 간 이력 오염 방지)
qa_eval = TaxQASystem(conversational_rag)
for tc in test_cases:
# RAG 체인 실행
result = qa_eval.ask(tc["question"])
# 검색된 문서의 미리보기 텍스트를 하나로 병합
context_text = " ".join([s["content"] for s in result["sources"]])
# 키워드 적중 판정: 검색 결과(context) 또는 답변(answer) 중
# 하나라도 포함되면 적중으로 판정한다.
# 답변에서도 검사하는 이유: LLM이 문서 내용을 요약·재구성하면서
# 원본과 다른 표현을 사용할 수 있기 때문이다.
found = [kw for kw in tc["expected_keywords"]
if kw in context_text or kw in result["answer"]]
# 적중률 계산 (0.0 ~ 1.0)
score = len(found) / len(tc["expected_keywords"])
# 평가 결과 출력
print(f"\nQ: {tc['question']}")
print(f" 키워드 적중률: {score:.0%} ({len(found)}/{len(tc['expected_keywords'])})")
print(f" 답변 길이: {len(result['answer'])}자")
# 테스트 간 대화 이력을 초기화하여 각 케이스를 독립적으로 평가
qa_eval.reset_history()
---
전체 시스템 아키텍처 최종 정리
┌─────────────────────────────────────────────────────────────────┐
│ TaxQASystem │
│ │
│ ask("직장인의 세금 계산 방법은?") │
│ │ │
│ │ ┌──────────────────────────────────┐ │
│ ├─→│ FR-06: 키워드 전처리 │ │
│ │ │ "직장인"→"근로소득자" │ │
│ │ │ "세금"→"소득세" │ │
│ │ └──────────────┬───────────────────┘ │
│ │ ↓ │
│ │ ┌──────────────────────────────────────────────────────┐ │
│ │ │ conversational_rag (create_retrieval_chain) │ │
│ │ │ │ │
│ │ │ ┌────────────────────────────────────────────────┐ │ │
│ │ │ │ history_aware_retriever │ │ │
│ │ │ │ │ │ │
│ │ │ │ FR-04: 대화 이력 → 질의 재구성 (LLM 1회차) │ │ │
│ │ │ │ ↓ │ │ │
│ │ │ │ FR-07: 하이브리드 검색 │ │ │
│ │ │ │ ├─ BM25 (0.3) ─┐ │ │ │
│ │ │ │ └─ 벡터 (0.7) ─┴→ RRF 통합 → 문서 반환 │ │ │
│ │ │ └───────────────────────────┬────────────────────┘ │ │
│ │ │ ↓ │ │
│ │ │ ┌────────────────────────────────────────────────┐ │ │
│ │ │ │ combine_chain (create_stuff_documents_chain) │ │ │
│ │ │ │ │ │ │
│ │ │ │ FR-02: context + chat_history → LLM 2회차 │ │ │
│ │ │ │ FR-05: 답변 + [신뢰도: 높음/중간/낮음] │ │ │
│ │ │ └───────────────────────────┬────────────────────┘ │ │
│ │ └──────────────────────────────┼───────────────────────┘ │
│ │ ↓ │
│ │ ┌──────────────────────────────────────────────────────┐ │
│ └─→│ FR-03: 결과 포맷팅 │ │
│ │ FR-04: 대화 이력 업데이트 │ │
│ │ → {"question", "processed_query", "answer", │ │
│ │ "sources", "num_sources"} │ │
│ └──────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
- 마이그레이션 핵심: `query` → `input`, `result` → `answer`, `source_documents` → `context`로 키 이름이 변경되었다. 프롬프트를 직접 제어할 수 있어 커스터마이징이 훨씬 유연하다.
'Study > LangChain' 카테고리의 다른 글
| 6. LangChain 검증, 고도화, 프로덕션 가이드 (0) | 2026.03.07 |
|---|---|
| 5. LangChain 에이전트 구축 가이드 - Tool 연동과 에이전트 패턴 (0) | 2026.03.03 |
| 4. LangChain 기본 활용 가이드 - 기본부터 체인 구성까지 (0) | 2026.02.23 |
| RAG를 활용한 LLM Application 개발 (feat. LangChain) - 학습 후기 (2) | 2026.01.18 |
| [인프런] 입문자를 위한 LangChain 기초 (0) | 2025.12.25 |
댓글