RAG 시스템 6편 - LangChain으로 RAG 구현 완벽 가이드

LangChain을 활용한 RAG(Retrieval-Augmented Generation) 시스템의 실전 구현 방법을 배웁니다. 문서 로딩부터 벡터 저장소 구축, 검색 기반 질의응답까지 단계별로 알아봅니다. 실무에서 바로 활용 가능한 코드와 팁을 제공합니다.

LangChain RAG 시스템 기본 구조 - 전체 아키텍처 이해하기
Document Loaders - 다양한 파일 형식 처리하기
Text Splitters - 문서를 효율적으로 청크화하기
Vector Stores - 벡터 데이터베이스 구축과 활용
Retrieval 전략 - 검색 품질 최적화하기
RetrievalQA Chain - 질의응답 체인 구성하기
ConversationalRetrievalChain - 대화 맥락을 유지하는 RAG
Custom Prompts - 프롬프트 최적화 전략
답을 모르면 "해당 정보를 찾을 수 없습니다. 추가 문의는 support@example.com으로 연락주세요"라고 하세요
Error Handling과 Fallbacks - 안정성 확보하기
Performance Optimization - 속도와 비용 최적화

1. LangChain RAG 시스템 기본 구조 - 전체 아키텍처 이해하기

시작하며

여러분이 사내 문서 검색 시스템을 만들어야 하는데, GPT만으로는 최신 정보를 반영할 수 없어서 고민한 적 있나요? 수천 개의 PDF 문서에서 정확한 답변을 찾아내야 하는데, 어디서부터 시작해야 할지 막막하셨을 겁니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. LLM은 훈련 시점까지의 데이터만 알고 있기 때문에, 여러분의 회사 내부 문서나 최신 뉴스 같은 정보는 전혀 모릅니다.

그래서 잘못된 답변을 하거나 아예 답변을 못 하는 경우가 생깁니다. 바로 이럴 때 필요한 것이 LangChain을 이용한 RAG 시스템입니다.

외부 문서를 검색해서 LLM에게 컨텍스트로 제공하면, 정확하고 최신의 답변을 얻을 수 있습니다.

개요

간단히 말해서, LangChain RAG 시스템은 "검색 + 생성"의 두 단계로 작동하는 질의응답 시스템입니다. 왜 이 시스템이 필요한지 실무 관점에서 설명하자면, LLM 단독으로는 할 수 없는 일들을 가능하게 해줍니다.

예를 들어, 회사의 내부 규정 문서를 기반으로 한 챗봇, 최신 제품 매뉴얼 검색 시스템, 법률 문서 분석 도구 같은 경우에 매우 유용합니다. 기존에는 키워드 기반 검색으로 문서를 찾아서 사람이 직접 읽어야 했다면, 이제는 자연어로 질문하면 AI가 관련 문서를 찾아서 요약된 답변까지 제공할 수 있습니다.

LangChain RAG의 핵심 특징은 세 가지입니다: (1) 문서를 벡터로 변환하여 의미 기반 검색이 가능하고, (2) 여러 데이터 소스를 통합할 수 있으며, (3) LLM과의 연동이 간단합니다. 이러한 특징들이 중요한 이유는 복잡한 시스템을 빠르게 구축하고, 유지보수도 쉽게 할 수 있기 때문입니다.

코드 예제

from langchain.document_loaders import TextLoader
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA

# 문서 로딩: 텍스트 파일을 읽어옵니다
loader = TextLoader("company_docs.txt", encoding="utf-8")
documents = loader.load()

# 임베딩 모델 초기화: 텍스트를 벡터로 변환
embeddings = OpenAIEmbeddings()

# 벡터 저장소 생성: 문서를 벡터화하여 저장
vectorstore = Chroma.from_documents(documents, embeddings)

# LLM 초기화: 답변 생성을 담당
llm = OpenAI(temperature=0)

# RAG 체인 구성: 검색 + 생성을 하나로 연결
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    return_source_documents=True  # 출처 문서도 함께 반환
)

# 질문하기
result = qa_chain({"query": "우리 회사의 휴가 정책은?"})
print(result["result"])

설명

이것이 하는 일: 이 코드는 텍스트 문서를 읽어서 검색 가능한 벡터 DB로 만들고, 질문에 대한 답변을 자동으로 생성하는 전체 파이프라인을 구축합니다. 첫 번째로, TextLoader와 OpenAIEmbeddings 부분은 문서를 AI가 이해할 수 있는 형태로 변환하는 것을 담당합니다.

TextLoader가 파일을 읽으면, 그 텍스트는 OpenAI의 임베딩 모델을 통해 1536차원의 벡터로 변환됩니다. 이렇게 하는 이유는 단순한 키워드 매칭이 아니라 의미적으로 유사한 문서를 찾기 위함입니다.

그 다음으로, Chroma.from_documents가 실행되면서 모든 문서가 벡터화되어 데이터베이스에 저장됩니다. 내부에서는 각 문서를 적절한 크기의 청크로 나누고, 각 청크마다 임베딩을 생성하여 인덱싱합니다.

이 과정이 중요한 이유는 나중에 빠른 검색이 가능하도록 최적화된 구조를 만들기 때문입니다. RetrievalQA.from_chain_type이 실행되면 검색과 생성을 하나의 체인으로 연결합니다.

질문이 들어오면 먼저 vectorstore에서 관련 문서를 검색하고, 그 문서들을 컨텍스트로 LLM에게 전달하여 최종 답변을 생성합니다. 여러분이 이 코드를 사용하면 몇 줄의 코드만으로 강력한 질의응답 시스템을 구축할 수 있습니다.

실무에서의 이점은 (1) 문서가 업데이트되어도 LLM 재훈련 없이 반영 가능, (2) 답변의 출처를 명확히 제시 가능, (3) 다양한 파일 포맷 지원으로 확장성이 높다는 점입니다.

실전 팁

💡 return_source_documents=True를 반드시 설정하세요. 답변의 출처를 보여줄 수 있어 신뢰도가 크게 향상됩니다. 특히 법률이나 의료 같은 민감한 분야에서는 필수입니다.

💡 temperature=0으로 설정하면 일관성 있는 답변을 얻을 수 있습니다. RAG 시스템은 창의적인 글쓰기가 아니라 정확한 정보 전달이 목적이므로, 랜덤성을 최소화하는 것이 좋습니다.

💡 문서 로딩 시 encoding="utf-8"을 명시하세요. 한글 문서를 다룰 때 인코딩 오류로 고생하는 경우가 많은데, 이것만으로 대부분의 문제를 해결할 수 있습니다.

💡 처음 벡터 저장소를 만들 때는 시간이 걸리므로, 한 번 만든 후에는 persist_directory를 사용해 디스크에 저장하고 재사용하세요. 매번 새로 만들면 비용과 시간이 낭비됩니다.

💡 실무에서는 vectorstore.as_retriever()에 search_kwargs={"k": 3}을 추가해 검색 결과 개수를 조절하세요. 너무 많으면 LLM의 컨텍스트 창을 넘길 수 있고, 너무 적으면 정보가 부족할 수 있습니다.

2. Document Loaders - 다양한 파일 형식 처리하기

시작하며

여러분이 RAG 시스템을 구축하려는데, 회사 문서가 PDF, Word, 웹페이지, 데이터베이스 등 여러 형식으로 흩어져 있어서 막막한 적 있나요? 각 파일 형식마다 다른 라이브러리를 찾아서 파싱 로직을 일일이 작성해야 한다면 정말 번거로울 겁니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. 특히 레거시 시스템을 다루는 경우, 오래된 HWP 파일부터 최신 Notion 페이지까지 다양한 소스를 통합해야 하는데, 각각의 파싱 방법이 달라서 개발 시간이 배로 늘어납니다.

바로 이럴 때 필요한 것이 LangChain의 Document Loaders입니다. 100가지 이상의 파일 형식과 데이터 소스를 통일된 인터페이스로 처리할 수 있어, 데이터 수집 파이프라인을 빠르게 구축할 수 있습니다.

개요

간단히 말해서, Document Loaders는 다양한 형식의 데이터를 LangChain이 이해할 수 있는 Document 객체로 변환하는 통일된 인터페이스입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 데이터 소스가 다양할수록 시스템의 가치가 높아지기 때문입니다.

예를 들어, 고객 지원 챗봇을 만든다면 FAQ PDF, 제품 매뉴얼 Word 파일, 고객 문의 데이터베이스, 최신 업데이트가 있는 웹페이지 등을 모두 활용해야 완성도 높은 시스템이 됩니다. 기존에는 각 파일 형식마다 PyPDF2, python-docx, BeautifulSoup 등 다른 라이브러리를 사용하고 파싱 결과를 표준화하는 코드를 직접 작성했다면, 이제는 해당하는 Loader를 import하고 .load()만 호출하면 됩니다.

Document Loaders의 핵심 특징은 세 가지입니다: (1) 통일된 인터페이스로 모든 소스를 동일하게 처리, (2) 메타데이터 자동 추출(파일명, 페이지 번호, URL 등), (3) lazy loading 지원으로 대용량 파일도 메모리 효율적으로 처리. 이러한 특징들이 중요한 이유는 개발 속도를 10배 이상 높이고, 코드의 일관성을 유지할 수 있기 때문입니다.

코드 예제

from langchain.document_loaders import (
    PyPDFLoader, UnstructuredWordDocumentLoader,
    WebBaseLoader, CSVLoader
)

# PDF 문서 로딩: 각 페이지가 별도 Document로 변환됩니다
pdf_loader = PyPDFLoader("product_manual.pdf")
pdf_docs = pdf_loader.load()
print(f"PDF 페이지 수: {len(pdf_docs)}")

# Word 문서 로딩: 전체 내용이 하나의 Document로 변환됩니다
word_loader = UnstructuredWordDocumentLoader("company_policy.docx")
word_docs = word_loader.load()

# 웹페이지 로딩: HTML에서 텍스트만 추출합니다
web_loader = WebBaseLoader("https://example.com/faq")
web_docs = web_loader.load()

# CSV 로딩: 각 행이 별도 Document로 변환됩니다
csv_loader = CSVLoader("customer_data.csv", encoding="utf-8")
csv_docs = csv_loader.load()

# 모든 문서 통합: 서로 다른 소스의 문서를 하나로 합칩니다
all_documents = pdf_docs + word_docs + web_docs + csv_docs
print(f"전체 문서 수: {len(all_documents)}")

# 메타데이터 확인: 출처 추적이 가능합니다
for doc in all_documents[:3]:
    print(f"출처: {doc.metadata.get('source', 'Unknown')}")

설명

이것이 하는 일: 이 코드는 서로 다른 네 가지 파일 형식을 읽어서 모두 동일한 Document 형식으로 변환하고, 하나의 리스트로 통합하는 멀티소스 데이터 파이프라인을 구축합니다. 첫 번째로, PyPDFLoader 부분은 PDF 파일을 페이지 단위로 분리하여 로딩합니다.

각 페이지가 별도의 Document 객체가 되고, metadata에는 페이지 번호와 파일 경로가 자동으로 저장됩니다. 이렇게 하는 이유는 나중에 "3페이지에서 찾았습니다"처럼 정확한 위치를 사용자에게 알려줄 수 있기 때문입니다.

그 다음으로, 각 Loader의 .load() 메서드가 실행되면서 내부적으로는 파일을 파싱하고, 텍스트를 추출하고, 메타데이터를 구성하는 복잡한 과정이 일어납니다. 예를 들어 WebBaseLoader는 HTML을 다운로드하고, JavaScript를 실행하지 않은 상태에서 텍스트만 깔끔하게 추출합니다.

UnstructuredWordDocumentLoader는 .docx 파일의 XML 구조를 파싱해서 텍스트와 서식 정보를 분리합니다. 마지막으로, all_documents = pdf_docs + word_docs + ...가 실행되면 서로 다른 소스의 문서들이 하나의 리스트로 통합됩니다.

이 리스트는 이후 벡터 저장소에 저장되거나 검색에 사용될 수 있습니다. 메타데이터를 확인하는 부분은 각 문서가 어디서 왔는지 추적할 수 있게 해줍니다.

여러분이 이 코드를 사용하면 데이터 소스를 추가할 때마다 단 3줄(import, 초기화, load)만 추가하면 됩니다. 실무에서의 이점은 (1) 새로운 파일 형식 추가가 매우 쉽고, (2) 모든 문서가 동일한 구조를 가져 후처리가 간단하며, (3) 메타데이터 덕분에 답변의 출처를 명확히 제시할 수 있다는 점입니다.

실전 팁

💡 대용량 PDF는 PyPDFLoader 대신 UnstructuredPDFLoader를 사용하세요. 이미지가 많거나 복잡한 레이아웃을 가진 PDF는 더 정확하게 파싱됩니다. 다만 속도는 조금 느릴 수 있으니 trade-off를 고려하세요.

💡 웹페이지 로딩 시 WebBaseLoader에 bs_kwargs={"parse_only": SoupStrainer("main")}을 전달하면 메인 콘텐츠만 추출할 수 있습니다. 네비게이션 메뉴나 광고 같은 불필요한 텍스트를 제거해서 검색 품질을 높일 수 있습니다.

💡 CSV 데이터는 CSVLoader의 csv_args 매개변수로 구분자나 인용 문자를 커스터마이징하세요. 예: CSVLoader("data.csv", csv_args={"delimiter": "\t"})로 탭 구분 파일도 처리 가능합니다.

💡 DirectoryLoader를 사용하면 폴더 전체를 한 번에 로딩할 수 있습니다. 예: DirectoryLoader("./docs", glob="**/*.pdf", loader_cls=PyPDFLoader)로 모든 하위 폴더의 PDF를 재귀적으로 로딩합니다.

💡 데이터베이스는 SQLDatabaseLoader를 사용하되, 쿼리 결과를 문자열로 변환하는 로직을 추가하세요. 예: "고객ID 123, 이름 홍길동, 주문일 2024-01-15" 형식으로 변환하면 LLM이 더 잘 이해합니다.

3. Text Splitters - 문서를 효율적으로 청크화하기

시작하며

여러분이 300페이지짜리 매뉴얼을 RAG 시스템에 넣었는데, 질문에 대한 답변이 엉뚱하게 나오거나 "토큰 제한을 초과했습니다" 에러가 발생한 적 있나요? 긴 문서를 그대로 벡터화하면 의미가 희석되고, LLM에게 전달할 때도 컨텍스트 창을 넘어버리는 문제가 생깁니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. 문서를 너무 크게 나누면 검색 정확도가 떨어지고, 너무 작게 나누면 문맥이 끊겨서 의미를 파악하기 어렵습니다.

적절한 크기로 나누는 것이 RAG 시스템 성능의 핵심입니다. 바로 이럴 때 필요한 것이 Text Splitters입니다.

문서의 구조와 의미를 고려하여 최적의 크기로 청크를 나누고, 청크 간 문맥도 유지할 수 있게 해줍니다.

개요

간단히 말해서, Text Splitters는 긴 문서를 검색과 처리에 적합한 크기의 청크(덩어리)로 지능적으로 분할하는 도구입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 임베딩 모델과 LLM 모두 입력 크기에 제한이 있고, 너무 긴 텍스트는 벡터화했을 때 의미가 뭉개지기 때문입니다.

예를 들어, 1만 단어짜리 기술 문서를 통째로 벡터화하면 특정 질문과의 유사도를 정확히 계산하기 어렵습니다. 하지만 각 섹션별로 나누면 "인증 방법"에 대한 질문에 인증 섹션만 정확히 매칭됩니다.

기존에는 단순히 글자 수나 단어 수로 문서를 자르거나, 정규식으로 문단을 나눴다면, 이제는 문서의 구조(제목, 문단, 문장)를 이해하고, 의미 단위로 나누며, 청크 간 오버랩을 두어 문맥도 보존할 수 있습니다. Text Splitters의 핵심 특징은 세 가지입니다: (1) 문서 타입별 최적화된 분할 전략(코드, 마크다운, 일반 텍스트 등), (2) chunk_overlap으로 청크 간 문맥 유지, (3) 토큰 기반 분할로 LLM 제한 정확히 준수.

이러한 특징들이 중요한 이유는 검색 품질과 답변 정확도를 직접적으로 좌우하기 때문입니다.

코드 예제

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import TextLoader

# 문서 로딩
loader = TextLoader("technical_doc.txt", encoding="utf-8")
documents = loader.load()

# Text Splitter 초기화: 청크 크기와 오버랩 설정
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,  # 각 청크의 최대 문자 수
    chunk_overlap=200,  # 청크 간 겹치는 문자 수
    length_function=len,  # 길이 측정 함수
    separators=["\n\n", "\n", ". ", " ", ""]  # 분할 우선순위
)

# 문서 분할 실행: 의미 단위로 청크를 나눕니다
chunks = text_splitter.split_documents(documents)
print(f"원본 문서 수: {len(documents)}")
print(f"분할된 청크 수: {len(chunks)}")

# 청크 확인: 각 청크의 크기와 내용을 확인
for i, chunk in enumerate(chunks[:3]):
    print(f"\n청크 {i+1} (길이: {len(chunk.page_content)}자)")
    print(f"내용: {chunk.page_content[:100]}...")
    print(f"메타데이터: {chunk.metadata}")

설명

이것이 하는 일: 이 코드는 긴 문서를 읽어서 의미를 보존하면서 검색에 최적화된 크기의 청크들로 분할하고, 각 청크가 독립적으로 이해될 수 있도록 오버랩을 적용합니다. 첫 번째로, RecursiveCharacterTextSplitter 초기화 부분은 분할 전략을 설정합니다.

chunk_size=1000은 각 청크가 최대 1000자를 넘지 않도록 하고, chunk_overlap=200은 이전 청크의 마지막 200자를 다음 청크의 시작에 포함시킵니다. 이렇게 하는 이유는 청크 경계에서 문맥이 끊기는 것을 방지하기 위함입니다.

예를 들어, "따라서 이 방법은"이라는 문장이 청크 경계에 걸리면, 오버랩 덕분에 양쪽 청크 모두에서 앞뒤 문맥을 파악할 수 있습니다. 그 다음으로, separators 매개변수가 실행되면서 재귀적 분할이 시작됩니다.

먼저 "\n\n"(문단 구분)으로 나누려고 시도하고, 그래도 chunk_size를 넘으면 "\n"(줄바꿈)으로, 그것도 안 되면 ". "(문장 끝)으로, 최종적으로 " "(공백)으로 분할합니다.

