벡터 리트리버와 RAG 평가 - BM25, 앙상블, LangGraph, 환각 검증

벡터 리트리버(Vector Retriever) 는 사용자 질의(Query) 를 임베딩 벡터로 바꾼 뒤, 벡터 DB에 저장된 청크 벡터와 유사도를 계산해 가장 관련 높은 결과를 찾는 구성 요소다. 키워드 일치가 아니라 의미적 유사성으로 고르기 때문에 “의미를 이해하는 검색기”에 가깝다. 코사인 유사도·내적·유클리드 거리 등을 쓰고, 검색 결과는 LLM과 붙여 RAG에서 답변 품질을 올리는 데 쓰인다.

---

1. Chroma에 넣고 similarity_search

앞 글에서 만든 docs(또는 semantic_docs로 바꿔 넣어도 된다)를 Chroma에 적재한다.

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

# large: 고품질(정확도 높음), 벡터 차원: 3072
# small: 벡터 차원: 1536
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=embeddings,
    persist_directory="./chroma_db",
)

# 질문을 벡터로 변환 → DB 문서와 유사도 비교 → 가장 비슷한 chunk k개
query = "개인 정보 보호와 관련하여 엔트로픽의 규제 사항을 우선 순위 높은 것부터 5가지를 정리하여 나열하시요"
results = vectorstore.similarity_search(query, k=3)
print(results)

print(results[0].page_content)
print(results[1].page_content)
print(results[2].page_content)

---

2. as_retriever

similarity_search 대신 리트리버 인터페이스로 쓰는 패턴이다.

vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
vector_result = vector_retriever.invoke(query)

print(vector_result[0].page_content)
print(vector_result[1].page_content)
print(vector_result[2].page_content)

---

3. BM25 리트리버

BM25는 질의와 문서의 키워드 일치 정도를 TF·IDF·문서 길이 등으로 점수화한다. 임베딩 없이 텍스트만으로 검색하므로 빠르고 비용이 거의 없지만, 의미 이해는 벡터 검색보다 약할 수 있다.

from langchain_community.retrievers import BM25Retriever

bm25_retriever = BM25Retriever.from_documents(docs)
bm25_retriever.k = 3
bm25_result = bm25_retriever.invoke(query)

print(bm25_result[0].page_content)
print(bm25_result[1].page_content)
print(bm25_result[2].page_content)

---

4. 앙상블 리트리버

앙상블 리트리버는 BM25와 벡터 리트리버를 가중치로 섞어, 키워드·의미 각각의 강점을 함께 쓴다.

from langchain_classic.retrievers import EnsembleRetriever

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.6, 0.4],  # 키워드 0.6, 의미 0.4
)

query = "AI 안전 등급(AI Safety Level)"
ensemble_result = ensemble_retriever.invoke(query)
bm25_result = bm25_retriever.invoke(query)
vector_result = vector_retriever.invoke(query)

print("[Ensemble Retriever]")
for doc in ensemble_result:
    print(doc.page_content)
    print()

print("[BM25 Retriever]")
for doc in bm25_result:
    print(doc.page_content)
    print()

print("[Vector Retriever]")
for doc in vector_result:
    print(doc.page_content)
    print()

---

5. LangGraph: 앙상블 검색 → 답변

MessagesState를 확장해 context를 두고, 검색 노드에서 앙상블 결과를 넣은 뒤 답변 노드에서 LLM을 호출한다.

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

class State(MessagesState):
    context: str

graph_builder = StateGraph(State)

def retriever(state: State):
    print("---------- RETRIEVER ----------")
    query_text = state["messages"][-1].content
    retrieved_docs = ensemble_retriever.invoke(query_text)
    context = retrieved_docs[0].page_content
    print("[CONTEXT]\n", context)
    return {"context": context}

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)  # 사용 중인 모델명으로 교체
prompt = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            """당신은 검색된 문서를 바탕으로 질문에 답하는 도우미입니다.
반드시 한국어로 답변하세요.
모르는 내용은 억지로 추측하지 말고 모른다고 답하세요.
[검색 문맥]
{context}
""",
        ),
        ("human", "{question}"),
    ],
)

def answer(state: State):
    print("---------- ANSWER ----------")
    query_text = state["messages"][-1].content
    context = state["context"]
    chain = prompt | llm
    response = chain.invoke({"context": context, "question": query_text})
    return {"messages": [response]}

graph_builder.add_node("retriever", retriever)
graph_builder.add_node("answer", answer)
graph_builder.add_edge(START, "retriever")
graph_builder.add_edge("retriever", "answer")
graph_builder.add_edge("answer", END)

graph = graph_builder.compile()

response = graph.invoke(
    {
        "messages": [HumanMessage(content="레스토랑 치폴레는 대화형 AI를 활용했더니 어떻게 되었어?")]
    }
)

for msg in response["messages"]:
    msg.pretty_print()

---

6. 환각 여부를 평가하는 RAG

내부 PDF만 근거로 답하게 하면서, 검색 필요 여부 → 검색 → 문서 관련성 → 답변 생성 → 근거성(grounding) → 답변 충족성을 단계적으로 평가하고, 실패 시 질의 재작성·재생성 루프를 도는 그래프다.

6-1. 준비

from langchain_community.document_loaders import PyPDFLoader
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_experimental.text_splitter import SemanticChunker

