[교육] LangChain & LangGraph 실습 - AI agent 만들기 - (3)

 

 

실습 - 회사 내규 검색 AI agent 만들기 

 

앞의 문서 기반 챗봇 AI agent 만들기에서 추가된 점 : 

- 여러 문서 파일을 기반으로 만들어진다. 

- Reranker을 사용한다. 

 

Reranker란?

벡터스토어의 검색 효율을 평가하기 위한 방법 ( RAG 문서 검색 평가 방법 중 하나 )

1차검색에서 가져온 후보문서들에 대해 점수를 다시 매겨 순서를 재조정하거나, 불필요한 문서를 제거하는 기법

즉, 정답에 가까운 문서를 상위 배치할 수 있어 할루시네이션을 줄일 수 있다. 하지만 구현 난이도가 어렵다는 단점이 있음.

---

LangChain을 사용한 방법 

 

1. 여러 문서일 때 문서 로드 및 전처리 

# PDF 로더 생성
hr_loader = PyPDFLoader(path+filename[0])
security_loader = PyPDFLoader(path+filename[1])
onboard_loader = PyPDFLoader(path+filename[2])
tools_loader = PyPDFLoader(path+filename[3])
culture_loader = PyPDFLoader(path+filename[4])

# 텍스트 스플리터 생성
splitter = RecursiveCharacterTextSplitter(chunk_size=100, 
                                          chunk_overlap=0,
                                          separators=["\n\n"])

