2026-03-20 강의 정리

---

목차

1. 오늘 강의 개요
2. 실습 데이터 준비 (샘플 문서 3종)
3. 환경 설정 & 라이브러리 import
4. Document 객체 & 문서 로딩
5. FAISS 벡터스토어 생성, 저장, 로드
6. Retriever 만들기
7. LCEL (LangChain Expression Language) 개념
8. RunnablePassthrough - 핵심 개념
9. format_docs 함수 - 문서 포매팅
10. RAG Chain 완성 구조
11. RAG Prompt 개선 (할루시네이션 방지)
12. RunnableParallel - 병렬 실행
13. Streaming
14. RunnableLambda + RunnableBranch - 조건 분기
15. similarity_search_with_score - 유사도 점수 활용
16. verify_query - threshold 기반 필터링
17. 핵심 개념 총정리
18. 자주 나오는 실수/주의사항

---

1. 오늘 강의 개요

오늘 강의는 RAG(Retrieval-Augmented Generation) 시스템 전체를 처음부터 끝까지 직접 구현하는 날이다.

RAG = “문서를 검색해서 LLM에 넣어주는” 시스템

전체 흐름

사용자 질문
    ↓
Retriever (FAISS에서 관련 문서 k개 검색)
    ↓
format_docs (검색된 문서를 하나의 텍스트로 합치기)
    ↓
Prompt (context + question 조합)
    ↓
LLM (gpt-4o-mini)
    ↓
StrOutputParser
    ↓
최종 답변

이 흐름을 LCEL(LangChain Expression Language)로 파이프(

) 연산자를 사용해 연결한다.

---

2. 실습 데이터 준비 (샘플 문서 3종)

오늘 실습에서는 직접 샘플 텍스트 파일 3개를 만들어서 사용한다.

파일 목록

• company_policy.txt → 사내규정
• ai_report.txt → AI보고서
• product_manual.txt → 제품매뉴얼

코드

from pathlib import Path

sample_texts = {
    'company_policy.txt': """주식회사 모두의연구소 사내 규정

제1조 (목적)
이 규정은 주식회사 모두의연구소의 임직원이 준수해야 할 기본적인 사항을 정하는 것을 목적으로 한다.

제2조 (근무시간)
1. 기본 근무시간은 오전 9시부터 오후 6시까지로 한다.
2. 유연근무제를 시행하며, 코어타임은 오전 10시부터 오후 4시까지이다.
3. 재택근무는 주 2회까지 가능하다.

제3조 (휴가)
1. 연차휴가는 근로기준법에 따라 부여한다.
2. 경조사 휴가는 별도 규정에 따른다.
3. 자기개발 휴가를 연 5일 추가 부여한다.

제4조 (교육)
1. 모든 임직원은 연간 40시간 이상의 교육을 이수해야 한다.
2. 외부 컨퍼런스 참석비를 연 200만원까지 지원한다.
3. 온라인 학습 플랫폼 이용료를 전액 지원한다.
""",
    'ai_report.txt': """...""",
    'product_manual.txt': """..."""
}

SAMPLE_DIR = Path('sample_data')
SAMPLE_DIR.mkdir(exist_ok=True)

for filename, content in sample_texts.items():
    (SAMPLE_DIR / filename).write_text(content, encoding='utf-8')

Path(‘sample_data’).mkdir(exist_ok=True)

• exist_ok=True : 폴더가 이미 있어도 에러 없이 넘어감

.write_text(content, encoding=’utf-8’)

• 파일에 텍스트를 쓸 때 반드시 encoding 지정

---

3. 환경 설정 & 라이브러리 import

코드

import os
import time
import numpy as np
from dotenv import load_dotenv
load_dotenv()

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_community.vectorstores import FAISS

llm = ChatOpenAI(model='gpt-4o-mini')
embeddings = OpenAIEmbeddings(model='text-embedding-3-small')