PDF_PATH = "/content/AI브리프_3월_260303.pdf"
PERSIST_DIR = "./chroma_ai_brief"
LLM_MODEL = "gpt-4o-mini"
EMBEDDING_MODEL = "text-embedding-3-small"

TOP_K = 3
MAX_REWRITE = 2
MAX_REGENERATE = 2

loader = PyPDFLoader(PDF_PATH)
pages = loader.load()

embeddings = OpenAIEmbeddings(model=EMBEDDING_MODEL)
text_splitter = SemanticChunker(embeddings)
docs = text_splitter.split_documents(pages)

print(f"원본 페이지 수: {len(pages)}")
print(f"청킹 후 문서 수: {len(docs)}")

vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=embeddings,
    persist_directory=PERSIST_DIR,
)
retriever = vectorstore.as_retriever(search_kwargs={"k": TOP_K})
llm = ChatOpenAI(model=LLM_MODEL, temperature=0)

RAG 한 사이클을 한 줄로 쓰면: 질문 수신 → 관련 문서 검색 → 결과를 LLM에 전달 → 답변 생성.

6-2. 상태·평가 스키마(요지)

• State: question, rewritten_question, documents, context, generation, rewrite_count, regenerate_count, relevance_passed, grounded_passed, answer_passed, need_retrieval, route_label, final_status 등
• 구조화 출력: RetrievalNeedGrade, RelevanceGrade, GroundingGrade, AnswerGrade처럼 yes/no만 받는 Pydantic 모델을 두고 llm.with_structured_output로 분기한다.

6-3. 노드와 분기(노트에 적어 둔 이름 그대로)

• decide_need_retrieval: 내부 PDF 검색이 필요한지 yes/no
• direct_answer: 검색 없이 일반 답변
• retrieve: rewritten_question으로 리트리버 호출, context 조합
• grade_documents: 검색 문서가 질문과 관련 있는지
• transform_query: 검색이 잘 되도록 질문 재작성, rewrite_count 증가
• generate: 컨텍스트 기반 답변 생성
• grade_grounding: 답이 문서에 근거하는지
• grade_answer: 답이 질문을 푸는지
• regenerate: 근거 실패 시 더 보수적으로 재생성, regenerate_count 증가
• give_up / fail_grounding / finalize_success: 종료 처리

조건 함수 예: 관련성 없고 재작성 횟수 초과면 give_up, 근거 없고 재생성 횟수 초과면 fail_grounding, 근거·답변 모두 통과면 finalize_success 후 종료.

전체 StateGraph 연결은 노트에 적은 대로 START → decide_need_retrieval → (retrieve | direct_answer) → … → END 형태로 이으면 된다. 구현 시에는 상태 키 이름을 한 가지 규칙으로 통일하는 것이 중요하다(예: rewritten_question 오타, response.content vs 잘못된 필드 참조 등).

6-4. 테스트 실행 뼈대

from langchain_core.messages import HumanMessage

def run_rag_test(question: str):
    print("\n" + "-" * 100)
    print("질문: ", question)
    print("-" * 100)

    inputs = {
        "messages": [HumanMessage(content=question)],
        "question": question,
        "rewritten_question": question,
        "documents": [],
        "context": "",
        "generation": "",
        "rewrite_count": 0,
        "regenerate_count": 0,
        "relevance_passed": False,
        "grounded_passed": False,
        "answer_passed": False,
        "need_retrieval": False,
        "route_label": "",
        "final_status": "",
    }

    result = graph.invoke(inputs)

    print("\n[최종 답변]")
    print(result.get("generation", ""))
    print("\n[평가 결과]")
    print("need_retrieval :", result.get("need_retrieval"))
    print("relevance_passed :", result.get("relevance_passed"))
    print("grounded_passed :", result.get("grounded_passed"))
    print("answer_passed :", result.get("answer_passed"))
    print("rewrite_count :", result.get("rewrite_count"))
    print("regenerate_count :", result.get("regenerate_count"))
    print("final_status :", result.get("final_status"))
    print("route_label :", result.get("route_label"))
    return result

노트에 묶어 둔 테스트 카테고리는 대략 다음과 같다.

묶음	의도
검색 불필요	인사, 일반 상식, 문서와 무관한 질문
검색 필요	첨부 PDF 기반 질문
질의 재작성	짧고 애매한 질문
답변 해결성	구체적 질문에 핵심이 맞는지
근거 스트레스	문서에 없을 법한 세부까지 요구

run_test_suite로 카테고리별 질문 리스트를 돌리면, 어디서 재작성·재생성 루프가 도는지 로그로 보기 좋다.

---

마치며

• 벡터 리트리버는 의미 기반 상위 k개 청크를 가져오는 표준이고, Chroma는 실습·프로토타입에 잘 맞는다.
• BM25와 벡터를 앙상블하면 키워드 누락·의미 누락을 동시에 줄이려는 설계가 된다.
• LangGraph로 검색·생성·다단계 평가를 노드로 쪼개면 환각 리스크를 줄이는 RAG를 구조적으로 다루기 쉽다.
• 전체 그래프 소스는 내 노트 원본을 그대로 두고, 여기서는 흐름과 테스트 관점만 옮겼다. 필드명·LLM 응답 필드(content)는 실행 전에 한번씩 점검하는 것이 좋다.

---

참고

• 벡터 데이터베이스