내부에서는 재귀 알고리즘이 동작하여 가능한 한 큰 의미 단위를 유지하면서 크기 제한을 맞춥니다. split_documents()가 실행되면 각 원본 Document가 여러 개의 작은 Document로 변환됩니다.

중요한 점은 원본의 메타데이터(파일명, 페이지 번호 등)가 모든 청크에 복사되어, 나중에 출처를 추적할 수 있다는 것입니다. 여러분이 이 코드를 사용하면 검색 정확도가 크게 향상됩니다.

실무에서의 이점은 (1) 질문과 관련된 정확한 부분만 검색되어 노이즈 감소, (2) 청크 오버랩으로 문맥 손실 최소화, (3) LLM 토큰 제한을 안정적으로 준수하여 에러 방지, (4) 답변 생성 속도 향상(작은 청크만 처리하므로)입니다.

실전 팁

💡 chunk_size는 임베딩 모델의 최대 토큰 수의 70% 정도로 설정하세요. OpenAI ada-002는 8191 토큰이므로, 약 5000자(영어 기준 ~1250 단어)가 적절합니다. 안전 마진을 두는 이유는 특수 문자나 토큰화 오차 때문입니다.

💡 chunk_overlap은 chunk_size의 10-20%로 설정하는 것이 일반적입니다. 너무 크면 중복 저장으로 인한 비용 증가, 너무 작으면 문맥 손실이 발생합니다. 실험을 통해 최적값을 찾으세요.

💡 코드 문서는 RecursiveCharacterTextSplitter.from_language(language="python", ...)를 사용하세요. 함수나 클래스 단위로 나누기 때문에 코드 검색 품질이 훨씬 좋습니다. 지원 언어: Python, JavaScript, Java, C, Go 등.

💡 마크다운 문서는 MarkdownHeaderTextSplitter를 사용하면 제목 구조를 메타데이터로 보존할 수 있습니다. 예: "# 설치 > ## 요구사항"이 메타데이터에 저장되어, "설치 요구사항"을 검색할 때 정확도가 높아집니다.

💡 토큰 수를 정확히 제어하려면 CharacterTextSplitter 대신 TokenTextSplitter를 사용하고, tokenizer에 tiktoken 라이브러리를 전달하세요. 특히 GPT-4처럼 토큰 비용이 높은 모델을 사용할 때 중요합니다.

4. Vector Stores - 벡터 데이터베이스 구축과 활용

시작하며

여러분이 수만 개의 문서를 검색 가능하게 만들었는데, 검색 속도가 너무 느리거나, 서버를 재시작할 때마다 처음부터 다시 벡터화해야 해서 고민한 적 있나요? 인메모리로만 관리하면 확장성이 떨어지고, 영구 저장이 안 되면 매번 비용이 발생합니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. 특히 프로덕션 환경에서는 수백만 개의 벡터를 빠르게 검색하고, 데이터를 영구 저장하며, 실시간으로 업데이트할 수 있어야 합니다.

단순한 리스트로는 이런 요구사항을 만족시킬 수 없습니다. 바로 이럴 때 필요한 것이 Vector Stores입니다.

벡터를 효율적으로 저장하고, 유사도 검색을 빠르게 수행하며, 디스크에 영구 저장까지 지원하는 전문 데이터베이스입니다.

개요

간단히 말해서, Vector Stores는 고차원 벡터를 저장하고 유사도 기반 검색을 수행하는 특수한 데이터베이스입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 일반 데이터베이스의 정확한 매칭(WHERE name = 'John')이 아니라 의미적 유사도 검색(가장 비슷한 문서 찾기)이 필요하기 때문입니다.

예를 들어, "환불 방법"이라는 질문에 "결제 취소 절차"라는 제목의 문서를 찾아야 하는데, 키워드는 하나도 겹치지 않지만 의미는 유사합니다. 벡터 검색만이 이를 가능하게 합니다.

기존에는 모든 벡터를 메모리에 로드하고 하나씩 코사인 유사도를 계산하는 브루트포스 방식을 사용했다면, 이제는 HNSW, IVF 같은 근사 최근접 이웃(ANN) 알고리즘으로 밀리초 단위의 빠른 검색이 가능합니다. Vector Stores의 핵심 특징은 세 가지입니다: (1) 빠른 유사도 검색(수백만 벡터에서 밀리초 응답), (2) 영구 저장과 인덱싱(디스크/클라우드 저장 지원), (3) 메타데이터 필터링(조건부 검색 가능).

이러한 특징들이 중요한 이유는 프로덕션 수준의 RAG 시스템을 구축하는 데 필수적이기 때문입니다.

코드 예제

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 문서 준비
loader = TextLoader("knowledge_base.txt", encoding="utf-8")
documents = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50)
chunks = text_splitter.split_documents(documents)

# 임베딩 모델 초기화
embeddings = OpenAIEmbeddings()

# 벡터 저장소 생성 및 영구 저장: persist_directory로 디스크에 저장
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"  # 디스크 저장 경로
)

# 유사도 검색: 질문과 가장 유사한 문서 k개 반환
query = "비밀번호를 재설정하는 방법은?"
results = vectorstore.similarity_search(query, k=3)

# 검색 결과 출력
for i, doc in enumerate(results):
    print(f"\n결과 {i+1} (유사도 점수 포함):")
    print(f"내용: {doc.page_content[:200]}...")
    print(f"출처: {doc.metadata.get('source', 'Unknown')}")

# 기존 벡터 저장소 로드: 서버 재시작 시 재사용
loaded_vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

설명

이것이 하는 일: 이 코드는 문서를 벡터 데이터베이스에 저장하고, 자연어 질문으로 의미적으로 유사한 문서를 검색하며, 결과를 디스크에 영구 저장하여 재사용 가능하게 만듭니다. 첫 번째로, Chroma.from_documents 부분은 모든 청크를 벡터화하고 인덱싱합니다.

내부적으로는 각 청크의 텍스트를 embeddings.embed_documents()로 벡터로 변환하고(API 호출), 그 벡터들을 Chroma의 내부 인덱스 구조에 저장합니다. persist_directory가 설정되어 있으면 SQLite 데이터베이스 파일로 디스크에 저장됩니다.

이렇게 하는 이유는 다음에 실행할 때 API 호출 없이 바로 로드할 수 있어 시간과 비용을 절약하기 때문입니다. 그 다음으로, similarity_search가 실행되면 질문 문자열이 벡터로 변환되고(embeddings.embed_query()), 저장된 모든 벡터와의 코사인 유사도가 계산됩니다.

Chroma는 내부적으로 HNSW(Hierarchical Navigable Small World) 알고리즘을 사용해 전체 벡터를 순회하지 않고도 가장 유사한 k개를 빠르게 찾습니다. k=3이므로 상위 3개 문서만 반환됩니다.

마지막으로, Chroma(persist_directory=...)로 로드하는 부분은 기존에 저장된 벡터 인덱스를 디스크에서 읽어옵니다. 이때 embedding_function을 동일하게 전달해야 새로운 쿼리를 벡터화할 수 있습니다.

이 과정은 수 초 만에 완료되어, 수만 개의 문서를 다시 벡터화하는 것(수 분 소요)에 비해 엄청나게 빠릅니다. 여러분이 이 코드를 사용하면 프로덕션 수준의 검색 시스템을 구축할 수 있습니다.

실무에서의 이점은 (1) 초기 구축 후 빠른 재시작(벡터 재생성 불필요), (2) 대규모 문서에서도 빠른 검색(밀리초 단위), (3) 점진적 업데이트 가능(새 문서만 추가), (4) 메모리 효율적(필요한 부분만 로드)입니다.

실전 팁

💡 Chroma 외에도 용도에 맞는 벡터 스토어를 선택하세요. Pinecone(클라우드, 대규모), FAISS(로컬, 초고속), Weaviate(하이브리드 검색), Qdrant(필터링 강력) 등이 있습니다. 무료 프로토타입은 Chroma, 프로덕션은 Pinecone 추천합니다.

💡 similarity_search_with_score()를 사용하면 유사도 점수도 함께 반환됩니다. 점수가 너무 낮으면(예: 0.5 이하) "관련 정보를 찾을 수 없습니다"라고 답변하는 로직을 추가하세요. 엉뚱한 답변을 방지할 수 있습니다.

💡 메타데이터 필터링을 활용하세요. 예: vectorstore.similarity_search(query, k=3, filter={"source": "manual.pdf"})로 특정 문서 내에서만 검색 가능합니다. 권한 제어나 카테고리별 검색에 유용합니다.

💡 대용량 데이터는 배치로 추가하세요. vectorstore.add_documents(chunks)를 한 번에 수천 개씩 호출하면 인덱싱 효율이 높아집니다. 한 번에 하나씩 추가하면 인덱스 재구성이 반복되어 느립니다.

💡 OpenAI 임베딩 API 비용이 부담된다면, HuggingFaceEmbeddings로 로컬 모델을 사용하세요. "sentence-transformers/all-MiniLM-L6-v2"는 무료이고 성능도 준수합니다. 다만 다국어 지원은 OpenAI가 더 좋습니다.

5. Retrieval 전략 - 검색 품질 최적화하기

시작하며

여러분이 RAG 시스템을 만들었는데, 간단한 질문은 잘 답변하지만 복잡한 질문이나 다단계 추론이 필요한 질문에는 엉뚱한 답변을 하는 경우를 경험한 적 있나요? 단순히 유사도가 높은 문서 몇 개를 가져오는 것만으로는 부족한 상황이 많습니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. "작년 대비 올해 매출 증가율은?"처럼 여러 문서를 종합해야 하는 질문이나, "AWS Lambda의 장단점 비교"처럼 다양한 관점의 문서가 필요한 질문에서는 기본 검색 전략으로는 한계가 있습니다.

바로 이럴 때 필요한 것이 고급 Retrieval 전략입니다. 검색 결과를 재순위화하거나, 여러 번 검색하거나, 쿼리를 확장하는 등의 기법으로 검색 품질을 크게 향상시킬 수 있습니다.

개요

간단히 말해서, Retrieval 전략은 단순한 벡터 유사도 검색을 넘어서 더 정확하고 포괄적인 문서를 찾기 위한 고급 기법들입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 실제 사용자의 질문은 단순하지 않고, 맥락이 부족하거나, 여러 정보를 종합해야 하는 경우가 많기 때문입니다.

예를 들어, "그거 어떻게 해?"라는 모호한 질문이나, "A와 B를 비교해줘"처럼 두 가지 개념이 섞인 질문, "최신 정책은?"처럼 시간 정보가 중요한 질문 등에서는 고급 전략이 필수입니다. 기존에는 similarity_search(query, k=3)로 상위 3개만 가져왔다면, 이제는 MMR(Maximum Marginal Relevance)로 다양성을 보장하거나, Contextual Compression으로 관련 부분만 추출하거나, Multi-Query로 여러 각도에서 검색할 수 있습니다.

고급 Retrieval 전략의 핵심 특징은 세 가지입니다: (1) 다양성 보장(중복되지 않는 관점의 문서 검색), (2) 정밀도 향상(불필요한 부분 제거), (3) 재현율 향상(더 많은 관련 문서 발견). 이러한 특징들이 중요한 이유는 답변의 품질과 사용자 만족도를 직접적으로 높이기 때문입니다.

코드 예제

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain.llms import OpenAI

# 기본 벡터 저장소와 리트리버 설정
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

# MMR 검색: 유사도와 다양성을 동시에 고려합니다
mmr_retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximum Marginal Relevance
    search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5}
)

# Contextual Compression: 문서에서 관련 부분만 추출합니다
llm = OpenAI(temperature=0)
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=mmr_retriever
)

# 검색 실행: 압축된 관련 정보만 가져옵니다
query = "LangChain에서 메모리를 사용하는 방법은?"
compressed_docs = compression_retriever.get_relevant_documents(query)

# 결과 확인
print(f"압축된 문서 수: {len(compressed_docs)}")
for i, doc in enumerate(compressed_docs):
    print(f"\n문서 {i+1} (압축됨):")
    print(f"내용: {doc.page_content}")
    print(f"출처: {doc.metadata.get('source', 'Unknown')}")

설명

이것이 하는 일: 이 코드는 단순한 유사도 검색을 넘어서, 다양한 관점의 문서를 가져오고(MMR), 각 문서에서 질문과 직접 관련된 부분만 추출하여(Compression) 고품질의 컨텍스트를 생성합니다. 첫 번째로, MMR 리트리버 설정 부분은 검색 전략을 세밀하게 조정합니다.

search_type="mmr"은 단순히 유사도가 높은 문서만 가져오는 것이 아니라, 이미 선택된 문서와 너무 비슷한 문서는 제외합니다. fetch_k=20은 먼저 20개의 후보를 가져오고, 그중에서 다양성을 고려해 k=5개를 최종 선택합니다.

lambda_mult=0.5는 유사도와 다양성의 균형점인데, 0에 가까우면 다양성 우선, 1에 가까우면 유사도 우선입니다. 이렇게 하는 이유는 같은 내용의 중복 문서를 피하고, 다양한 각도의 정보를 제공하기 위함입니다.

그 다음으로, ContextualCompressionRetriever가 실행되면 두 단계의 처리가 일어납니다. 먼저 base_retriever(MMR)가 5개의 다양한 문서를 가져오고, 그 다음 compressor(LLM)가 각 문서를 분석하여 질문과 직접 관련된 문장이나 문단만 추출합니다.

예를 들어, 1000자짜리 문서에서 실제로 질문에 답하는 200자만 남기고 나머지는 제거합니다. 내부적으로는 LLM에게 "이 문서에서 질문과 관련된 부분만 추출하라"는 프롬프트를 보냅니다.

get_relevant_documents()가 실행되면 최종적으로 압축되고 다양성이 보장된 고품질 문서들이 반환됩니다. 이 문서들은 불필요한 정보가 제거되어 토큰 수가 적고, 중복이 없어 효율적이며, 여러 관점을 포함하여 포괄적입니다.

여러분이 이 코드를 사용하면 답변 품질이 눈에 띄게 향상됩니다. 실무에서의 이점은 (1) LLM에 전달되는 토큰 수 감소로 비용 절감, (2) 불필요한 정보가 없어 답변 정확도 향상, (3) 다양한 관점으로 편향 감소, (4) 복잡한 질문에 대한 포괄적인 답변 가능입니다.

실전 팁

💡 MMR의 lambda_mult 값은 도메인에 따라 조정하세요. 뉴스 검색처럼 다양한 관점이 중요하면 0.3, 기술 문서처럼 정확도가 중요하면 0.7로 설정합니다. A/B 테스트로 최적값을 찾으세요.

💡 Contextual Compression은 비용이 높으므로(문서마다 LLM 호출), fetch_k를 너무 크게 설정하지 마세요. 보통 k의 2-4배가 적당합니다. 예: k=5이면 fetch_k=10-20.

💡 Self-Query Retriever를 사용하면 자연어 질문을 메타데이터 필터로 자동 변환합니다. 예: "2024년 이후의 기술 문서"를 {year: {$gte: 2024}, category: "tech"}로 변환해서 검색합니다.

💡 Multi-Query Retriever는 하나의 질문을 여러 각도로 다시 작성하여 검색합니다. "LangChain 설치 방법"을 "How to install LangChain", "LangChain setup guide", "LangChain 시작하기"로 확장해서 재현율을 높입니다.

💡 Ensemble Retriever로 BM25(키워드 검색)와 벡터 검색을 결합하세요. 고유명사나 정확한 용어는 BM25가, 의미적 유사성은 벡터 검색이 잘 찾으므로 둘을 가중 평균하면 최고의 결과를 얻습니다.

6. RetrievalQA Chain - 질의응답 체인 구성하기

시작하며

여러분이 벡터 저장소에서 관련 문서를 검색하는 것까지는 성공했는데, 그 문서들을 LLM에 어떻게 전달하고, 프롬프트를 어떻게 구성하며, 답변을 어떻게 포맷팅해야 할지 막막한 적 있나요? 검색과 생성을 매번 수동으로 연결하면 코드가 복잡해지고 일관성을 유지하기 어렵습니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. 특히 프롬프트 엔지니어링, 컨텍스트 관리, 에러 핸들링 등을 고려하면 RAG 시스템 구축이 생각보다 훨씬 복잡해집니다.

매번 같은 로직을 반복 작성하는 것도 비효율적입니다. 바로 이럴 때 필요한 것이 RetrievalQA Chain입니다.

검색부터 답변 생성까지 전체 파이프라인을 하나의 체인으로 묶어, 간단한 코드로 강력한 질의응답 시스템을 만들 수 있습니다.

개요

간단히 말해서, RetrievalQA Chain은 문서 검색과 LLM 기반 답변 생성을 하나로 통합한 End-to-End 질의응답 파이프라인입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, RAG의 모든 단계(검색 → 컨텍스트 구성 → 프롬프트 생성 → LLM 호출 → 답변 추출)를 자동화하여 개발 속도를 10배 이상 높이기 때문입니다.

예를 들어, 고객 지원 챗봇을 만든다면 사용자 질문을 받아서 답변을 반환하는 전체 로직을 단 몇 줄로 구현할 수 있습니다. 기존에는 retriever.get_relevant_documents() → 문서 포맷팅 → 프롬프트 템플릿 작성 → llm.predict() → 답변 파싱의 각 단계를 직접 코딩했다면, 이제는 RetrievalQA.from_chain_type()로 모든 것이 자동화됩니다.

RetrievalQA Chain의 핵심 특징은 세 가지입니다: (1) 다양한 체인 타입 지원(stuff, map_reduce, refine 등), (2) 커스터마이징 가능한 프롬프트 템플릿, (3) 출처 문서 반환으로 투명성 보장. 이러한 특징들이 중요한 이유는 프로덕션 수준의 신뢰할 수 있는 시스템을 빠르게 구축할 수 있기 때문입니다.

코드 예제

from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.prompts import PromptTemplate

# 벡터 저장소 로드
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

# LLM 초기화
llm = OpenAI(temperature=0, model_name="gpt-3.5-turbo")

# 커스텀 프롬프트 템플릿: 답변 형식을 제어합니다
prompt_template = """아래 컨텍스트를 사용하여 질문에 답변하세요.
답을 모르면 "정보가 부족합니다"라고 답하고, 추측하지 마세요.
답변은 최대 3문장으로 간결하게 작성하세요.

컨텍스트: {context}

질문: {question}

답변:"""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# RetrievalQA 체인 생성
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # 모든 문서를 하나의 프롬프트에 포함
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    return_source_documents=True,  # 출처 반환
    chain_type_kwargs={"prompt": PROMPT}  # 커스텀 프롬프트 적용
)

