6. LangChain 검증, 고도화, 프로덕션 가이드
1. 평가(Evaluation) 체계
1.1. 평가가 필요한 이유
- RAG 시스템은 여러 컴포넌트(문서 로딩, 청킹, 임베딩, 검색, 생성)가 순차적으로 동작하는 파이프라인이다. 각 단계에서 품질 손실이 발생할 수 있으며, 이러한 손실은 최종 답변 품질에 누적적으로 영향을 미친다. 따라서 체계적인 평가(Evaluation) 없이는 어떤 단계에서 문제가 발생하는지 파악하기 어렵고, 개선 방향을 잡기도 힘들다.
- 평가는 단순히 "답변이 맞는가?"를 확인하는 것이 아니라, 파이프라인의 각 단계별 품질을 정량적으로 측정하여 병목 지점을 식별하고 개선 효과를 수치로 확인하는 과정이다.
[RAG 파이프라인 품질 손실 지점]
문서 로딩 → [품질 손실 1: 불완전한 파싱]
↓ 예: PDF 테이블 깨짐, 특수문자 누락, 인코딩 오류
청킹 → [품질 손실 2: 문맥 단절]
↓ 예: 관련 내용이 서로 다른 청크로 분리, 중요 정보 잘림
임베딩 → [품질 손실 3: 의미 손실]
↓ 예: 도메인 용어의 의미를 정확히 포착하지 못함
검색 → [품질 손실 4: 관련 없는 문서 검색]
↓ 예: 유사한 키워드지만 다른 맥락의 문서 반환
생성 → [품질 손실 5: 환각, 부정확한 답변]
예: 검색된 문서에 없는 내용을 생성, 문서 내용 왜곡
1.1.1. 평가를 하지 않으면 발생하는 문제
| 문제 | 설명 | 결과 |
| 병목 지점 미식별 | 어떤 단계에서 품질이 떨어지는지 모름 | 비효율적 개선 시도 |
| 개선 효과 측정 불가 | 변경 전후 비교 기준이 없음 | 감에 의존한 판단 |
| 회귀(Regression) 감지 불가 | 새 설정이 기존보다 나빠졌는지 확인 불가 | 품질 저하 방치 |
| 프로덕션 장애 예방 불가 | 배포 전 품질 기준 미달 감지 불가 | 사용자에게 저품질 답변 제공 |
| 의사소통 어려움 | 품질을 수치로 보고할 수 없음 | 이해관계자 설득 실패 |
- 실무 권장: RAG 시스템 개발 초기부터 평가 체계를 구축해야 한다. 평가 없이 파이프라인을 개선하는 것은 "계기판 없이 비행기를 조종하는 것"과 같다. 최소 3가지 평가자(정확도, 환각, 유용성)를 먼저 구현하고, 이후 도메인에 맞는 평가자를 추가하는 것이 효율적이다.
1.2. 평가 대상과 메트릭
- RAG 시스템의 평가 메트릭은 크게 검색 품질 메트릭과 생성 품질 메트릭으로 나뉜다. 검색 품질은 올바른 문서를 찾았는지, 생성 품질은 찾은 문서를 기반으로 올바른 답변을 생성했는지를 평가한다.
| 평가 대상 | 메트릭 | 측정 방법 | 상세 설명 |
| 검색 품질 | Precision@K | 정답 문서 포함 여부 | 검색된 K개 문서 중 실제 관련 문서의 비율. 높을수록 노이즈가 적음 |
| 검색 품질 | Recall@K | 정답 문서 포함 여부 | 전체 관련 문서 중 검색된 비율. 높을수록 관련 문서를 빠뜨리지 않음 |
| 검색 품질 | MRR (Mean Reciprocal Rank) | 첫 번째 정답 위치 | 첫 번째 관련 문서가 검색 결과 상위에 있을수록 높은 점수 |
| 답변 정확도 | Answer Accuracy | 참조 답변과 비교 | 생성된 답변이 정답과 일치하는 정도 |
| 답변 유용성 | Helpfulness Score | LLM Judge 평가 | 사용자 관점에서 답변이 실질적으로 도움이 되는지 |
| 환각 탐지 | Hallucination Rate | 컨텍스트 기반 검증 | 컨텍스트에 없는 내용을 생성한 비율. 낮을수록 좋음 |
| 답변 충실도 | Faithfulness | 컨텍스트와 답변 일치도 | 답변의 모든 주장이 컨텍스트에 근거하는 비율 |
| 응답 시간 | Latency (ms) | 엔드투엔드 측정 | 질의 입력부터 답변 생성까지의 시간 |
| 비용 | Cost per Query | API 호출 비용 추적 | 질의 1건당 발생하는 LLM API 비용 |
1.2.1. 검색 메트릭 상세: Precision@K vs Recall@K
- 검색 평가의 핵심 메트릭인 Precision과 Recall은 서로 트레이드오프 관계에 있다.
전체 관련 문서: [A, B, C, D, E] (5개)
검색 결과 (K=4): [A, B, F, G] (4개)
Precision@4 = 관련 문서 / 검색 결과 = 2/4 = 0.50 (A, B만 관련)
Recall@4 = 검색된 관련 문서 / 전체 관련 문서 = 2/5 = 0.40 (A, B만 검색됨)
→ F, G는 관련 없는 문서 (Precision 저하)
→ C, D, E는 놓친 관련 문서 (Recall 저하)| 메트릭 | 높이려면 | 부작용 |
| Precision | K를 줄이고 임계값을 높여 정밀하게 검색 | Recall 감소 (관련 문서 놓침) |
| Recall | K를 늘리고 검색 범위를 넓힘 | Precision 감소 (노이즈 증가) |
- 실무 권장: 일반적인 RAG Q&A 시스템에서는 Precision을 우선시한다. 관련 없는 문서가 컨텍스트에 포함되면 LLM이 잘못된 답변을 생성할 가능성이 높아지기 때문이다. 반면, 법률이나 의료처럼 관련 문서를 절대 놓쳐서는 안 되는 도메인에서는 Recall을 우선시한다.
1.2.2. MRR (Mean Reciprocal Rank) 상세
- MRR은 "첫 번째 정답이 검색 결과에서 몇 번째에 위치하는가"를 측정한다. 사용자가 보통 상위 결과만 확인하므로, 관련 문서가 최상위에 있을수록 좋다.
질의 1: 첫 번째 관련 문서가 1번째 → RR = 1/1 = 1.0
질의 2: 첫 번째 관련 문서가 3번째 → RR = 1/3 = 0.33
질의 3: 첫 번째 관련 문서가 2번째 → RR = 1/2 = 0.5
MRR = (1.0 + 0.33 + 0.5) / 3 = 0.61
2. LangSmith 기반 평가 (단계 1: 평가 환경 구축)
2.1. LangSmith 개요와 설정
- LangSmith는 LangChain 팀이 만든 LLM 애플리케이션 개발/디버깅/평가/모니터링 통합 플랫폼이다. LangChain과 긴밀하게 통합되어 있어, 환경변수 설정만으로 파이프라인의 모든 단계를 자동으로 추적(trace)할 수 있다.
2.1.1. LangSmith가 제공하는 핵심 기능
| 기능 | 설명 | 사용 시점 |
| 추적(Tracing) | 모든 LLM 호출, 검색, 체인 실행을 자동 기록하여 상세한 실행 흐름과 지연 시간을 추적 | 개발/디버깅 시, 병목 지점 분석 |
| 데이터셋 관리 | 평가용 Q&A 쌍을 체계적으로 관리하고 라벨링, 분류, 버전 관리를 지원 | 평가 데이터 구축 및 업데이트 시 |
| 평가 실행 | 여러 평가자(LLM Judge, 메트릭 계산)를 병렬 실행하고 결과 비교/시각화 제공 | A/B 테스트, 설정 비교, 최적화 시 |
| 모니터링 | 프로덕션 실행의 성능/비용/오류를 실시간 대시보드로 추적하고 알림 설정 | 운영 중, SLA 준수 확인 |
| 피드백 수집 | 사용자/전문가 피드백을 자동 수집하여 평가 데이터로 전환하고 모델 개선 루프 구성 | 지속적 개선, 인간-in-the-loop 시 |
import os
# LangSmith 추적 활성화를 위한 환경변수 설정
os.environ["LANGCHAIN_TRACING_V2"] = "true" # 추적 기능 활성화 (V2는 최신 추적 프로토콜)
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-api-key" # LangSmith API 키
os.environ["LANGCHAIN_PROJECT"] = "tax-rag-evaluation" # 프로젝트명 (대시보드에서 구분용)
# os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com" # 기본값, 변경 불필요
- 파라미터 정의 기준: `LANGCHAIN_PROJECT`는 실험 목적에 따라 구분하여 설정한다. 예: `tax-rag-v1-chunk500`, `tax-rag-v2-reranking` 등으로 실험별 프로젝트를 분리하면 대시보드에서 비교가 용이하다. 하나의 프로젝트 안에서 `experiment_prefix`로 세부 실험을 구분할 수도 있다.
2.1.2. LangSmith 추적 동작 원리
[사용자 질의]
↓
[LangChain 체인 실행] ──→ [LangSmith 서버로 자동 전송]
├─ Retriever 호출 기록 ├─ 입력/출력
├─ LLM 호출 기록 ├─ 토큰 사용량
├─ 실행 시간 기록 ├─ 비용 정보
└─ 에러 기록 └─ 체인 구조 시각화
↓
[답변 반환]
- 환경변수를 설정하면, LangChain의 모든 Runnable(`ChatOpenAI`, `Retriever`, `Chain` 등)이 실행될 때마다 자동으로 LangSmith 서버에 추적 데이터를 전송한다. 별도의 코드 변경 없이 동작하는 것이 핵심 장점이다.
- 실무 권장: 개발 환경에서는 항상 `LANGCHAIN_TRACING_V2=true`를 설정하여 모든 실행을 추적한다. 프로덕션에서는 비용과 성능을 고려하여 샘플링(예: 10%만 추적)하거나, 오류 발생 시에만 추적하도록 설정할 수 있다.
2.2. 평가 데이터셋 생성
- 평가 데이터셋은 RAG 시스템의 품질을 측정하는 기준이 되는 Q&A 쌍의 모음이다. "입력(질문)"과 "기대 출력(참조 답변)"으로 구성되며, 일부 평가자는 "기대 출처 문서"도 필요로 한다.
2.2.1. 좋은 평가 데이터셋의 조건
| 조건 | 설명 | 예시 |
| 대표성 | 실제 사용자 질의 패턴을 반영하여 일반화 가능 | 단순 질의, 복합 질의, 비교 질의, 팔로업 질의 등 실제 로그 기반 |
| 다양성 | 다양한 주제, 난이도, 형식의 질문을 포함 | 쉬운 사실 질문~어려운 분석 질문, 단문/장문, 다양한 도메인(법률, 기술, 일상 등) |
| 정확성 | 참조 답변이 도메인 전문가 검증 또는 신뢰할 수 있는 출처 기반 | 전문가가 직접 작성한 정답, 공식 문서 발췌, 다중 출처 교차 검증 |
| 충분한 크기 | 통계적으로 의미 있는 샘플 수로 신뢰도 확보 | 최소 50개 이상, 이상적 100-1000개, 대규모 평가시 10K+ |
| 엣지 케이스 포함 | 시스템 한계와 예외 상황을 테스트 | 답변 불가 질문(정보 부족), 모호한 질문, 노이즈 포함 질문, 공격적/편향 질문 |
from langsmith import Client
# LangSmith 클라이언트 초기화 (환경변수에서 API 키 자동 로드)
client = Client()
# 데이터셋 생성
# - dataset_name: 대시보드에서 식별하기 위한 고유한 이름
# - description: 데이터셋의 목적과 범위를 설명
dataset = client.create_dataset(
dataset_name="tax-qa-eval-dataset",
description="세법 RAG 시스템 평가용 Q&A 데이터셋"
)
# 평가 예시 추가
# - input: RAG 시스템에 전달될 입력 (질문)
# - output: 기대되는 정답 (참조 답변) - 평가 시 비교 기준으로 사용
examples = [
{
"input": {"question": "근로소득세 세율은?"},
"output": {"answer": "근로소득세는 과세표준에 따라 6%~45%의 누진세율이 적용됩니다."}
},
{
"input": {"question": "연말정산 시 공제되는 항목은?"},
"output": {"answer": "근로소득공제, 인적공제, 특별소득공제, 세액공제 등이 있습니다."}
},
{
"input": {"question": "부가가치세 신고 기한은?"},
"output": {"answer": "일반과세자는 1월 25일(1기 확정), 7월 25일(2기 확정)까지입니다."}
}
]
# 각 예시를 데이터셋에 추가
for ex in examples:
client.create_example(
inputs=ex["input"], # 입력 데이터 (dict 형태)
outputs=ex["output"], # 기대 출력 (dict 형태)
dataset_id=dataset.id # 소속 데이터셋 ID
)
2.2.2. 평가 데이터셋 구축 전략
- 평가 데이터셋을 효과적으로 구축하기 위한 전략은 다음과 같다.
| 전략 | 설명 | 장점 | 단점 |
| 수동 구축 | 도메인 전문가가 직접 Q&A 쌍을 작성하고 검증 | 최고 수준의 정확도와 도메인 특화, 신뢰도 높음 | 시간/비용 많이 소요, 확장성 제한적 |
| LLM 기반 생성 | LLM에게 문서나 지침을 주고 대량의 Q&A 쌍 자동 생성 | 빠르고 대량 생산 가능, 초기 데이터셋 빠르게 구축 | 환각/오류 발생 가능성, 후속 검증 필수 |
| 사용자 로그 활용 | 실제 사용자 질의 로그를 수집하여 Q&A 데이터셋화 | 실제 사용 패턴 100% 반영, 대표성 우수 | 고품질 참조 답변 별도 작성 필요, 초기 데이터 부족시 어려움 |
| 혼합 방식 | LLM으로 초안 생성 후 전문가 검증 또는 사용자 피드백 루프 | 효율성과 정확도의 최적 균형, 지속적 개선 가능 | 복잡한 워크플로우 설계 및 관리 필요, 초기 설정 비용 발생 |
- 실무 권장: 초기에는 LLM으로 50~100개의 Q&A 쌍을 자동 생성한 뒤, 도메인 전문가가 검증/수정하는 혼합 방식이 가장 효율적이다. 이후 프로덕션 운영 중 수집되는 실제 사용자 질의를 추가하여 데이터셋을 지속적으로 확장한다.
2.3. RagBot 평가 클래스
- LangSmith로 평가를 수행하려면, RAG 시스템의 실행 과정을 추적 가능한 형태로 구성해야 한다. `@traceable()` 데코레이터를 사용하면 각 메서드의 입력/출력이 자동으로 LangSmith에 기록되어, 검색-컨텍스트 구성-생성의 각 단계를 개별적으로 분석할 수 있다.
from langsmith import traceable
class RagBot:
"""
평가 가능한 RAG 봇
- @traceable() 데코레이터: 해당 메서드의 실행을 LangSmith에 자동 기록
- 각 단계(검색, 컨텍스트 구성, 생성)가 개별 추적되어 병목 분석 가능
"""
def __init__(self, retriever, llm):
self.retriever = retriever # 벡터 DB 검색기
self.llm = llm # 답변 생성용 LLM
@traceable() # 최상위 실행으로 추적 (하위 단계의 부모 노드가 됨)
def get_answer(self, question: str) -> dict:
"""질문에 답변하고 추적 정보 포함"""
# 1단계: 관련 문서 검색
docs = self.retrieve_docs(question)
# 2단계: 검색된 문서를 하나의 컨텍스트 문자열로 결합
context = self.format_context(docs)
# 3단계: LLM으로 최종 답변 생성
answer = self.generate_answer(question, context)
# 평가에 필요한 모든 정보를 딕셔너리로 반환
return {
"answer": answer, # 생성된 답변
"contexts": [doc.page_content for doc in docs], # 검색된 문서 내용 (환각 평가에 사용)
"sources": [doc.metadata.get("source", "") for doc in docs] # 출처 정보 (검색 품질 평가에 사용)
}
@traceable() # 검색 단계 개별 추적 (실행 시간, 반환 문서 수 등)
def retrieve_docs(self, question: str):
return self.retriever.invoke(question)
@traceable() # 컨텍스트 구성 단계 추적
def format_context(self, docs):
return "\n\n".join([doc.page_content for doc in docs])
@traceable() # 생성 단계 추적 (토큰 사용량, 응답 시간 등)
def generate_answer(self, question: str, context: str) -> str:
response = self.llm.invoke(
f"컨텍스트를 기반으로 답하세요.\n컨텍스트: {context}\n질문: {question}\n답변:"
)
return response.content
- 파라미터 정의 기준: `@traceable()` 데코레이터의 선택적 파라미터로 `name`(추적 이름), `run_type`(실행 유형: chain, llm, retriever, tool 등), `tags`(필터링용 태그)를 지정할 수 있다. 예: `@traceable(name="문서_검색", run_type="retriever", tags=["v1"])`
2.3.1. @traceable()과 LangSmith 추적 구조
[get_answer] ─── 최상위 실행 (Run)
├── [retrieve_docs] ─── 자식 실행 1 (검색)
│ └── 실행 시간: 120ms, 반환 문서: 4개
├── [format_context] ─── 자식 실행 2 (컨텍스트 구성)
│ └── 실행 시간: 1ms, 출력 길이: 3200자
└── [generate_answer] ─── 자식 실행 3 (생성)
└── 실행 시간: 2100ms, 토큰: 입력 850 / 출력 120
- LangSmith 대시보드에서 이 구조를 시각적으로 확인할 수 있으며, 어떤 단계에서 시간이 많이 소요되는지, 어떤 문서가 검색되었는지 등을 상세히 분석할 수 있다.
2.4. 평가자(Evaluator) 구성
- 평가자(Evaluator)는 RAG 시스템의 출력을 자동으로 채점하는 함수이다. LangSmith의 `evaluate()` 함수에 평가자를 전달하면, 데이터셋의 각 예시에 대해 자동으로 평가를 수행한다.
2.4.1. 평가자의 동작 원리
[평가 데이터셋] [RAG 시스템] [평가자]
│ │ │
├─ 질문 ──────────────→ ├─ 답변 생성 ──────────→ ├─ 답변 vs 참조 비교
├─ 참조 답변 ├─ 검색 문서 ├─ 환각 여부 확인
└─ 기대 출처 └─ 출처 정보 └─ 점수 반환
- 평가자 함수는 `run`(RAG 시스템의 실행 결과)과 `example`(데이터셋의 평가 예시)을 인자로 받아 점수를 반환한다.
from langchain import hub
from langchain_openai import ChatOpenAI
# 평가 전용 LLM (평가의 일관성을 위해 temperature=0으로 설정)
# gpt-4o를 사용하는 이유: 평가 정확도가 높아야 하므로 최고 성능 모델 사용
eval_llm = ChatOpenAI(model="gpt-4o", temperature=0)
# ──────────────────────────────────────────────────
# 평가자 1: 답변 정확도 (참조 답변 비교)
# ──────────────────────────────────────────────────
def answer_accuracy_evaluator(run, example) -> dict:
"""
예측 답변과 참조 답변의 일치도 평가
- run: RAG 시스템의 실행 결과 (run.outputs에 답변 포함)
- example: 데이터셋의 평가 예시 (example.outputs에 참조 답변 포함)
- 반환: {"key": "메트릭명", "score": 점수}
"""
# LangChain Hub에서 사전 정의된 평가 프롬프트 가져오기
# 이 프롬프트는 LLM에게 "예측 답변이 참조 답변과 의미적으로 일치하는지" 판단하도록 지시
prompt = hub.pull("langchain-ai/rag-answer-vs-reference")
chain = prompt | eval_llm
# RAG 시스템이 생성한 답변
prediction = run.outputs.get("answer", "")
# 데이터셋에 정의된 참조(정답) 답변
reference = example.outputs.get("answer", "")
# 원본 질문 (맥락 이해를 위해 평가자에게도 전달)
question = example.inputs.get("question", "")
result = chain.invoke({
"question": question,
"answer": prediction,
"reference": reference
})
# LLM의 판정 결과에서 점수 추출 (CORRECT → 1, INCORRECT → 0)
score = 1 if "CORRECT" in result.content.upper() else 0
return {"key": "answer_accuracy", "score": score}
# ──────────────────────────────────────────────────
# 평가자 2: 답변 유용성
# ──────────────────────────────────────────────────
def answer_helpfulness_evaluator(run, example) -> dict:
"""
답변의 유용성 평가
- 참조 답변 없이, 질문에 대해 답변이 실질적으로 도움이 되는지 판단
- 정확하지만 유용하지 않은 답변 감지 (예: 너무 짧거나 모호한 답변)
"""
prompt = hub.pull("langchain-ai/rag-answer-helpfulness")
chain = prompt | eval_llm
prediction = run.outputs.get("answer", "")
question = example.inputs.get("question", "")
result = chain.invoke({
"question": question,
"answer": prediction,
})
score = 1 if "HELPFUL" in result.content.upper() else 0
return {"key": "helpfulness", "score": score}
# ──────────────────────────────────────────────────
# 평가자 3: 환각 탐지
# ──────────────────────────────────────────────────
def hallucination_evaluator(run, example) -> dict:
"""
답변이 컨텍스트를 기반으로 하는지 확인 (환각 탐지)
- 컨텍스트에 없는 내용을 답변에 포함했는지 검사
- RAG에서 가장 중요한 평가 항목 중 하나
- 환각이 감지되면 score=0, 환각이 없으면 score=1
"""
prompt = hub.pull("langchain-ai/rag-answer-hallucination")
chain = prompt | eval_llm
prediction = run.outputs.get("answer", "")
# 검색된 컨텍스트 (RagBot.get_answer()에서 반환한 contexts 리스트)
contexts = run.outputs.get("contexts", [])
question = example.inputs.get("question", "")
result = chain.invoke({
"question": question,
"answer": prediction,
"context": "\n".join(contexts) # 여러 컨텍스트를 하나의 문자열로 결합
})
# 환각이 감지되면 0점, 감지되지 않으면 1점 (역전 점수)
score = 0 if "HALLUCINATION" in result.content.upper() else 1
return {"key": "no_hallucination", "score": score}
2.4.2. 평가자 유형 비교
| 평가자 유형 | 설명 | 장점 | 단점 | 예시 |
| LLM Judge | 강력한 LLM이 기준 답변과 생성 답변을 비교하여 다차원 평가 수행 | 자연어 이해 기반 유연한 판단, 의미적 유사성 비교 가능, 대규모 자동화 | API 비용 발생, 평가 일관성 변동성(모델 버전에 따라 다름), 프롬프트 민감도 | 정확도, 유용성, 환각 탐지, 충실도(Faithfulness) |
| 규칙 기반 | 미리 정의한 규칙과 정규식을 통한 자동 평가 | 초고속 처리, 비용 제로, 완벽한 재현성 및 일관성 | 복잡한 의미 이해 불가, 경직된 기준만 적용 | 키워드 포함 여부, 응답 길이 제한 준수, 형식 검증, 토큰 수 체크 |
| 인간 평가 | 도메인 전문가 또는 크라우드소싱 작업자가 직접 점수 매김 | 최고 수준의 정확도와 맥락 이해, 미묘한 뉘앙스 포착 | 높은 시간/비용 소요, 주관성으로 인한 일관성 부족, 확장성 제한 | 전문가 검토, 사용자 경험 평가, 문화적 적절성 확인 |
| 통계 기반 | BLEU, ROUGE, BERTScore 등 전통적/학습 기반 문자열 유사도 메트릭 | 즉시 계산 가능, 비용 없음, 표준화된 벤치마크 | 의미적·맥락적 차이 포착 불가, 단어/문법 수준 한계 | 기계번역 품질(BLEU), 요약 충실도(ROUGE), 임베딩 유사도(BERTScore) |
- 실무 권장: LLM Judge를 주력으로 사용하되, 규칙 기반 평가자를 보조로 추가한다. 예를 들어, 답변 길이가 10자 미만이면 자동으로 0점을 부여하는 규칙 기반 평가자를 추가하면 LLM 호출 비용을 절약할 수 있다. 주기적으로 인간 평가와 LLM Judge 결과를 비교하여 LLM Judge의 품질을 검증한다.
2.5. 평가 실행
- 모든 준비(데이터셋, RAG 시스템, 평가자)가 완료되면 `evaluate()` 함수로 평가를 실행한다. 이 함수는 데이터셋의 각 예시에 대해 RAG 시스템을 실행하고, 모든 평가자로 결과를 채점한 뒤, LangSmith 대시보드에 결과를 기록한다.
from langsmith import evaluate
# RAG 봇 인스턴스 생성
rag_bot = RagBot(retriever=retriever, llm=llm)
# 평가 대상 함수 정의
# - LangSmith의 evaluate()는 이 함수에 데이터셋의 각 입력(inputs)을 전달
# - 함수의 반환값(dict)이 평가자의 run.outputs가 됨
def predict(inputs: dict) -> dict:
return rag_bot.get_answer(inputs["question"])
# 평가 실행
results = evaluate(
predict, # 평가할 함수 (RAG 시스템)
data="tax-qa-eval-dataset", # 평가 데이터셋 이름 (LangSmith에 등록된 이름)
evaluators=[ # 사용할 평가자 리스트
answer_accuracy_evaluator, # 답변 정확도
answer_helpfulness_evaluator, # 답변 유용성
hallucination_evaluator, # 환각 탐지
],
experiment_prefix="tax-rag-v1", # 실험 이름 접두사 (대시보드에서 실험 구분용)
max_concurrency=4 # 동시 평가 수 (병렬 처리로 속도 향상)
)
print(f"실험 결과: {results}")
# LangSmith 대시보드에서 상세 결과 확인 가능:
# - 각 예시별 점수
# - 평가자별 평균 점수
# - 실패한 예시 상세 분석
- 파라미터 정의 기준: `max_concurrency`는 API 레이트 리밋과 시스템 리소스를 고려하여 설정한다. OpenAI API의 기본 레이트 리밋 내에서 보통 4~8이 적절하다. 너무 높으면 429(Rate Limit) 오류가 발생하고, 너무 낮으면 평가 시간이 오래 걸린다.
2.5.1. 평가 결과 해석
실험: tax-rag-v1-20240115-143022
──────────────────────────────────
평가자 | 평균 점수 | 표준편차
─────────────────┼──────────┼────────
answer_accuracy | 0.80 | 0.40
helpfulness | 0.85 | 0.36
no_hallucination | 0.90 | 0.30
──────────────────────────────────
→ 해석:
- 정확도 80%: 10개 중 2개는 부정확 → 검색 품질 또는 프롬프트 개선 필요
- 유용성 85%: 대부분 유용하지만 일부 불충분 → 답변 상세도 개선
- 환각률 10%: 10개 중 1개에서 환각 발생 → 컨텍스트 검증 강화 필요
3. 커스텀 평가 (단계 2: 도메인별 평가 구현)
- LangSmith의 기본 평가자(Hub에서 가져오는 프롬프트 기반)만으로는 도메인 특화된 품질 기준을 충분히 평가하기 어려운 경우가 있다. 커스텀 평가자를 구현하면 도메인의 특수한 요구사항(검색 품질, 답변 충실도, 답변 완전성 등)을 정밀하게 측정할 수 있다.
3.1. 검색 품질 평가
- 검색 품질 평가는 RAG 파이프라인에서 가장 중요한 평가 중 하나이다. 아무리 LLM이 뛰어나도, 관련 없는 문서가 검색되면 좋은 답변을 생성할 수 없다. 검색 품질은 Precision@K(정밀도)와 Recall@K(재현율)로 측정한다.
def retrieval_precision_evaluator(run, example) -> dict:
"""
검색된 문서 중 관련 문서 비율 (Precision@K)
- 높을수록 검색 결과에 노이즈가 적음
- 예: 4개 검색 중 3개가 관련 → Precision = 0.75
"""
# RAG 시스템이 반환한 출처 목록
retrieved_sources = run.outputs.get("sources", [])
# 데이터셋에 정의된 기대 출처 목록
expected_sources = example.outputs.get("expected_sources", [])
# 검색 결과가 없으면 0점
if not retrieved_sources:
return {"key": "retrieval_precision", "score": 0}
# 검색된 문서 중 기대 출처에 포함된 문서 수
relevant = sum(1 for s in retrieved_sources if s in expected_sources)
# Precision = 관련 문서 수 / 전체 검색 문서 수
precision = relevant / len(retrieved_sources)
return {"key": "retrieval_precision", "score": precision}
def retrieval_recall_evaluator(run, example) -> dict:
"""
정답 문서 중 검색된 비율 (Recall@K)
- 높을수록 관련 문서를 빠뜨리지 않음
- 예: 정답 문서 5개 중 3개 검색됨 → Recall = 0.6
"""
retrieved_sources = run.outputs.get("sources", [])
expected_sources = example.outputs.get("expected_sources", [])
# 기대 출처가 없으면 (모든 문서가 관련) 1점
if not expected_sources:
return {"key": "retrieval_recall", "score": 1}
# 기대 출처 중 실제 검색된 문서 수
found = sum(1 for s in expected_sources if s in retrieved_sources)
# Recall = 검색된 관련 문서 수 / 전체 관련 문서 수
recall = found / len(expected_sources)
return {"key": "retrieval_recall", "score": recall}
- 실무 권장: 검색 품질 평가를 위해서는 데이터셋에 `expected_sources` 필드를 추가해야 한다. 각 질문에 대해 "어떤 문서가 검색되어야 정답인가"를 미리 정의하는 것이다. 이 과정이 번거롭지만, 검색 품질을 정량적으로 측정하는 유일한 방법이다.
3.2. 답변 충실도 평가 (Faithfulness)
- 답변 충실도(Faithfulness)는 생성된 답변이 검색된 컨텍스트의 정보만을 사용했는지 평가한다. 환각 탐지와 유사하지만, 0/1 이진 판정이 아닌 연속적인 점수(0.0~1.0)로 충실도의 정도를 세밀하게 측정한다.
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
def faithfulness_evaluator(run, example) -> dict:
"""
답변이 컨텍스트의 정보만 사용했는지 평가
- 1.0: 답변의 모든 주장이 컨텍스트에 근거
- 0.5: 일부는 근거하지만 일부는 외부 지식 사용
- 0.0: 대부분의 주장이 컨텍스트에 근거하지 않음
"""
prompt = ChatPromptTemplate.from_template("""
다음 답변이 주어진 컨텍스트의 정보만을 사용하여 작성되었는지 평가하세요.
컨텍스트:
{context}
답변:
{answer}
평가 기준:
- 1.0: 답변의 모든 주장이 컨텍스트에 근거
- 0.5: 일부 주장은 컨텍스트에 근거하지만 일부는 외부 지식 사용
- 0.0: 대부분의 주장이 컨텍스트에 근거하지 않음
점수(숫자만):
""")
chain = prompt | eval_llm | StrOutputParser()
contexts = run.outputs.get("contexts", [])
answer = run.outputs.get("answer", "")
result = chain.invoke({
"context": "\n".join(contexts),
"answer": answer
})
# 문자열 응답에서 숫자 점수 추출 (파싱 실패 시 0.0)
try:
score = float(result.strip())
except ValueError:
score = 0.0
return {"key": "faithfulness", "score": score}
3.2.1. 환각 탐지 vs 충실도 평가 비교
| 항목 | 환각 탐지 (Hallucination) | 충실도 (Faithfulness) |
| 점수 유형 | 이진 (0 또는 1, 또는 비율 %) | 연속 (0.0~1.0, 0%~100%) |
| 판단 기준 | 컨텍스트에 없는 사실/주장을 생성했는가 여부 | 답변의 모든 주장/정보가 주어진 컨텍스트에 근거하고 일치하는 정도 |
| 세밀도 | 낮음 (존재/비존재만 판별, 전체 문단 단위) | 높음 (개별 주장별 점수화, 부분적 충실도 측정 가능) |
| 용도 | 고위험 환각 사례 필터링 및 경고 시스템 | 세밀한 품질 분석, RAG 파이프라인 최적화 |
| 권장 사용 | 프로덕션 모니터링, 실시간 검출 | 개발/최적화 단계, 모델 비교 및 개선 |
3.3. 답변 완전성 평가
- 답변 완전성(Completeness)은 답변이 질문의 모든 측면을 다루는지 평가한다. 정확하지만 불완전한 답변(질문의 일부만 답하는 경우)을 감지하는 데 유용하다.
def completeness_evaluator(run, example) -> dict:
"""
답변이 질문의 모든 측면을 다루는지 평가
- 복합 질의("A와 B의 차이점과 각각의 세율은?")에서 특히 중요
- 질문의 모든 하위 질문에 대해 답변이 포함되어 있는지 확인
"""
prompt = ChatPromptTemplate.from_template("""
질문의 모든 측면이 답변에서 다뤄졌는지 평가하세요.
질문: {question}
답변: {answer}
평가:
- 1.0: 질문의 모든 측면을 완벽히 다룸
- 0.7: 대부분 다루되 일부 누락
- 0.4: 부분적으로만 답변
- 0.0: 질문에 거의 답하지 못함
점수(숫자만):
""")
chain = prompt | eval_llm | StrOutputParser()
question = example.inputs.get("question", "")
answer = run.outputs.get("answer", "")
result = chain.invoke({"question": question, "answer": answer})
try:
score = float(result.strip())
except ValueError:
score = 0.0
return {"key": "completeness", "score": score}
3.4. RAGAS 프레임워크 활용
- RAGAS(Retrieval Augmented Generation Assessment)는 RAG 시스템의 자동 평가를 위한 오픈소스 프레임워크이다. LangSmith와 함께 사용하거나 독립적으로 사용 가능하다. RAGAS의 핵심 강점은 "참조 답변 없이도" 일부 메트릭을 평가할 수 있다는 점이다.
3.4.1. RAGAS 평가 파이프라인
[평가 데이터]
├─ question (질문)
├─ answer (RAG 시스템의 답변)
├─ contexts (검색된 문서들)
└─ ground_truth (정답, 일부 메트릭에만 필요)
↓
[RAGAS 평가 엔진]
├─ faithfulness: 답변이 컨텍스트에 근거하는가?
├─ answer_relevancy: 답변이 질문에 관련있는가?
├─ context_precision: 관련 컨텍스트가 상위에 있는가?
└─ context_recall: 정답 정보가 컨텍스트에 포함되는가?
↓
[점수 반환] (각 메트릭 0.0~1.0)# pip install ragas
from ragas import evaluate
from ragas.metrics import (
faithfulness, # 답변이 컨텍스트에 충실한지 (참조 답변 불필요)
answer_relevancy, # 답변이 질문에 관련있는지 (참조 답변 불필요)
context_precision, # 검색된 컨텍스트의 정밀도 (참조 답변 필요)
context_recall, # 검색된 컨텍스트의 재현율 (참조 답변 필요)
)
from datasets import Dataset
# 평가 데이터 준비 (HuggingFace Dataset 형식)
eval_data = {
"question": ["근로소득세 세율은?", "연말정산 공제 항목은?"],
"answer": ["6%~45% 누진세율입니다.", "근로소득공제, 인적공제 등이 있습니다."],
"contexts": [
["근로소득세는 과세표준에 따라 6%~45%의 8단계 누진세율이 적용됩니다."],
["연말정산 시 근로소득공제, 인적공제, 특별소득공제 등을 적용합니다."]
],
"ground_truth": ["6%~45%의 누진세율", "근로소득공제, 인적공제, 특별소득공제, 세액공제"]
}
dataset = Dataset.from_dict(eval_data)
# 평가 실행 (내부적으로 LLM을 호출하여 각 메트릭을 계산)
result = evaluate(
dataset=dataset,
metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)
print(result)
# → {'faithfulness': 0.95, 'answer_relevancy': 0.88, ...}
- RAGAS 핵심 메트릭:
| 메트릭 | 평가 대상 | 참조 답변 필요 | 상세 설명 |
| faithfulness | 답변이 컨텍스트에만 근거하는지 (환각 여부) | 불필요 | 답변의 개별 주장(claim)을 추출한 후, 각 주장이 주어진 컨텍스트에서 논리적으로 추론 가능한지 LLM Judge로 검증. 비율로 산출 |
| answer_relevancy | 답변이 질문에 직접 관련 있는지 | 불필요 | 답변 내용에서 질문을 역으로 재생성(reverse generation)하고, 원본 질문과의 의미적 유사도(semantic similarity)를 측정 |
| context_precision | 검색된 컨텍스트 중 정답에 기여하는 문서가 상위에 위치하는지 | 필요 | 상위 K개 컨텍스트 중 실제 정답 생성에 필요한 청크의 비율을 평가. 검색 품질 직접 측정 |
| context_recall | 정답에 필요한 모든 정보가 검색 컨텍스트에 포함되는지 | 필요 | 참조 정답의 각 문장/주장이 검색된 컨텍스트에서 뒷받침되는 비율 계산. 정보 누락 탐지 |
| answer_correctness | 답변의 전체적인 사실적·의미적 정확도 | 필요 | 참조 정답과의 사실적 유사성(factual similarity)과 의미적 유사성(semantic similarity)을 가중 평균. 종합 정확도 지표 |
- 실무 권장: RAGAS는 "참조 답변 없이 평가 가능한 메트릭"(faithfulness, answer_relevancy)이 있어, 대규모 평가 데이터셋을 구축하기 어려운 초기 단계에서 특히 유용하다. LangSmith의 커스텀 평가자와 병행하면 더욱 포괄적인 품질 평가가 가능하다.
3.5. LangSmith 비용 추적
- LLM 기반 시스템의 운영에서 비용 관리는 핵심 과제이다. LangSmith는 각 실행의 토큰 사용량과 비용을 자동으로 추적하여, 어떤 단계에서 비용이 많이 발생하는지 파악할 수 있게 한다.
from langsmith import Client
client = Client()
# 프로젝트의 실행 기록에서 비용 정보 확인
runs = client.list_runs(
project_name="tax-rag-evaluation",
execution_order=1, # 최상위 실행만 조회 (자식 실행 제외)
)
total_cost = 0
total_tokens = 0
for run in runs:
if run.total_cost:
total_cost += run.total_cost
# 토큰 사용량도 추적 가능
if run.total_tokens:
total_tokens += run.total_tokens
print(f"총 비용: ${total_cost:.4f}")
print(f"총 토큰: {total_tokens:,}")
print(f"질의당 평균 비용: ${total_cost/max(1, len(list(runs))):.4f}")
# LangSmith 대시보드에서 시각적으로 확인 가능:
# - 실행별 토큰 사용량 (입력 토큰 / 출력 토큰 분리)
# - 모델별 비용 분석 (gpt-4o vs gpt-4o-mini 등)
# - 기간별 비용 추이 (일별/주별/월별 차트)
# - 단계별 비용 분석 (검색 vs 생성 vs 평가)
3.5.1. 비용 최적화 체크포인트
| 체크포인트 | 확인 사항 | 최적화 방법 |
| LLM 모델 선택 | 고비용 모델이 모든 단계에 필요한가? 과도한 모델 사용 여부 | 생성/복잡 추론: gpt-4o/gpt-4o-2024-11-20, 평가/분류/요약/라우팅: gpt-4o-mini, 임베딩: text-embedding-3-small |
| 검색 K 값 | K가 과도하게 커서 노이즈 컨텍스트 증가? 불필요한 토큰 소비 | K=3~5로 시작해 평가하며 조정, context_precision으로 최적 K 탐색, 최대 10~20 제한 |
| 캐싱 적용 | 동일/유사 질의 반복 처리? 검색/임베딩 재계산 낭비 | 질의 임베딩 기반 semantic 캐시(Redis/Vector DB), LLM 응답 전체 캐싱, TTL 1~24시간 설정 |
| 프롬프트 길이 | 시스템 프롬프트/컨텍스트가 토큰 제한 초과 또는 과도하게 긴가? | 필수 지침만 유지(500토큰 이내), Few-shot 예시 1~2개, 긴 컨텍스트 압축/요약 우선 적용 |
| 평가 빈도 | 모든 질의에 전체 평가 실행? 비용 폭증 여부 | 샘플링(1~10%), A/B 테스트 시 무작위 분배, 프로덕션은 경고 임계값 초과 시만 풀 평가 |
4. A/B 테스트와 실험 (단계 3: 비교 실험)
- RAG 시스템의 성능을 개선하려면 다양한 설정(chunk_size, 모델, 검색 전략 등)을 체계적으로 비교 실험해야 한다. LangSmith의 `evaluate()` 함수와 `experiment_prefix`를 활용하면, 각 설정별 실험 결과를 대시보드에서 나란히 비교할 수 있다.
4.1. A/B 테스트의 핵심 원칙
| 원칙 | 설명 | 예시 |
| 단일 변수 변경 | 한 번에 딱 하나의 파라미터/컴포넌트만 변경하여 인과관계 명확화 | chunk_size=512→1024만 변경 (embedding_model, top_k 등 모두 동일 유지) |
| 동일 데이터셋 | 모든 실험에서 정확히 동일한 평가 데이터셋 사용 (순서 랜덤화 가능) | "tax-qa-eval-dataset-v1" 고정 사용, split 동일 (train/eval 동일 seed) |
| 동일 평가자 | 모든 실험에서 동일한 평가 메트릭과 Judge 모델/프롬프트 적용 | gpt-4o-mini + faithfulness + answer_correctness 고정, 프롬프트 버전 명시 |
| 충분한 샘플 | 통계적 유의미성을 위한 최소 샘플 수 확보 (p-value < 0.05 목표) | 최소 50개 이상, 이상적 100~500개, 신뢰구간 계산으로 검증 |
| 재현 가능성 | 모든 실험 설정, seed, 환경을 완벽히 기록하여 누구나 재현 가능 | config YAML 파일 + seed=42 고정 + Git commit hash 기록 + Docker 환경 |
4.2. 설정 변경에 따른 비교 실험
- chunk_size는 RAG 파이프라인의 성능에 큰 영향을 미치는 핵심 하이퍼파라미터이다. 최적값은 도메인과 질의 유형에 따라 다르므로, 여러 값을 실험하여 비교해야 한다.
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langsmith import evaluate
# 실험 1: chunk_size 비교
# - 동일한 문서, 임베딩 모델, LLM을 사용하고 chunk_size만 변경
# - chunk_overlap은 chunk_size의 약 10%로 설정 (일반적 비율)
configs = [
{"name": "chunk_500", "chunk_size": 500, "chunk_overlap": 50},
{"name": "chunk_1000", "chunk_size": 1000, "chunk_overlap": 100},
{"name": "chunk_1500", "chunk_size": 1500, "chunk_overlap": 200},
{"name": "chunk_2000", "chunk_size": 2000, "chunk_overlap": 300},
]
for config in configs:
# 각 설정으로 RAG 파이프라인을 처음부터 구성
# [전제 조건] loader는 미리 생성된 DocumentLoader 인스턴스
# 예: loader = PyPDFLoader("tax_law.pdf") 또는 TextLoader("regulations.txt")
splitter = RecursiveCharacterTextSplitter(
chunk_size=config["chunk_size"],
chunk_overlap=config["chunk_overlap"]
)
# 문서를 해당 설정으로 분할
docs = loader.load_and_split(splitter)
# 분할된 문서를 벡터 DB에 저장 (새로운 컬렉션)
db = Chroma.from_documents(docs, embedding)
# 검색기 생성 (K=4로 고정하여 다른 변수 통제)
retriever = db.as_retriever(search_kwargs={"k": 4})
# RAG 봇 인스턴스 생성
rag_bot = RagBot(retriever=retriever, llm=llm)
# 평가 대상 함수 (클로저로 rag_bot 캡처)
def predict(inputs):
return rag_bot.get_answer(inputs["question"])
# 평가 실행 - experiment_prefix로 각 설정을 구분
evaluate(
predict,
data="tax-qa-eval-dataset",
evaluators=[answer_accuracy_evaluator, hallucination_evaluator],
experiment_prefix=f"chunk-{config['name']}", # 예: "chunk-chunk_500"
)
# → LangSmith 대시보드에서 chunk_500, chunk_1000, chunk_1500, chunk_2000의
# 정확도와 환각률을 나란히 비교 가능
- 실무 권장: chunk_size 실험 시, 검색 품질(Precision, Recall)과 답변 품질(정확도, 환각)을 모두 측정한다. chunk_size가 크면 Recall은 높아지지만 Precision과 환각이 증가할 수 있다. 반대로 작으면 Precision은 높아지지만 문맥 부족으로 정확도가 떨어질 수 있다.
4.3. 모델 비교 실험
- LLM 모델에 따른 성능 차이를 비교한다. 동일한 검색 결과에 대해 어떤 모델이 더 정확하고 충실한 답변을 생성하는지 확인한다.
from langchain_openai import ChatOpenAI
# 비교할 모델 목록
# - temperature=0: 평가의 일관성을 위해 결정적(deterministic) 출력 사용
models = [
{"name": "gpt-4o-mini", "model": "gpt-4o-mini"},
{"name": "gpt-4o", "model": "gpt-4o"},
]
for model_config in models:
# 각 모델로 LLM 인스턴스 생성 (나머지 설정은 동일)
llm = ChatOpenAI(model=model_config["model"], temperature=0)
rag_bot = RagBot(retriever=retriever, llm=llm)
def predict(inputs):
return rag_bot.get_answer(inputs["question"])
evaluate(
predict,
data="tax-qa-eval-dataset",
evaluators=[answer_accuracy_evaluator, faithfulness_evaluator],
experiment_prefix=f"model-{model_config['name']}",
)
# → gpt-4o-mini vs gpt-4o의 정확도와 충실도 비교
# → 비용 대비 성능 분석 (gpt-4o-mini가 80%의 비용으로 95%의 성능이면 채택)
4.3.1. 모델 비교 시 고려사항
| 항목 | gpt-4o-mini | gpt-4o | 판단 기준 |
| 비용 | 매우 저렴 (input $0.15/M, output $0.60/M) | 기준 (input $2.50/M, output $10.00/M) | 전체 예산, 쿼리당 비용 허용 범위 |
| 속도 | 매우 빠름 (지연 200-500ms, 초고속 처리) | 보통 (지연 500-1500ms) | 실시간 응답 요구 (RTT <1s), 처리량 (QPS) |
| 정확도 | 대부분 작업 충분 (표준 RAG 평가 85-92%) | 최고 수준 (92-97%) | 도메인 전문성, 허용 오차율 |
| 환각 | 약간 높음 (faithfulness 88-92%) | 매우 낮음 (94-97%) | 신뢰성/정확성 최우선 요구사항 |
| 복합 추론 | 제한적 (단순~중급 추론 가능) | 우수 (다단계/법률 등 복잡 추론) | 질의 복잡도 (단순 사실 vs 법률 분석) |
- 실무 권장: 먼저 gpt-4o-mini로 기준선(baseline)을 설정하고, 품질이 부족한 경우에만 gpt-4o로 전환한다. 비용 대비 성능(cost-performance ratio)을 수치로 비교하여 의사결정한다. 대부분의 단순 Q&A에서는 gpt-4o-mini가 충분하고, 복잡한 추론이 필요한 질의에서만 gpt-4o가 의미 있는 차이를 보인다.
4.4. 검색 전략 비교
- 검색 전략(search_type, K 값, score_threshold 등)에 따른 성능 차이를 비교한다.
# 비교할 검색 설정 목록
retriever_configs = [
# 기본 유사도 검색: K=3 (적은 문서, 높은 정밀도)
{"name": "similarity-k3", "search_type": "similarity", "k": 3},
# 기본 유사도 검색: K=6 (많은 문서, 높은 재현율)
{"name": "similarity-k6", "search_type": "similarity", "k": 6},
# MMR: 관련성 + 다양성 균형 (유사 문서가 많은 도메인에 적합)
{"name": "mmr-k4", "search_type": "mmr", "k": 4},
# 임계값 기반: 유사도 0.7 이상만 반환 (품질 보장, 결과 수 불확실)
{"name": "threshold-07", "search_type": "similarity_score_threshold",
"score_threshold": 0.7},
]
for rc in retriever_configs:
# search_kwargs 구성
search_kwargs = {"k": rc.get("k", 4)}
if "score_threshold" in rc:
search_kwargs["score_threshold"] = rc["score_threshold"]
# 검색기 생성
retriever = db.as_retriever(
search_type=rc["search_type"],
search_kwargs=search_kwargs
)
rag_bot = RagBot(retriever=retriever, llm=llm)
def predict(inputs):
return rag_bot.get_answer(inputs["question"])
evaluate(
predict,
data="tax-qa-eval-dataset",
evaluators=[retrieval_precision_evaluator, answer_accuracy_evaluator],
experiment_prefix=f"retriever-{rc['name']}",
)
4.4.1. 검색 전략별 특성 비교
| 전략 | 결과 수 | 정밀도 | 재현율 | 적합한 상황 |
| similarity, K=3 | 고정 3개 | 매우 높음 (90%+) | 낮음 (관련 정보 누락 위험) | 단순 사실 Q&A, 문서 간 명확 구분, 빠른 응답 우선 |
| similarity, K=6 | 고정 6개 | 보통 (80-85%) | 높음 (포괄적 정보 확보) | 복합/다중 문서 참조 질의, 정답 분산된 경우 |
| MMR, K=4 | 고정 4개 | 높음 (85-90%, 다양성 보장) | 보통 (중복 제거로 균형) | 유사 문서 다수 도메인 (법률, 규정, FAQ), 최적 다양성 필요 |
| threshold=0.7 | 가변 (0~N개) | 매우 높음 (95%+, 엄격 필터링) | 불확실 (임계 미달 시 빈 컨텍스트) | 품질/정확도 최우선, "답변 불가" 허용 가능한 고위험 도메인 |
- 실무 권장: similarity(K=4)를 기본으로 시작하고, 검색 결과에 중복이 많으면 MMR을, 노이즈가 많으면 threshold를, 문맥이 부족하면 K를 늘리는 방향으로 조정한다. 법률/의료 도메인처럼 정확성이 중요한 경우 threshold + MMR 조합이 효과적이다
5. 고도화 기법 (단계 4: 성능 향상)
- 기본 RAG 파이프라인의 평가 결과를 기반으로, 성능을 향상시키는 고도화 기법들을 적용한다. 각 기법은 독립적으로 적용할 수 있으며, 적용 후 반드시 평가를 통해 개선 효과를 수치로 확인해야 한다.
[고도화 기법 적용 순서 (권장)]
1. Re-ranking (검색 품질 직접 향상)
↓
2. 대화 이력 기반 질의 재작성 (대화형 시스템인 경우)
↓
3. 답변 검증 Self-check (환각 감소)
↓
4. Fallback 전략 (안정성 확보)
↓
5. 캐싱 (비용/속도 최적화)
↓
6. 비동기 처리 (처리량 향상)
↓
7. Rate Limiting (비용 제어)
5.1. Re-ranking
- Re-ranking(재순위화)은 1차 벡터 검색으로 넓게 가져온 후보 문서들을 더 정밀한 모델로 재평가하여 최종 순위를 매기는 기법이다. 벡터 검색의 Bi-encoder 방식은 빠르지만 정밀도가 제한적인데, Re-ranking은 Cross-encoder 방식으로 질의-문서 쌍을 직접 비교하여 관련성을 더 정확하게 평가한다.
5.1.1. Re-ranking의 2단계 검색 전략
[사용자 질의]
↓
[1단계: Bi-encoder (벡터 검색)] ─── 빠르지만 대략적
│ 질의 벡터 ←→ 문서 벡터 (독립적으로 인코딩)
│ K=20개 후보 문서 검색 (넓게 검색)
↓
[2단계: Cross-encoder (Re-ranking)] ─── 느리지만 정밀
│ (질의, 문서) 쌍을 함께 입력하여 관련도 점수 산출
│ 상위 3~5개 선택 (정밀하게 필터링)
↓
[최종 상위 K개 문서 반환]from langchain.retrievers import ContextualCompressionRetriever
# CrossEncoderReranker 사용 시:
# pip install sentence-transformers
# from langchain_community.cross_encoders import HuggingFaceCrossEncoder
class LLMReranker:
"""
LLM 기반 재순위 매기기
- Cross-encoder 모델 대신 LLM을 사용하여 관련도를 평가
- 별도의 Re-ranker 모델 없이 기존 LLM만으로 구현 가능
- 단점: LLM 호출 비용 발생, 속도 느림
"""
def __init__(self, llm):
self.llm = llm # 관련도 평가에 사용할 LLM
def rerank(self, query: str, documents: list[Document], top_k: int = 3) -> list[Document]:
"""
검색된 문서들을 질의와의 관련도 순으로 재정렬
- query: 사용자 질의
- documents: 1차 검색으로 가져온 후보 문서 리스트
- top_k: 최종 반환할 상위 문서 수
"""
scored_docs = []
for doc in documents:
# 각 문서에 대해 질의와의 관련도 점수를 LLM으로 평가
score = self._score_relevance(query, doc.page_content)
scored_docs.append((doc, score))
# 점수 내림차순 정렬 (높은 관련도 → 낮은 관련도)
scored_docs.sort(key=lambda x: x[1], reverse=True)
# 상위 top_k개만 반환
return [doc for doc, _ in scored_docs[:top_k]]
def _score_relevance(self, query: str, content: str) -> float:
"""
질의와 문서의 관련도를 0~10 점수로 평가
- content[:500]: 문서 앞부분 500자만 사용 (토큰 비용 절약)
"""
response = self.llm.invoke(
f"질문과 문서의 관련도를 0~10 점수로 평가하세요.\n"
f"질문: {query}\n문서: {content[:500]}\n점수(숫자만):"
)
try:
return float(response.content.strip())
except ValueError:
return 0.0 # 파싱 실패 시 최하점
5.1.2. Re-ranking 방식 비교
| 방식 | 속도 | 정확도 | 비용 | 구현 난이도 |
| Cross-encoder 모델 (Cohere Rerank, bge-reranker-base) | 빠름 (초당 50-200 쿼리, 배치 처리 가능) | 높음 (NDCG@10 0.75-0.85) | 전용 API: Cohere $2-5/M 쿼리, HuggingFace 무료/로컬 | 쉬움 (LangChain 통합 5줄 코드, API 키만) |
| LLM 기반 Re-ranking (gpt-4o-mini 등 프롬프트 비교) | 느림 (쿼리당 2-10초, 순차 처리) | 높음 (맥락 이해 우수, 0.80+) | LLM 토큰 비용 (K=10 기준 $0.01-0.05/쿼리) | 중간 (프롬프트 엔지니어링 + 배치 최적화 필요) |
| ColBERT (토큰 단위 late interaction) | 중간 (인덱싱 후 0.5-2초/쿼리) | 매우 높음 (SOTA 수준, NDCG 0.85-0.90+) | 로컬 GPU 무료, 클라우드 인프라 비용 | 어려움 (ColBERTv2 구현, 인덱싱/쿼리 최적화 복잡) |
- 실무 권장: 프로덕션에서는 Cohere Rerank API나 bge-reranker 같은 전용 Re-ranker 모델을 사용하는 것이 비용/성능 면에서 효율적이다. LLM 기반 Re-ranking은 프로토타입이나 소규모 시스템에서 별도 모델 없이 빠르게 적용할 때 유용하다. 1차 검색의 K를 평소의 3~5배로 늘린 뒤(예: K=20), Re-ranking으로 상위 3~5개를 선별하는 것이 일반적인 패턴이다.
5.2. 대화 이력 기반 질의 재작성
- 대화형 RAG 시스템에서는 사용자가 이전 대화를 참조하는 질의를 할 수 있다. 예를 들어 "소득세 세율은?"이라고 질문한 뒤 "그러면 공제 항목은?"이라고 물으면, "그러면"이 소득세를 가리키는 것이다. 질의 재작성(Query Rewriting)은 대화 이력을 반영하여 불완전한 질의를 독립적인 검색 질의로 변환한다.
5.2.1. 질의 재작성 동작 원리
[대화 이력]
사용자: 소득세 세율은?
AI: 6%~45%의 누진세율입니다.
사용자: 그러면 공제 항목은? ← "그러면"이 소득세를 참조
[질의 재작성]
"그러면 공제 항목은?" → "소득세의 공제 항목은?"
↓
독립적인 검색 질의로 변환되어 정확한 검색 가능class ConversationAwareRetriever:
"""
대화 맥락을 반영하여 질의를 재작성하는 검색기
- 대화 이력이 있으면 LLM으로 질의를 재작성
- 대화 이력이 없으면 원본 질의 그대로 검색
"""
def __init__(self, retriever, llm):
self.retriever = retriever # 기본 검색기 (벡터 DB)
self.llm = llm # 질의 재작성용 LLM
def retrieve(self, question: str, chat_history: list = None) -> list[Document]:
"""
대화 이력을 반영하여 검색 수행
- question: 현재 사용자 질의
- chat_history: 이전 대화 메시지 리스트 (순서대로)
"""
if chat_history:
# 대화 이력을 반영하여 질의 재작성
rewritten = self._rewrite_with_context(question, chat_history)
else:
rewritten = question
return self.retriever.invoke(rewritten)
def _rewrite_with_context(self, question: str, history: list) -> str:
"""
대화 이력을 참고하여 질의를 독립적인 검색 질의로 재작성
- history[-6:]: 최근 3턴(6개 메시지)만 참조 (토큰 절약 + 최근 맥락 집중)
- 홀수 인덱스 = 사용자, 짝수 인덱스 = AI (대화 순서)
"""
history_text = "\n".join([
f"{'사용자' if i%2==0 else 'AI'}: {msg}"
for i, msg in enumerate(history[-6:]) # 최근 3턴만 사용
])
response = self.llm.invoke(
f"대화 이력을 참고하여 현재 질문을 독립적인 검색 질의로 재작성하세요.\n\n"
f"대화 이력:\n{history_text}\n\n"
f"현재 질문: {question}\n\n재작성된 질의:"
)
return response.content.strip()
- 파라미터 정의 기준: `history[-6:]`에서 6은 최근 3턴(사용자 3개 + AI 3개 = 6개 메시지)을 의미한다. 너무 많은 이력을 포함하면 토큰 비용이 증가하고 오래된 맥락이 노이즈가 될 수 있으며, 너무 적으면 맥락을 놓칠 수 있다. 일반적으로 최근 3~5턴이 적절하다.
5.3. 답변 검증 (Self-check)
- 답변 검증(Self-check)은 LLM이 생성한 답변을 다시 LLM에게 검증시키는 후처리 단계이다. 생성된 답변의 환각 여부와 완전성을 자동으로 검사하여, 품질이 낮으면 재생성하거나 "답변 불가"로 처리한다.
5.3.1. Self-check 동작 흐름
[질의 + 컨텍스트] → [LLM 답변 생성] → [Self-check]
├─ 환각 체크: 컨텍스트에 근거하는가?
└─ 완전성 체크: 질문을 완전히 다루는가?
↓
[is_valid = True?]
├─ Yes → 답변 반환
└─ No → 재생성 또는 "답변 불가" 처리class AnswerVerifier:
"""
생성된 답변의 품질을 자체 검증
- 환각 체크: 답변의 모든 주장이 컨텍스트에 근거하는지
- 완전성 체크: 답변이 질문의 모든 측면을 다루는지
- 종합 판단: 두 점수가 모두 임계값 이상이면 유효한 답변
"""
def __init__(self, llm):
self.llm = llm # 검증용 LLM (생성용과 같은 모델 사용 가능)
def verify(self, question: str, answer: str, context: str) -> dict:
"""
답변 품질 종합 검증
- question: 원본 질문
- answer: LLM이 생성한 답변
- context: 검색된 컨텍스트 (답변의 근거)
"""
# 1. 환각 체크 - 컨텍스트에 없는 내용을 생성했는지 확인
hallucination = self._check_hallucination(answer, context)
# 2. 완전성 체크 - 질문의 모든 측면을 다루는지 확인
completeness = self._check_completeness(question, answer)
# 3. 종합 판단 - 두 기준 모두 통과해야 유효
# 환각 점수 > 0.7: 대부분의 주장이 컨텍스트에 근거
# 완전성 점수 > 0.5: 질문의 핵심 측면을 최소한 다룸
is_valid = hallucination["score"] > 0.7 and completeness["score"] > 0.5
return {
"is_valid": is_valid, # 최종 판정 (True=유효, False=재생성 필요)
"hallucination": hallucination, # 환각 점수 상세
"completeness": completeness, # 완전성 점수 상세
"should_regenerate": not is_valid # 재생성 필요 여부
}
def _check_hallucination(self, answer: str, context: str) -> dict:
"""환각 검사: 답변이 컨텍스트에 근거하는 정도 (0~1)"""
response = self.llm.invoke(
f"답변의 모든 주장이 컨텍스트에 근거하는지 확인하세요.\n"
f"컨텍스트: {context}\n답변: {answer}\n"
f"점수(0~1, 1=환각 없음):"
)
try:
score = float(response.content.strip())
except ValueError:
score = 0.5 # 파싱 실패 시 중간값
return {"score": score}
def _check_completeness(self, question: str, answer: str) -> dict:
"""완전성 검사: 답변이 질문을 완전히 다루는 정도 (0~1)"""
response = self.llm.invoke(
f"답변이 질문을 완전히 다루는지 평가하세요.\n"
f"질문: {question}\n답변: {answer}\n"
f"점수(0~1, 1=완전):"
)
try:
score = float(response.content.strip())
except ValueError:
score = 0.5
return {"score": score}
- 실무 권장: Self-check는 추가적인 LLM 호출이 필요하므로 비용과 지연시간이 증가한다. 모든 질의에 적용하기보다는, 프로덕션에서 샘플링(예: 10%의 질의에만 적용)하거나, 사용자가 "답변에 문제가 있다"고 피드백한 경우에만 적용하는 전략이 효율적이다. 또는 환각 임계값(0.7)을 먼저 적용하고, 통과하지 못한 경우에만 재생성하는 조건부 적용도 가능하다.
5.4. Fallback 전략
- Fallback(대체) 전략은 기본 체인이 실패(에러, 타임아웃, 빈 응답 등)할 때 대체 체인으로 자동 전환하는 안정성 확보 패턴이다. LangChain의 `.with_fallbacks()` 메서드를 사용하면 간단하게 구현할 수 있다.
5.4.1. Fallback 동작 원리
[사용자 질의]
↓
[Primary Chain (gpt-4o + 벡터 검색)]
├─ 성공 → 답변 반환
└─ 실패 (에러, 타임아웃 등)
↓
[Fallback Chain (gpt-4o-mini + 기본 응답)]
├─ 성공 → 대체 답변 반환
└─ 실패 → 최종 에러 반환from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
# 기본 체인: 고성능 모델 + 벡터 검색 기반 답변
# - gpt-4o: 최고 품질 답변 생성
# - retriever | format_docs: 벡터 DB에서 관련 문서를 검색하고 텍스트로 변환
primary_chain = (
{"context": retriever | format_docs, "question": RunnablePassthrough()}
| prompt
| ChatOpenAI(model="gpt-4o", temperature=0)
| StrOutputParser()
)
# 대체 체인: 경량 모델 + 검색 없이 응답
# - 검색이 실패해도 기본적인 안내를 제공
# - gpt-4o-mini: 빠르고 저렴한 대체 응답
fallback_chain = (
{"context": lambda _: "관련 문서를 찾지 못했습니다.", "question": RunnablePassthrough()}
| prompt
| ChatOpenAI(model="gpt-4o-mini", temperature=0)
| StrOutputParser()
)
# .with_fallbacks()로 기본 체인에 대체 체인 연결
# - primary_chain이 예외를 발생시키면 자동으로 fallback_chain 실행
robust_chain = primary_chain.with_fallbacks([fallback_chain])
5.4.2. Fallback 전략 설계 패턴
| 패턴 | 설명 | 적합한 상황 |
| 모델 다운그레이드 | 고성능 모델 실패 시 저사양/대체 모델로 자동 전환 (gpt-4o → gpt-4o-mini/claude-3.5-sonnet) | API 장애, 레이트 리밋 초과, 비용 초과 시 graceful degradation |
| 검색 제거 | RAG 검색 단계 생략하고 모델 내장 지식 또는 기본 지침으로 응답 생성 | 벡터 DB 연결 장애, 임베딩 API 다운, 검색 타임아웃 |
| 사전 정의 응답 | 미리 작성된 안전 메시지 또는 기본 템플릿 반환 ("죄송합니다. 현재 서비스 점검 중입니다.") | 전면 장애, 모든 컴포넌트 실패, 유지보수 시간 |
| 다중 Fallback | 체인1 실패 → 체인2 시도 → 체인3 → 기본 응답 (Router + SequentialChain) | 높은 가용성 요구 (99.9% SLA), 미션 크리티컬 서비스, 트래픽 피크 대응 |
- 실무 권장: Fallback 체인은 반드시 기본 체인보다 단순하고 안정적이어야 한다. 기본 체인의 실패 원인이 API 장애라면 같은 API를 사용하는 Fallback은 무의미하다. 최소한의 안내 메시지를 반환하는 최종 Fallback을 항상 포함하는 것이 좋다.
5.5. 캐싱 전략
- 동일한 질의가 반복되는 경우, LLM API를 매번 호출하는 것은 비용과 시간 낭비이다. 캐싱(Caching)은 이전 호출 결과를 저장해두고 동일한 입력이 들어오면 저장된 결과를 반환하는 최적화 기법이다.
5.5.1. 캐시 유형 비교
| 캐시 유형 | 저장 위치 | 지속성 | 속도 | 적합한 환경 |
| InMemoryCache | 메모리 (프로세스 내 dict/TTLTree) | 프로세스 종료 시 삭제 (재시작 무효화) | 가장 빠름 (<1ms 조회) | 개발/테스트, 단일 사용자 로컬 실행 |
| SQLiteCache | 로컬 파일 시스템 (langchain.db 등) | 영구 저장 (파일 백업 가능) | 빠름 (1-5ms, 디스크 I/O) | 단일 서버 프로덕션, 오프라인/엣지 환경, 소규모 트래픽 |
| RedisCache | Redis 서버 (클러스터 지원) | 영구 저장 + TTL 자동 만료, replication | 매우 빠름 (<1ms, 네트워크) | 다중 서버/컨테이너 프로덕션, 고트래픽 (QPS 10K+), 분산 환경 |
from langchain_community.cache import InMemoryCache, SQLiteCache
# 참고: langchain.cache에서도 import 가능하지만, langchain_community.cache가 최신 권장 경로
from langchain.globals import set_llm_cache
# 인메모리 캐시 (개발용)
# - 프로그램 재시작 시 캐시가 사라짐
# - 메모리 사용량에 주의 (대량 캐시 시)
set_llm_cache(InMemoryCache())
# SQLite 캐시 (프로덕션용)
# - 파일 기반으로 영구 저장
# - 프로그램 재시작 후에도 캐시 유지
# - database_path: 캐시 DB 파일 경로
set_llm_cache(SQLiteCache(database_path=".langchain_cache.db"))
# 동일한 질의에 대해 캐시된 결과 반환 (API 호출 절약)
# 첫 번째 호출: LLM API 호출 → 결과 캐시에 저장 → 반환 (느림)
# 두 번째 동일 호출: 캐시에서 즉시 반환 (빠름, 비용 없음)
- 실무 권장: 개발 환경에서는 `InMemoryCache`로 빠르게 테스트하고, 프로덕션에서는 `SQLiteCache`(단일 서버) 또는 `RedisCache`(다중 서버)를 사용한다. 캐시 적중률(hit rate)을 모니터링하여 캐싱 효과를 확인한다. 문서가 업데이트되면 캐시를 무효화(invalidation)해야 하므로, TTL(Time-To-Live) 설정이나 수동 캐시 클리어 메커니즘을 함께 구현한다.
5.5.2. .with_retry() (일시적 장애 자동 재시도)
- API 호출 시 네트워크 오류, 레이트 리밋 등 일시적 장애에 대한 자동 재시도를 설정한다. `.with_fallbacks()`가 대체 체인으로 전환하는 것이라면, `.with_retry()`는 동일 체인을 재시도한다.
5.5.3. 지수 백오프(Exponential Backoff) 원리
[요청 실패]
↓
[1차 재시도] ← 1초 대기 (+ 랜덤 지터)
↓ (실패)
[2차 재시도] ← 2초 대기 (+ 랜덤 지터)
↓ (실패)
[3차 재시도] ← 4초 대기 (+ 랜덤 지터)
↓ (실패)
[최종 실패] → 예외 발생 (또는 Fallback 체인으로 전환)
- 지수 백오프는 재시도 간격을 점점 늘려서 서버에 과부하를 주지 않으면서 복구를 기다리는 전략이다. 지터(jitter)는 여러 클라이언트가 동시에 재시도하는 "썬더링 허드(thundering herd)" 문제를 방지하기 위한 랜덤 지연이다.
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
# .with_retry()로 일시적 장애 시 자동 재시도 설정
resilient_llm = llm.with_retry(
stop_after_attempt=3, # 최대 3회 시도 (1회 원본 + 2회 재시도)
wait_exponential_jitter=True, # 지수 백오프 + 지터: 1초 → 2초 → 4초 (+ 랜덤)
)
# 체인에서 사용 - LLM 호출 단계에만 재시도 적용
prompt = ChatPromptTemplate.from_messages([
("system", "세법 전문가로서 답변하세요."),
("human", "{question}")
])
# 체인의 LLM 단계만 재시도 가능하게 구성
chain = prompt | resilient_llm | StrOutputParser()
# .with_fallbacks()와 조합: 재시도 후에도 실패하면 대체 모델 사용
# 이 패턴이 가장 견고한 장애 대응 전략
fallback_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
robust_llm = resilient_llm.with_fallbacks([fallback_llm])
robust_chain = prompt | robust_llm | StrOutputParser()
result = robust_chain.invoke({"question": "근로소득세 계산 방법은?"})
print(result)
- 장애 대응 전략 비교:
- `.with_retry()`: 동일 모델/체인 재시도 (일시적 장애 - 네트워크 오류, 429 Rate Limit)
- `.with_fallbacks()`: 대체 모델/체인으로 전환 (지속적 장애 - 서비스 중단, 모델 에러)
- 조합 사용 권장: 먼저 재시도(3회) → 그래도 실패 시 대체 모델로 전환
5.6. 비동기 처리
- 여러 질의를 동시에 처리해야 하는 경우, 순차 처리(동기)보다 비동기 병렬 처리가 훨씬 효율적이다. LangChain의 모든 Runnable은 `.ainvoke()`, `.abatch()` 같은 비동기 메서드를 제공한다.
5.6.1. 동기 vs 비동기 처리 비교
[동기 처리] 총 6초 (각 2초 × 3개)
질의1 ──[2초]──→ 답변1
질의2 ──[2초]──→ 답변2
질의3 ──[2초]──→ 답변3
[비동기 처리] 총 2초 (3개 동시)
질의1 ──[2초]──→ 답변1
질의2 ──[2초]──→ 답변2
질의3 ──[2초]──→ 답변3import asyncio
async def async_rag_pipeline(questions: list[str]) -> list[str]:
"""
여러 질의를 비동기로 병렬 처리
- asyncio.gather(): 모든 비동기 작업을 동시에 실행하고 결과를 모아 반환
- rag_chain.ainvoke(): RAG 체인의 비동기 버전 (.invoke()의 async 버전)
"""
# 각 질의에 대한 비동기 작업 생성
tasks = [rag_chain.ainvoke({"input": q}) for q in questions]
# 모든 작업을 동시에 실행하고 완료될 때까지 대기
results = await asyncio.gather(*tasks)
return [r["answer"] for r in results]
# 실행
questions = ["소득세란?", "법인세율?", "부가세 신고?"]
answers = asyncio.run(async_rag_pipeline(questions))
- 실무 권장: 비동기 처리 시 `max_concurrency`를 함께 설정하여 동시 요청 수를 제한한다. API 레이트 리밋을 초과하면 오히려 429 에러로 성능이 저하될 수 있다. `asyncio.Semaphore`를 사용하거나 LangChain의 `.abatch(config={"max_concurrency": N})`를 활용한다.
5.7. Rate Limiting (요청 속도 제한)
- 프로덕션 환경에서 API 비용 제어와 서버 보호를 위해 요청 속도를 제한한다. LangChain은 `InMemoryRateLimiter`를 통해 토큰 버킷(Token Bucket) 알고리즘 기반의 요청 속도 제한을 내장 지원한다.
5.7.1. 토큰 버킷 알고리즘
[토큰 버킷]
버킷 크기 (max_bucket_size=5)
┌─────┐
│●●●●●│ ← 최대 5개 토큰 저장 가능
└──┬──┘
│
초당 2개 토큰 추가 (requests_per_second=2)
│
요청 시 토큰 1개 소비
│
토큰 없으면 대기 (check_every_n_seconds=0.1 간격으로 확인)
→ 버스트: 한 번에 최대 5개 요청 가능 (버킷에 토큰이 있으면)
→ 지속: 초당 최대 2개 요청 (토큰 생성 속도에 의해 제한)from langchain_openai import ChatOpenAI
from langchain_core.rate_limiters import InMemoryRateLimiter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import time
# InMemoryRateLimiter: 토큰 버킷 알고리즘 기반 요청 속도 제한
rate_limiter = InMemoryRateLimiter(
requests_per_second=2, # 초당 2개 요청으로 제한 (지속적 처리 속도)
check_every_n_seconds=0.1, # 100ms마다 토큰 확인 (체크 간격)
max_bucket_size=5, # 최대 5개까지 버스트 허용 (순간 최대 동시 요청)
)
# LLM에 rate_limiter 적용
# - rate_limiter 파라미터로 전달하면 해당 LLM의 모든 호출에 자동 적용
llm = ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
rate_limiter=rate_limiter, # 요청 속도 자동 제한
)
# 체인 구성
prompt = ChatPromptTemplate.from_messages([
("system", "간결하게 답하세요."),
("human", "{question}")
])
chain = prompt | llm | StrOutputParser()
# 대량 질의 처리 시 자동으로 속도 제한 적용
questions = ["소득세란?", "법인세란?", "부가세란?", "상속세란?", "증여세란?"]
start = time.time()
results = chain.batch(
[{"question": q} for q in questions],
config={"max_concurrency": 3} # 동시 실행 수도 제한 가능
)
elapsed = time.time() - start
print(f"{len(questions)}개 질의 처리: {elapsed:.1f}초")
for q, r in zip(questions, results):
print(f" Q: {q} → A: {r[:50]}...")
- Rate Limiting 적용 위치:
- `rate_limiter` 파라미터: LLM 수준에서 제한 (권장 - 모든 호출에 자동 적용)
- `max_concurrency`: `.batch()` / `.abatch()` 호출 시 동시 실행 수 제한 (배치 작업에 적합)
- API Gateway: 서비스 수준에서 제한 (FastAPI + slowapi, nginx 등 - 외부 요청 제어)
6. 프로덕션 배포 (단계 5: 서비스화)
- 개발과 검증이 완료된 RAG 시스템을 실제 사용자에게 서비스하기 위해 API로 배포하고, UI를 연결하며, 모니터링 체계를 구축하는 단계이다.
6.1. LangServe로 API 서비스화
- LangServe는 LangChain 체인을 FastAPI 기반의 REST API로 쉽게 배포할 수 있게 해주는 라이브러리이다. `add_routes()` 한 줄로 체인을 API 엔드포인트로 노출할 수 있다.
6.1.1. LangServe가 자동 생성하는 엔드포인트
| 엔드포인트 | 설명 | 용도 |
| POST /tax-qa/invoke | 단일 JSON 입력을 받아 동기/비동기 처리 후 JSON 응답 반환 (LangServe 기본) | 일반 사용자 질의, 단일 요청 처리, API 클라이언트 통합 |
| POST /tax-qa/batch | 여러 입력 배열을 일괄 받아 병렬 처리 후 결과 리스트 반환 | 대량 평가, 배치 처리, 데이터셋 검증, A/B 테스트 |
| POST /tax-qa/stream | 입력에 대한 실시간 토큰 스트리밍 응답 (SSE 지원) | 긴 답변 실시간 출력, 채팅 UI, 사용자 대기 시간 최소화 |
| GET /tax-qa/playground | 브라우저 기반 인터랙티브 테스트 UI 제공 (LangServe 자동 생성) | 개발/디버깅, 프롬프트 테스트, 실시간 추적 확인 |
| POST /tax-qa/feedback | 응답 ID와 사용자 평가(점수/코멘트) 수집 후 DB 저장 | 인간 피드백 루프, 모델 개선 데이터셋 구축, 품질 모니터링 |
# server.py
from fastapi import FastAPI
from langserve import add_routes
# FastAPI 앱 생성
app = FastAPI(
title="Tax RAG API", # API 문서 제목 (Swagger UI에 표시)
version="1.0", # API 버전
description="세법 RAG 기반 질의응답 API" # API 설명
)
# RAG 체인을 REST API로 노출
# - path: URL 경로 접두사 (/tax-qa/invoke, /tax-qa/stream 등 자동 생성)
# - enable_feedback_endpoint: 사용자 피드백 수집 엔드포인트 활성화
# - enable_public_trace_link_endpoint: LangSmith 추적 링크 제공
add_routes(
app,
rag_chain, # 배포할 LangChain 체인
path="/tax-qa", # API 경로
enable_feedback_endpoint=True, # 피드백 엔드포인트 활성화
enable_public_trace_link_endpoint=True, # 추적 링크 활성화
)
# 에이전트도 별도 API로 노출 가능
add_routes(
app,
agent_executor,
path="/tax-agent",
)
# 실행: uvicorn server:app --host 0.0.0.0 --port 8000
# Swagger 문서: http://localhost:8000/docs
# Playground: http://localhost:8000/tax-qa/playground
- 실무 권장: LangServe는 빠른 프로토타이핑에 적합하지만, 프로덕션에서는 인증/인가, 로깅, 에러 핸들링 등을 FastAPI 미들웨어로 추가해야 한다. 대규모 트래픽에서는 LangServe 대신 FastAPI에서 직접 체인을 호출하는 방식이 더 유연하다.
6.2. Streamlit UI 통합
- Streamlit은 Python 코드만으로 대화형 웹 UI를 빠르게 구축할 수 있는 프레임워크이다. RAG 시스템의 데모, 내부 도구, PoC(Proof of Concept) 구현에 적합하다.
# app.py
import streamlit as st
from langchain_openai import ChatOpenAI
from langchain.callbacks import StreamlitCallbackHandler
# 페이지 제목 설정
st.title("세법 상담 AI")
# 세션 상태에 대화 이력 초기화
# - st.session_state: Streamlit의 상태 관리 (페이지 새로고침 간 데이터 유지)
# - messages: 대화 이력을 저장하는 리스트
if "messages" not in st.session_state:
st.session_state.messages = []
# 이전 메시지 표시 (대화 이력을 화면에 렌더링)
for message in st.session_state.messages:
with st.chat_message(message["role"]): # "user" 또는 "assistant" 아이콘
st.markdown(message["content"])
# 사용자 입력 받기
# - st.chat_input(): 채팅 입력창 표시
# - 왈러스 연산자(:=)로 입력값이 있을 때만 실행
if prompt := st.chat_input("세법 관련 질문을 입력하세요"):
# 사용자 메시지를 이력에 추가
st.session_state.messages.append({"role": "user", "content": prompt})
with st.chat_message("assistant"):
# StreamlitCallbackHandler: LangChain 실행 과정을 실시간으로 UI에 표시
# - 검색 중, 생성 중 등의 중간 단계를 사용자에게 보여줌
st_callback = StreamlitCallbackHandler(st.container())
response = agent_executor.invoke(
{"input": prompt},
{"callbacks": [st_callback]} # 콜백으로 실시간 표시
)
answer = response["output"]
st.markdown(answer)
# AI 응답을 이력에 추가
st.session_state.messages.append({"role": "assistant", "content": answer})
6.2.1. Streamlit vs Gradio vs 직접 구현 비교
| 항목 | Streamlit | Gradio | FastAPI + React |
| 학습 곡선 | 매우 낮음 (Python만, 10줄로 UI 완성) | 매우 낮음 (함수 데코레이터, 5줄 데모) | 높음 (백엔드+프론트엔드 분리 개발 필요) |
| 커스터마이징 | 중간 (CSS/컴포넌트 확장 가능, 제한적 레이아웃) | 중간 (테마/블록 편집기, JS 제한적) | 매우 높음 (완전 자유로운 UI/UX, 상태관리 포함) |
| 적합 용도 | 데이터 대시보드, 내부 PoC/프로토타입, LLM 체인 테스트 | ML 모델 데모 공유, HuggingFace 통합, 커뮤니티 데모 | 프로덕션 서비스, 사용자 인증/권한, 대규모 트래픽 |
| 배포 | Streamlit Cloud (무료/쉬움), Docker/AWS | HuggingFace Spaces (무료 호스팅), Docker | 완전 자유 (Vercel/Netlify 프론트, Railway/Docker 백엔드) |
| 성능 | 보통 (Python 싱글 스레드, 100QPS 한계) | 보통 (동일 Python 기반, 웹소켓 스트림) | 높음 (비동기/스케일링, 10K+ QPS, 로드밸런싱) |
6.3. 모니터링 체크리스트
- 프로덕션 RAG 시스템은 지속적인 모니터링이 필수이다. 성능 저하, 비용 급증, 오류 발생을 조기에 감지하여 대응해야 한다.
| 모니터링 항목 | 도구 | 임계값 (예시) | 알림 조건 |
| 응답 시간 | LangSmith / Prometheus + Grafana | P95 < 5초, P99 < 10초 | P95 > 8초 지속 5분, P99 > 15초 즉시 PagerDuty |
| 환각률 | LangSmith Evaluator + 커스텀 Dashboard | 일간 < 3%, 주간 < 5% | 일간 > 7% 또는 3일 연속 상승 트렌드 시 Slack #alerts |
| 검색 적중률 | LangSmith Tracing + context_precision 메트릭 | > 85% (K=5 기준) | 주간 평균 < 75% 또는 급락(10%↓) 시 알림 |
| API 비용 | OpenAI Dashboard + AWS Cost Explorer | 월 $1,000 이내, 일 $40 이내 | 일 예산 80% 초과 또는 토큰 사용량 20%↑ 시 경고 |
| 에러율 | Sentry / LangSmith Errors / CloudWatch | < 0.5% (4XX/5XX) | 5분 내 > 2% 또는 특정 에러(LLM timeout) 10회↑ 시 즉시 |
| 사용자 만족도 | /feedback API + DB 집계 + NPS 계산 | 주간 평균 > 4.2/5.0 | 주간 < 3.8 또는 1점 리뷰 5개↑ 시 매니저 알림 |
6.3.1. 모니터링 구축 우선순위
[1단계: 필수] 에러 로깅 + 응답 시간 측정
↓
[2단계: 권장] LangSmith 추적 + 비용 모니터링
↓
[3단계: 고도화] 품질 메트릭 자동 측정 + 사용자 피드백 수집
↓
[4단계: 최적화] 대시보드 구축 + 자동 알림 + 자동 스케일링
- 실무 권장: 최소한 에러 로깅과 응답 시간 측정은 배포 전에 반드시 구축한다. LangSmith 추적은 환경변수 설정만으로 활성화되므로 추가 구현 비용이 거의 없다. 비용 모니터링은 예상치 못한 비용 폭증(예: 무한 루프, 대량 봇 요청)을 방지하기 위해 필수이다.
7. 전체 프로덕션 파이프라인 체크리스트
- RAG 시스템을 개발부터 운영까지 체계적으로 진행하기 위한 단계별 체크리스트이다. 각 단계를 순서대로 완료하면서, 이전 단계에서 미비한 항목이 있으면 보완한다.
[전체 흐름]
개발 단계 → 검증 단계 → 고도화 단계 → 배포 단계 → 운영 단계
│ │ │ │ │
│ │ │ │ └─ 지속적 품질 관리
│ │ │ └─ 서비스 안정성 확보
│ │ └─ 성능 향상 기법 적용
│ └─ 품질 기준 충족 확인
└─ 기본 파이프라인 + 평가 체계 구축
7.1. 개발 단계
- [ ] 평가 데이터셋 구축 (최소 50개 Q&A 쌍)
- [ ] 기본 RAG 파이프라인 구현
- [ ] LangSmith 추적 설정
- [ ] 기본 평가자 3종 구현 (정확도, 유용성, 환각)
- [ ] chunk_size, k 값 등 하이퍼파라미터 실험
7.2. 검증 단계
- [ ] 평가 점수 기준선 설정 (정확도 > 80%, 환각 < 5%)
- [ ] A/B 테스트로 설정 비교
- [ ] 도메인 전문가 검토
- [ ] 엣지 케이스 테스트 (빈 검색 결과, 긴 질의, 다국어 등)
- [ ] 검색 품질 별도 평가 (Precision, Recall)
7.3. 고도화 단계
- [ ] Re-ranking 적용
- [ ] 질의 변환/확장 적용
- [ ] 답변 자체 검증 적용
- [ ] Fallback 전략 구현
- [ ] 캐싱 적용
- [ ] 비동기 처리 적용
7.4. 배포 단계
- [ ] API 서비스 구현 (LangServe / FastAPI)
- [ ] 인증/인가 설정
- [ ] Rate limiting 설정
- [ ] 모니터링/알림 설정
- [ ] 로깅 구조 설계
- [ ] 문서 업데이트 파이프라인 구축
- [ ] 롤백 전략 준비
7.5. 운영 단계
- [ ] 정기 평가 실행 (주간/월간)
- [ ] 사용자 피드백 수집 및 반영
- [ ] 문서 최신화 프로세스
- [ ] 비용 모니터링 및 최적화
- [ ] 성능 병목 분석 및 개선
- 실무 권장: 체크리스트의 모든 항목을 한 번에 완료하려 하지 말고, 각 단계를 완료한 뒤 다음 단계로 진행한다. 특히 개발 단계에서 평가 체계를 먼저 구축하는 것이 핵심이다. 평가 없이 고도화를 진행하면, 개선 효과를 측정할 수 없어 비효율적인 시행착오를 반복하게 된다.
8. 실무 프로젝트: RAG 시스템 평가 및 고도화 파이프라인
- 본 문서에서 학습한 검증/고도화 기법을 통합하여, RAG 시스템의 자동 평가 → 병목 진단 → 최적화 → 재평가 사이클을 구현한다.
8.1. 시나리오 및 요구 명세
| 항목 | 요구사항 |
| 목표 | 기존 RAG 파이프라인의 품질을 정량적으로 측정하고 자동 개선 |
| 평가 지표 | Retrieval Recall, Answer Faithfulness, Answer Relevancy |
| 최적화 대상 | chunk_size, k 값, 프롬프트 템플릿 |
| 자동화 | 평가 → 진단 → 파라미터 조정 → 재평가 루프 |
8.2. 핵심 구현: 자동 평가 및 진단 시스템
# ============================================================
# RAG 품질 자동 평가 및 진단 시스템
# ============================================================
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
import json
# ── 평가 데이터셋 정의 ──
# 실무에서는 도메인 전문가와 함께 20~50개 이상의 Q&A 쌍을 구성
eval_dataset = [
{
"question": "연차 휴가는 며칠인가요?",
"expected_answer": "15일",
"expected_context_keywords": ["제1조", "연차휴가", "15일"]
},
{
"question": "출장 숙박비 한도는?",
"expected_answer": "10만원",
"expected_context_keywords": ["제4조", "숙박비", "10만원"]
},
{
"question": "배우자 출산 휴가는 며칠?",
"expected_answer": "10일",
"expected_context_keywords": ["제2조", "출산", "10일"]
},
]
def evaluate_rag_pipeline(chain, retriever, dataset: list) -> dict:
"""RAG 파이프라인의 품질을 3가지 지표로 평가
Returns:
dict: {
"retrieval_recall": float, # 검색 재현율 (0~1)
"answer_relevancy": float, # 답변 관련성 (0~1)
"answer_faithfulness": float, # 답변 충실도 (0~1)
"details": list # 개별 평가 결과
}
"""
results = {"details": [], "retrieval_recall": 0, "answer_relevancy": 0, "answer_faithfulness": 0}
for item in dataset:
q = item["question"]
# 1. 검색 평가: 기대 키워드가 검색 결과에 포함되었는지
docs = retriever.invoke(q)
context_text = " ".join(d.page_content for d in docs)
keyword_hits = sum(1 for kw in item["expected_context_keywords"] if kw in context_text)
recall = keyword_hits / len(item["expected_context_keywords"])
# 2. 답변 생성
answer = chain.invoke(q)
# 3. 답변 관련성: 기대 답변 키워드가 실제 답변에 포함되었는지
relevancy = 1.0 if item["expected_answer"] in answer else 0.0
# 4. 답변 충실도: 답변이 검색 문서에 기반하는지 (간이 검사)
# 답변의 핵심 수치/키워드가 컨텍스트에도 존재하면 충실
faithfulness = 1.0 if item["expected_answer"] in context_text else 0.0
results["details"].append({
"question": q, "recall": recall,
"relevancy": relevancy, "faithfulness": faithfulness
})
# 평균 점수 계산
n = len(dataset)
results["retrieval_recall"] = sum(d["recall"] for d in results["details"]) / n
results["answer_relevancy"] = sum(d["relevancy"] for d in results["details"]) / n
results["answer_faithfulness"] = sum(d["faithfulness"] for d in results["details"]) / n
return results
def diagnose_and_suggest(eval_results: dict) -> list[str]:
"""평가 결과를 분석하여 개선 방향을 제안
Returns:
list[str]: 개선 제안 목록
"""
suggestions = []
if eval_results["retrieval_recall"] < 0.7:
suggestions.append("검색 재현율이 낮습니다 → chunk_size 축소 또는 k 값 증가를 시도하세요")
if eval_results["answer_relevancy"] < 0.7:
suggestions.append("답변 관련성이 낮습니다 → 프롬프트에 '간결하게 핵심만' 지시를 추가하세요")
if eval_results["answer_faithfulness"] < 0.8:
suggestions.append("답변 충실도가 낮습니다 → 프롬프트에 '컨텍스트 외 내용 금지' 규칙을 강화하세요")
if not suggestions:
suggestions.append("모든 지표가 양호합니다. 현재 설정을 유지하세요.")
return suggestions
# ── 실행 예시 ──
# (chain, retriever는 기존 RAG 파이프라인에서 가져옴)
# eval_results = evaluate_rag_pipeline(qa_chain, retriever, eval_dataset)
# print(f"검색 재현율: {eval_results['retrieval_recall']:.0%}")
# print(f"답변 관련성: {eval_results['answer_relevancy']:.0%}")
# print(f"답변 충실도: {eval_results['answer_faithfulness']:.0%}")
# for s in diagnose_and_suggest(eval_results):
# print(f" → {s}")
- 예상 실행 결과:
검색 재현율: 89%
답변 관련성: 100%
답변 충실도: 100%
→ 모든 지표가 양호합니다. 현재 설정을 유지하세요.
- 실무 권장: 이 평가 시스템을 CI/CD 파이프라인에 통합하면, 프롬프트나 검색 파라미터 변경 시 자동으로 품질 회귀를 감지할 수 있다. 평가 데이터셋은 최소 20개 이상, 다양한 질의 유형을 포함하도록 구성하라.
'Study > LangChain' 카테고리의 다른 글
| 8. LangChain 실무 설계 고려사항 총정리 (0) | 2026.03.08 |
|---|---|
| 7. LangChain 커스텀 함수 구현 샘플 모음 - RAG + LangChain 전 단계 (0) | 2026.03.07 |
| 5. LangChain 에이전트 구축 가이드 - Tool 연동과 에이전트 패턴 (0) | 2026.03.03 |
| 4. LangChain 기본 활용 가이드 - 기본부터 체인 구성까지 (0) | 2026.02.23 |
| 1.1. create_retrieval_chain 완전 가이드 (0) | 2026.02.18 |
댓글