# 문서 전처리 함수 생성
# pdf 파일에서 불필요한 공백을 제거하고 문서의 길이를 조정하는 함수 (ex. 의미적으로 바뀌는 부분은 문단바꿈으로 구분)
# 문서에 맞게 전처리 함수 설계해야 함
def cleaning_docs(docs):
    docs = docs.load()
    lens = None
    for idx, doc in enumerate(docs):
        corpus = doc.page_content.replace("\xa0", "").replace("  ", " ").split("\n")
        if lens is None:
            lens = []
            for sentence in corpus:
                lens.append(len(sentence))
            length = sorted(lens)[len(lens)//2]
        else:
            pass

        cleaning_corpus = []
        for sentence in corpus[:-2]: # 마지막 두 문장은 제거 (페이지넘버 제거)
            if len(sentence) >= length:
                cleaning_corpus.append(sentence)
            else:
                cleaning_corpus.append(sentence+"\n\n")   
        docs[idx].page_content = "".join(cleaning_corpus)

    return docs

# 문서 전처리
hr_docs = cleaning_docs(hr_loader)
security_docs = cleaning_docs(security_loader)
onboard_docs = cleaning_docs(onboard_loader)
tools_docs = cleaning_docs(tools_loader)
culture_docs = cleaning_docs(culture_loader)

# 텍스트 스플리터를 이용한 문서 분할
hr_docs = splitter.split_documents(hr_docs)
security_docs = splitter.split_documents(security_docs)
onboard_docs = splitter.split_documents(onboard_docs)
tools_docs = splitter.split_documents(tools_docs)
culture_docs = splitter.split_documents(culture_docs)

# 문서 병합 -> 문서의 양이 크지 않기 때문에 병합해서 사용함 
total_docs = hr_docs+security_docs+onboard_docs+tools_docs+culture_docs

# 통합된 벡터스토어 생성
vector_store = FAISS.from_documents(embedding=embeddings, documents=total_docs)

각각의 파일에 전처리, 스플리터를 적용한 후, (문서의 양이 방대하지 않다면) 분할된 모든 문서를 하나의 리스트로 병합하여 이를 기반으로 단일 벡터스토어를 구축한다.

 

리트리버 vs 리랭커 리트리버 차이

# 그냥 리트리버
retriever = vector_store.as_retriever() 
compressor = LLMChainExtractor.from_llm(llm) 

# 리랭커 역할 리트리버
reranked_retriever = ContextualCompressionRetriever(
    base_retriever=retriever,
    base_compressor=compressor
)

query = "연차는 몇 일 사용할 수 있나요?"

docs = retriever.invoke(query)
docs = reranked_retriever.invoke(query)

 

 

그냥 리트리버를 사용했을 땐 필요없는 문장도 가져오지만 reranker를 사용하면 정답에 가까운 문장 하나를 가져온다.
하지만 버려진 문장들에 대한 정보는 알 수 없다.

 

프롬프트를 만들고 체인을 생성한다.

# 리트리버만 리랭크 리트리버로 변경, 외 기존 체인과 동일 
chain = {
    "context": reranked_retriever | RunnableLambda(format_docs),
    "query": RunnablePassthrough()
} | prompt | llm

 

query = "연차는 몇 일 사용할 수 있나요?"
result = chain.invoke(query)
print(result.content)

 

-> 법정 연차휴가는 1년 15일이 부여되며, 이후 근속연수에 따라 가산됩니다. 따라서 사용 가능한 연차는 근속연수에 따라 달라질 수 있습니다.

 

# 점수 기반 Reranker
model = CrossEncoder("BAAI/bge-reranker-base") 

def rerank(docs, top_n=5):
    pairs = [[query, doc.page_content] for doc in docs]
    scores = model.predict(pairs)
    reranked = sorted(zip(scores, docs), key=lambda x: x[0], reverse=True)
    result = [doc.page_content+f",  Score : {score}" for score, doc in reranked[:top_n]]
    return "\n".join(result)
    
query = "복지가 어떤 것이 있나요?"
docs = retriever.invoke(query) # 검색되는 과정을 보여줌
 
result = rerank(docs) #문서 관련도 점수를 나타냄 (높을수록 관련도가 높다)

# rerank 함수 포함 체인 생성
chain = {
    "context": retriever | RunnableLambda(rerank),
    "query": RunnablePassthrough()
} | prompt | llm

query = "퇴사를 계획중인데 어떻게 하면 되나요?"
response = chain.invoke(query)
print(response.content)

 

reranker를 사용하여 조금 더 정확한 답변을 얻을 수 있다.

 

---

LangGraph 를 이용한 방법 

 

문서로드 - 텍스트 전처리 및 분할 - 벡터스토어 - 리트리버 - LLM 기반 Reranker 설정 까지 랭체인 실습과 동일하게 진행 후

 

state 설정부터 해준다. 

class State(TypedDict):
    query : Annotated[str, "User Question"]
    answer : Annotated[str, "LLM response"]
    document : Annotated[Document, "Retrieve Response"]
    retrieval_type : Annotated[str, "Document Category"]

 

이 실습에서는 5개의 문서를 각각 벡터스토어에 넣고 이후 리트리버 설정도 하나씩 해주었다. 

먼저 질문에 가장 가까운 문서를 선택하고 그 문서에서 reranker를 사용해서 답변의 품질을 높이는 RAG 시스템을 만드는 과정이다.

 

# llm에게 영어로 질문을 하고, 리트리버를 통해 문서를 가져온 후, 그 문서의 카테고리를 확인하는 상태 그래프를 생성합니다.
class RetriverChecker(BaseModel):
    """
    질문의 의도를 파악하고 가까운 문서를 나타내주는 프롬프트 작성
    """

    retrieval_type : Literal["HR", "Security", "Onboard", "Tools", "Culture"] 
    = Field(..., description="""영어 버전""")
# llm에 구조화된 출력 설정
retriever_checker = llm.with_structured_output(RetriverChecker)

# 질문의 의도에 따라 분류를 해줌
result = retriever_checker.invoke("입사 첫 날인데 어떤 것을 해야 하나요?")
print(result.retrieval_type)

def retriever_check(state: State):
    prompt = PromptTemplate.from_template(
    """
    ~질문의 의도에 따라 문서 선택 프롬프트 작성~
    
    질문 : {query}

    """
    )

    chain = prompt | retriever_checker #리트리버 체커가 뱉는 게 구조화된 출력

    result = chain.invoke({"query":state["query"]})

    return {"retrieval_type" : result.retrieval_type}

 

이후 reranker, response 함수를 구현하고 그래프의 노드와 엣지를 설정해준다.

 

- 노드들은 상태 그래프의 각 단계를 나타내며, 엣지는 상태 간의 전환을 정의합니다.
- retriever_check 노드는 사용자의 질문을 받아 리트리버를 통해 문서 카테고리를 확인합니다.
- reranker 노드는 확인된 카테고리에 따라 적절한 리트리버를 사용하여 문서를 검색합니다.
- response 노드는 검색된 문서를 기반으로 사용자의 질문에 대한 답변을 생성합니다.

# 노드와 엣지 구성 
graph_builder.add_node("retriever_check", retriever_check)
graph_builder.add_node("reranker", reranker)
graph_builder.add_node("response", response)

graph_builder.add_edge(START, "retriever_check")
graph_builder.add_edge("retriever_check", "reranker")
graph_builder.add_edge("reranker", "response")
graph_builder.add_edge("response", END);

 

어떤 과정으로 답변을 하는지 확인할 수 있다.