[LLM] LangChain과 RAG을 활용한 간단한 LLM 기반 챗봇 구현하기

LangChain?

LangChain은 대형 언어 모델(LLM)을 효율적으로 활용할 수 있도록 지원하는 프레임워크로, AI 기반 애플리케이션 개발을 보다 체계적이고 유연하게 설계할 수 있도록 돕는다. LangChain은 다양한 LLM과 연동할 수 있으며, 이를 통해 AI 모델을 쉽게 호출하고, 대화형 응답을 생성하거나 데이터 분석 및 자동화 프로세스를 구축하는 데 활용할 수 있다.
- ChatGPT

 

LangChain은 다양한 LLM 모델을 손쉽게 호출하고 사용할 수 있게 지원한다. 특히 OpenAI API를 직접 활용하여 프롬프트를 전달하는 방식보다 LangChain을 활용하면 LLM을 활용하는 측면에서 확장성도 좋고 코드 활용도 훨씬 간편하다.

공부하다 궁금해서 검색해봄

 

실제 코드를 놓고 비교해도 아래와 같이 더 직관적으로 이해하기 쉬운 형태이고, LLM 모델이 가져오는 응답 역시 별다른 파싱이 필요없기 때문에 응답 활용이 훨씬 자유롭다.

 

API 직접 활용 방식의 result는 response로부터 구구절절(지금은 deprecated된 형식) 파싱해와야 하는 반면,

LangChain의 result는 invoke 한번이면 끝난다. 추가적으로, 변수 활용 측면에서도 자유롭기 때문에 본 챗봇 구현에도 LangChain을 최대한 활용하고자 했다.

 

챗봇 설계 목적

청소년 대상 금융 교육용 챗봇을 설계하기 위해 교육자료가 될 수 있는 최신 뉴스/교육자료 등을 벡터DB(chroma db 사용)에 우선 저장해놓고, 챗봇에서 질의문을 던지면

 

1. 쿼리와 저장된 컨텐츠들 별 유형과의 적합성을 파악한다.

2. 가장 적합한 DB에서 자료를 가져와 답변한다.

 

요 방식으로 설계하였다.

 

자료를 크롤링하는 코드는 뉴스/증권보고서/교육자료 별 각각 다른 사람들이 작성했기 때문에 편의상 개별적인 DB를 구축하게 되었고, 통합해서 자료를 가져오는 과정에서 처음에는 쿼리와 연관성있는 모든 자료를 검색해서 가져오게 해야겠다! 했는데 토큰 길이가 폭발해버리는 이슈로 인해.. ㅠㅠ

 

불가피하게 입력쿼리로부터 LLM 모델이 연관성을 파악하게 설계해서 어느 DB를 찾을지 분류부터 하고,

분류된 결과의 DB로부터 자료를 가져와 답할 수 있는 방향성으로 코드를 설계했다.

 

UI 연결하기 이전단계까지 구현을 맡아서, 나는 콘솔로 테스트를 진행했다.

 

코드

모듈화 해달라고 부탁을 받아서 ChatBot class를 만들고 그 안에 모든 작동 함수를 때려넣었다.

 

1. query로부터 연관된 DB유형 분류

from langchain_core.prompts import ChatPromptTemplate

def classify_query(self, query: str) -> str:
    prompt = ChatPromptTemplate.from_messages([
        ('system', 
        """
        다음 질의를 읽고 관련된 정보 유형을 한 단어로 출력하세요.
        가능한 값: "edu", "news", "report", "all", "nothing".
        
        - edu: 경제 용어나 개념 등 배경지식
        - news: 최신 산업 동향
        - report: 기업 분석, 재무 정보 등
        - all/nothing: 범주 구분이 불명확할 때

        "nothing"과 "all"은 최소한으로 사용하고,
        가능하면 edu, news, report 중 하나로 분류하세요.
        """),
        ('user', f"질의: {query}\n답변:")
    ])

    chain = prompt | self.llm
    classification = chain.invoke({"query": query}).content.strip().lower()
    return classification

작동 방식

각각 모으는 자료로부터 얻을 수 있는 정보들의 성격이 다 다르기 때문에, 유형을 분류하는 함수를 정의했다.