# 질문 실행
query = "LangChain에서 체인을 어떻게 구성하나요?"
result = qa_chain({"query": query})

# 결과 출력
print(f"질문: {query}")
print(f"\n답변: {result['result']}")
print(f"\n출처 문서 수: {len(result['source_documents'])}")
for i, doc in enumerate(result['source_documents']):
    print(f"\n출처 {i+1}: {doc.metadata.get('source', 'Unknown')}")

설명

이것이 하는 일: 이 코드는 질문을 받으면 자동으로 관련 문서를 검색하고, 그 문서들을 프롬프트에 포함시켜 LLM에 전달하며, 생성된 답변과 출처 문서를 함께 반환하는 완전한 질의응답 시스템을 구축합니다. 첫 번째로, PromptTemplate 정의 부분은 LLM에게 어떻게 답변할지 지시합니다.

{context}에는 검색된 문서들이, {question}에는 사용자의 질문이 자동으로 채워집니다. "답을 모르면 추측하지 마세요"라는 지시는 할루시네이션을 방지하고, "최대 3문장"은 답변 길이를 제어합니다.

이렇게 하는 이유는 프롬프트가 답변의 품질과 형식을 직접적으로 결정하기 때문입니다. 실무에서는 도메인에 맞게 프롬프트를 세밀하게 조정하는 것이 핵심입니다.

그 다음으로, RetrievalQA.from_chain_type()이 실행되면 내부적으로 복잡한 체인이 구성됩니다. chain_type="stuff"는 검색된 모든 문서(k=3개)를 하나의 문자열로 연결하여 프롬프트의 {context}에 넣는 방식입니다.

다른 옵션으로는 "map_reduce"(각 문서를 개별 처리 후 결합), "refine"(순차적으로 답변 개선), "map_rerank"(여러 답변 중 최고 점수 선택)가 있습니다. "stuff"가 가장 간단하고 빠르지만, 문서가 많으면 토큰 제한을 넘을 수 있습니다.

qa_chain({"query": query})가 실행되면 다음 순서로 처리됩니다: (1) query를 벡터로 변환, (2) 벡터 저장소에서 상위 k=3개 문서 검색, (3) 검색된 문서들을 context로, 질문을 question으로 프롬프트에 삽입, (4) 완성된 프롬프트를 LLM에 전달, (5) LLM의 응답을 result['result']에, 검색된 문서를 result['source_documents']에 저장하여 반환. 여러분이 이 코드를 사용하면 프로덕션 레벨의 질의응답 시스템을 10분 만에 구축할 수 있습니다.

실무에서의 이점은 (1) 전체 파이프라인이 하나의 객체로 관리되어 유지보수 용이, (2) 프롬프트만 변경하면 답변 스타일 커스터마이징 가능, (3) 출처 제공으로 사용자 신뢰도 향상, (4) 에러 핸들링과 재시도 로직이 내장되어 안정성 높음입니다.

실전 팁

💡 문서가 많을 때는 chain_type="map_reduce"를 사용하세요. 각 문서를 병렬로 요약한 후 최종 답변을 생성하므로, "stuff"의 토큰 제한 문제를 피할 수 있습니다. 다만 LLM 호출 횟수가 늘어 비용은 증가합니다.

💡 프롬프트에 "답변은 한국어로 작성하세요"를 추가하세요. LLM이 영어 문서를 참조할 때 영어로 답변하는 것을 방지합니다. 다국어 환경에서는 필수입니다.

💡 verbose=True를 RetrievalQA에 추가하면 내부 동작(검색된 문서, 생성된 프롬프트 등)을 볼 수 있습니다. 디버깅이나 프롬프트 튜닝 시 매우 유용합니다.

💡 ConversationalRetrievalChain을 사용하면 대화 히스토리를 유지할 수 있습니다. "그것의 장점은?"처럼 이전 대화를 참조하는 질문도 처리 가능합니다. 챗봇 구축 시 필수입니다.

💡 답변 품질이 낮으면 search_kwargs={"k": 5}로 더 많은 문서를 검색하거나, retriever에 score_threshold를 설정해 유사도가 낮은 문서는 제외하세요. 예: search_kwargs={"k": 5, "score_threshold": 0.7}.

7. ConversationalRetrievalChain - 대화 맥락을 유지하는 RAG

시작하며

여러분이 RAG 챗봇을 만들었는데, "LangChain이 뭐야?" → "그것의 장점은?" → "어디서 사용돼?"처럼 연속된 질문에서 "그것"이 뭔지 몰라서 엉뚱한 답변을 하는 경우를 경험한 적 있나요? 각 질문을 독립적으로 처리하면 대화의 흐름이 끊겨서 사용자 경험이 나빠집니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. 특히 고객 지원 챗봇이나 교육용 튜터링 시스템처럼 여러 턴의 대화가 필요한 경우, 이전 대화 맥락을 기억하지 못하면 매번 질문을 완전한 문장으로 다시 작성해야 하는 불편함이 생깁니다.

바로 이럴 때 필요한 것이 ConversationalRetrievalChain입니다. 대화 히스토리를 관리하고, 현재 질문을 이전 맥락과 결합하여 재구성한 후 검색하므로, 자연스러운 대화형 RAG 시스템을 만들 수 있습니다.

개요

간단히 말해서, ConversationalRetrievalChain은 대화 히스토리를 유지하면서 RAG 검색을 수행하는 상태 유지형(stateful) 질의응답 체인입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 실제 사용자는 완전한 문장으로 질문하지 않고, "그거", "저번에 말한", "더 자세히"처럼 대명사와 생략을 많이 사용하기 때문입니다.

예를 들어, 기술 지원 챗봇에서 "Lambda 함수 만드는 방법" → "비용은?" → "예제 보여줘"처럼 연속 질문이 일반적인데, 맥락 없이는 "비용은?"이 무엇의 비용인지 알 수 없습니다. 기존에는 각 질문마다 전체 대화를 수동으로 포함시키거나, 사용자에게 매번 완전한 문장을 요구했다면, 이제는 체인이 자동으로 대화를 추적하고, 질문을 재구성하며, 관련 문서를 검색합니다.

ConversationalRetrievalChain의 핵심 특징은 세 가지입니다: (1) 자동 대화 히스토리 관리(메모리 통합), (2) 질문 재구성(현재 질문 + 이전 맥락 → 독립적인 질문), (3) 스트리밍 지원(답변이 생성되는 대로 실시간 표시). 이러한 특징들이 중요한 이유는 ChatGPT처럼 자연스러운 대화형 인터페이스를 제공할 수 있기 때문입니다.

코드 예제

from langchain.chains import ConversationalRetrievalChain
from langchain.memory import ConversationBufferMemory
from langchain.llms import OpenAI
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 벡터 저장소 설정
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

# 메모리 초기화: 대화 히스토리를 저장합니다
memory = ConversationBufferMemory(
    memory_key="chat_history",  # 프롬프트에서 사용할 키
    return_messages=True,  # 메시지 객체로 반환
    output_key="answer"  # 답변을 저장할 키
)

# LLM 초기화
llm = OpenAI(temperature=0)

# ConversationalRetrievalChain 생성
qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    memory=memory,
    return_source_documents=True,
    verbose=True  # 내부 동작 확인
)

# 대화 시뮬레이션: 연속된 질문들
queries = [
    "LangChain에서 체인이란 무엇인가요?",
    "그것의 주요 장점은 뭔가요?",  # "그것" = 체인
    "실제 예제를 보여주세요"  # 앞의 맥락 유지
]

for query in queries:
    print(f"\n{'='*60}")
    print(f"질문: {query}")
    result = qa_chain({"question": query})
    print(f"답변: {result['answer']}")
    # 대화 히스토리는 자동으로 memory에 저장됨

# 대화 히스토리 확인
print(f"\n{'='*60}")
print("전체 대화 히스토리:")
print(memory.load_memory_variables({}))

설명

이것이 하는 일: 이 코드는 여러 턴의 대화를 기억하면서, 각 질문을 이전 맥락과 결합하여 완전한 질문으로 재구성하고, 그에 맞는 문서를 검색하여 답변하는 대화형 RAG 시스템을 구축합니다. 첫 번째로, ConversationBufferMemory 초기화 부분은 대화 저장소를 만듭니다.

memory_key="chat_history"는 프롬프트 템플릿에서 대화 내역을 참조할 때 사용하는 변수명이고, return_messages=True는 대화를 문자열이 아닌 메시지 객체(HumanMessage, AIMessage)로 저장합니다. output_key="answer"는 체인의 출력 중 어느 부분을 메모리에 저장할지 지정합니다.

이렇게 하는 이유는 대화 히스토리를 구조화된 형태로 유지하여, 나중에 요약하거나 검색할 때 효율적이기 때문입니다. 그 다음으로, ConversationalRetrievalChain.from_llm()이 실행되면 내부적으로 두 개의 서브 체인이 생성됩니다.

첫 번째는 "Question Generator Chain"으로, 현재 질문과 chat_history를 받아서 독립적인 질문(standalone question)을 생성합니다. 예를 들어, 질문이 "그것의 장점은?"이고 히스토리에 "LangChain 체인"이 있으면, "LangChain 체인의 장점은?"으로 재구성합니다.

두 번째는 "Document Chain"으로, 재구성된 질문으로 문서를 검색하고 답변을 생성합니다. qa_chain({"question": query})가 실행되면 다음 순서로 처리됩니다: (1) memory에서 chat_history 로드, (2) 현재 질문 + chat_history → LLM → 독립적인 질문 생성, (3) 독립적인 질문으로 벡터 검색, (4) 검색된 문서 + 원래 질문(맥락 포함) → LLM → 답변 생성, (5) (질문, 답변) 쌍을 memory에 저장.

중요한 점은 검색은 재구성된 질문으로 하지만, 답변 생성은 원래 질문으로 한다는 것입니다. 이렇게 하면 검색 정확도는 높이고, 답변은 자연스럽게 유지할 수 있습니다.

여러분이 이 코드를 사용하면 ChatGPT 같은 자연스러운 대화 경험을 제공할 수 있습니다. 실무에서의 이점은 (1) 사용자가 짧은 질문으로도 소통 가능, (2) 대화 흐름이 자연스러워 사용자 만족도 향상, (3) 메모리 관리가 자동화되어 개발 간편, (4) 대화 히스토리를 분석하여 사용자 의도 파악 가능입니다.

💡 메모리 타입을 용도에 맞게 선택하세요. ConversationBufferMemory(전체 저장), ConversationSummaryMemory(요약 저장), ConversationBufferWindowMemory(최근 k개만 저장) 등이 있습니다.

긴 대화는 SummaryMemory로 토큰 절약 가능합니다. 💡 condense_question_prompt를 커스터마이징하면 질문 재구성 방식을 제어할 수 있습니다.

예: "이전 대화를 고려하여, 다음 질문을 독립적인 질문으로 다시 작성하세요"처럼 명확한 지시를 추가하세요. 💡 max_tokens_limit을 설정해 대화 히스토리가 너무 길어지는 것을 방지하세요.

예: ConversationSummaryMemory(llm=llm, max_token_limit=500)로 500 토큰 초과 시 자동 요약합니다. 💡 get_chat_history 함수를 전달하면 대화 포맷을 커스터마이징할 수 있습니다.

예: 타임스탬프 추가, 사용자 이름 포함 등. chain = ConversationalRetrievalChain.from_llm(..., get_chat_history=lambda h: ...) 💡 스트리밍을 활성화하려면 llm에 streaming=True, callbacks=[StreamingStdOutCallbackHandler()]를 추가하세요.

답변이 생성되는 대로 실시간 출력되어 UX가 크게 향상됩니다.

8. Custom Prompts - 프롬프트 최적화 전략

시작하며

여러분이 RAG 시스템을 구축했는데, 답변이 너무 길거나, 형식이 일관되지 않거나, 불필요한 정보가 포함되는 경우를 경험한 적 있나요? 기본 프롬프트는 범용적이라서 여러분의 특정 도메인이나 사용 사례에 최적화되어 있지 않습니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. 법률 문서 검색 시스템에서는 정확한 조항 인용이 필요하고, 의료 Q&A에서는 면책 조항이 필수이며, 고객 지원에서는 친근한 톤이 중요합니다.

하나의 프롬프트로 모든 요구사항을 만족시킬 수 없습니다. 바로 이럴 때 필요한 것이 Custom Prompts입니다.

도메인 지식, 답변 형식, 톤앤매너, 제약사항 등을 프롬프트에 명시하여 RAG 시스템의 출력을 정밀하게 제어할 수 있습니다.

개요

간단히 말해서, Custom Prompts는 LangChain의 기본 프롬프트를 여러분의 요구사항에 맞게 커스터마이징하여 답변 품질과 형식을 제어하는 기법입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 프롬프트가 LLM 출력의 90%를 결정하기 때문입니다.

같은 문서를 검색해도 프롬프트에 따라 "간단히 말하면..."처럼 시작하거나, "1. 2.

3."처럼 번호 매기거나, "출처: [문서명]"을 포함하는 등 완전히 다른 결과를 얻을 수 있습니다. 예를 들어, 기술 지원에서는 단계별 가이드가 필요하고, 영업에서는 설득력 있는 답변이 필요합니다.

기존에는 LangChain의 기본 프롬프트를 그대로 사용하거나, 전체 체인을 다시 작성해야 했다면, 이제는 PromptTemplate만 교체하면 됩니다. Custom Prompts의 핵심 특징은 세 가지입니다: (1) 변수 기반 템플릿(동적 콘텐츠 삽입), (2) Few-shot 예제 포함(원하는 형식 학습), (3) 조건부 로직(입력에 따라 프롬프트 변경).

이러한 특징들이 중요한 이유는 동일한 RAG 시스템을 다양한 상황에 맞게 조정할 수 있기 때문입니다.

코드 예제

from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 벡터 저장소 설정
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