각 모듈 역할

• ChatOpenAI : LLM (대화 생성)
• OpenAIEmbeddings : 텍스트 → 벡터(숫자 배열)로 변환
• Document : LangChain이 다루는 문서 단위 객체
• ChatPromptTemplate : 프롬프트 템플릿 생성
• StrOutputParser : LLM 출력을 문자열로 파싱
• FAISS : Facebook이 만든 벡터 유사도 검색 라이브러리

---

4. Document 객체 & 문서 로딩

Document 구조

Document(
    page_content = "실제 텍스트 내용",
    metadata = {
        'category': '사내규정',
        'source': 'company_policy.txt'
    }
)

page_content : 임베딩되고 검색에 사용되는 실제 텍스트

metadata : 부가 정보 (어느 파일인지, 카테고리 등). 검색 결과에서 출처 표시에 활용

문서 로딩 & 분할 코드

from pathlib import Path

data_dir = Path("sample_data")

files = {
    data_dir / "company_policy.txt" : '사내규정',
    data_dir / 'product_manual.txt' : '제품매뉴얼',
    data_dir / 'ai_report.txt'      : 'AI보고서'
}

documents = []
for fpath, category in files.items():
    text = fpath.read_text()
    for section in text.strip().split('\n\n'):  # 빈 줄 기준으로 문단 분리
        if section.strip():
            documents.append(Document(
                page_content = section.strip(),
                metadata = {'category': category, 'source': fpath.name}
            ))

text.split(‘\n\n’) 으로 빈 줄 기준 문단 분리 → 각 문단이 하나의 Document가 됨

len(documents) → 13개 (3개 파일에서 문단 단위로 분리됨)

중요 포인트 - 왜 문단 단위로 나누나?

• LLM에게 너무 긴 텍스트를 통째로 주면 비효율적
• 관련 있는 부분만 쏙 뽑아서 LLM에게 전달하는 게 RAG의 핵심
• 이걸 “청킹(Chunking)” 이라고 함

---

5. FAISS 벡터스토어 생성, 저장, 로드

FAISS란?

• Facebook AI Similarity Search
• 텍스트를 벡터(숫자 배열)로 변환해서 저장
• 질문이 들어오면 가장 유사한 벡터(=문서)를 빠르게 찾아줌

벡터스토어 생성

vectorstore = FAISS.from_documents(documents, embeddings)

• documents : Document 객체 리스트
• embeddings : 텍스트를 벡터로 변환해주는 임베딩 모델
• 이 한 줄이 내부적으로 각 document의 page_content를 임베딩 API에 보내서 벡터화함

벡터스토어 저장

vectorstore.save_local('faiss_docs')

• 로컬에 파일로 저장 (faiss_docs/ 폴더 생성됨)
• 다음에 다시 임베딩하지 않고 바로 불러서 쓸 수 있음 (비용, 시간 절약)

벡터스토어 불러오기

loaded_vs = FAISS.load_local(
    'faiss_docs',
    embeddings,
    allow_dangerous_deserialization=True  # 반드시 True 설정
)
loaded_vs.index.ntotal  # → 13 (저장된 벡터 수)

allow_dangerous_deserialization=True 는 필수 파라미터 (LangChain이 보안 경고를 주기 때문에 명시적으로 True 줘야 함)

.index.ntotal : 저장된 벡터의 총 개수 확인

---

6. Retriever 만들기

코드

retriever = vectorstore.as_retriever(search_kwargs={'k': 3})

as_retriever() : vectorstore를 retriever 인터페이스로 변환

search_kwargs={‘k’: 3} : 검색 결과 상위 k개 반환 (기본값 4)

사용 예시

retriever.invoke('재택근무 규정')
# → 관련 Document 3개 리스트 반환

k 값에 따른 차이