물론 OpenAI API를 최대한 활용하기 위해 코사인 유사도 판별법이나 BERT같은 classification 모델을 사용하는 대신 질의의 가변성을 생각하여 가장 확실한 LLM 모델에게 맡겼다.

 

용어 설명 / 배경지식과 같이 개념적인 부분에 대한 성격이 강하면 교육자료에서,

산업 / 기업들의 '최신' 동향을 파악하고자 하는 경우 뉴스기사에서,

그 외 기업들의 대략적인 현재 정보를 파악하고자 하는 경우 증권보고서에서 자료를 가져올 수 있도록 하였다.

 

경제관련이나 딱 하나로 답하기 모호한 쿼리가 들어올 경우 all, 그 외 성격의 쿼리가 들어올 경우

1. 일단 경제 관련 용어인지 생각해보고

2. 영 아니다 싶으면 대답을 하지 않게

하는 규칙을 만들기 위해 nothing도 함께 설계해줬다. 그리고 최대한 두개의 label로 분류하는 일이 없도록 프롬프트에 명시했다.

 

'SK', '삼전' 처럼 기업명만 치면 증권보고서에서 정보를 찾아주고,

'SK 최근 호재', '환율 변동 추이' 이런 '최신/현황'과 같은 흐름적인 키워드를 강조한 쿼리가 들어오면 뉴스기사로 잘 분류해준다.

 

작동 원리

prompt = ChatPromptTemplate.from_messages([
        ('system', 
        """
        ~~~~
        """),
        ('user', f"질의: {query}\n답변:")
    ])

 

ChatPromptTemplate.from_messages : 일반적으로 LangChain을 활용한 대화형 prompt를 작성할 때 많이 쓰는 함수 중 하나이다. 배열형으로 message를 입력할 수 있는데, 각 메시지는 역할(role)과 내용(content)으로 구성된다. 구성만 지킨다면 안에 메시지 작성 방식도 비교적 다양한 편이다.

 

위의 예시처럼 튜플들의 배열 형태로 "system", "user" 메시지를 따로따로 나눠 작성해주는 방법도 있고, 

from langchain.schema.messages import SystemMessage, HumanMessage, AIMessage

prompt_messages = [
    SystemMessage(content=
        f"""
        시스템메시지~~~
        """),
    HumanMessage(content=
        f"""
        인간메시지~~~~
        """)
    ]

SystemMessage, HumanMessage를 불러와 각 역할 별 메시지를 정의하는 방법도 존재한다. 목적에 맞게, 편한대로 사용하면 된다.

 

또한 f-string을 활용하여 프롬프트 자체에 변수 선언을 해주고, 나중에 invoke를 통해 chain을 실행할 때 변수에 대한 값을 바인딩이 가능한데, 이를 통해 프롬프트를 자유롭게 변형하며 활용할 수 있기 때문에 특히 챗봇같은 동적 시스템을 설계할 때 유용하게 쓸 수 있다.

 

chain = prompt | self.llm
classification = chain.invoke({"query": query}).content.strip().lower()

이후 chain이라는 변수에 프롬프트와 미리 선언했던 llm 모델(gpt-4o 사용)을 비트연산자 '|'를 오버라이딩한 체이닝 연산자를 활용하여 LangChain 구조를 할당해준다.

 

여기서 등장하는 체이닝 연산자 '|'는 LCEL(LangChain Expression Language)에서 등장하는 개념으로, 서로 다른 구성요소인 프롬프트와 LLM 모델과 Parser(여기서 사용하진 않았지만 자주 사용함), 아니면 또다른 LLM 모델 등등을 엮어 구성 요소들을 연결하는 역할을 한다.

일반적으로 프롬프트 | 모델 | 파서 와 같이 선언하는데, 정해진 규칙은 아니지만 데이터의 전달 흐름을 표현한 것이기 때문에 보통 위와 같은 방식으로 설계한다.

chain1 = prompt1 | llm | JsonOutputParser()
chain2 =(
     chain1 | prompt2 | llm | StrOutputParser()
)

반드시 저 형식을 지켜야된다는 것이 아닐 뿐, 위와 같이 더 복합적인 방식으로 정의하여 다양한 인공지능 모델들을 확장 사용할 수 있다!

 

prompt 내에 {query} 라는 변수가 있었는데, invoke할때 안에 dictionary형 인자로 넘겨주기만 하면 간단하게 다양한 쿼리에 대해 분류가 가능하다.

 