# 도메인 특화 프롬프트: 기술 지원용
tech_support_template = """당신은 친절한 기술 지원 전문가입니다.
아래 문서를 참고하여 사용자의 질문에 단계별로 답변하세요.

중요 규칙:

4. 답을 모르면 "해당 정보를 찾을 수 없습니다. 추가 문의는 support@example.com으로 연락주세요"라고 하세요

설명

이것이 하는 일: 이 코드는 기술 지원과 법률 문서라는 서로 다른 도메인에 최적화된 프롬프트 템플릿을 정의하고, 선택한 템플릿에 따라 동일한 검색 결과도 완전히 다른 형식의 답변을 생성합니다. 첫 번째로, tech_support_template 정의 부분은 LLM의 역할, 답변 형식, 제약사항을 명시합니다.

"당신은 친절한 기술 지원 전문가입니다"는 톤을 설정하고, "번호 매긴 단계로 구성"은 형식을 지정하며, "기술 용어는 쉽게 풀어서"는 대상 독자를 고려합니다. 이렇게 하는 이유는 LLM이 모호한 지시보다 명확하고 구체적인 지시를 훨씬 잘 따르기 때문입니다.

"답을 모르면..." 부분은 할루시네이션을 방지하는 안전장치입니다. 그 다음으로, legal_template의 Few-shot 예제 부분은 LLM에게 원하는 답변 형식을 직접 보여줍니다.

"예시:"로 실제 질문과 이상적인 답변을 제공하면, LLM은 그 패턴을 학습하여 비슷한 형식으로 답변합니다. 내부적으로는 In-Context Learning이 작동하여, 몇 개의 예제만으로도 복잡한 형식을 따를 수 있습니다.

법률 문서에서는 "[출처: 제12조 3항]" 같은 정확한 인용이 필수이므로, 예제에 포함시킵니다. PromptTemplate 객체가 생성되면 {context}와 {question}이라는 두 개의 플레이스홀더를 가진 템플릿이 만들어집니다.

나중에 체인이 실행될 때, {context}는 검색된 문서들로, {question}은 사용자 질문으로 자동으로 채워집니다. chain_type_kwargs={"prompt": PROMPT}로 전달하면 기본 프롬프트 대신 이 커스텀 프롬프트가 사용됩니다.

여러분이 이 코드를 사용하면 동일한 RAG 시스템을 다양한 도메인에 맞게 조정할 수 있습니다. 실무에서의 이점은 (1) 프롬프트만 바꾸면 되므로 유연성 극대화, (2) 도메인 전문성을 코드가 아닌 프롬프트로 표현하여 비개발자도 수정 가능, (3) A/B 테스트로 최적 프롬프트 발견 가능, (4) 일관된 답변 형식으로 사용자 경험 향상입니다.

실전 팁

💡 프롬프트에 "답변은 한국어로 작성하세요"를 명시하세요. 영어 문서를 참조할 때 LLM이 영어로 답변하는 것을 방지합니다. 특히 다국어 환경에서 필수입니다.

💡 Few-shot 예제는 2-3개가 적당합니다. 너무 많으면 토큰을 낭비하고, 너무 적으면 패턴 학습이 안 됩니다. 가장 대표적이고 다양한 예제를 선택하세요.

💡 출력 형식을 구조화하려면 "JSON 형식으로 답변하세요: {"answer": "...", "confidence": 0.9, "sources": [...]}"처럼 명시하세요. 파싱이 쉬워져 후처리가 간편합니다.

💡 프롬프트 버전 관리를 하세요. Git에 프롬프트 파일을 저장하고, 변경 이력을 추적하면 성능 회귀를 방지하고, 최적 버전을 찾을 수 있습니다.

💡 LangSmith나 Weights & Biases로 프롬프트 성능을 측정하세요. 여러 프롬프트를 테스트 세트에 실행하고, 정확도/관련성/형식 준수율을 비교하여 데이터 기반으로 선택합니다.

9. Error Handling과 Fallbacks - 안정성 확보하기

시작하며

여러분이 RAG 시스템을 프로덕션에 배포했는데, 간혹 OpenAI API가 타임아웃되거나, 검색 결과가 없거나, LLM이 이상한 답변을 하는 경우를 경험한 적 있나요? 에러 처리 없이는 시스템이 멈추거나, 사용자에게 에러 메시지가 그대로 노출됩니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. API 호출 실패, 네트워크 오류, 토큰 제한 초과, 빈 검색 결과 등 다양한 엣지 케이스가 있는데, 이를 처리하지 않으면 사용자 경험이 나빠지고, 시스템 신뢰도가 떨어집니다.

바로 이럴 때 필요한 것이 Error Handling과 Fallbacks입니다. 예상 가능한 오류를 사전에 처리하고, 주 시스템 실패 시 백업 방안을 제공하여 안정적인 서비스를 유지할 수 있습니다.

개요

간단히 말해서, Error Handling은 예외 상황을 감지하고 적절히 대응하는 것이고, Fallbacks은 주 시스템 실패 시 대체 방안을 자동으로 실행하는 메커니즘입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 외부 API(OpenAI, Pinecone 등)에 의존하는 시스템은 네트워크 불안정, 서비스 장애, 요금 제한 등의 이유로 실패할 수 있기 때문입니다.

예를 들어, OpenAI API가 과부하로 응답이 10초 이상 걸리면 사용자는 떠나버립니다. 또한 "검색 결과 없음" 같은 정상적인 케이스도 우아하게 처리해야 합니다.

기존에는 try-except로 개별 에러를 잡고, 수동으로 재시도 로직을 작성했다면, 이제는 LangChain의 built-in retry, fallback, timeout 설정과 validation 체인을 활용할 수 있습니다. Error Handling의 핵심 특징은 세 가지입니다: (1) 자동 재시도(일시적 오류 복구), (2) Fallback 체인(대체 LLM이나 하드코딩된 응답), (3) Graceful degradation(부분 실패 시에도 최선의 답변).

이러한 특징들이 중요한 이유는 99.9% 가용성을 달성하고, 사용자에게 항상 응답을 제공하기 위함입니다.

코드 예제

from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.callbacks import get_openai_callback
import time

# 벡터 저장소 설정
embeddings = OpenAIEmbeddings()
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)

# Fallback LLM: 주 모델 실패 시 사용할 간단한 모델
primary_llm = OpenAI(model_name="gpt-4", temperature=0, request_timeout=10)
fallback_llm = OpenAI(model_name="gpt-3.5-turbo", temperature=0, request_timeout=5)

def create_qa_chain_with_retry(llm, max_retries=3):
    """재시도 로직을 포함한 QA 체인 생성"""
    qa_chain = RetrievalQA.from_chain_type(
        llm=llm,
        retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
        return_source_documents=True
    )
    return qa_chain

def query_with_fallback(query, verbose=True):
    """Fallback과 에러 처리를 포함한 안전한 쿼리 실행"""
    # 1차 시도: Primary LLM
    for attempt in range(3):
        try:
            if verbose:
                print(f"[시도 {attempt+1}] Primary LLM (GPT-4) 사용...")

            qa_chain = create_qa_chain_with_retry(primary_llm)

            with get_openai_callback() as cb:
                result = qa_chain({"query": query})
                if verbose:
                    print(f"성공! 토큰 사용: {cb.total_tokens}, 비용: ${cb.total_cost:.4f}")

                # 검색 결과 검증
                if not result.get('source_documents'):
                    return {"answer": "관련 정보를 찾을 수 없습니다. 질문을 다시 작성해주세요.", "source": "no_results"}

                return {"answer": result['result'], "sources": result['source_documents']}

        except Exception as e:
            if verbose:
                print(f"[오류] {type(e).__name__}: {str(e)}")

            if attempt < 2:  # 마지막 시도가 아니면 재시도
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            else:
                break  # 3회 실패 시 Fallback으로

    # 2차 시도: Fallback LLM
    try:
        if verbose:
            print("[Fallback] GPT-3.5-turbo 사용...")

        qa_chain = create_qa_chain_with_retry(fallback_llm)
        result = qa_chain({"query": query})
        return {"answer": result['result'], "sources": result.get('source_documents', []), "fallback": True}

    except Exception as e:
        # 최종 실패: 하드코딩된 응답
        if verbose:
            print(f"[최종 실패] {type(e).__name__}: {str(e)}")

        return {
            "answer": "죄송합니다. 일시적인 오류로 답변을 생성할 수 없습니다. 잠시 후 다시 시도해주세요.",
            "error": str(e),
            "fallback": "hard_coded"
        }

# 실행 예시
query = "LangChain의 핵심 개념은?"
response = query_with_fallback(query, verbose=True)
print(f"\n최종 답변: {response['answer']}")
if response.get('fallback'):
    print(f"[경고] Fallback 사용됨: {response.get('fallback')}")

설명

이것이 하는 일: 이 코드는 GPT-4로 답변을 시도하되, 실패 시 자동으로 재시도하고, 3회 실패하면 GPT-3.5로 전환하며, 그마저 실패하면 사용자에게 정중한 오류 메시지를 반환하는 다층 방어 시스템을 구축합니다. 첫 번째로, create_qa_chain_with_retry 함수는 재시도가 가능한 QA 체인을 생성합니다.

request_timeout=10은 API 호출이 10초를 넘으면 자동으로 취소하여 무한 대기를 방지합니다. 이렇게 하는 이유는 네트워크가 느린 환경에서도 시스템이 멈추지 않게 하기 위함입니다.

LangChain은 내부적으로 tenacity 라이브러리를 사용해 일부 오류(RateLimitError 등)를 자동으로 재시도하지만, 명시적으로 제어하는 것이 더 안전합니다. 그 다음으로, query_with_fallback의 재시도 루프가 실행됩니다.

for attempt in range(3)으로 최대 3회 시도하고, 실패할 때마다 time.sleep(2 ** attempt)로 대기 시간을 지수적으로 늘립니다(2초, 4초, 8초). 이것을 Exponential Backoff라고 하며, 서버 과부하 상황에서 재시도 폭주를 방지합니다.

get_openai_callback()은 각 호출의 토큰 사용량과 비용을 추적하여 로깅이나 빌링에 사용할 수 있습니다. 검색 결과 검증 부분(if not result.get('source_documents'))은 기술적 오류가 아닌 논리적 실패를 처리합니다.

벡터 스토어에서 유사한 문서를 찾지 못한 경우 "관련 정보를 찾을 수 없습니다"라고 정직하게 답변하여 할루시네이션을 방지합니다. Fallback LLM 부분은 primary_llm이 3회 모두 실패했을 때만 실행됩니다.

GPT-3.5는 GPT-4보다 10배 빠르고 20배 저렴하지만 품질은 약간 낮습니다. 프로덕션에서는 99%는 GPT-4로, 1% 실패 케이스만 GPT-3.5로 처리하여 비용과 품질의 균형을 맞춥니다.

여러분이 이 코드를 사용하면 프로덕션 수준의 안정성을 확보할 수 있습니다. 실무에서의 이점은 (1) 일시적 네트워크 오류에도 서비스 중단 없음, (2) 비용 최적화(실패 시에만 저렴한 모델 사용), (3) 사용자에게 항상 응답 제공(빈 화면 없음), (4) 오류 로그로 시스템 모니터링 가능입니다.

실전 팁

💡 Exponential backoff의 최대 대기 시간을 설정하세요. time.sleep(min(2 ** attempt, 30))으로 최대 30초로 제한하여 무한정 대기를 방지합니다.

💡 LangSmith나 Sentry를 통합하여 오류를 실시간 모니터링하세요. 예외 발생 시 자동으로 알림을 받고, 오류 패턴을 분석하여 근본 원인을 해결할 수 있습니다.

💡 Circuit Breaker 패턴을 구현하세요. 연속 5회 실패 시 1분간 주 시스템을 차단하고 Fallback만 사용하여, 다운된 서비스에 계속 요청하는 것을 방지합니다.

💡 검색 결과가 없을 때는 "더 구체적인 질문을 해주세요" 대신, 유사한 질문 예시를 제공하세요. 예: "다음 중 하나를 찾으시나요? 1) 설치 방법 2) 사용 예제 3) 문제 해결"

💡 LLM 응답 검증 체인을 추가하세요. 답변에 욕설, 민감 정보, 할루시네이션이 포함되었는지 검사하고, 문제가 있으면 "답변을 생성할 수 없습니다"로 대체합니다.

10. Performance Optimization - 속도와 비용 최적화

시작하며

여러분이 RAG 시스템을 배포했는데, 사용자가 늘어나면서 응답 시간이 5초에서 20초로 늘어나거나, OpenAI API 요금이 월 수백 달러로 급증하는 경험을 한 적 있나요? 프로토타입에서는 괜찮았지만, 프로덕션에서는 성능과 비용이 치명적입니다.

이런 문제는 실제 개발 현장에서 자주 발생합니다. 매번 문서를 다시 벡터화하거나, 불필요하게 큰 청크를 사용하거나, 캐싱을 활용하지 않으면 리소스가 낭비됩니다.

특히 트래픽이 많은 서비스에서는 1초 단축이 수만 달러의 비용 절감으로 이어집니다. 바로 이럴 때 필요한 것이 Performance Optimization입니다.

캐싱, 배치 처리, 모델 선택, 청크 크기 조정 등의 기법으로 속도를 10배 높이고 비용을 80% 절감할 수 있습니다.

개요

간단히 말해서, Performance Optimization은 RAG 시스템의 응답 속도를 높이고, API 호출 비용을 줄이며, 리소스 사용을 최적화하는 일련의 기법들입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 사용자는 3초 이내 응답을 기대하고, 느리면 떠나버리기 때문입니다.

또한 OpenAI API는 토큰당 과금되므로, 불필요한 토큰 사용은 직접적인 비용 증가로 이어집니다. 예를 들어, 하루 1만 건의 쿼리를 처리하는 서비스에서 쿼리당 평균 토큰을 2000에서 1000으로 줄이면 월 비용이 $3000에서 $1500로 50% 절감됩니다.

기존에는 매번 API를 호출하고, 모든 문서를 다시 처리하며, 단일 프로세스로 순차 처리했다면, 이제는 Redis 캐싱, 벡터 저장소 재사용, 비동기 배치 처리, 모델 다운사이징 등을 활용할 수 있습니다. Performance Optimization의 핵심 특징은 세 가지입니다: (1) 응답 캐싱(동일 질문 재사용), (2) 배치 처리(여러 요청 동시 처리), (3) 비용 최적화(작은 모델, 짧은 청크).

이러한 특징들이 중요한 이유는 확장 가능한(scalable) 시스템을 만들고, 운영 비용을 관리 가능한 수준으로 유지하기 위함입니다.

코드 예제

from langchain.cache import RedisCache
from langchain.globals import set_llm_cache
from langchain.llms import OpenAI
from langchain.embeddings.cache import CacheBackedEmbeddings
from langchain.storage import LocalFileStore
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
import redis

# 1. LLM 응답 캐싱: 동일한 질문은 캐시에서 가져옵니다
redis_client = redis.Redis(host='localhost', port=6379, db=0)
set_llm_cache(RedisCache(redis_client))

# 2. 임베딩 캐싱: 동일한 텍스트는 다시 벡터화하지 않습니다
underlying_embeddings = OpenAIEmbeddings()
store = LocalFileStore("./embedding_cache/")
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings,
    store,
    namespace=underlying_embeddings.model
)

# 3. 벡터 저장소 재사용: 이미 구축된 인덱스 로드
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=cached_embeddings
)

# 4. 비용 최적화: 더 저렴한 모델 사용
llm = OpenAI(
    model_name="gpt-3.5-turbo",  # GPT-4 대신 3.5 사용 (20배 저렴)
    temperature=0,
    max_tokens=300  # 답변 길이 제한으로 토큰 절약
)

# 5. 검색 최적화: 필요한 만큼만 검색
optimized_retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",  # 유사도 임계값 이하 제외
    search_kwargs={
        "k": 3,  # 최소한의 문서만 검색
        "score_threshold": 0.7  # 70% 이하 유사도는 제외
    }
)

# QA 체인 생성
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    retriever=optimized_retriever,
    return_source_documents=True
)

# 배치 처리 예시: 여러 질문을 동시에 처리
queries = [
    "LangChain이란?",
    "RAG 시스템의 장점은?",
    "벡터 저장소 선택 기준은?"
]

import asyncio
from concurrent.futures import ThreadPoolExecutor

def process_query(query):
    """단일 쿼리 처리"""
    return qa_chain({"query": query})

# 병렬 처리: 3배 빠른 처리
with ThreadPoolExecutor(max_workers=3) as executor:
    results = list(executor.map(process_query, queries))

for i, result in enumerate(results):
    print(f"\n질문 {i+1}: {queries[i]}")
    print(f"답변: {result['result'][:100]}...")

설명

이것이 하는 일: 이 코드는 Redis와 파일 시스템에 캐시를 구축하여 중복 API 호출을 제거하고, 저렴한 모델과 토큰 제한으로 비용을 줄이며, 병렬 처리로 처리량을 3배 높이는 종합 최적화 시스템을 구축합니다. 첫 번째로, RedisCache 설정은 LLM의 모든 응답을 Redis에 저장합니다.

동일한 프롬프트가 다시 들어오면 OpenAI API를 호출하지 않고 Redis에서 즉시 반환합니다. 예를 들어, "LangChain이란?"이라는 질문이 하루에 100번 들어오면, 첫 번째만 API를 호출하고 나머지 99번은 캐시에서 가져와 비용을 99% 절감합니다.

이렇게 하는 이유는 FAQ나 인기 있는 질문은 답변이 거의 바뀌지 않기 때문입니다. 그 다음으로, CacheBackedEmbeddings는 텍스트를 벡터로 변환할 때 캐싱합니다.

"LangChain 설명"이라는 텍스트를 한 번 벡터화하면, 다음에는 API 호출 없이 파일에서 로드합니다. 내부적으로는 텍스트의 해시값을 키로 사용하여 ./embedding_cache/ 폴더에 pickle 파일로 저장합니다.

문서를 추가할 때 이미 벡터화된 부분은 재사용하므로, 대규모 업데이트도 빠릅니다. similarity_score_threshold 설정은 유사도가 0.7(70%) 미만인 문서는 아예 반환하지 않습니다.

이렇게 하면 관련 없는 문서를 LLM에 전달하지 않아 (1) 토큰 사용 감소, (2) 답변 품질 향상(노이즈 제거), (3) 처리 속도 향상의 효과를 얻습니다. 일반적으로 0.7-0.8이 적절하며, 도메인에 따라 조정합니다.

ThreadPoolExecutor는 여러 쿼리를 동시에 처리합니다. max_workers=3이면 3개의 쿼리를 병렬로 실행하여, 순차 처리 대비 3배 빠릅니다.

주의할 점은 OpenAI API의 rate limit(분당 요청 수 제한)을 고려해야 한다는 것입니다. 너무 많은 worker는 RateLimitError를 유발할 수 있으므로, 실험을 통해 최적값을 찾으세요.

여러분이 이 코드를 사용하면 대규모 트래픽에도 안정적인 서비스를 제공할 수 있습니다. 실무에서의 이점은 (1) 응답 시간 80% 단축(캐시 히트 시), (2) API 비용 50-90% 절감(중복 호출 제거), (3) 처리량 3배 증가(병렬 처리), (4) 확장성 확보(Redis 클러스터로 수평 확장 가능)입니다.

실전 팁

💡 캐시 만료 시간(TTL)을 설정하세요. RedisCache에 ttl=3600 (1시간)을 추가하면, 오래된 정보가 계속 제공되는 것을 방지합니다. 실시간성이 중요한 뉴스는 짧게, 정적인 문서는 길게 설정합니다.

💡 Semantic Cache를 사용하면 비슷한 질문도 캐시에서 가져옵니다. "LangChain이란?"과 "LangChain 설명해줘"를 같은 질문으로 인식하여 캐시 히트율을 높입니다. GPTCache 라이브러리 참고하세요.

💡 streaming=True로 답변을 스트리밍하세요. 전체 답변을 기다리지 않고 생성되는 대로 표시하여 체감 속도를 크게 높입니다. 사용자는 3초 대기보다 즉시 시작하는 1초짜리 스트림을 선호합니다.

💡 청크 크기를 최적화하세요. 너무 크면 검색 정확도 저하, 너무 작으면 LLM 호출 증가입니다. A/B 테스트로 chunk_size=500-1000 범위에서 최적값을 찾으세요.

💡 비동기 처리를 도입하세요. asyncio와 aiohttp로 I/O 바운드 작업(API 호출, DB 쿼리)을 비동기화하면 동일한 리소스로 10배 많은 요청을 처리할 수 있습니다. LangChain도 async 메서드를 지원합니다.

#Python#LangChain#RAG#VectorStore#Embeddings#AI

RAG 시스템 6편 - LangChain으로 RAG 구현 완벽 가이드

# RAG 시스템 6편 - LangChain으로 RAG 구현 완벽 가이드 LangChain을 활용한 RAG(Retrieval-Augmented Generation) 시스템의 실전 구현 방법을 배웁니다. 문서 로딩부터 벡터 저장소 구축, 검색 기반 질의응답까지 단계별로 알아봅니다. 실무에서 바로 활용 가능한 코드와 팁을 제공합니다. --- ## 목차 1. [LangChain RAG 시스템 기본 구조 - 전체 아키텍처 이해하기](#langchain-rag-시스템-기본-구조---전체-아키텍처-이해하기) 2. [Document Loaders - 다양한 파일 형식 처리하기](#document-loaders---다양한-파일-형식-처리하기) 3. [Text Splitters - 문서를 효율적으로 청크화하기](#text-splitters---문서를-효율적으로-청크화하기) 4. [Vector Stores - 벡터 데이터베이스 구축과 활용](#vector-stores---벡터-데이터베이스-구축과-활용) 5. [Retrieval 전략 - 검색 품질 최적화하기](#retrieval-전략---검색-품질-최적화하기) 6. [RetrievalQA Chain - 질의응답 체인 구성하기](#retrievalqa-chain---질의응답-체인-구성하기) 7. [ConversationalRetrievalChain - 대화 맥락을 유지하는 RAG](#conversationalretrievalchain---대화-맥락을-유지하는-rag) 8. [Custom Prompts - 프롬프트 최적화 전략](#custom-prompts---프롬프트-최적화-전략) 4. [답을 모르면 "해당 정보를 찾을 수 없습니다. 추가 문의는 support@example.com으로 연락주세요"라고 하세요](#답을-모르면-"해당-정보를-찾을-수-없습니다.-추가-문의는-support@example.com으로-연락주세요"라고-하세요) 9. [Error Handling과 Fallbacks - 안정성 확보하기](#error-handling과-fallbacks---안정성-확보하기) 10. [Performance Optimization - 속도와 비용 최적화](#performance-optimization---속도와-비용-최적화) --- ## 1. LangChain RAG 시스템 기본 구조 - 전체 아키텍처 이해하기 ### 시작하며 여러분이 사내 문서 검색 시스템을 만들어야 하는데, GPT만으로는 최신 정보를 반영할 수 없어서 고민한 적 있나요? 수천 개의 PDF 문서에서 정확한 답변을 찾아내야 하는데, 어디서부터 시작해야 할지 막막하셨을 겁니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. LLM은 훈련 시점까지의 데이터만 알고 있기 때문에, 여러분의 회사 내부 문서나 최신 뉴스 같은 정보는 전혀 모릅니다. 그래서 잘못된 답변을 하거나 아예 답변을 못 하는 경우가 생깁니다. 바로 이럴 때 필요한 것이 LangChain을 이용한 RAG 시스템입니다. 외부 문서를 검색해서 LLM에게 컨텍스트로 제공하면, 정확하고 최신의 답변을 얻을 수 있습니다. ### 개요 간단히 말해서, LangChain RAG 시스템은 "검색 + 생성"의 두 단계로 작동하는 질의응답 시스템입니다. 왜 이 시스템이 필요한지 실무 관점에서 설명하자면, LLM 단독으로는 할 수 없는 일들을 가능하게 해줍니다. 예를 들어, 회사의 내부 규정 문서를 기반으로 한 챗봇, 최신 제품 매뉴얼 검색 시스템, 법률 문서 분석 도구 같은 경우에 매우 유용합니다. 기존에는 키워드 기반 검색으로 문서를 찾아서 사람이 직접 읽어야 했다면, 이제는 자연어로 질문하면 AI가 관련 문서를 찾아서 요약된 답변까지 제공할 수 있습니다. LangChain RAG의 핵심 특징은 세 가지입니다: (1) 문서를 벡터로 변환하여 의미 기반 검색이 가능하고, (2) 여러 데이터 소스를 통합할 수 있으며, (3) LLM과의 연동이 간단합니다. 이러한 특징들이 중요한 이유는 복잡한 시스템을 빠르게 구축하고, 유지보수도 쉽게 할 수 있기 때문입니다. ### 코드 예제 ```python from langchain.document_loaders import TextLoader from langchain.embeddings import OpenAIEmbeddings from langchain.vectorstores import Chroma from langchain.llms import OpenAI from langchain.chains import RetrievalQA # 문서 로딩: 텍스트 파일을 읽어옵니다 loader = TextLoader("company_docs.txt", encoding="utf-8") documents = loader.load() # 임베딩 모델 초기화: 텍스트를 벡터로 변환 embeddings = OpenAIEmbeddings() # 벡터 저장소 생성: 문서를 벡터화하여 저장 vectorstore = Chroma.from_documents(documents, embeddings) # LLM 초기화: 답변 생성을 담당 llm = OpenAI(temperature=0) # RAG 체인 구성: 검색 + 생성을 하나로 연결 qa_chain = RetrievalQA.from_chain_type( llm=llm, retriever=vectorstore.as_retriever(), return_source_documents=True # 출처 문서도 함께 반환 ) # 질문하기 result = qa_chain({"query": "우리 회사의 휴가 정책은?"}) print(result["result"]) ``` ### 설명 이것이 하는 일: 이 코드는 텍스트 문서를 읽어서 검색 가능한 벡터 DB로 만들고, 질문에 대한 답변을 자동으로 생성하는 전체 파이프라인을 구축합니다. 첫 번째로, TextLoader와 OpenAIEmbeddings 부분은 문서를 AI가 이해할 수 있는 형태로 변환하는 것을 담당합니다. TextLoader가 파일을 읽으면, 그 텍스트는 OpenAI의 임베딩 모델을 통해 1536차원의 벡터로 변환됩니다. 이렇게 하는 이유는 단순한 키워드 매칭이 아니라 의미적으로 유사한 문서를 찾기 위함입니다. 그 다음으로, Chroma.from_documents가 실행되면서 모든 문서가 벡터화되어 데이터베이스에 저장됩니다. 내부에서는 각 문서를 적절한 크기의 청크로 나누고, 각 청크마다 임베딩을 생성하여 인덱싱합니다. 이 과정이 중요한 이유는 나중에 빠른 검색이 가능하도록 최적화된 구조를 만들기 때문입니다. RetrievalQA.from_chain_type이 실행되면 검색과 생성을 하나의 체인으로 연결합니다. 질문이 들어오면 먼저 vectorstore에서 관련 문서를 검색하고, 그 문서들을 컨텍스트로 LLM에게 전달하여 최종 답변을 생성합니다. 여러분이 이 코드를 사용하면 몇 줄의 코드만으로 강력한 질의응답 시스템을 구축할 수 있습니다. 실무에서의 이점은 (1) 문서가 업데이트되어도 LLM 재훈련 없이 반영 가능, (2) 답변의 출처를 명확히 제시 가능, (3) 다양한 파일 포맷 지원으로 확장성이 높다는 점입니다. ### 실전 팁 💡 return_source_documents=True를 반드시 설정하세요. 답변의 출처를 보여줄 수 있어 신뢰도가 크게 향상됩니다. 특히 법률이나 의료 같은 민감한 분야에서는 필수입니다. 💡 temperature=0으로 설정하면 일관성 있는 답변을 얻을 수 있습니다. RAG 시스템은 창의적인 글쓰기가 아니라 정확한 정보 전달이 목적이므로, 랜덤성을 최소화하는 것이 좋습니다. 💡 문서 로딩 시 encoding="utf-8"을 명시하세요. 한글 문서를 다룰 때 인코딩 오류로 고생하는 경우가 많은데, 이것만으로 대부분의 문제를 해결할 수 있습니다. 💡 처음 벡터 저장소를 만들 때는 시간이 걸리므로, 한 번 만든 후에는 persist_directory를 사용해 디스크에 저장하고 재사용하세요. 매번 새로 만들면 비용과 시간이 낭비됩니다. 💡 실무에서는 vectorstore.as_retriever()에 search_kwargs={"k": 3}을 추가해 검색 결과 개수를 조절하세요. 너무 많으면 LLM의 컨텍스트 창을 넘길 수 있고, 너무 적으면 정보가 부족할 수 있습니다. --- ## 2. Document Loaders - 다양한 파일 형식 처리하기 ### 시작하며 여러분이 RAG 시스템을 구축하려는데, 회사 문서가 PDF, Word, 웹페이지, 데이터베이스 등 여러 형식으로 흩어져 있어서 막막한 적 있나요? 각 파일 형식마다 다른 라이브러리를 찾아서 파싱 로직을 일일이 작성해야 한다면 정말 번거로울 겁니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. 특히 레거시 시스템을 다루는 경우, 오래된 HWP 파일부터 최신 Notion 페이지까지 다양한 소스를 통합해야 하는데, 각각의 파싱 방법이 달라서 개발 시간이 배로 늘어납니다. 바로 이럴 때 필요한 것이 LangChain의 Document Loaders입니다. 100가지 이상의 파일 형식과 데이터 소스를 통일된 인터페이스로 처리할 수 있어, 데이터 수집 파이프라인을 빠르게 구축할 수 있습니다. ### 개요 간단히 말해서, Document Loaders는 다양한 형식의 데이터를 LangChain이 이해할 수 있는 Document 객체로 변환하는 통일된 인터페이스입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 데이터 소스가 다양할수록 시스템의 가치가 높아지기 때문입니다. 예를 들어, 고객 지원 챗봇을 만든다면 FAQ PDF, 제품 매뉴얼 Word 파일, 고객 문의 데이터베이스, 최신 업데이트가 있는 웹페이지 등을 모두 활용해야 완성도 높은 시스템이 됩니다. 기존에는 각 파일 형식마다 PyPDF2, python-docx, BeautifulSoup 등 다른 라이브러리를 사용하고 파싱 결과를 표준화하는 코드를 직접 작성했다면, 이제는 해당하는 Loader를 import하고 .load()만 호출하면 됩니다. Document Loaders의 핵심 특징은 세 가지입니다: (1) 통일된 인터페이스로 모든 소스를 동일하게 처리, (2) 메타데이터 자동 추출(파일명, 페이지 번호, URL 등), (3) lazy loading 지원으로 대용량 파일도 메모리 효율적으로 처리. 이러한 특징들이 중요한 이유는 개발 속도를 10배 이상 높이고, 코드의 일관성을 유지할 수 있기 때문입니다. ### 코드 예제 ```python from langchain.document_loaders import ( PyPDFLoader, UnstructuredWordDocumentLoader, WebBaseLoader, CSVLoader ) # PDF 문서 로딩: 각 페이지가 별도 Document로 변환됩니다 pdf_loader = PyPDFLoader("product_manual.pdf") pdf_docs = pdf_loader.load() print(f"PDF 페이지 수: {len(pdf_docs)}") # Word 문서 로딩: 전체 내용이 하나의 Document로 변환됩니다 word_loader = UnstructuredWordDocumentLoader("company_policy.docx") word_docs = word_loader.load() # 웹페이지 로딩: HTML에서 텍스트만 추출합니다 web_loader = WebBaseLoader("https://example.com/faq") web_docs = web_loader.load() # CSV 로딩: 각 행이 별도 Document로 변환됩니다 csv_loader = CSVLoader("customer_data.csv", encoding="utf-8") csv_docs = csv_loader.load() # 모든 문서 통합: 서로 다른 소스의 문서를 하나로 합칩니다 all_documents = pdf_docs + word_docs + web_docs + csv_docs print(f"전체 문서 수: {len(all_documents)}") # 메타데이터 확인: 출처 추적이 가능합니다 for doc in all_documents[:3]: print(f"출처: {doc.metadata.get('source', 'Unknown')}") ``` ### 설명 이것이 하는 일: 이 코드는 서로 다른 네 가지 파일 형식을 읽어서 모두 동일한 Document 형식으로 변환하고, 하나의 리스트로 통합하는 멀티소스 데이터 파이프라인을 구축합니다. 첫 번째로, PyPDFLoader 부분은 PDF 파일을 페이지 단위로 분리하여 로딩합니다. 각 페이지가 별도의 Document 객체가 되고, metadata에는 페이지 번호와 파일 경로가 자동으로 저장됩니다. 이렇게 하는 이유는 나중에 "3페이지에서 찾았습니다"처럼 정확한 위치를 사용자에게 알려줄 수 있기 때문입니다. 그 다음으로, 각 Loader의 .load() 메서드가 실행되면서 내부적으로는 파일을 파싱하고, 텍스트를 추출하고, 메타데이터를 구성하는 복잡한 과정이 일어납니다. 예를 들어 WebBaseLoader는 HTML을 다운로드하고, JavaScript를 실행하지 않은 상태에서 텍스트만 깔끔하게 추출합니다. UnstructuredWordDocumentLoader는 .docx 파일의 XML 구조를 파싱해서 텍스트와 서식 정보를 분리합니다. 마지막으로, all_documents = pdf_docs + word_docs + ...가 실행되면 서로 다른 소스의 문서들이 하나의 리스트로 통합됩니다. 이 리스트는 이후 벡터 저장소에 저장되거나 검색에 사용될 수 있습니다. 메타데이터를 확인하는 부분은 각 문서가 어디서 왔는지 추적할 수 있게 해줍니다. 여러분이 이 코드를 사용하면 데이터 소스를 추가할 때마다 단 3줄(import, 초기화, load)만 추가하면 됩니다. 실무에서의 이점은 (1) 새로운 파일 형식 추가가 매우 쉽고, (2) 모든 문서가 동일한 구조를 가져 후처리가 간단하며, (3) 메타데이터 덕분에 답변의 출처를 명확히 제시할 수 있다는 점입니다. ### 실전 팁 💡 대용량 PDF는 PyPDFLoader 대신 UnstructuredPDFLoader를 사용하세요. 이미지가 많거나 복잡한 레이아웃을 가진 PDF는 더 정확하게 파싱됩니다. 다만 속도는 조금 느릴 수 있으니 trade-off를 고려하세요. 💡 웹페이지 로딩 시 WebBaseLoader에 bs_kwargs={"parse_only": SoupStrainer("main")}을 전달하면 메인 콘텐츠만 추출할 수 있습니다. 네비게이션 메뉴나 광고 같은 불필요한 텍스트를 제거해서 검색 품질을 높일 수 있습니다. 💡 CSV 데이터는 CSVLoader의 csv_args 매개변수로 구분자나 인용 문자를 커스터마이징하세요. 예: CSVLoader("data.csv", csv_args={"delimiter": "\t"})로 탭 구분 파일도 처리 가능합니다. 💡 DirectoryLoader를 사용하면 폴더 전체를 한 번에 로딩할 수 있습니다. 예: DirectoryLoader("./docs", glob="**/*.pdf", loader_cls=PyPDFLoader)로 모든 하위 폴더의 PDF를 재귀적으로 로딩합니다. 💡 데이터베이스는 SQLDatabaseLoader를 사용하되, 쿼리 결과를 문자열로 변환하는 로직을 추가하세요. 예: "고객ID 123, 이름 홍길동, 주문일 2024-01-15" 형식으로 변환하면 LLM이 더 잘 이해합니다. --- ## 3. Text Splitters - 문서를 효율적으로 청크화하기 ### 시작하며 여러분이 300페이지짜리 매뉴얼을 RAG 시스템에 넣었는데, 질문에 대한 답변이 엉뚱하게 나오거나 "토큰 제한을 초과했습니다" 에러가 발생한 적 있나요? 긴 문서를 그대로 벡터화하면 의미가 희석되고, LLM에게 전달할 때도 컨텍스트 창을 넘어버리는 문제가 생깁니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. 문서를 너무 크게 나누면 검색 정확도가 떨어지고, 너무 작게 나누면 문맥이 끊겨서 의미를 파악하기 어렵습니다. 적절한 크기로 나누는 것이 RAG 시스템 성능의 핵심입니다. 바로 이럴 때 필요한 것이 Text Splitters입니다. 문서의 구조와 의미를 고려하여 최적의 크기로 청크를 나누고, 청크 간 문맥도 유지할 수 있게 해줍니다. ### 개요 간단히 말해서, Text Splitters는 긴 문서를 검색과 처리에 적합한 크기의 청크(덩어리)로 지능적으로 분할하는 도구입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 임베딩 모델과 LLM 모두 입력 크기에 제한이 있고, 너무 긴 텍스트는 벡터화했을 때 의미가 뭉개지기 때문입니다. 예를 들어, 1만 단어짜리 기술 문서를 통째로 벡터화하면 특정 질문과의 유사도를 정확히 계산하기 어렵습니다. 하지만 각 섹션별로 나누면 "인증 방법"에 대한 질문에 인증 섹션만 정확히 매칭됩니다. 기존에는 단순히 글자 수나 단어 수로 문서를 자르거나, 정규식으로 문단을 나눴다면, 이제는 문서의 구조(제목, 문단, 문장)를 이해하고, 의미 단위로 나누며, 청크 간 오버랩을 두어 문맥도 보존할 수 있습니다. Text Splitters의 핵심 특징은 세 가지입니다: (1) 문서 타입별 최적화된 분할 전략(코드, 마크다운, 일반 텍스트 등), (2) chunk_overlap으로 청크 간 문맥 유지, (3) 토큰 기반 분할로 LLM 제한 정확히 준수. 이러한 특징들이 중요한 이유는 검색 품질과 답변 정확도를 직접적으로 좌우하기 때문입니다. ### 코드 예제 ```python from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain.document_loaders import TextLoader # 문서 로딩 loader = TextLoader("technical_doc.txt", encoding="utf-8") documents = loader.load() # Text Splitter 초기화: 청크 크기와 오버랩 설정 text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, # 각 청크의 최대 문자 수 chunk_overlap=200, # 청크 간 겹치는 문자 수 length_function=len, # 길이 측정 함수 separators=["\n\n", "\n", ". ", " ", ""] # 분할 우선순위 ) # 문서 분할 실행: 의미 단위로 청크를 나눕니다 chunks = text_splitter.split_documents(documents) print(f"원본 문서 수: {len(documents)}") print(f"분할된 청크 수: {len(chunks)}") # 청크 확인: 각 청크의 크기와 내용을 확인 for i, chunk in enumerate(chunks[:3]): print(f"\n청크 {i+1} (길이: {len(chunk.page_content)}자)") print(f"내용: {chunk.page_content[:100]}...") print(f"메타데이터: {chunk.metadata}") ``` ### 설명 이것이 하는 일: 이 코드는 긴 문서를 읽어서 의미를 보존하면서 검색에 최적화된 크기의 청크들로 분할하고, 각 청크가 독립적으로 이해될 수 있도록 오버랩을 적용합니다. 첫 번째로, RecursiveCharacterTextSplitter 초기화 부분은 분할 전략을 설정합니다. chunk_size=1000은 각 청크가 최대 1000자를 넘지 않도록 하고, chunk_overlap=200은 이전 청크의 마지막 200자를 다음 청크의 시작에 포함시킵니다. 이렇게 하는 이유는 청크 경계에서 문맥이 끊기는 것을 방지하기 위함입니다. 예를 들어, "따라서 이 방법은"이라는 문장이 청크 경계에 걸리면, 오버랩 덕분에 양쪽 청크 모두에서 앞뒤 문맥을 파악할 수 있습니다. 그 다음으로, separators 매개변수가 실행되면서 재귀적 분할이 시작됩니다. 먼저 "\n\n"(문단 구분)으로 나누려고 시도하고, 그래도 chunk_size를 넘으면 "\n"(줄바꿈)으로, 그것도 안 되면 ". "(문장 끝)으로, 최종적으로 " "(공백)으로 분할합니다. 내부에서는 재귀 알고리즘이 동작하여 가능한 한 큰 의미 단위를 유지하면서 크기 제한을 맞춥니다. split_documents()가 실행되면 각 원본 Document가 여러 개의 작은 Document로 변환됩니다. 중요한 점은 원본의 메타데이터(파일명, 페이지 번호 등)가 모든 청크에 복사되어, 나중에 출처를 추적할 수 있다는 것입니다. 여러분이 이 코드를 사용하면 검색 정확도가 크게 향상됩니다. 실무에서의 이점은 (1) 질문과 관련된 정확한 부분만 검색되어 노이즈 감소, (2) 청크 오버랩으로 문맥 손실 최소화, (3) LLM 토큰 제한을 안정적으로 준수하여 에러 방지, (4) 답변 생성 속도 향상(작은 청크만 처리하므로)입니다. ### 실전 팁 💡 chunk_size는 임베딩 모델의 최대 토큰 수의 70% 정도로 설정하세요. OpenAI ada-002는 8191 토큰이므로, 약 5000자(영어 기준 ~1250 단어)가 적절합니다. 안전 마진을 두는 이유는 특수 문자나 토큰화 오차 때문입니다. 💡 chunk_overlap은 chunk_size의 10-20%로 설정하는 것이 일반적입니다. 너무 크면 중복 저장으로 인한 비용 증가, 너무 작으면 문맥 손실이 발생합니다. 실험을 통해 최적값을 찾으세요. 💡 코드 문서는 RecursiveCharacterTextSplitter.from_language(language="python", ...)를 사용하세요. 함수나 클래스 단위로 나누기 때문에 코드 검색 품질이 훨씬 좋습니다. 지원 언어: Python, JavaScript, Java, C, Go 등. 💡 마크다운 문서는 MarkdownHeaderTextSplitter를 사용하면 제목 구조를 메타데이터로 보존할 수 있습니다. 예: "# 설치 > ## 요구사항"이 메타데이터에 저장되어, "설치 요구사항"을 검색할 때 정확도가 높아집니다. 💡 토큰 수를 정확히 제어하려면 CharacterTextSplitter 대신 TokenTextSplitter를 사용하고, tokenizer에 tiktoken 라이브러리를 전달하세요. 특히 GPT-4처럼 토큰 비용이 높은 모델을 사용할 때 중요합니다. --- ## 4. Vector Stores - 벡터 데이터베이스 구축과 활용 ### 시작하며 여러분이 수만 개의 문서를 검색 가능하게 만들었는데, 검색 속도가 너무 느리거나, 서버를 재시작할 때마다 처음부터 다시 벡터화해야 해서 고민한 적 있나요? 인메모리로만 관리하면 확장성이 떨어지고, 영구 저장이 안 되면 매번 비용이 발생합니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. 특히 프로덕션 환경에서는 수백만 개의 벡터를 빠르게 검색하고, 데이터를 영구 저장하며, 실시간으로 업데이트할 수 있어야 합니다. 단순한 리스트로는 이런 요구사항을 만족시킬 수 없습니다. 바로 이럴 때 필요한 것이 Vector Stores입니다. 벡터를 효율적으로 저장하고, 유사도 검색을 빠르게 수행하며, 디스크에 영구 저장까지 지원하는 전문 데이터베이스입니다. ### 개요 간단히 말해서, Vector Stores는 고차원 벡터를 저장하고 유사도 기반 검색을 수행하는 특수한 데이터베이스입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 일반 데이터베이스의 정확한 매칭(WHERE name = 'John')이 아니라 의미적 유사도 검색(가장 비슷한 문서 찾기)이 필요하기 때문입니다. 예를 들어, "환불 방법"이라는 질문에 "결제 취소 절차"라는 제목의 문서를 찾아야 하는데, 키워드는 하나도 겹치지 않지만 의미는 유사합니다. 벡터 검색만이 이를 가능하게 합니다. 기존에는 모든 벡터를 메모리에 로드하고 하나씩 코사인 유사도를 계산하는 브루트포스 방식을 사용했다면, 이제는 HNSW, IVF 같은 근사 최근접 이웃(ANN) 알고리즘으로 밀리초 단위의 빠른 검색이 가능합니다. Vector Stores의 핵심 특징은 세 가지입니다: (1) 빠른 유사도 검색(수백만 벡터에서 밀리초 응답), (2) 영구 저장과 인덱싱(디스크/클라우드 저장 지원), (3) 메타데이터 필터링(조건부 검색 가능). 이러한 특징들이 중요한 이유는 프로덕션 수준의 RAG 시스템을 구축하는 데 필수적이기 때문입니다. ### 코드 예제 ```python from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.document_loaders import TextLoader from langchain.text_splitter import RecursiveCharacterTextSplitter # 문서 준비 loader = TextLoader("knowledge_base.txt", encoding="utf-8") documents = loader.load() text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) chunks = text_splitter.split_documents(documents) # 임베딩 모델 초기화 embeddings = OpenAIEmbeddings() # 벡터 저장소 생성 및 영구 저장: persist_directory로 디스크에 저장 vectorstore = Chroma.from_documents( documents=chunks, embedding=embeddings, persist_directory="./chroma_db" # 디스크 저장 경로 ) # 유사도 검색: 질문과 가장 유사한 문서 k개 반환 query = "비밀번호를 재설정하는 방법은?" results = vectorstore.similarity_search(query, k=3) # 검색 결과 출력 for i, doc in enumerate(results): print(f"\n결과 {i+1} (유사도 점수 포함):") print(f"내용: {doc.page_content[:200]}...") print(f"출처: {doc.metadata.get('source', 'Unknown')}") # 기존 벡터 저장소 로드: 서버 재시작 시 재사용 loaded_vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=embeddings ) ``` ### 설명 이것이 하는 일: 이 코드는 문서를 벡터 데이터베이스에 저장하고, 자연어 질문으로 의미적으로 유사한 문서를 검색하며, 결과를 디스크에 영구 저장하여 재사용 가능하게 만듭니다. 첫 번째로, Chroma.from_documents 부분은 모든 청크를 벡터화하고 인덱싱합니다. 내부적으로는 각 청크의 텍스트를 embeddings.embed_documents()로 벡터로 변환하고(API 호출), 그 벡터들을 Chroma의 내부 인덱스 구조에 저장합니다. persist_directory가 설정되어 있으면 SQLite 데이터베이스 파일로 디스크에 저장됩니다. 이렇게 하는 이유는 다음에 실행할 때 API 호출 없이 바로 로드할 수 있어 시간과 비용을 절약하기 때문입니다. 그 다음으로, similarity_search가 실행되면 질문 문자열이 벡터로 변환되고(embeddings.embed_query()), 저장된 모든 벡터와의 코사인 유사도가 계산됩니다. Chroma는 내부적으로 HNSW(Hierarchical Navigable Small World) 알고리즘을 사용해 전체 벡터를 순회하지 않고도 가장 유사한 k개를 빠르게 찾습니다. k=3이므로 상위 3개 문서만 반환됩니다. 마지막으로, Chroma(persist_directory=...)로 로드하는 부분은 기존에 저장된 벡터 인덱스를 디스크에서 읽어옵니다. 이때 embedding_function을 동일하게 전달해야 새로운 쿼리를 벡터화할 수 있습니다. 이 과정은 수 초 만에 완료되어, 수만 개의 문서를 다시 벡터화하는 것(수 분 소요)에 비해 엄청나게 빠릅니다. 여러분이 이 코드를 사용하면 프로덕션 수준의 검색 시스템을 구축할 수 있습니다. 실무에서의 이점은 (1) 초기 구축 후 빠른 재시작(벡터 재생성 불필요), (2) 대규모 문서에서도 빠른 검색(밀리초 단위), (3) 점진적 업데이트 가능(새 문서만 추가), (4) 메모리 효율적(필요한 부분만 로드)입니다. ### 실전 팁 💡 Chroma 외에도 용도에 맞는 벡터 스토어를 선택하세요. Pinecone(클라우드, 대규모), FAISS(로컬, 초고속), Weaviate(하이브리드 검색), Qdrant(필터링 강력) 등이 있습니다. 무료 프로토타입은 Chroma, 프로덕션은 Pinecone 추천합니다. 💡 similarity_search_with_score()를 사용하면 유사도 점수도 함께 반환됩니다. 점수가 너무 낮으면(예: 0.5 이하) "관련 정보를 찾을 수 없습니다"라고 답변하는 로직을 추가하세요. 엉뚱한 답변을 방지할 수 있습니다. 💡 메타데이터 필터링을 활용하세요. 예: vectorstore.similarity_search(query, k=3, filter={"source": "manual.pdf"})로 특정 문서 내에서만 검색 가능합니다. 권한 제어나 카테고리별 검색에 유용합니다. 💡 대용량 데이터는 배치로 추가하세요. vectorstore.add_documents(chunks)를 한 번에 수천 개씩 호출하면 인덱싱 효율이 높아집니다. 한 번에 하나씩 추가하면 인덱스 재구성이 반복되어 느립니다. 💡 OpenAI 임베딩 API 비용이 부담된다면, HuggingFaceEmbeddings로 로컬 모델을 사용하세요. "sentence-transformers/all-MiniLM-L6-v2"는 무료이고 성능도 준수합니다. 다만 다국어 지원은 OpenAI가 더 좋습니다. --- ## 5. Retrieval 전략 - 검색 품질 최적화하기 ### 시작하며 여러분이 RAG 시스템을 만들었는데, 간단한 질문은 잘 답변하지만 복잡한 질문이나 다단계 추론이 필요한 질문에는 엉뚱한 답변을 하는 경우를 경험한 적 있나요? 단순히 유사도가 높은 문서 몇 개를 가져오는 것만으로는 부족한 상황이 많습니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. "작년 대비 올해 매출 증가율은?"처럼 여러 문서를 종합해야 하는 질문이나, "AWS Lambda의 장단점 비교"처럼 다양한 관점의 문서가 필요한 질문에서는 기본 검색 전략으로는 한계가 있습니다. 바로 이럴 때 필요한 것이 고급 Retrieval 전략입니다. 검색 결과를 재순위화하거나, 여러 번 검색하거나, 쿼리를 확장하는 등의 기법으로 검색 품질을 크게 향상시킬 수 있습니다. ### 개요 간단히 말해서, Retrieval 전략은 단순한 벡터 유사도 검색을 넘어서 더 정확하고 포괄적인 문서를 찾기 위한 고급 기법들입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 실제 사용자의 질문은 단순하지 않고, 맥락이 부족하거나, 여러 정보를 종합해야 하는 경우가 많기 때문입니다. 예를 들어, "그거 어떻게 해?"라는 모호한 질문이나, "A와 B를 비교해줘"처럼 두 가지 개념이 섞인 질문, "최신 정책은?"처럼 시간 정보가 중요한 질문 등에서는 고급 전략이 필수입니다. 기존에는 similarity_search(query, k=3)로 상위 3개만 가져왔다면, 이제는 MMR(Maximum Marginal Relevance)로 다양성을 보장하거나, Contextual Compression으로 관련 부분만 추출하거나, Multi-Query로 여러 각도에서 검색할 수 있습니다. 고급 Retrieval 전략의 핵심 특징은 세 가지입니다: (1) 다양성 보장(중복되지 않는 관점의 문서 검색), (2) 정밀도 향상(불필요한 부분 제거), (3) 재현율 향상(더 많은 관련 문서 발견). 이러한 특징들이 중요한 이유는 답변의 품질과 사용자 만족도를 직접적으로 높이기 때문입니다. ### 코드 예제 ```python from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.retrievers import ContextualCompressionRetriever from langchain.retrievers.document_compressors import LLMChainExtractor from langchain.llms import OpenAI # 기본 벡터 저장소와 리트리버 설정 embeddings = OpenAIEmbeddings() vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) # MMR 검색: 유사도와 다양성을 동시에 고려합니다 mmr_retriever = vectorstore.as_retriever( search_type="mmr", # Maximum Marginal Relevance search_kwargs={"k": 5, "fetch_k": 20, "lambda_mult": 0.5} ) # Contextual Compression: 문서에서 관련 부분만 추출합니다 llm = OpenAI(temperature=0) compressor = LLMChainExtractor.from_llm(llm) compression_retriever = ContextualCompressionRetriever( base_compressor=compressor, base_retriever=mmr_retriever ) # 검색 실행: 압축된 관련 정보만 가져옵니다 query = "LangChain에서 메모리를 사용하는 방법은?" compressed_docs = compression_retriever.get_relevant_documents(query) # 결과 확인 print(f"압축된 문서 수: {len(compressed_docs)}") for i, doc in enumerate(compressed_docs): print(f"\n문서 {i+1} (압축됨):") print(f"내용: {doc.page_content}") print(f"출처: {doc.metadata.get('source', 'Unknown')}") ``` ### 설명 이것이 하는 일: 이 코드는 단순한 유사도 검색을 넘어서, 다양한 관점의 문서를 가져오고(MMR), 각 문서에서 질문과 직접 관련된 부분만 추출하여(Compression) 고품질의 컨텍스트를 생성합니다. 첫 번째로, MMR 리트리버 설정 부분은 검색 전략을 세밀하게 조정합니다. search_type="mmr"은 단순히 유사도가 높은 문서만 가져오는 것이 아니라, 이미 선택된 문서와 너무 비슷한 문서는 제외합니다. fetch_k=20은 먼저 20개의 후보를 가져오고, 그중에서 다양성을 고려해 k=5개를 최종 선택합니다. lambda_mult=0.5는 유사도와 다양성의 균형점인데, 0에 가까우면 다양성 우선, 1에 가까우면 유사도 우선입니다. 이렇게 하는 이유는 같은 내용의 중복 문서를 피하고, 다양한 각도의 정보를 제공하기 위함입니다. 그 다음으로, ContextualCompressionRetriever가 실행되면 두 단계의 처리가 일어납니다. 먼저 base_retriever(MMR)가 5개의 다양한 문서를 가져오고, 그 다음 compressor(LLM)가 각 문서를 분석하여 질문과 직접 관련된 문장이나 문단만 추출합니다. 예를 들어, 1000자짜리 문서에서 실제로 질문에 답하는 200자만 남기고 나머지는 제거합니다. 내부적으로는 LLM에게 "이 문서에서 질문과 관련된 부분만 추출하라"는 프롬프트를 보냅니다. get_relevant_documents()가 실행되면 최종적으로 압축되고 다양성이 보장된 고품질 문서들이 반환됩니다. 이 문서들은 불필요한 정보가 제거되어 토큰 수가 적고, 중복이 없어 효율적이며, 여러 관점을 포함하여 포괄적입니다. 여러분이 이 코드를 사용하면 답변 품질이 눈에 띄게 향상됩니다. 실무에서의 이점은 (1) LLM에 전달되는 토큰 수 감소로 비용 절감, (2) 불필요한 정보가 없어 답변 정확도 향상, (3) 다양한 관점으로 편향 감소, (4) 복잡한 질문에 대한 포괄적인 답변 가능입니다. ### 실전 팁 💡 MMR의 lambda_mult 값은 도메인에 따라 조정하세요. 뉴스 검색처럼 다양한 관점이 중요하면 0.3, 기술 문서처럼 정확도가 중요하면 0.7로 설정합니다. A/B 테스트로 최적값을 찾으세요. 💡 Contextual Compression은 비용이 높으므로(문서마다 LLM 호출), fetch_k를 너무 크게 설정하지 마세요. 보통 k의 2-4배가 적당합니다. 예: k=5이면 fetch_k=10-20. 💡 Self-Query Retriever를 사용하면 자연어 질문을 메타데이터 필터로 자동 변환합니다. 예: "2024년 이후의 기술 문서"를 {year: {$gte: 2024}, category: "tech"}로 변환해서 검색합니다. 💡 Multi-Query Retriever는 하나의 질문을 여러 각도로 다시 작성하여 검색합니다. "LangChain 설치 방법"을 "How to install LangChain", "LangChain setup guide", "LangChain 시작하기"로 확장해서 재현율을 높입니다. 💡 Ensemble Retriever로 BM25(키워드 검색)와 벡터 검색을 결합하세요. 고유명사나 정확한 용어는 BM25가, 의미적 유사성은 벡터 검색이 잘 찾으므로 둘을 가중 평균하면 최고의 결과를 얻습니다. --- ## 6. RetrievalQA Chain - 질의응답 체인 구성하기 ### 시작하며 여러분이 벡터 저장소에서 관련 문서를 검색하는 것까지는 성공했는데, 그 문서들을 LLM에 어떻게 전달하고, 프롬프트를 어떻게 구성하며, 답변을 어떻게 포맷팅해야 할지 막막한 적 있나요? 검색과 생성을 매번 수동으로 연결하면 코드가 복잡해지고 일관성을 유지하기 어렵습니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. 특히 프롬프트 엔지니어링, 컨텍스트 관리, 에러 핸들링 등을 고려하면 RAG 시스템 구축이 생각보다 훨씬 복잡해집니다. 매번 같은 로직을 반복 작성하는 것도 비효율적입니다. 바로 이럴 때 필요한 것이 RetrievalQA Chain입니다. 검색부터 답변 생성까지 전체 파이프라인을 하나의 체인으로 묶어, 간단한 코드로 강력한 질의응답 시스템을 만들 수 있습니다. ### 개요 간단히 말해서, RetrievalQA Chain은 문서 검색과 LLM 기반 답변 생성을 하나로 통합한 End-to-End 질의응답 파이프라인입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, RAG의 모든 단계(검색 → 컨텍스트 구성 → 프롬프트 생성 → LLM 호출 → 답변 추출)를 자동화하여 개발 속도를 10배 이상 높이기 때문입니다. 예를 들어, 고객 지원 챗봇을 만든다면 사용자 질문을 받아서 답변을 반환하는 전체 로직을 단 몇 줄로 구현할 수 있습니다. 기존에는 retriever.get_relevant_documents() → 문서 포맷팅 → 프롬프트 템플릿 작성 → llm.predict() → 답변 파싱의 각 단계를 직접 코딩했다면, 이제는 RetrievalQA.from_chain_type()로 모든 것이 자동화됩니다. RetrievalQA Chain의 핵심 특징은 세 가지입니다: (1) 다양한 체인 타입 지원(stuff, map_reduce, refine 등), (2) 커스터마이징 가능한 프롬프트 템플릿, (3) 출처 문서 반환으로 투명성 보장. 이러한 특징들이 중요한 이유는 프로덕션 수준의 신뢰할 수 있는 시스템을 빠르게 구축할 수 있기 때문입니다. ### 코드 예제 ```python from langchain.chains import RetrievalQA from langchain.llms import OpenAI from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.prompts import PromptTemplate # 벡터 저장소 로드 embeddings = OpenAIEmbeddings() vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) # LLM 초기화 llm = OpenAI(temperature=0, model_name="gpt-3.5-turbo") # 커스텀 프롬프트 템플릿: 답변 형식을 제어합니다 prompt_template = """아래 컨텍스트를 사용하여 질문에 답변하세요. 답을 모르면 "정보가 부족합니다"라고 답하고, 추측하지 마세요. 답변은 최대 3문장으로 간결하게 작성하세요. 컨텍스트: {context} 질문: {question} 답변:""" PROMPT = PromptTemplate( template=prompt_template, input_variables=["context", "question"] ) # RetrievalQA 체인 생성 qa_chain = RetrievalQA.from_chain_type( llm=llm, chain_type="stuff", # 모든 문서를 하나의 프롬프트에 포함 retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), return_source_documents=True, # 출처 반환 chain_type_kwargs={"prompt": PROMPT} # 커스텀 프롬프트 적용 ) # 질문 실행 query = "LangChain에서 체인을 어떻게 구성하나요?" result = qa_chain({"query": query}) # 결과 출력 print(f"질문: {query}") print(f"\n답변: {result['result']}") print(f"\n출처 문서 수: {len(result['source_documents'])}") for i, doc in enumerate(result['source_documents']): print(f"\n출처 {i+1}: {doc.metadata.get('source', 'Unknown')}") ``` ### 설명 이것이 하는 일: 이 코드는 질문을 받으면 자동으로 관련 문서를 검색하고, 그 문서들을 프롬프트에 포함시켜 LLM에 전달하며, 생성된 답변과 출처 문서를 함께 반환하는 완전한 질의응답 시스템을 구축합니다. 첫 번째로, PromptTemplate 정의 부분은 LLM에게 어떻게 답변할지 지시합니다. {context}에는 검색된 문서들이, {question}에는 사용자의 질문이 자동으로 채워집니다. "답을 모르면 추측하지 마세요"라는 지시는 할루시네이션을 방지하고, "최대 3문장"은 답변 길이를 제어합니다. 이렇게 하는 이유는 프롬프트가 답변의 품질과 형식을 직접적으로 결정하기 때문입니다. 실무에서는 도메인에 맞게 프롬프트를 세밀하게 조정하는 것이 핵심입니다. 그 다음으로, RetrievalQA.from_chain_type()이 실행되면 내부적으로 복잡한 체인이 구성됩니다. chain_type="stuff"는 검색된 모든 문서(k=3개)를 하나의 문자열로 연결하여 프롬프트의 {context}에 넣는 방식입니다. 다른 옵션으로는 "map_reduce"(각 문서를 개별 처리 후 결합), "refine"(순차적으로 답변 개선), "map_rerank"(여러 답변 중 최고 점수 선택)가 있습니다. "stuff"가 가장 간단하고 빠르지만, 문서가 많으면 토큰 제한을 넘을 수 있습니다. qa_chain({"query": query})가 실행되면 다음 순서로 처리됩니다: (1) query를 벡터로 변환, (2) 벡터 저장소에서 상위 k=3개 문서 검색, (3) 검색된 문서들을 context로, 질문을 question으로 프롬프트에 삽입, (4) 완성된 프롬프트를 LLM에 전달, (5) LLM의 응답을 result['result']에, 검색된 문서를 result['source_documents']에 저장하여 반환. 여러분이 이 코드를 사용하면 프로덕션 레벨의 질의응답 시스템을 10분 만에 구축할 수 있습니다. 실무에서의 이점은 (1) 전체 파이프라인이 하나의 객체로 관리되어 유지보수 용이, (2) 프롬프트만 변경하면 답변 스타일 커스터마이징 가능, (3) 출처 제공으로 사용자 신뢰도 향상, (4) 에러 핸들링과 재시도 로직이 내장되어 안정성 높음입니다. ### 실전 팁 💡 문서가 많을 때는 chain_type="map_reduce"를 사용하세요. 각 문서를 병렬로 요약한 후 최종 답변을 생성하므로, "stuff"의 토큰 제한 문제를 피할 수 있습니다. 다만 LLM 호출 횟수가 늘어 비용은 증가합니다. 💡 프롬프트에 "답변은 한국어로 작성하세요"를 추가하세요. LLM이 영어 문서를 참조할 때 영어로 답변하는 것을 방지합니다. 다국어 환경에서는 필수입니다. 💡 verbose=True를 RetrievalQA에 추가하면 내부 동작(검색된 문서, 생성된 프롬프트 등)을 볼 수 있습니다. 디버깅이나 프롬프트 튜닝 시 매우 유용합니다. 💡 ConversationalRetrievalChain을 사용하면 대화 히스토리를 유지할 수 있습니다. "그것의 장점은?"처럼 이전 대화를 참조하는 질문도 처리 가능합니다. 챗봇 구축 시 필수입니다. 💡 답변 품질이 낮으면 search_kwargs={"k": 5}로 더 많은 문서를 검색하거나, retriever에 score_threshold를 설정해 유사도가 낮은 문서는 제외하세요. 예: search_kwargs={"k": 5, "score_threshold": 0.7}. --- ## 7. ConversationalRetrievalChain - 대화 맥락을 유지하는 RAG ### 시작하며 여러분이 RAG 챗봇을 만들었는데, "LangChain이 뭐야?" → "그것의 장점은?" → "어디서 사용돼?"처럼 연속된 질문에서 "그것"이 뭔지 몰라서 엉뚱한 답변을 하는 경우를 경험한 적 있나요? 각 질문을 독립적으로 처리하면 대화의 흐름이 끊겨서 사용자 경험이 나빠집니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. 특히 고객 지원 챗봇이나 교육용 튜터링 시스템처럼 여러 턴의 대화가 필요한 경우, 이전 대화 맥락을 기억하지 못하면 매번 질문을 완전한 문장으로 다시 작성해야 하는 불편함이 생깁니다. 바로 이럴 때 필요한 것이 ConversationalRetrievalChain입니다. 대화 히스토리를 관리하고, 현재 질문을 이전 맥락과 결합하여 재구성한 후 검색하므로, 자연스러운 대화형 RAG 시스템을 만들 수 있습니다. ### 개요 간단히 말해서, ConversationalRetrievalChain은 대화 히스토리를 유지하면서 RAG 검색을 수행하는 상태 유지형(stateful) 질의응답 체인입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 실제 사용자는 완전한 문장으로 질문하지 않고, "그거", "저번에 말한", "더 자세히"처럼 대명사와 생략을 많이 사용하기 때문입니다. 예를 들어, 기술 지원 챗봇에서 "Lambda 함수 만드는 방법" → "비용은?" → "예제 보여줘"처럼 연속 질문이 일반적인데, 맥락 없이는 "비용은?"이 무엇의 비용인지 알 수 없습니다. 기존에는 각 질문마다 전체 대화를 수동으로 포함시키거나, 사용자에게 매번 완전한 문장을 요구했다면, 이제는 체인이 자동으로 대화를 추적하고, 질문을 재구성하며, 관련 문서를 검색합니다. ConversationalRetrievalChain의 핵심 특징은 세 가지입니다: (1) 자동 대화 히스토리 관리(메모리 통합), (2) 질문 재구성(현재 질문 + 이전 맥락 → 독립적인 질문), (3) 스트리밍 지원(답변이 생성되는 대로 실시간 표시). 이러한 특징들이 중요한 이유는 ChatGPT처럼 자연스러운 대화형 인터페이스를 제공할 수 있기 때문입니다. ### 코드 예제 ```python from langchain.chains import ConversationalRetrievalChain from langchain.memory import ConversationBufferMemory from langchain.llms import OpenAI from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings # 벡터 저장소 설정 embeddings = OpenAIEmbeddings() vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) # 메모리 초기화: 대화 히스토리를 저장합니다 memory = ConversationBufferMemory( memory_key="chat_history", # 프롬프트에서 사용할 키 return_messages=True, # 메시지 객체로 반환 output_key="answer" # 답변을 저장할 키 ) # LLM 초기화 llm = OpenAI(temperature=0) # ConversationalRetrievalChain 생성 qa_chain = ConversationalRetrievalChain.from_llm( llm=llm, retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), memory=memory, return_source_documents=True, verbose=True # 내부 동작 확인 ) # 대화 시뮬레이션: 연속된 질문들 queries = [ "LangChain에서 체인이란 무엇인가요?", "그것의 주요 장점은 뭔가요?", # "그것" = 체인 "실제 예제를 보여주세요" # 앞의 맥락 유지 ] for query in queries: print(f"\n{'='*60}") print(f"질문: {query}") result = qa_chain({"question": query}) print(f"답변: {result['answer']}") # 대화 히스토리는 자동으로 memory에 저장됨 # 대화 히스토리 확인 print(f"\n{'='*60}") print("전체 대화 히스토리:") print(memory.load_memory_variables({})) ``` ### 설명 이것이 하는 일: 이 코드는 여러 턴의 대화를 기억하면서, 각 질문을 이전 맥락과 결합하여 완전한 질문으로 재구성하고, 그에 맞는 문서를 검색하여 답변하는 대화형 RAG 시스템을 구축합니다. 첫 번째로, ConversationBufferMemory 초기화 부분은 대화 저장소를 만듭니다. memory_key="chat_history"는 프롬프트 템플릿에서 대화 내역을 참조할 때 사용하는 변수명이고, return_messages=True는 대화를 문자열이 아닌 메시지 객체(HumanMessage, AIMessage)로 저장합니다. output_key="answer"는 체인의 출력 중 어느 부분을 메모리에 저장할지 지정합니다. 이렇게 하는 이유는 대화 히스토리를 구조화된 형태로 유지하여, 나중에 요약하거나 검색할 때 효율적이기 때문입니다. 그 다음으로, ConversationalRetrievalChain.from_llm()이 실행되면 내부적으로 두 개의 서브 체인이 생성됩니다. 첫 번째는 "Question Generator Chain"으로, 현재 질문과 chat_history를 받아서 독립적인 질문(standalone question)을 생성합니다. 예를 들어, 질문이 "그것의 장점은?"이고 히스토리에 "LangChain 체인"이 있으면, "LangChain 체인의 장점은?"으로 재구성합니다. 두 번째는 "Document Chain"으로, 재구성된 질문으로 문서를 검색하고 답변을 생성합니다. qa_chain({"question": query})가 실행되면 다음 순서로 처리됩니다: (1) memory에서 chat_history 로드, (2) 현재 질문 + chat_history → LLM → 독립적인 질문 생성, (3) 독립적인 질문으로 벡터 검색, (4) 검색된 문서 + 원래 질문(맥락 포함) → LLM → 답변 생성, (5) (질문, 답변) 쌍을 memory에 저장. 중요한 점은 검색은 재구성된 질문으로 하지만, 답변 생성은 원래 질문으로 한다는 것입니다. 이렇게 하면 검색 정확도는 높이고, 답변은 자연스럽게 유지할 수 있습니다. 여러분이 이 코드를 사용하면 ChatGPT 같은 자연스러운 대화 경험을 제공할 수 있습니다. 실무에서의 이점은 (1) 사용자가 짧은 질문으로도 소통 가능, (2) 대화 흐름이 자연스러워 사용자 만족도 향상, (3) 메모리 관리가 자동화되어 개발 간편, (4) 대화 히스토리를 분석하여 사용자 의도 파악 가능입니다. 💡 메모리 타입을 용도에 맞게 선택하세요. ConversationBufferMemory(전체 저장), ConversationSummaryMemory(요약 저장), ConversationBufferWindowMemory(최근 k개만 저장) 등이 있습니다. 긴 대화는 SummaryMemory로 토큰 절약 가능합니다. 💡 condense_question_prompt를 커스터마이징하면 질문 재구성 방식을 제어할 수 있습니다. 예: "이전 대화를 고려하여, 다음 질문을 독립적인 질문으로 다시 작성하세요"처럼 명확한 지시를 추가하세요. 💡 max_tokens_limit을 설정해 대화 히스토리가 너무 길어지는 것을 방지하세요. 예: ConversationSummaryMemory(llm=llm, max_token_limit=500)로 500 토큰 초과 시 자동 요약합니다. 💡 get_chat_history 함수를 전달하면 대화 포맷을 커스터마이징할 수 있습니다. 예: 타임스탬프 추가, 사용자 이름 포함 등. chain = ConversationalRetrievalChain.from_llm(..., get_chat_history=lambda h: ...) 💡 스트리밍을 활성화하려면 llm에 streaming=True, callbacks=[StreamingStdOutCallbackHandler()]를 추가하세요. 답변이 생성되는 대로 실시간 출력되어 UX가 크게 향상됩니다. --- ## 8. Custom Prompts - 프롬프트 최적화 전략 ### 시작하며 여러분이 RAG 시스템을 구축했는데, 답변이 너무 길거나, 형식이 일관되지 않거나, 불필요한 정보가 포함되는 경우를 경험한 적 있나요? 기본 프롬프트는 범용적이라서 여러분의 특정 도메인이나 사용 사례에 최적화되어 있지 않습니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. 법률 문서 검색 시스템에서는 정확한 조항 인용이 필요하고, 의료 Q&A에서는 면책 조항이 필수이며, 고객 지원에서는 친근한 톤이 중요합니다. 하나의 프롬프트로 모든 요구사항을 만족시킬 수 없습니다. 바로 이럴 때 필요한 것이 Custom Prompts입니다. 도메인 지식, 답변 형식, 톤앤매너, 제약사항 등을 프롬프트에 명시하여 RAG 시스템의 출력을 정밀하게 제어할 수 있습니다. ### 개요 간단히 말해서, Custom Prompts는 LangChain의 기본 프롬프트를 여러분의 요구사항에 맞게 커스터마이징하여 답변 품질과 형식을 제어하는 기법입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 프롬프트가 LLM 출력의 90%를 결정하기 때문입니다. 같은 문서를 검색해도 프롬프트에 따라 "간단히 말하면..."처럼 시작하거나, "1. 2. 3."처럼 번호 매기거나, "출처: [문서명]"을 포함하는 등 완전히 다른 결과를 얻을 수 있습니다. 예를 들어, 기술 지원에서는 단계별 가이드가 필요하고, 영업에서는 설득력 있는 답변이 필요합니다. 기존에는 LangChain의 기본 프롬프트를 그대로 사용하거나, 전체 체인을 다시 작성해야 했다면, 이제는 PromptTemplate만 교체하면 됩니다. Custom Prompts의 핵심 특징은 세 가지입니다: (1) 변수 기반 템플릿(동적 콘텐츠 삽입), (2) Few-shot 예제 포함(원하는 형식 학습), (3) 조건부 로직(입력에 따라 프롬프트 변경). 이러한 특징들이 중요한 이유는 동일한 RAG 시스템을 다양한 상황에 맞게 조정할 수 있기 때문입니다. ### 코드 예제 ```python from langchain.prompts import PromptTemplate from langchain.chains import RetrievalQA from langchain.llms import OpenAI from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings # 벡터 저장소 설정 embeddings = OpenAIEmbeddings() vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) # 도메인 특화 프롬프트: 기술 지원용 tech_support_template = """당신은 친절한 기술 지원 전문가입니다. 아래 문서를 참고하여 사용자의 질문에 단계별로 답변하세요. 중요 규칙: ``` --- ## 4. 답을 모르면 "해당 정보를 찾을 수 없습니다. 추가 문의는 support@example.com으로 연락주세요"라고 하세요 ### 설명 이것이 하는 일: 이 코드는 기술 지원과 법률 문서라는 서로 다른 도메인에 최적화된 프롬프트 템플릿을 정의하고, 선택한 템플릿에 따라 동일한 검색 결과도 완전히 다른 형식의 답변을 생성합니다. 첫 번째로, tech_support_template 정의 부분은 LLM의 역할, 답변 형식, 제약사항을 명시합니다. "당신은 친절한 기술 지원 전문가입니다"는 톤을 설정하고, "번호 매긴 단계로 구성"은 형식을 지정하며, "기술 용어는 쉽게 풀어서"는 대상 독자를 고려합니다. 이렇게 하는 이유는 LLM이 모호한 지시보다 명확하고 구체적인 지시를 훨씬 잘 따르기 때문입니다. "답을 모르면..." 부분은 할루시네이션을 방지하는 안전장치입니다. 그 다음으로, legal_template의 Few-shot 예제 부분은 LLM에게 원하는 답변 형식을 직접 보여줍니다. "예시:"로 실제 질문과 이상적인 답변을 제공하면, LLM은 그 패턴을 학습하여 비슷한 형식으로 답변합니다. 내부적으로는 In-Context Learning이 작동하여, 몇 개의 예제만으로도 복잡한 형식을 따를 수 있습니다. 법률 문서에서는 "[출처: 제12조 3항]" 같은 정확한 인용이 필수이므로, 예제에 포함시킵니다. PromptTemplate 객체가 생성되면 {context}와 {question}이라는 두 개의 플레이스홀더를 가진 템플릿이 만들어집니다. 나중에 체인이 실행될 때, {context}는 검색된 문서들로, {question}은 사용자 질문으로 자동으로 채워집니다. chain_type_kwargs={"prompt": PROMPT}로 전달하면 기본 프롬프트 대신 이 커스텀 프롬프트가 사용됩니다. 여러분이 이 코드를 사용하면 동일한 RAG 시스템을 다양한 도메인에 맞게 조정할 수 있습니다. 실무에서의 이점은 (1) 프롬프트만 바꾸면 되므로 유연성 극대화, (2) 도메인 전문성을 코드가 아닌 프롬프트로 표현하여 비개발자도 수정 가능, (3) A/B 테스트로 최적 프롬프트 발견 가능, (4) 일관된 답변 형식으로 사용자 경험 향상입니다. ### 실전 팁 💡 프롬프트에 "답변은 한국어로 작성하세요"를 명시하세요. 영어 문서를 참조할 때 LLM이 영어로 답변하는 것을 방지합니다. 특히 다국어 환경에서 필수입니다. 💡 Few-shot 예제는 2-3개가 적당합니다. 너무 많으면 토큰을 낭비하고, 너무 적으면 패턴 학습이 안 됩니다. 가장 대표적이고 다양한 예제를 선택하세요. 💡 출력 형식을 구조화하려면 "JSON 형식으로 답변하세요: {\"answer\": \"...\", \"confidence\": 0.9, \"sources\": [...]}"처럼 명시하세요. 파싱이 쉬워져 후처리가 간편합니다. 💡 프롬프트 버전 관리를 하세요. Git에 프롬프트 파일을 저장하고, 변경 이력을 추적하면 성능 회귀를 방지하고, 최적 버전을 찾을 수 있습니다. 💡 LangSmith나 Weights & Biases로 프롬프트 성능을 측정하세요. 여러 프롬프트를 테스트 세트에 실행하고, 정확도/관련성/형식 준수율을 비교하여 데이터 기반으로 선택합니다. --- ## 9. Error Handling과 Fallbacks - 안정성 확보하기 ### 시작하며 여러분이 RAG 시스템을 프로덕션에 배포했는데, 간혹 OpenAI API가 타임아웃되거나, 검색 결과가 없거나, LLM이 이상한 답변을 하는 경우를 경험한 적 있나요? 에러 처리 없이는 시스템이 멈추거나, 사용자에게 에러 메시지가 그대로 노출됩니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. API 호출 실패, 네트워크 오류, 토큰 제한 초과, 빈 검색 결과 등 다양한 엣지 케이스가 있는데, 이를 처리하지 않으면 사용자 경험이 나빠지고, 시스템 신뢰도가 떨어집니다. 바로 이럴 때 필요한 것이 Error Handling과 Fallbacks입니다. 예상 가능한 오류를 사전에 처리하고, 주 시스템 실패 시 백업 방안을 제공하여 안정적인 서비스를 유지할 수 있습니다. ### 개요 간단히 말해서, Error Handling은 예외 상황을 감지하고 적절히 대응하는 것이고, Fallbacks은 주 시스템 실패 시 대체 방안을 자동으로 실행하는 메커니즘입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 외부 API(OpenAI, Pinecone 등)에 의존하는 시스템은 네트워크 불안정, 서비스 장애, 요금 제한 등의 이유로 실패할 수 있기 때문입니다. 예를 들어, OpenAI API가 과부하로 응답이 10초 이상 걸리면 사용자는 떠나버립니다. 또한 "검색 결과 없음" 같은 정상적인 케이스도 우아하게 처리해야 합니다. 기존에는 try-except로 개별 에러를 잡고, 수동으로 재시도 로직을 작성했다면, 이제는 LangChain의 built-in retry, fallback, timeout 설정과 validation 체인을 활용할 수 있습니다. Error Handling의 핵심 특징은 세 가지입니다: (1) 자동 재시도(일시적 오류 복구), (2) Fallback 체인(대체 LLM이나 하드코딩된 응답), (3) Graceful degradation(부분 실패 시에도 최선의 답변). 이러한 특징들이 중요한 이유는 99.9% 가용성을 달성하고, 사용자에게 항상 응답을 제공하기 위함입니다. ### 코드 예제 ```python from langchain.chains import RetrievalQA from langchain.llms import OpenAI from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.callbacks import get_openai_callback import time # 벡터 저장소 설정 embeddings = OpenAIEmbeddings() vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embeddings) # Fallback LLM: 주 모델 실패 시 사용할 간단한 모델 primary_llm = OpenAI(model_name="gpt-4", temperature=0, request_timeout=10) fallback_llm = OpenAI(model_name="gpt-3.5-turbo", temperature=0, request_timeout=5) def create_qa_chain_with_retry(llm, max_retries=3): """재시도 로직을 포함한 QA 체인 생성""" qa_chain = RetrievalQA.from_chain_type( llm=llm, retriever=vectorstore.as_retriever(search_kwargs={"k": 3}), return_source_documents=True ) return qa_chain def query_with_fallback(query, verbose=True): """Fallback과 에러 처리를 포함한 안전한 쿼리 실행""" # 1차 시도: Primary LLM for attempt in range(3): try: if verbose: print(f"[시도 {attempt+1}] Primary LLM (GPT-4) 사용...") qa_chain = create_qa_chain_with_retry(primary_llm) with get_openai_callback() as cb: result = qa_chain({"query": query}) if verbose: print(f"성공! 토큰 사용: {cb.total_tokens}, 비용: ${cb.total_cost:.4f}") # 검색 결과 검증 if not result.get('source_documents'): return {"answer": "관련 정보를 찾을 수 없습니다. 질문을 다시 작성해주세요.", "source": "no_results"} return {"answer": result['result'], "sources": result['source_documents']} except Exception as e: if verbose: print(f"[오류] {type(e).__name__}: {str(e)}") if attempt < 2: # 마지막 시도가 아니면 재시도 time.sleep(2 ** attempt) # Exponential backoff continue else: break # 3회 실패 시 Fallback으로 # 2차 시도: Fallback LLM try: if verbose: print("[Fallback] GPT-3.5-turbo 사용...") qa_chain = create_qa_chain_with_retry(fallback_llm) result = qa_chain({"query": query}) return {"answer": result['result'], "sources": result.get('source_documents', []), "fallback": True} except Exception as e: # 최종 실패: 하드코딩된 응답 if verbose: print(f"[최종 실패] {type(e).__name__}: {str(e)}") return { "answer": "죄송합니다. 일시적인 오류로 답변을 생성할 수 없습니다. 잠시 후 다시 시도해주세요.", "error": str(e), "fallback": "hard_coded" } # 실행 예시 query = "LangChain의 핵심 개념은?" response = query_with_fallback(query, verbose=True) print(f"\n최종 답변: {response['answer']}") if response.get('fallback'): print(f"[경고] Fallback 사용됨: {response.get('fallback')}") ``` ### 설명 이것이 하는 일: 이 코드는 GPT-4로 답변을 시도하되, 실패 시 자동으로 재시도하고, 3회 실패하면 GPT-3.5로 전환하며, 그마저 실패하면 사용자에게 정중한 오류 메시지를 반환하는 다층 방어 시스템을 구축합니다. 첫 번째로, create_qa_chain_with_retry 함수는 재시도가 가능한 QA 체인을 생성합니다. request_timeout=10은 API 호출이 10초를 넘으면 자동으로 취소하여 무한 대기를 방지합니다. 이렇게 하는 이유는 네트워크가 느린 환경에서도 시스템이 멈추지 않게 하기 위함입니다. LangChain은 내부적으로 tenacity 라이브러리를 사용해 일부 오류(RateLimitError 등)를 자동으로 재시도하지만, 명시적으로 제어하는 것이 더 안전합니다. 그 다음으로, query_with_fallback의 재시도 루프가 실행됩니다. for attempt in range(3)으로 최대 3회 시도하고, 실패할 때마다 time.sleep(2 ** attempt)로 대기 시간을 지수적으로 늘립니다(2초, 4초, 8초). 이것을 Exponential Backoff라고 하며, 서버 과부하 상황에서 재시도 폭주를 방지합니다. get_openai_callback()은 각 호출의 토큰 사용량과 비용을 추적하여 로깅이나 빌링에 사용할 수 있습니다. 검색 결과 검증 부분(if not result.get('source_documents'))은 기술적 오류가 아닌 논리적 실패를 처리합니다. 벡터 스토어에서 유사한 문서를 찾지 못한 경우 "관련 정보를 찾을 수 없습니다"라고 정직하게 답변하여 할루시네이션을 방지합니다. Fallback LLM 부분은 primary_llm이 3회 모두 실패했을 때만 실행됩니다. GPT-3.5는 GPT-4보다 10배 빠르고 20배 저렴하지만 품질은 약간 낮습니다. 프로덕션에서는 99%는 GPT-4로, 1% 실패 케이스만 GPT-3.5로 처리하여 비용과 품질의 균형을 맞춥니다. 여러분이 이 코드를 사용하면 프로덕션 수준의 안정성을 확보할 수 있습니다. 실무에서의 이점은 (1) 일시적 네트워크 오류에도 서비스 중단 없음, (2) 비용 최적화(실패 시에만 저렴한 모델 사용), (3) 사용자에게 항상 응답 제공(빈 화면 없음), (4) 오류 로그로 시스템 모니터링 가능입니다. ### 실전 팁 💡 Exponential backoff의 최대 대기 시간을 설정하세요. time.sleep(min(2 ** attempt, 30))으로 최대 30초로 제한하여 무한정 대기를 방지합니다. 💡 LangSmith나 Sentry를 통합하여 오류를 실시간 모니터링하세요. 예외 발생 시 자동으로 알림을 받고, 오류 패턴을 분석하여 근본 원인을 해결할 수 있습니다. 💡 Circuit Breaker 패턴을 구현하세요. 연속 5회 실패 시 1분간 주 시스템을 차단하고 Fallback만 사용하여, 다운된 서비스에 계속 요청하는 것을 방지합니다. 💡 검색 결과가 없을 때는 "더 구체적인 질문을 해주세요" 대신, 유사한 질문 예시를 제공하세요. 예: "다음 중 하나를 찾으시나요? 1) 설치 방법 2) 사용 예제 3) 문제 해결" 💡 LLM 응답 검증 체인을 추가하세요. 답변에 욕설, 민감 정보, 할루시네이션이 포함되었는지 검사하고, 문제가 있으면 "답변을 생성할 수 없습니다"로 대체합니다. --- ## 10. Performance Optimization - 속도와 비용 최적화 ### 시작하며 여러분이 RAG 시스템을 배포했는데, 사용자가 늘어나면서 응답 시간이 5초에서 20초로 늘어나거나, OpenAI API 요금이 월 수백 달러로 급증하는 경험을 한 적 있나요? 프로토타입에서는 괜찮았지만, 프로덕션에서는 성능과 비용이 치명적입니다. 이런 문제는 실제 개발 현장에서 자주 발생합니다. 매번 문서를 다시 벡터화하거나, 불필요하게 큰 청크를 사용하거나, 캐싱을 활용하지 않으면 리소스가 낭비됩니다. 특히 트래픽이 많은 서비스에서는 1초 단축이 수만 달러의 비용 절감으로 이어집니다. 바로 이럴 때 필요한 것이 Performance Optimization입니다. 캐싱, 배치 처리, 모델 선택, 청크 크기 조정 등의 기법으로 속도를 10배 높이고 비용을 80% 절감할 수 있습니다. ### 개요 간단히 말해서, Performance Optimization은 RAG 시스템의 응답 속도를 높이고, API 호출 비용을 줄이며, 리소스 사용을 최적화하는 일련의 기법들입니다. 왜 이 개념이 필요한지 실무 관점에서 설명하자면, 사용자는 3초 이내 응답을 기대하고, 느리면 떠나버리기 때문입니다. 또한 OpenAI API는 토큰당 과금되므로, 불필요한 토큰 사용은 직접적인 비용 증가로 이어집니다. 예를 들어, 하루 1만 건의 쿼리를 처리하는 서비스에서 쿼리당 평균 토큰을 2000에서 1000으로 줄이면 월 비용이 $3000에서 $1500로 50% 절감됩니다. 기존에는 매번 API를 호출하고, 모든 문서를 다시 처리하며, 단일 프로세스로 순차 처리했다면, 이제는 Redis 캐싱, 벡터 저장소 재사용, 비동기 배치 처리, 모델 다운사이징 등을 활용할 수 있습니다. Performance Optimization의 핵심 특징은 세 가지입니다: (1) 응답 캐싱(동일 질문 재사용), (2) 배치 처리(여러 요청 동시 처리), (3) 비용 최적화(작은 모델, 짧은 청크). 이러한 특징들이 중요한 이유는 확장 가능한(scalable) 시스템을 만들고, 운영 비용을 관리 가능한 수준으로 유지하기 위함입니다. ### 코드 예제 ```python from langchain.cache import RedisCache from langchain.globals import set_llm_cache from langchain.llms import OpenAI from langchain.embeddings.cache import CacheBackedEmbeddings from langchain.storage import LocalFileStore from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.chains import RetrievalQA import redis # 1. LLM 응답 캐싱: 동일한 질문은 캐시에서 가져옵니다 redis_client = redis.Redis(host='localhost', port=6379, db=0) set_llm_cache(RedisCache(redis_client)) # 2. 임베딩 캐싱: 동일한 텍스트는 다시 벡터화하지 않습니다 underlying_embeddings = OpenAIEmbeddings() store = LocalFileStore("./embedding_cache/") cached_embeddings = CacheBackedEmbeddings.from_bytes_store( underlying_embeddings, store, namespace=underlying_embeddings.model ) # 3. 벡터 저장소 재사용: 이미 구축된 인덱스 로드 vectorstore = Chroma( persist_directory="./chroma_db", embedding_function=cached_embeddings ) # 4. 비용 최적화: 더 저렴한 모델 사용 llm = OpenAI( model_name="gpt-3.5-turbo", # GPT-4 대신 3.5 사용 (20배 저렴) temperature=0, max_tokens=300 # 답변 길이 제한으로 토큰 절약 ) # 5. 검색 최적화: 필요한 만큼만 검색 optimized_retriever = vectorstore.as_retriever( search_type="similarity_score_threshold", # 유사도 임계값 이하 제외 search_kwargs={ "k": 3, # 최소한의 문서만 검색 "score_threshold": 0.7 # 70% 이하 유사도는 제외 } ) # QA 체인 생성 qa_chain = RetrievalQA.from_chain_type( llm=llm, retriever=optimized_retriever, return_source_documents=True ) # 배치 처리 예시: 여러 질문을 동시에 처리 queries = [ "LangChain이란?", "RAG 시스템의 장점은?", "벡터 저장소 선택 기준은?" ] import asyncio from concurrent.futures import ThreadPoolExecutor def process_query(query): """단일 쿼리 처리""" return qa_chain({"query": query}) # 병렬 처리: 3배 빠른 처리 with ThreadPoolExecutor(max_workers=3) as executor: results = list(executor.map(process_query, queries)) for i, result in enumerate(results): print(f"\n질문 {i+1}: {queries[i]}") print(f"답변: {result['result'][:100]}...") ``` ### 설명 이것이 하는 일: 이 코드는 Redis와 파일 시스템에 캐시를 구축하여 중복 API 호출을 제거하고, 저렴한 모델과 토큰 제한으로 비용을 줄이며, 병렬 처리로 처리량을 3배 높이는 종합 최적화 시스템을 구축합니다. 첫 번째로, RedisCache 설정은 LLM의 모든 응답을 Redis에 저장합니다. 동일한 프롬프트가 다시 들어오면 OpenAI API를 호출하지 않고 Redis에서 즉시 반환합니다. 예를 들어, "LangChain이란?"이라는 질문이 하루에 100번 들어오면, 첫 번째만 API를 호출하고 나머지 99번은 캐시에서 가져와 비용을 99% 절감합니다. 이렇게 하는 이유는 FAQ나 인기 있는 질문은 답변이 거의 바뀌지 않기 때문입니다. 그 다음으로, CacheBackedEmbeddings는 텍스트를 벡터로 변환할 때 캐싱합니다. "LangChain 설명"이라는 텍스트를 한 번 벡터화하면, 다음에는 API 호출 없이 파일에서 로드합니다. 내부적으로는 텍스트의 해시값을 키로 사용하여 ./embedding_cache/ 폴더에 pickle 파일로 저장합니다. 문서를 추가할 때 이미 벡터화된 부분은 재사용하므로, 대규모 업데이트도 빠릅니다. similarity_score_threshold 설정은 유사도가 0.7(70%) 미만인 문서는 아예 반환하지 않습니다. 이렇게 하면 관련 없는 문서를 LLM에 전달하지 않아 (1) 토큰 사용 감소, (2) 답변 품질 향상(노이즈 제거), (3) 처리 속도 향상의 효과를 얻습니다. 일반적으로 0.7-0.8이 적절하며, 도메인에 따라 조정합니다. ThreadPoolExecutor는 여러 쿼리를 동시에 처리합니다. max_workers=3이면 3개의 쿼리를 병렬로 실행하여, 순차 처리 대비 3배 빠릅니다. 주의할 점은 OpenAI API의 rate limit(분당 요청 수 제한)을 고려해야 한다는 것입니다. 너무 많은 worker는 RateLimitError를 유발할 수 있으므로, 실험을 통해 최적값을 찾으세요. 여러분이 이 코드를 사용하면 대규모 트래픽에도 안정적인 서비스를 제공할 수 있습니다. 실무에서의 이점은 (1) 응답 시간 80% 단축(캐시 히트 시), (2) API 비용 50-90% 절감(중복 호출 제거), (3) 처리량 3배 증가(병렬 처리), (4) 확장성 확보(Redis 클러스터로 수평 확장 가능)입니다. ### 실전 팁 💡 캐시 만료 시간(TTL)을 설정하세요. RedisCache에 ttl=3600 (1시간)을 추가하면, 오래된 정보가 계속 제공되는 것을 방지합니다. 실시간성이 중요한 뉴스는 짧게, 정적인 문서는 길게 설정합니다. 💡 Semantic Cache를 사용하면 비슷한 질문도 캐시에서 가져옵니다. "LangChain이란?"과 "LangChain 설명해줘"를 같은 질문으로 인식하여 캐시 히트율을 높입니다. GPTCache 라이브러리 참고하세요. 💡 streaming=True로 답변을 스트리밍하세요. 전체 답변을 기다리지 않고 생성되는 대로 표시하여 체감 속도를 크게 높입니다. 사용자는 3초 대기보다 즉시 시작하는 1초짜리 스트림을 선호합니다. 💡 청크 크기를 최적화하세요. 너무 크면 검색 정확도 저하, 너무 작으면 LLM 호출 증가입니다. A/B 테스트로 chunk_size=500-1000 범위에서 최적값을 찾으세요. 💡 비동기 처리를 도입하세요. asyncio와 aiohttp로 I/O 바운드 작업(API 호출, DB 쿼리)을 비동기화하면 동일한 리소스로 10배 많은 요청을 처리할 수 있습니다. LangChain도 async 메서드를 지원합니다. ---