• k=1 : 가장 관련도 높은 문서 1개만 → 간결한 답변, 정보 누락 가능
• k=3 : 상위 3개 → 더 풍부한 컨텍스트
• k가 크면 LLM에게 더 많은 정보를 주지만, 토큰 소모 증가

실습: k=1 vs k=3 비교

retriever_k1 = vectorstore.as_retriever(search_kwargs={'k': 1})
retriever_k3 = vectorstore.as_retriever(search_kwargs={'k': 3})
# 같은 질문을 두 retriever로 각각 체인 만들어서 답변 비교해보기

---

7. LCEL (LangChain Expression Language) 개념

LCEL이란?

• LangChain Expression Language

• 파이프 연산자(

  )를 사용해서 체인을 직관적으로 연결하는 방식

기본 예시

# LCEL 방식
simple_chain = simple_prompt | llm | StrOutputParser()
result = simple_chain.invoke({'question': '대한민국의 수도는?'})
print(result)  # → "대한민국의 수도는 서울입니다."

파이프(|) 연산자의 의미

A | B | C

= A의 출력이 B의 입력으로 → B의 출력이 C의 입력으로

즉, 데이터가 왼쪽에서 오른쪽으로 흘러가는 것

비유

prompt | llm | parser
= 틀 만들기 → LLM에 보내기 → 결과 파싱하기
= 공장의 컨베이어 벨트처럼 단계별로 처리

LCEL에서 쓸 수 있는 주요 Runnable 들

• ChatPromptTemplate : 프롬프트 생성
• ChatOpenAI : LLM 호출
• StrOutputParser : 출력 파싱
• RunnablePassthrough : 입력 그대로 통과 (오늘의 핵심!)
• RunnableParallel : 여러 체인 병렬 실행
• RunnableLambda : 일반 파이썬 함수를 Runnable로
• RunnableBranch : 조건에 따라 다른 체인 실행

---

8. RunnablePassthrough - 핵심 개념 (★★★ 가장 중요!)

개념

• 입력을 그대로 통과시키는 Runnable
• “아무것도 안 하고 그냥 전달”

코드

from langchain_core.runnables import RunnablePassthrough

passthrough = RunnablePassthrough()

passthrough.invoke('안녕하세요')  # → '안녕하세요'
passthrough.invoke(123)           # → 123
passthrough.invoke({'a': 1})      # → {'a': 1}

그런데 왜 쓰나? - RAG에서의 역할

RAG Chain을 만들 때 이런 구조가 필요하다:

LLM에게 넘겨줄 것: { “context”: 검색결과, “question”: 원래질문 }

문제: 파이프로 연결하면 입력이 변형되는데, 원래 질문을 그대로 보존하려면?

rag_chain = (
    {
        "context" : retriever | format_docs,   # 질문 → 검색 → 포매팅
        "question": RunnablePassthrough()       # 질문 → 그대로 통과
    }
    | rag_prompt
    | llm
    | StrOutputParser()
)

딕셔너리 { } 안에 여러 키를 정의하면 → 각각 실행되어 딕셔너리 형태로 합쳐짐

“context” 키: 질문을 retriever에 넣고 검색 → format_docs로 포매팅

“question” 키: 질문을 그대로 통과시켜서 prompt의 {question}에 전달

시각적 표현

"재택근무 규정이 어떻게 되나요?"
            ↓
┌─────────────────────────────────────────────────────┐
│  {                                                  │
│    "context": retriever | format_docs               │
│                → "[사내규정] 재택근무는 주2회..."    │
│    "question": RunnablePassthrough()                │
│                → "재택근무 규정이 어떻게 되나요?"   │
│  }                                                  │
└─────────────────────────────────────────────────────┘
            ↓
rag_prompt (context + question을 합쳐서 프롬프트 생성)
            ↓
llm
            ↓
StrOutputParser()
            ↓
"재택근무는 주 2회까지 가능합니다."

---

9. format_docs 함수 - 문서 포매팅

문제 상황