2. 분류된 질의문과 DB 조회 결과를 기반으로 답변 생성

def generate_answer(self, query: str, classification: str, context: str) -> str:
    # 분류가 nothing일 경우: 경제 여부 판단
    if classification == "nothing":
        prompt = ChatPromptTemplate.from_messages([
            SystemMessage(
                """
                너는 주식 교육 및 종목 추천 정보를 제공하는 청소년 맞춤 경제 교육 챗봇이야.
                사용자의 질의가 경제 관련이면 쉽게 설명해주고, 아니라면 '경제와 관련된 질문을 입력해주세요!' 라고 답변해.
                말투는 존댓말을 유지해줘.
                """
            ),
            HumanMessage(f"질의: {query}\n답변:")
        ])
    else:
    prompt = ChatPromptTemplate.from_messages([
            SystemMessage(
                f"""
                너는 주식 교육 및 추천 종목에 관한 정보를 제공하는 청소년 맞춤 경제 교육 챗봇이야.
                아래는 참고할 수 있는 문서 정보야:\n\n{context}
                
                위 정보를 참고하여 사용자의 질의에 대해 쉽게 설명해줘.
                특히 뉴스 기반이라면 최근 동향을 포함해줘.
                필요하다면 추천 종목도 함께 알려줘.
                말투는 존댓말을 유지해줘.
                """
            ),
            HumanMessage(f"질의: {query}\n답변:")
        ])

    chain = prompt | self.llm
    return chain.invoke({"query": query, "combined_info": context}).content

작동 방식

쿼리 분류 결과가 nothing일 경우, 경제 관련으로 답할 수 있는지 검토하고 알아서 컷할 수 있게 먼저 분기했다.

그 외의 결과에 대해서는, prompt에 벡터DB에서 찾은 결과를 {context} 변수에 담아 참고 자료로 활용하여 RAG(검색 증강 생성)를 가능하게 설계했다.

 

사실 해당 코드를 작성할 때 LangChain에 좀 익숙해지고 싶어서 이전에 쿼리 분류기에서는 시스템과 유저 메시지를 작성할 때 튜플로 정의했고, 여기서는 SystemMessage와 HumanMessage로 작성했다.

뭘 써도 둘다 똑같이 작동하니까 편하게 쓰면 된다.

 

이전과 똑같이 chain으로 엮어내 invoke하는 방식을 사용했기 때문에 코드 작동 원리는 패스한다.

 

실행 결과

챗봇의 목적에 적합한 키워드를 입력했다면, 위와 같이 생각보다 더 괜찮은 답변을 가져오는 것을 확인할 수 있다.

최신 뉴스를 크롤링했기 때문에, 챗지피티에 바로 질문하면 나올 수 없는 내용까지 답으로 가져온다.

수집한 데이터의 퀄리티가 답안에 가장 중요한 영향을 끼치기 때문에, 적절한 데이터를 수집하고 잘 정제하는 것이 제일 중요하다.

 

또한, 앞서 미리 처리해준 코드 덕분에 이상한걸 질문해도 이렇게 먹금도 잘한다.

만족스럽다. 👍

 

전체 코드

import os
import openai
from dotenv import load_dotenv

from langchain_chroma import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain_openai import ChatOpenAI
from langchain.schema.messages import SystemMessage, HumanMessage
from langchain_core.prompts import ChatPromptTemplate


