실습 4 — RetrievalQA 와 LCEL 로 근거 붙인 답의 짜임 보기

import warnings
import zlib

warnings.filterwarnings("ignore", category=DeprecationWarning)

from langchain_community.vectorstores import FAISS
from langchain_core.embeddings import Embeddings


class BigramEmbeddings(Embeddings):
    """글자 두 개 묶음(바이그램)을 세어 숫자 목록을 만드는 연습용 임베딩"""

    def __init__(self, size=512):
        self.size = size

    def _vector(self, text):
        text = text.replace(" ", "").replace("\n", "")
        vec = [0.0] * self.size
        for i in range(len(text) - 1):
            # 글자 두 개를 crc32 로 0~511 사이의 칸 번호로 바꿔 그 칸을 1 올린다
            vec[zlib.crc32(text[i:i + 2].encode("utf-8")) % self.size] += 1.0
        norm = sum(v * v for v in vec) ** 0.5 or 1.0
        return [v / norm for v in vec]

    def embed_documents(self, texts):
        return [self._vector(t) for t in texts]

    def embed_query(self, text):
        return self._vector(text)


# 실습 3이 저장한 조항 단위 색인을 같은 임베딩으로 불러온다(내가 만든 색인이라 pickle 을 허락한다)
db = FAISS.load_local("out_ch06_faiss", BigramEmbeddings(), allow_dangerous_deserialization=True)
# as_retriever: 벡터DB 를 검색기로 바꾼다. search_kwargs 의 k 는 돌려줄 조각 수
retriever = db.as_retriever(search_kwargs={"k": 2})

question = "출발 5일 전에 취소하면 수수료가 얼마예요?"
docs = retriever.invoke(question)
print(type(retriever).__name__, "/ 찾은 조각", len(docs), "개")
for d in docs:
    print(f"  {d.metadata['article']}({d.metadata['page']}쪽)", d.page_content[:45], "…")

from langchain_classic.chains import RetrievalQA
from langchain_core.language_models.fake_chat_models import FakeListChatModel

# 가짜 모델: 무엇을 받든 정해 둔 답을 차례로 돌려준다
fake = FakeListChatModel(responses=["출발 3~6일 전 취소는 상품 금액의 30%를 수수료로 뗍니다."])

qa = RetrievalQA.from_chain_type(
    llm=fake,                        # 답을 쓸 모델
    retriever=retriever,             # 근거를 찾을 검색기
    chain_type="stuff",              # 찾은 조각을 모두 이어 붙여(stuff) 프롬프트 하나에 넣는다
    return_source_documents=True,    # 답과 함께 근거 조각도 돌려준다
)
out = qa.invoke({"query": question})
print("열쇠:", list(out))
print("답:", out["result"])
print("근거:", [d.metadata["article"] for d in out["source_documents"]])

# RetrievalQA 안에 든 기본 프롬프트(영어)의 시스템 지시 첫 두 줄
system_template = qa.combine_documents_chain.llm_chain.prompt.messages[0].prompt.template
print(system_template.split("\n")[:2])

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableLambda, RunnablePassthrough


def format_docs(docs):
    """조각 목록을 '[제5조 · 2쪽] 내용' 줄로 이어 붙인다"""
    return "\n".join(f"[{d.metadata['article']} · {d.metadata['page']}쪽] {d.page_content}" for d in docs)


prompt = ChatPromptTemplate.from_messages([
    ("system",
     "너는 구름다리여행의 약관 안내 직원이다. 아래 [근거]에 적힌 내용으로만 두 문장 이내로 답한다.\n"
     "근거에 없는 내용이면 '약관에서 찾지 못했습니다.'라고만 답한다.\n\n[근거]\n{context}"),
    ("human", "{question}"),
])


def peek(prompt_value):
    """모델이 받을 시스템 메시지를 찍어 보고 그대로 넘긴다"""
    print("--- 모델이 받는 시스템 메시지 ---")
    print(prompt_value.to_messages()[0].content)
    print("---")
    return prompt_value


fake = FakeListChatModel(responses=["출발 3~6일 전 취소는 상품 금액의 30%를 수수료로 뗍니다."])
# 맨 앞 사전: 같은 질문이 context 칸(검색기 → 이어 붙이기)과 question 칸(그대로)으로 동시에 들어간다
chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | RunnableLambda(peek)
    | fake
    | StrOutputParser()
)
print("답:", chain.invoke(question))

import json

from langchain_core.runnables import RunnableParallel

fake = FakeListChatModel(responses=[
    "출발 3~6일 전 취소는 상품 금액의 30%를 수수료로 뗍니다.",
    "국내선은 15kg 짐 한 개가 포함되고, 기내에는 7kg 이하 가방 하나를 들고 탈 수 있습니다.",
])
answer_chain = prompt | fake | StrOutputParser()

# 1) 검색 결과(docs)와 질문을 먼저 사전으로 받고 2) assign 으로 그 사전에 answer 칸을 더한다
rag = RunnableParallel(docs=retriever, question=RunnablePassthrough()).assign(
    answer=lambda x: answer_chain.invoke({"context": format_docs(x["docs"]), "question": x["question"]})
)

log = []
for q in ["출발 5일 전에 취소하면 수수료가 얼마예요?", "짐은 몇 kg까지 실을 수 있나요?"]:
    out = rag.invoke(q)
    # 출처는 모델이 아니라 우리 코드가 metadata 에서 붙인다
    sources = [f"{d.metadata['article']}({d.metadata['page']}쪽)" for d in out["docs"]]
    print("질문:", q)
    print("  답:", out["answer"])
    print("  근거:", ", ".join(sources))
    log.append({"질문": q, "답": out["answer"], "근거": sources})

with open("out_ch06_qa_log.json", "w", encoding="utf-8") as f:
    json.dump(log, f, ensure_ascii=False, indent=2)
print("저장: out_ch06_qa_log.json")