• retriever.invoke()는 Document 객체 리스트를 반환
• 하지만 LLM 프롬프트에는 텍스트(문자열)를 넣어야 함
• → Document 리스트를 하나의 문자열로 합쳐야 함

format_docs 함수

def format_docs(docs):
    return '\n--\n'.join(
        f"[{d.metadata.get('category', '')}] {d.page_content}"
        for d in docs
    )

결과 예시

[사내규정] 제2조 (근무시간)
1. 기본 근무시간은 오전 9시부터 오후 6시까지로 한다.
2. 유연근무제를 시행하며, 코어타임은 오전 10시부터 오후 4시까지이다.
3. 재택근무는 주 2회까지 가능하다.
--
[사내규정] 제4조 (교육)
1. 모든 임직원은 연간 40시간 이상의 교육을 이수해야 한다.
...

[카테고리] 를 앞에 붙여서 LLM이 어떤 문서인지 파악 가능

‘\n–\n’ 로 문서 사이를 구분 → LLM이 각 문서를 구분해서 읽을 수 있음

d.metadata.get(‘category’, ‘’) : metadata에 ‘category’ 없으면 빈문자열

체인에서 format_docs 사용

retriever_chain = retriever | format_docs
context_text = retriever_chain.invoke('재택근무')
# retriever가 반환한 Document 리스트를 format_docs가 문자열로 변환

일반 파이썬 함수도 파이프(|) 연산자로 연결 가능! (LangChain이 자동으로 RunnableLambda로 감싸줌)

---

10. RAG Chain 완성 구조

RAG Prompt 정의

rag_prompt = ChatPromptTemplate.from_messages([
    ("system", "사내 도우미 챗봇입니다. 참고문서:\n{context}\n\n문서 기반으로 답변해주세요."),
    ("user", "{question}")
])

{context} : format_docs가 만든 문자열이 들어감

{question} : 사용자 원래 질문이 들어감

RAG Chain 전체 코드