class ChatBot:
    def __init__(self):
        load_dotenv()
        openai.api_key = os.getenv("OPENAI_API_KEY")
        os.environ["OPEN_API_KEY"] = openai.api_key  # 혹시 다른 라이브러리에서 사용할 경우 대비

        self.llm = ChatOpenAI(model_name="gpt-4o", temperature=0.2, max_tokens=1024)

        # 각 Chroma DB 로드
        self.edu_db = self.load_vector_store("chroma_edu_db")
        self.news_db = self.load_vector_store("chroma_news_db")
        self.report_db = self.load_vector_store("chroma_report_db")

    def load_vector_store(self, persist_directory: str) -> Chroma:
        """지정된 디렉토리에서 Chroma 벡터스토어를 로드합니다."""
        return Chroma(
            persist_directory=persist_directory,
            embedding_function=OpenAIEmbeddings()
        )

    def classify_query(self, query: str) -> str:
        """사용자 질의를 edu, news, report 중 하나로 분류합니다."""

        prompt = ChatPromptTemplate.from_messages([
            ('system', 
            """
            다음 질의를 읽고 관련된 정보 유형을 한 단어로 출력하세요.
            가능한 값: "edu", "news", "report", "all", "nothing".
            
            - edu: 경제 용어나 개념 등 배경지식
            - news: 최신 산업 동향
            - report: 기업 분석, 재무 정보 등
            - all/nothing: 범주 구분이 불명확할 때

            "nothing"과 "all"은 최소한으로 사용하고,
            가능하면 edu, news, report 중 하나로 분류하세요.
            """),
            ('user', f"질의: {query}\n답변:")
        ])

        chain = prompt | self.llm
        classification = chain.invoke({"query": query}).content.strip().lower()
        return classification

    def search_documents(self, query: str, classification: str) -> str:
        """분류 결과에 따라 적절한 DB에서 문서를 검색해 내용을 구성합니다."""

        combined_info = ""

        if classification in ["edu", "all"]:
            edu_docs = self.edu_db.similarity_search(query, k=3)
            if edu_docs:
                combined_info += "【유튜브 경제 교육 자료 (청소년용)】\n"
                combined_info += "\n".join(doc.page_content for doc in edu_docs) + "\n\n"

        if classification in ["news", "all"]:
            news_docs = self.news_db.similarity_search(query, k=3)
            if news_docs:
                combined_info += "【경제 관련 뉴스】\n"
                combined_info += "\n".join(doc.page_content for doc in news_docs) + "\n\n"

        if classification in ["report", "all"]:
            report_docs = self.report_db.similarity_search(query, k=3)
            if report_docs:
                combined_info += "【증권 보고서】\n"
                combined_info += "\n".join(doc.page_content for doc in report_docs) + "\n\n"

        return combined_info

    def generate_answer(self, query: str, classification: str, context: str) -> str:
        """분류된 질의와 관련 문서 정보를 바탕으로 GPT로 답변을 생성합니다."""

        # 분류가 nothing일 경우: 경제 여부 판단
        if classification == "nothing":
            prompt = ChatPromptTemplate.from_messages([
                SystemMessage(
                    """
                    너는 주식 교육 및 종목 추천 정보를 제공하는 청소년 맞춤 경제 교육 챗봇이야.
                    사용자의 질의가 경제 관련이면 쉽게 설명해주고, 아니라면 '경제와 관련된 질문을 입력해주세요!' 라고 답변해.
                    말투는 존댓말을 유지해줘.
                    """
                ),
                HumanMessage(f"질의: {query}\n답변:")
            ])
        else:
            prompt = ChatPromptTemplate.from_messages([
                SystemMessage(
                    f"""
                    너는 주식 교육 및 추천 종목에 관한 정보를 제공하는 청소년 맞춤 경제 교육 챗봇이야.
                    아래는 참고할 수 있는 문서 정보야:\n\n{context}
                    
                    위 정보를 참고하여 사용자의 질의에 대해 쉽게 설명해줘.
                    특히 뉴스 기반이라면 최근 동향을 포함해줘.
                    필요하다면 추천 종목도 함께 알려줘.
                    말투는 존댓말을 유지해줘.
                    """
                ),
                HumanMessage(f"질의: {query}\n답변:")
            ])

        chain = prompt | self.llm
        return chain.invoke({"query": query, "combined_info": context}).content

    def run_query(self, query: str) -> str:
        """전체 질의 처리 프로세스"""
        classification = self.classify_query(query)
        print(f"\n[질의 분류 결과]: {classification}")

        context = self.search_documents(query, classification)
        return self.generate_answer(query, classification, context)

    def input_query(self, user_query: str = ""):
        print("\n[사용자 질의]:", user_query)
        response = self.run_query(user_query)
        print("\n[챗봇 응답]:\n", response)


if __name__ == "__main__":
    chatbot = ChatBot()
    while True:
        user_input = input("\n질의를 입력하세요 (종료하려면 'quit' 입력): ")
        if user_input.strip().lower() == 'quit':
            print("챗봇을 종료합니다.")
            break
        chatbot.input_query(user_input)