카테고리: AI

언어: Python

태그: Python, LangChain, RAG, VectorStore, Embeddings, AI

작성자: AI Generated

프리미엄 콘텐츠 - 3개월 무료 체험 가능

RAG 시스템 6편 - LangChain으로 RAG 구현 완벽 가이드

목차

1. LangChain RAG 시스템 기본 구조 - 전체 아키텍처 이해하기

시작하며

개요

코드 예제

설명

실전 팁

2. Document Loaders - 다양한 파일 형식 처리하기

시작하며

개요

코드 예제

설명

실전 팁

3. Text Splitters - 문서를 효율적으로 청크화하기

시작하며

개요

코드 예제

설명

실전 팁

4. Vector Stores - 벡터 데이터베이스 구축과 활용

시작하며

개요

코드 예제

설명

실전 팁

5. Retrieval 전략 - 검색 품질 최적화하기

시작하며

개요

코드 예제

설명

실전 팁

6. RetrievalQA Chain - 질의응답 체인 구성하기

시작하며

개요

코드 예제

설명

실전 팁

7. ConversationalRetrievalChain - 대화 맥락을 유지하는 RAG

시작하며

개요

코드 예제

설명

8. Custom Prompts - 프롬프트 최적화 전략

시작하며

개요

코드 예제

4. 답을 모르면 "해당 정보를 찾을 수 없습니다. 추가 문의는 support@example.com으로 연락주세요"라고 하세요

설명

실전 팁

9. Error Handling과 Fallbacks - 안정성 확보하기

시작하며

개요

코드 예제

설명

실전 팁

10. Performance Optimization - 속도와 비용 최적화

시작하며

개요

코드 예제

설명

실전 팁

이 카드뉴스가 포함된 코스

AI RAG 시스템 개발

댓글 (0)

함께 보면 좋은 카드 뉴스

VPC 네트워크의 기초 - CIDR과 서브넷 설계 완벽 가이드

AWS 리소스 정리와 비용 관리 완벽 가이드

AWS 고가용성과 내결함성 아키텍처 설계 기초

이스티오 기반 마이크로서비스 플랫폼 완벽 가이드

오토스케일링 완벽 가이드