rag_chain = (
    {
        "context" : retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | rag_prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke('재택근무 규정이 어떻게 되나요?')
print(answer)
# → "재택근무는 주 2회까지 가능하다는 규정이 있습니다."

데이터 흐름 상세

invoke("재택근무 규정이 어떻게 되나요?")
    ↓
딕셔너리 단계:
  - context  = retriever.invoke("재택근무...") | format_docs
             = "[사내규정] 제2조 (근무시간)\n..."
  - question = "재택근무 규정이 어떻게 되나요?"
    ↓
rag_prompt.invoke({"context": "...", "question": "..."})
= ChatPromptValue(messages=[SystemMessage("..."), HumanMessage("...")])
    ↓
llm.invoke(messages)
= AIMessage("재택근무는 주 2회까지...")
    ↓
StrOutputParser().invoke(AIMessage)
= "재택근무는 주 2회까지..."

k=1로 만들기 (실습)

retriever_k1 = vectorstore.as_retriever(search_kwargs={'k': 1})
rag_chain_k1 = (
    {
        "context" : retriever_k1 | format_docs,
        "question": RunnablePassthrough()
    }
    | rag_prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain_k1.invoke('재택근무 규정이 어떻게 되나요?')
# → "재택근무는 주 2회까지 가능하다는 규정이 있습니다." (k=3보다 간결)

여러 질문 한번에 테스트

question_list = ['스마트홈 허브 초기 설정 방법', 'ai 산업 성장률', '교육비 지원 한도']
for q in question_list:
    print(f"Q: {q}")
    print(f"A: {rag_chain.invoke(q)}")
    print("======")

---

11. RAG Prompt 개선 (할루시네이션 방지)

문제 상황

• 기본 RAG chain은 문서에 없는 내용을 물어봐도 LLM이 알고 있는 내용으로 답변
• 예: “2002년 월드컵 우승팀” → 문서에 없지만 LLM이 답변해버림

해결책: 프롬프트에 제약 추가

rag_prompt = ChatPromptTemplate.from_messages([
    ("system",
     "사내 도우미 챗봇입니다. 참고문서:\n{context}\n\n"
     "문서 기반으로 답변해주세요."
     "참고 문서에 없는 내용이면, '해당 정보가 없습니다' 라고 답변하세요."),
    ("user", "{question}")
])

rag_chain.invoke('2002년 월드컵 우승팀은 어디인가요?')
# → "해당 정보가 없습니다."

하지만 이 방법의 한계:

• retriever는 어쨌든 “가장 유사한” 문서를 가져옴 (관련 없어도)
• LLM이 프롬프트 지시를 항상 완벽하게 따르지는 않음
• 더 확실한 방법이 필요 → 유사도 점수(score) 기반 필터링 (섹션 15, 16)

---

12. RunnableParallel - 병렬 실행

개념

• 여러 체인을 동시에(병렬로) 실행하고 결과를 딕셔너리로 반환
• 답변 + 출처 문서를 같이 반환할 때 유용

기본 사용법

from langchain_core.runnables import RunnableParallel

rag_chain_with_sources = RunnableParallel(
    answer          = rag_chain,        # 답변 생성
    source_document = retriever         # 출처 문서 반환
)

result = rag_chain_with_sources.invoke('재택근무 규정이 어떻게 되나요?')
# result = {
#     'answer': '재택근무는 주 2회까지 가능하다고 규정되어 있습니다.',
#     'source_document': [Document(...), Document(...), Document(...)]
# }

출처 문서 출력

result = rag_chain_with_sources.invoke('재택근무 규정이 어떻게 되나요?')
for doc in result['source_document']:
    print(f"[{doc.metadata['category']}] {doc.page_content[:100]}")

3개 결과 동시에 반환하기 (실습)

new_chain = RunnableParallel(
    answer           = rag_chain,
    source_documents = retriever,
    context          = retriever | format_docs  # 텍스트로 변환된 컨텍스트도 같이
)

result = new_chain.invoke('재택근무 규정을 알려주세요')
# result = {
#     'answer': '재택근무는 주 2회까지 가능합니다.',
#     'source_documents': [Document(...), ...],
#     'context': '[사내규정] 제2조 ...'
# }

RunnableParallel은 딕셔너리 { } 로도 표현 가능 (위 RAG chain에서 쓴 방식)

두 방식은 동일:

RunnableParallel(answer=rag_chain, source=retriever)
{"answer": rag_chain, "source": retriever}  # 딕셔너리 방식

---

13. Streaming

개념

• LLM 답변을 토큰 단위로 실시간으로 받는 방식
• .invoke() 는 전체 답변이 완성된 뒤에 한번에 반환
• .stream() 은 생성되는 즉시 청크(chunk) 단위로 반환

코드

# 답변을 streaming으로 받기
for chunk in rag_chain.stream('재택근무 규정이 어떻게 되나요?'):
    print(chunk, end="", flush=True)
# → 재택근무는 주  2회까지 가능하다고 규정되어 있습니다.
#   (실제로는 단어/토큰 단위로 실시간 출력됨)

end=”” : 출력할 때 줄바꿈 없이

flush=True : 버퍼를 즉시 비워서 바로 출력

사용자 경험 향상 - ChatGPT처럼 타이핑하는 느낌으로 출력

---

14. RunnableLambda + RunnableBranch - 조건 분기

배경 - 왜 필요한가?

• retriever는 항상 k개의 문서를 반환 (관련 없어도)
• 프롬프트 지시만으로는 완벽하지 않음
• 유사도 점수를 직접 확인해서 관련 없으면 아예 LLM 호출 안 하는 방법

RunnableLambda

• 일반 파이썬 함수를 Runnable로 명시적으로 변환
• (format_docs처럼 자동으로 변환되는 경우도 있지만 복잡한 로직은 명시적으로 사용)

RunnableBranch

• 조건에 따라 다른 체인으로 분기
• 구조: RunnableBranch( (조건1, 실행1), (조건2, 실행2), 기본실행 )

전체 코드

from langchain_core.runnables import RunnableBranch, RunnableLambda

def check_and_prepare(question):
    """유사도 검색 + 점수 확인 + 데이터 준비"""
    results = vectorstore.similarity_search_with_score(question, k=3)
    docs = [doc for doc, _ in results]
    return {
        'question': question,
        'context' : format_docs(docs),
        'score'   : results[0][1]      # 가장 유사한 문서의 점수
    }

safe_chain = (
    RunnableLambda(check_and_prepare)
    | RunnableBranch(
        # 조건: score > 1.5 이면 (관련 문서 없음) → 즉시 "없다" 반환
        (lambda x: x['score'] > 1.5, lambda x: '해당 정보가 없습니다'),
        # 기본: score <= 1.5 이면 → 정상 RAG 처리
        rag_prompt | llm | StrOutputParser()
    )
)

safe_chain.invoke('2002년 월드컵 우승팀은 어디인가요?')
# → '해당 정보가 없습니다'   (score가 1.5 초과)

safe_chain.invoke('재택근무 규정을 알려주세요?')
# → '재택근무는 주 2회까지 가능하다고 규정되어 있습니다.'

FAISS 유사도 점수 이해

• FAISS는 기본적으로 L2 거리(유클리드 거리) 사용
• 점수가 낮을수록 = 더 유사함
• 점수가 높을수록 = 덜 유사함 (거리가 멀다)
• 일반적으로:
  • score < 1.0 : 매우 관련 있음
  • 1.0 ~ 1.5 : 어느정도 관련
  • score > 1.5 : 관련 없음 (문서에 없는 내용)

threshold 값(1.5)은 데이터셋에 따라 조정 필요

---

15. similarity_search_with_score - 유사도 점수 활용

기본 사용법

# 유사도 점수와 함께 검색
results = vectorstore.similarity_search_with_score('재택근무 규정', k=3)
# → [(Document, score), (Document, score), (Document, score)]

# 결과 출력
for doc, score in results:
    print(f"score: {score:.4f}")
    print(f"[{doc.metadata['category']}] {doc.page_content[:50]}")

실제 출력 예시

# '재택근무 규정' 검색 결과
score: 0.9968   [사내규정] 제2조 (근무시간) ...
score: 1.1234   [사내규정] 제3조 (휴가) ...
score: 1.2345   [사내규정] 제1조 (목적) ...

# '2002년 월드컵' 검색 결과 (관련 없음)
score: 1.7xxx   [AI보고서] 1. 개요 ...  ← score가 높음 = 관련 없음

카테고리별 score 테스트

test_queries = [
    {"query": "재택근무 몇 회?"},
    {"query": "스마트홈 초기 설정"},
    {"query": "RAG 기술 동향"},
]

for tq in test_queries:
    results = vectorstore.similarity_search_with_score(tq['query'], k=1)
    top_cat   = results[0][0].metadata['category']
    top_score = results[0][1]
    context   = results[0][0].page_content[:50]
    print(f"{tq['query']} → [{top_cat}] score={top_score:.4f}")

결과:

• “재택근무 몇 회?” → [사내규정] score=0.9943 (매우 관련)
• “스마트홈 초기 설정” → [제품매뉴얼] score=0.8596 (매우 관련)
• “RAG 기술 동향” → [AI보고서] score=낮음 (관련)

---

16. verify_query - threshold 기반 필터링 함수

코드

def verify_query(query, threshold=1.3, k=3):
    """
    query  : 검색 질문
    threshold : 이 점수 초과면 '없음' 반환
    k      : 검색할 문서 수
    """
    results = vectorstore.similarity_search_with_score(query, k=k)

    best_doc   = None
    best_score = float('inf')   # 초기값을 무한대로 (어떤 score보다 크게)

    for doc, score in results:
        if score < best_score:   # 더 낮은(= 더 유사한) score 찾기
            best_score = score
            best_doc   = doc

    if best_score > threshold:
        return '해당 정보가 없습니다'

    return best_doc.page_content

# 테스트
verify_query('2002년 월드컵')
# → '해당 정보가 없습니다'   (best_score > 1.3)

verify_query('재택근무 규정')
# → '제2조 (근무시간)\n1. 기본 근무시간은...'  (best_score < 1.3)

float(‘inf’) 사용 이유

• 초기 best_score를 “어떤 실제 score보다 큰 값”으로 설정해야 첫 번째 비교에서 반드시 업데이트됨
• 0이나 1로 초기화하면 실제 score가 더 클 경우 업데이트 안 됨

RunnableBranch 방식과의 차이

• RunnableBranch 방식: LCEL 파이프라인 내에서 분기 처리
• verify_query 방식 : 순수 파이썬 함수로 처리, 더 직관적

두 방식 모두 같은 목적(관련 없는 질문 필터링)이지만 구현 방식이 다름

---

17. 핵심 개념 총정리

RAG 전체 파이프라인

┌──────────────────────────────────────────────────────────────────────┐
│                         RAG Pipeline                                 │
│                                                                      │
│  텍스트 파일들                                                        │
│      ↓                                                               │
│  Document 객체들 (page_content + metadata)                           │
│      ↓                                                               │
│  OpenAIEmbeddings → 벡터화 → FAISS 저장                              │
│                                                                      │
│  사용자 질문 → retriever.invoke() → 관련 Document k개                │
│                    ↓                                                 │
│              format_docs() → 하나의 문자열                           │
│                    ↓                                                 │
│  ChatPromptTemplate({context, question})                             │
│                    ↓                                                 │
│              ChatOpenAI (LLM)                                        │
│                    ↓                                                 │
│              StrOutputParser() → 최종 답변                           │
└──────────────────────────────────────────────────────────────────────┘

오늘 배운 LCEL Runnable 정리

Runnable	역할	사용 예
RunnablePassthrough	입력을 그대로 통과	“question”: RunnablePassthrough()
RunnableParallel	여러 체인 병렬 실행, 딕셔너리	answer=chain, source=retriever
RunnableLambda	파이썬 함수를 Runnable로	RunnableLambda(check_and_prepare)
RunnableBranch	조건에 따라 분기	(조건, 실행1), 기본실행

FAISS 메서드 정리

메서드	반환값
FAISS.from_documents(docs, emb)	vectorstore 생성
vectorstore.save_local(path)	로컬 저장
FAISS.load_local(path, emb, …)	로컬에서 로드
vectorstore.as_retriever(…)	Retriever 인터페이스 반환
retriever.invoke(query)	[Document, …] 반환
vectorstore.similarity_search_with_score(query, k)	[(Document, score), …] 반환
vectorstore.index.ntotal	저장된 벡터 총 수

---

18. 자주 나오는 실수/주의사항

실수 1: voctorstore (오타)

# 틀린 코드 (강의 중 나온 오타)
docs = voctorstore.similarity_search(question, k=3)  # NameError!

# 올바른 코드
docs = vectorstore.similarity_search(question, k=3)

실수 2: allow_dangerous_deserialization 빠뜨리기

# 틀린 코드 → 에러 발생
loaded_vs = FAISS.load_local('faiss_docs', embeddings)

# 올바른 코드
loaded_vs = FAISS.load_local('faiss_docs', embeddings, allow_dangerous_deserialization=True)

실수 3: RunnablePassthrough 없이 RAG chain 구성

# 틀린 코드 → question이 전달 안 됨
rag_chain = (
    {"context": retriever | format_docs}   # question 키 없음!
    | rag_prompt
    | llm
)

# 올바른 코드
rag_chain = (
    {
        "context" : retriever | format_docs,
        "question": RunnablePassthrough()   # 반드시 필요!
    }
    | rag_prompt
    | llm
    | StrOutputParser()
)

실수 4: format_docs를 거치지 않고 Document 리스트를 바로 프롬프트에 넣기

# 틀린 코드 → context가 Document 객체 리스트가 됨
{"context": retriever, "question": RunnablePassthrough()}

# 올바른 코드
{"context": retriever | format_docs, "question": RunnablePassthrough()}

실수 5: threshold 방향 혼동

# FAISS L2 거리 기준:
# 점수 낮음 = 유사함 (가까움)
# 점수 높음 = 다름 (멀음)

# 따라서 필터링은:
if score > threshold:  # threshold 초과 = 관련 없음
    return "해당 정보가 없습니다"

# 이걸 반대로 쓰면 완전히 반대 동작!

포인트 - RunnableParallel을 쓰는 두 가지 방법

# 방법 1: 명시적 RunnableParallel 클래스 사용
RunnableParallel(answer=rag_chain, source=retriever)

# 방법 2: 딕셔너리 방식 (LCEL에서 자동 변환)
{"answer": rag_chain, "source": retriever}

# 두 방법은 완전히 동일하게 동작

---

[보충] RAG에서 Retriever가 “관련 없는” 문서를 가져오는 이유

retriever는 항상 k개의 문서를 반환한다. “2002년 월드컵”을 검색해도 DB에 있는 문서 중 “가장 유사한” k개를 반환한다. → 실제로는 전혀 관련 없는 문서가 반환됨

이를 확인해보면:

retriever.invoke('2002년 월드컵')
# → [AI보고서 관련 문서들] 이 나옴 (관련 없지만 그나마 가장 유사한 것들)

해결 방법 3가지:

1. 프롬프트 제약 : “문서에 없으면 없다고 답해” (가장 간단, 가장 불확실)
2. RunnableBranch: score 확인 후 분기 (LCEL 파이프라인 내 처리)
3. verify_query : 순수 파이썬 함수로 처리 (가장 직관적)

실무에서는 2번이나 3번 방식을 사용하는 것이 권장됨

---

[보충] LCEL 체인 invoke vs stream vs batch

# 단일 질문 처리
result = chain.invoke("질문")

# 스트리밍 처리 (실시간 출력)
for chunk in chain.stream("질문"):
    print(chunk, end="", flush=True)

# 여러 질문 배치 처리 (병렬)
results = chain.batch(["질문1", "질문2", "질문3"])

invoke : 전체 응답 완성 후 한번에 반환

stream : 토큰 단위 실시간 반환 (UX 좋음)

batch : 여러 입력 병렬 처리 (효율적)

---

[보충] 오늘 배운 내용의 실무 활용

오늘 만든 시스템으로 할 수 있는 것

1. 사내 문서 Q&A 봇
   • 사내 규정, 매뉴얼, 보고서 등을 FAISS에 넣고
   • 직원들이 자연어로 질문하면 관련 내용 답변
2. 고객 지원 챗봇
   • 제품 매뉴얼, FAQ를 벡터스토어에 저장
   • 고객 질문에 자동 답변
3. 문서 기반 리서치 도구
   • 논문, 보고서 등을 인덱싱
   • 관련 정보 빠르게 검색 및 요약

실무에서 발전시킬 부분

• 문서 청킹: 단순 \n\n 분리 → RecursiveCharacterTextSplitter 사용
• 임베딩 모델: text-embedding-3-small → text-embedding-3-large
• 벡터스토어: FAISS → Chroma, Pinecone (대규모 데이터)
• Reranking: 검색 결과 재정렬로 정확도 향상
• Conversation Memory: 이전 대화 기억하는 RAG

---

끝