728x90
👉RAG와 LangChain을 활용한 Chatbot 구현(1) 게시물 보러가기
지난 게시물에서 RAG의 개념, RAG를 사용했을 때의 장점, RAG 구현을 위한 프레임워크 langchain의 활용법을 알아보고 질문에 간단하게 답변할 수 있는 RAG를 구축해보았다.
지난 시간에 구축했던 RAG를 좀 더 보강하여 답변 신뢰도와 품질을 좀 더 높여보자.
Chat history 추가하기
RAG에서 대화 기록(chat history)를 사용해서 이전의 문맥(context)를 토대로 더 자연스럽고 일관된 답변을 생성하는 것이 가능하다. chat history를 사용하지 않을 경우 아래와 같이 비슷한 맥락의 질문임에도 불구하고 답변을 제대로 하지 못한다는 단점이 있다.
"1루 주자와 2루 주자가 동시에 베이스를 훔치는 것을 뭐라고 하나요?"와 "그럼 한 명이 하는 건?"은 사람이 보기에 비슷한 문맥의 문장임에 AI 모델에서 이를 인지하지 못하는 것을 알 수 있다. chat history를 제공하여 이전의 대화 기록을 참고하도록 하여 AI에게 추론 능력을 부여하 고 적은 정보로 정확한 답변을 제공받고 자연스러운 대화 또한 이어나갈 수 있다.
기존의 RAG에 chat history를 적용해보자. 먼저 이전의 대화 내용을 chat history로 저장하는 것부터 시작해보자. retriever는 이전의 retriever를 활용하여 chat history를 토대로 문맥 흐름을 계속 유지할 수 있는 history_aware_retriever를 생성한다(create_history_aware_retriever를 사용하는데 langchain 0.2v의 공식문서에서 사용법 확인이 가능하다). history_aware_retriever를 통해 채팅 기록을 기반으로 검색이 가능하고, 채팅 기록은 key값인 session_id와 value 값인 채팅기록(ChatMessageHistory)로 이루어진 해시 테이블 구조의 chat_storage로 관리한다.
from langchain_core.prompts import MessagesPlaceholder
from langchain.chains import create_history_aware_retriever
chat_storage = {}
def get_llm():
llm = ChatUpstage()
return llm
# chat history를 반영하여 질문을 재구성하는 시스템 설정
def get_history_retriever():
llm = get_llm()
retriever = get_retriever()
# chat history를 반영하여 질문을 재구성하는 시스템 프롬프트
contextualize_q_system_prompt = (
"Given a chat history and the latest user question "
"which might reference context in the chat history, "
"formulate a standalone question which can be understood "
"without the chat history. Do NOT answer the question, "
"just reformulate it if needed and otherwise return it as is."
)
# 채팅 기록을 반영하여 QA를 위한 프롬프트
contextualize_q_prompt = ChatPromptTemplate.from_messages(
[
("system", contextualize_q_system_prompt),
MessagesPlaceholder("chat_history"), # 메시지 리스트를 전달
("human", "{input}")
]
)
history_aware_retriever = create_history_aware_retriever(llm, retriever, contextualize_q_prompt)
return history_aware_retriever
채팅 기록을 반영하는 history_retriever를 구현했으면 질답을 위한 QA 체인을 추가하고 RAG 체인을 완성해보자.
주의해야 할 점은 채팅 기록을 기반으로 데이터를 불러오는 history_retriever가 추가 되었기 때문에 RAG 체인 내부에서 chat_storage에서 채팅 기록을 불러오고 저장하는 로직 또한 추가되어야 한다. 해당 기능을 담당하는 부분이 get_session_history(세션에 대한 채팅 기록이 있으면 기록을 반환하고 없으면 저장 후 반환)이다.
chat history를 반영하는 rag chain을 구현하기 위해서는 RunnableWithMessageHistory 클래스를 활용한다.
✅RunnableWithMessageHistory:
대화 히스토리 데이터를 자동으로 관리하여 Runnable 컴포넌트를 생성하고 실행할 수 있도록 돕는 클래스. 구현을 위해 아래의 총 5개의 인자를 활용한다.
1. runnable: 받은 인자를 기반으로 실행할 체인
2. get_session_history: 세션ID 키 값에 해당하는 채팅 기록 데이터
3. input_messages_key: 유저의 입력이 포함되는 키 값
4. history_messages_key: 유저의 채팅 기록을 저장할 키 값
5. output_messages_key: 출력 결과를 저장할 키 값
# 세션 기록을 가져오기 위한 함수 설정
def get_session_history(session_id: str) -> BaseChatMessageHistory:
# 세션이 없으면 해당 세션에 대한 채팅기록 생성
if session_id not in chat_storage:
chat_storage[session_id] = ChatMessageHistory()
return chat_storage[session_id]
# RAG 체인을 생성
def get_rag_chain():
llm = get_llm()
# 모델의 역할을 정의해주기 위한 시스템 프롬프트
system_prompt = ("""
당신은 최고의 KBO 리그 야구 규칙 전문가입니다.
아래의 문서를 참고해서 사용자의 질문에 대한 답변을 해주세요.
답변 시에 출처에 대해서도 명확하게 밝혀주시고, 사용자가 이해하기 쉽게 설명해주세요.
답변의 길이는 2-3줄 정도로 제한해주세요.
\n\n
문서: {context}
""")
# 채팅 기록 반영하여 QA를 위한 프롬프트
qa_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
MessagesPlaceholder("chat_history"), # 메시지 리스트(채팅 기록)를 전달
("human", "{input}")
]
)
history_aware_retriever = get_history_retriever() # 기록 기반 검색을 위한 retriever
qa_chain = create_stuff_documents_chain(llm, qa_prompt) # 모델에게 검색한 문서 리스트를 전달
rag_chain = create_retrieval_chain(history_aware_retriever, qa_chain) # RAG 체인 생성
conversational_rag_chain = RunnableWithMessageHistory(
rag_chain, # 실행할 RAG 체인이나 LLM 체인
get_session_history, # 세션_ID 기반의 채팅 기록
input_messages_key="input", # 유저의 입력이 담긴 키
history_messages_key="chat_history", # 채팅 기록을 전달할 키
output_messages_key="answer" # 출력 결과를 저장할 키
).pick("answer") # AI의 응답만 선택해서 반환
# 채팅 history를 포함한 retriever를 활용하여 rag_chain 반환
return conversational_rag_chain이후에 rag_chain을 invoke하고 앞서 작성한 모든 체인을 동작시키기 위한 get_ai_response함수를 아래와 작성해보자. 앞서 작성한 filter_chain, dictionary_chain, rag_chain을 체이닝해서 파이프라인을 만들어주고 해당 파이프라인을 invoke 한다. 이 때 filter_chain에서 쿼리를 "question" 키 값으로 처리하기 때문에 question에 사용자 질문을 저장해서 넘겨주고 conversational_rag_chain 동작을 위해서는 session_id가 필수이기 때문에 config의 configurable 값을 통해 session_id를 넘겨주도록 하자.
def get_ai_response():
filter_chain = get_filter()
dictionary_chain = get_dictionary_chain()
rag_chain = get_rag_chain()
baseball_chain = filter_chain | {"input": dictionary_chain} | rag_chain
ai_response = baseball_chain.invoke(
{"question": "야구에서 주자 한명이 베이스를 훔치는 것을 뭐라고 하나요?"},
config={"configurable": {"session_id": "abc123"}}
)
return ai_response지금까지 작성한 코드를 하나의 .py 파일로 옮기면 아래와 같다. 아래의 llm.py 파일을 실행시켜보자.
# llm.py
from langchain_upstage import ChatUpstage, UpstageEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
chat_storage = {}
def get_llm():
llm = ChatUpstage()
return llm
# 질문이 주제와 관련이 있는지 확인하는 필터 설정
def get_filter():
llm = get_llm()
filter_prompt = ChatPromptTemplate.from_template(f"""
유저의 질문이 야구와 관련된 질문인지 확인하고, 야구와 관련된 질문일 경우에는 "YES" 라는 답변을, 그렇지 않은 경우에는 "NO" 라는 답변을 리턴해주세요.
질문: {{question}}
""")
filtering_chain = filter_prompt | llm | StrOutputParser()
return filtering_chain
# 사전을 참고하여 유저의 질문을 재구성하는 시스템 설정
def get_dictionary_chain():
llm = get_llm()
dictionary = [
"베이스에 있는 선수를 나타내는 표현 -> 주자",
"타석에 있는 선수를 나타내는 표현 -> 타자",
"마운드 위에서 공을 던지는 선수를 나타내는 표현 -> 투수",
"뭐라고 하나요?, 뭐라고 부르나요(무엇이냐고 물을 경우) -> 뜻하는 용어가 무엇인가요?"
]
dictionary_prompt = ChatPromptTemplate.from_template(f"""
사용자의 질문을 확인하고 사전을 참고하여 사용자의 질문을 변경해주세요.
만약 변경할 필요가 없다고 판단되면 변경하지 않고 질문을 그대로 리턴해주세요.
사전: {dictionary}
질문: {{question}}
""")
dictionary_chain = dictionary_prompt | llm | StrOutputParser()
return dictionary_chain
# 문서를 불러오기 위한 retriever 설정
def get_retriever():
index_name = "baseball-rules-index"
# 데이터를 벡터화할 임베딩 모델 설정
embedding = UpstageEmbeddings(model="solar-embedding-1-large")
database = PineconeVectorStore.from_existing_index(index_name=index_name, embedding=embedding)
retriever = database.as_retriever(search_kwargs={"k": 4})
return retriever
# chat history를 반영하여 질문을 재구성하는 시스템 설정
def get_history_retriever():
llm = get_llm()
retriever = get_retriever()
# chat history를 반영하여 질문을 재구성하는 시스템 프롬프트
contextualize_q_system_prompt = (
"Given a chat history and the latest user question "
"which might reference context in the chat history, "
"formulate a standalone question which can be understood "
"without the chat history. Do NOT answer the question, "
"just reformulate it if needed and otherwise return it as is."
)
# 채팅 기록을 반영하여 QA를 위한 프롬프트
contextualize_q_prompt = ChatPromptTemplate.from_messages(
[
("system", contextualize_q_system_prompt),
MessagesPlaceholder("chat_history"), # 메시지 리스트를 전달
("human", "{input}")
]
)
history_aware_retriever = create_history_aware_retriever(llm, retriever, contextualize_q_prompt)
return history_aware_retriever
# 세션 기록을 가져오기 위한 함수 설정
def get_session_history(session_id: str) -> BaseChatMessageHistory:
# 세션이 없으면 해당 세션에 대한 채팅기록 생성
if session_id not in chat_storage:
chat_storage[session_id] = ChatMessageHistory()
return chat_storage[session_id]
# RAG 체인을 생성
def get_rag_chain():
llm = get_llm()
# 모델의 역할을 정의해주기 위한 시스템 프롬프트
system_prompt = ("""
당신은 최고의 KBO 리그 야구 규칙 전문가입니다.
아래의 문서를 참고해서 사용자의 질문에 대한 답변을 해주세요.
답변 시에 출처에 대해서도 명확하게 밝혀주시고, 사용자가 이해하기 쉽게 설명해주세요.
답변의 길이는 2-3줄 정도로 제한해주세요.
\n\n
문서: {context}
""")
# 채팅 기록 반영하여 QA를 위한 프롬프트
qa_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
MessagesPlaceholder("chat_history"), # 메시지 리스트(채팅 기록)를 전달
("human", "{input}")
]
)
history_aware_retriever = get_history_retriever() # 기록 기반 검색을 위한 retriever
qa_chain = create_stuff_documents_chain(llm, qa_prompt) # 모델에게 검색한 문서 리스트를 전달
rag_chain = create_retrieval_chain(history_aware_retriever, qa_chain) # RAG 체인 생성
conversational_rag_chain = RunnableWithMessageHistory(
rag_chain, # 실행할 RAG 체인이나 LLM 체인
get_session_history, # 세션_ID 기반의 채팅 기록
input_messages_key="input", # 유저의 입력이 담긴 키
history_messages_key="chat_history", # 채팅 기록을 전달할 키
output_messages_key="answer" # 출력 결과를 저장할 키
).pick("answer")
# 채팅 history를 포함한 retriever를 활용하여 rag_chain 반환
return conversational_rag_chain
def get_ai_response():
filter_chain = get_filter()
dictionary_chain = get_dictionary_chain()
rag_chain = get_rag_chain()
baseball_chain = filter_chain | {"input": dictionary_chain} | rag_chain
ai_response = baseball_chain.invoke(
{"question": "야구에서 주자 한명이 베이스를 훔치는 것을 뭐라고 하나요?"},
config={"configurable": {"session_id": "abc123"}}
)
return ai_response
ai_response = get_ai_response()
print(ai_response)
문제가 생겼다. 기존 파이프라인의 흐름을 정리해보자면 아래와 같은데, dictionary_chain의 결과값인 input을 보면 질문이 없거나 명확하지 않다는 메시지를 확인할 수 있다. filter chain을 통해 입력된 query의 결과물인 YES 나 NO 가 dictionary chain에 입력된다. dictionary chain에서는 "YES" 나 "NO" 쿼리를 dictionary를 기반으로 재구성해야 하는데, 당연히 불가능하다. 따라서 "질문이 없거나 잘못된 형식인 것 같습니다...."와 같은 답변을 출력하고 이 답변을 rag chain에 넘겨주는 것이다. 즉 filter_chain에서 dictionary_chain으로 데이터를 넘기는 과정에 문제가 있는 것이다.
사실 생각해보면 chat history를 활용하는 시점에서 filter chain은 필요가 없고 사용했을 때에 오히려 문제가 될 수 있다.
1️⃣첫 번째 질문: "야구에서 주자 한명이 도루하는 것을 뭐라고 해?"
2️⃣두 번째 질문: "그럼 2명이 하는건?"
두 질문이 서로 같은 문맥의 질문임에도 불구하고 filter chain(가장 먼저 동작하는 체인, history retriever보다 먼저 동작한다)에서 두 번째 질문을 야구 규칙과 연관성이 없는 질문으로 판단하게 된다. 따라서 filter chain을 제거하고 해당 기능을 rag chain에 위임하는 것도 나쁘지 않은 선택이다. filter chain을 삭제하고 rag chain에서 필터링이 가능하도록 프롬프트를 수정하자.
# RAG 체인을 생성
def get_rag_chain():
llm = get_llm()
# 모델의 역할을 정의해주기 위한 시스템 프롬프트
system_prompt = ("""
당신은 최고의 KBO 리그 야구 규칙 전문가입니다.
아래의 문서를 참고해서 사용자의 질문에 대한 답변을 해주세요.
먄약에 야구 규칙과 관련이 없다면 "🚫야구 규칙에 대한 질문만 답변이 가능합니다."라고 답변해주세요.
만약에 답변할 수 없다면 모른다고 답변해주세요.
답변 시에 출처에 대해서도 명확하게 밝혀주시고, 사용자가 이해하기 쉽게 설명해주세요.
답변의 길이는 2-3줄 정도로 제한해주세요.
\n\n
문서: {context}
""")
# 채팅 기록 반영하여 QA를 위한 프롬프트
qa_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
MessagesPlaceholder("chat_history"), # 메시지 리스트(채팅 기록)를 전달
("human", "{input}")
]
)
history_aware_retriever = get_history_retriever() # 기록 기반 검색을 위한 retriever
qa_chain = create_stuff_documents_chain(llm, qa_prompt) # 모델에게 검색한 문서 리스트를 전달
rag_chain = create_retrieval_chain(history_aware_retriever, qa_chain) # RAG 체인 생성
conversational_rag_chain = RunnableWithMessageHistory(
rag_chain, # 실행할 RAG 체인이나 LLM 체인
get_session_history, # 세션_ID 기반의 채팅 기록
input_messages_key="input", # 유저의 입력이 담긴 키
history_messages_key="chat_history", # 채팅 기록을 전달할 키
output_messages_key="answer" # 출력 결과를 저장할 키
).pick("answer")
# 채팅 history를 포함한 retriever를 활용하여 rag_chain 반환
return conversational_rag_chain
def get_ai_response():
dictionary_chain = get_dictionary_chain()
rag_chain = get_rag_chain()
baseball_chain = {"input": dictionary_chain} | rag_chain
ai_response = baseball_chain.invoke(
{"question": "야구에서 주자 한명이 베이스를 훔치는 것을 뭐라고 하나요?"},
config={"configurable": {"session_id": "abc123"}}
)
return ai_response두 질문에 대해 답변을 잘 출력하는 것을 확인할 수 있다.
🚨여러 번 해보니 문제가 생겼는데, 첫 번째 질문에서는 제대로 답변을 하지만 이전 채팅 기록을 토대로 문맥을 유추해서 답변을 해야 하는 "그럼 두 명이 동시에 하는 건 뭐라고 해?" 같은 질문에는 엉뚱한 답변을 하거나 제대로 답하지 못하는 경우도 종종 있다.
이러한 문제가 생기는 이유는 dictionary chain에서 "두 명이 동시에 하는건 뭐라고 해?"와 같은 쿼리를 보강을 한답시고 프롬프트를 통해 변경시키기 때문이다. 쿼리가 모호하더라도 dictionary chain 이후의 history retriever에서 chat history를 참고하여 문맥 파악이 가능하기 때문에 dictionary chain에서 약간의 용어 수정을 통해 쿼리를 보강하되 아예 모호한 질문에 대해서는 수정이 발생하지 않도록 해야한다.
모호한 질문에 대해서는 문맥 파악을 통해 쿼리 수정을 하도록 dictionary chain의 프롬프트를 조금 수정해보자.
# 사전을 참고하여 유저의 질문을 재구성하는 시스템 설정
def get_dictionary_chain():
llm = get_llm()
dictionary = [
"베이스에 있는 선수를 나타내는 표현 -> 주자",
"타석에 있는 선수를 나타내는 표현 -> 타자",
"마운드 위에서 공을 던지는 선수를 나타내는 표현 -> 투수"
]
dictionary_prompt = ChatPromptTemplate.from_template(f"""
사용자의 질문을 확인하고 사전을 참고하여 사용자의 질문을 변경해주세요.
만약 변경할 필요가 없다고 판단되면 변경하지 않고 질문을 그대로 리턴해주세요.
질문이나 주어가 모호할 경우 history retriever에서 history chat을 참고하여 유추할 수 있도록 질문을 변경하지 않고 그대로 리턴해주세요.
사전: {dictionary}
질문: {{question}}
""")
dictionary_chain = dictionary_prompt | llm | StrOutputParser()
return dictionary_chain결과가 잘 나오는 것을 확인 가능하다.
Few-shot 추가하기
코드를 여러 번 돌려보면 답변이 일관적이지 않다는 것을 확인할 수 있다("더블 스틸", ~~하는 것을 "더블스틸"이라고 합니다... 등). 사용자 입장에서도 일관적인 형태를 유지하는 답변을 보는 것이 편하니 Few-shot을 추가해보자(shot이 없는 것을 Zero-shot, 하나인 것을 One-shot, Single-shot으로 부르기도 한다).
Few-shot이란 쉽게 말해 예시 질문과 답변을 주고 이런 식으로 답변하면 돼! 라고 모델에게 미리 가르치는 것이다. 모델이 새로운 질문을 받을 때 미리 학습한 예시를 참고하도록 하는 기법이다.
느낀 바로는 아래의 장점이 있다.
1. hallucination(환각) 현상이 발생할 가능성을 낮출 수 있다. 주제를 벗어나는 키워드에 대해서는 답변하지 않도록 예시를 통해 학습시킬 수 있다.
2. 답변 양식을 통일 시킬 수 있다. 답변 양식에 맞춰 일관된 답변 형식을 유저에게 제공하여 가독성을 높일 수 있다.
3. 그 밖의 자잘한 모델의 오작동 방지(문맥 파악 오류 등)와 더 정확한 답변 생성
이전에 진행했던 프로젝트 Cquis에서도 OPENAI 모델 프롬프트를 작성했었는데 그 때는 Few-shot의 개념도 잘 모르는 채로 나름대로 hallucination을 잡겠다고 노력했었다.
few shot은 최종 qa_prompt에서 참고하는 것이기 때문에 ChatPromptTemplate.from_messages의 인자로 추가해주면 프롬프트 실행 시에 추가한 few shot 관련 프롬프트를 참고해서 답변을 생성한다. example_prompt를 통해 미리 대화 형식의 틀을 제공해주고 answer_examples를 통해 질답 예시를 미리 학습시킨다.
# RAG 체인을 생성
def get_rag_chain():
llm = get_llm()
# few-shot을 위한 프롬프트 설정
example_prompt = ChatPromptTemplate.from_messages(
[
("human", "{input}"),
("ai", "{answer}")
]
)
# few-shot을 위한 프롬프트 설정
few_shot_prompt = FewShotChatMessagePromptTemplate(
example_prompt=example_prompt,
examples=answer_examples
)
# 모델의 역할을 정의해주기 위한 시스템 프롬프트
system_prompt = ("""
당신은 최고의 KBO 리그 야구 규칙 전문가입니다.
아래의 문서를 참고해서 사용자의 질문에 대한 답변을 해주세요.
먄약에 야구 규칙과 관련이 없다면 "🚫야구 규칙에 대한 질문만 답변이 가능합니다."라고 답변해주세요.
만약에 답변할 수 없다면 모른다고 답변해주세요.
답변 시에 출처에 대해서도 명확하게 밝혀주시고, 사용자가 이해하기 쉽게 설명해주세요.
답변의 길이는 2-3줄 정도로 제한해주세요.
\n\n
문서: {context}
""")
# 채팅 기록 반영하여 QA를 위한 프롬프트
qa_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
few_shot_prompt, # few-shot을 참고하도록 전달
MessagesPlaceholder("chat_history"), # 메시지 리스트(채팅 기록)를 전달
("human", "{input}")
]
)
history_aware_retriever = get_history_retriever() # 기록 기반 검색을 위한 retriever
qa_chain = create_stuff_documents_chain(llm, qa_prompt) # 모델에게 검색한 문서 리스트를 전달
rag_chain = create_retrieval_chain(history_aware_retriever, qa_chain) # RAG 체인 생성
conversational_rag_chain = RunnableWithMessageHistory(
rag_chain, # 실행할 RAG 체인이나 LLM 체인
get_session_history, # 세션_ID 기반의 채팅 기록
input_messages_key="input", # 유저의 입력이 담긴 키
history_messages_key="chat_history", # 채팅 기록을 전달할 키
output_messages_key="answer" # 출력 결과를 저장할 키
).pick("answer")
# 채팅 history를 포함한 retriever를 활용하여 rag_chain 반환
return conversational_rag_chain아래와 같은 질답 예시 문서를 따로 작성해서 llm.py에 import 시켰다.
# fewshot_doc.py
answer_examples = [
{
"input": "주자가 두 명이 동시에 도루하는 것을 뭐라고 해?",
"answer": "주자 두 명이 동시에 도루하는 것은 '더블 스틸'이라고 합니다!\n\n"
"더블 스틸(Double Steal)은 두 명 이상의 주자가 동시에 다음 베이스로 도루하는 전략적 플레이입니다. "
"특히, 수비 팀이 한 주자에게 집중할 때 다른 주자가 성공적으로 진루할 가능성이 높아지는 전술입니다.\n"
"[출처: KBO 규칙 문서]"
},
{
"input": "투수가 반칙 동작을 하면 뭐라고 해?",
"answer": "투수가 반칙 동작을 하면 '보크'라고 합니다!\n\n"
"보크(Balk)는 투수가 주자가 있는 상황에서 규칙에 어긋나는 불법 동작을 했을 때 선언됩니다. "
"보크가 선언되면 모든 주자는 한 베이스씩 진루하게 되며, 이는 투수의 페이크 동작 방지를 위한 규정입니다.\n"
"[출처: KBO 규칙 문서]"
},
{
"input": "경기 중 타자가 삼진을 당했지만 1루로 달릴 수 있는 상황은?",
"answer": "타자가 삼진을 당했지만 1루로 달릴 수 있는 상황은 '낫아웃 삼진'입니다!\n\n"
"낫아웃 삼진(Dropped Third Strike)은 스트라이크 아웃이 선언되었지만 포수가 공을 놓쳐서 잡지 못한 경우를 의미합니다. "
"이때, 1루에 주자가 없거나 2아웃 상황이라면 타자는 1루로 달려 출루를 시도할 수 있습니다.\n"
"[출처: KBO 규칙 문서]"
},
{
"input": "주자가 1루와 2루 사이에서 태그 아웃되지 않으려고 왔다 갔다 하는 것은?",
"answer": "주자가 1루와 2루 사이에서 태그 아웃되지 않으려고 왔다 갔다 하는 것은 '런다운'이라고 합니다!\n\n"
"런다운(Rundown)은 주자가 야수들의 송구 플레이 사이에서 아웃을 피하기 위해 좌우로 움직이며 도망가는 상황을 의미합니다. "
"수비팀은 주자를 협력하여 빠르게 태그 아웃하는 것이 중요하며, 이를 '런다운 플레이'라고도 합니다.\n"
"[출처: KBO 규칙 문서]"
},
# 답변할 수 없는 질문 예시
{
"input": "오늘 KBO 리그 경기 결과 알려줘.",
"answer": "🚫야구 규칙에 대한 질문만 답변이 가능합니다."
},
{
"input": "2024년 KBO 리그에서 가장 많은 홈런을 친 선수는 누구야?",
"answer": "🚫야구 규칙에 대한 질문만 답변이 가능합니다."
},
{
"input": "KBO 리그에서 가장 인기 있는 팀은 어디야?",
"answer": "🚫야구 규칙에 대한 질문만 답변이 가능합니다."
},
{
"input": "야구장에서 가장 맛있는 음식 추천해줘.",
"answer": "🚫야구 규칙에 대한 질문만 답변이 가능합니다."
}
]few shot 적용 후에 답변을 테스트 하기 위해서 아래의 질문 set을 실행시켰다. chat history를 참고하여 동작하도록 일부러 두 번째 질문은 맥락을 유추해야 답변할 수 있도록 작성했다.
Question Set 1.
1️⃣주자가 베이스를 훔치는 것을 뭐라고 해?
2️⃣그럼 주자 두 명이 동시에 하는 건 뭐라고 해?
Question Set 2.
1️⃣인필드 플라이에 대해 설명해줘.
2️⃣해당 상황에서 주자는 어떻게 행동하는지 설명해줘.
지금까지 작성한 코드는 아래와 같다(llm.py, fewshot_doc.py)
# llm.py
from langchain_upstage import ChatUpstage, UpstageEmbeddings
from langchain_pinecone import PineconeVectorStore
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder, FewShotChatMessagePromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from fewshot_doc import answer_examples
chat_storage = {}
def get_llm():
llm = ChatUpstage()
return llm
# 사전을 참고하여 유저의 질문을 재구성하는 시스템 설정
def get_dictionary_chain():
llm = get_llm()
dictionary = [
"베이스에 있는 선수를 나타내는 표현 -> 주자",
"타석에 있는 선수를 나타내는 표현 -> 타자",
"마운드 위에서 공을 던지는 선수를 나타내는 표현 -> 투수"
]
dictionary_prompt = ChatPromptTemplate.from_template(f"""
사용자의 질문을 확인하고 사전을 참고하여 사용자의 질문을 변경해주세요.
만약 변경할 필요가 없다고 판단되면 변경하지 않고 질문을 그대로 리턴해주세요.
질문이나 주어가 모호할 경우 history retriever에서 history chat을 참고하여 유추할 수 있도록 질문을 변경하지 않고 그대로 리턴해주세요.
사전: {dictionary}
질문: {{question}}
""")
dictionary_chain = dictionary_prompt | llm | StrOutputParser()
return dictionary_chain
# 문서를 불러오기 위한 retriever 설정
def get_retriever():
index_name = "baseball-rules-index"
# 데이터를 벡터화할 임베딩 모델 설정
embedding = UpstageEmbeddings(model="solar-embedding-1-large")
database = PineconeVectorStore.from_existing_index(index_name=index_name, embedding=embedding)
retriever = database.as_retriever(search_kwargs={"k": 4})
return retriever
# chat history를 반영하여 질문을 재구성하는 시스템 설정
def get_history_retriever():
llm = get_llm()
retriever = get_retriever()
# chat history를 반영하여 질문을 재구성하는 시스템 프롬프트
contextualize_q_system_prompt = (
"Given a chat history and the latest user question "
"which might reference context in the chat history, "
"formulate a standalone question which can be understood "
"without the chat history. Do NOT answer the question, "
"just reformulate it if needed and otherwise return it as is."
)
# 채팅 기록을 반영하여 QA를 위한 프롬프트
contextualize_q_prompt = ChatPromptTemplate.from_messages(
[
("system", contextualize_q_system_prompt),
MessagesPlaceholder("chat_history"), # 메시지 리스트를 전달
("human", "{input}")
]
)
history_aware_retriever = create_history_aware_retriever(llm, retriever, contextualize_q_prompt)
return history_aware_retriever
# 세션 기록을 가져오기 위한 함수 설정
def get_session_history(session_id: str) -> BaseChatMessageHistory:
# 세션이 없으면 해당 세션에 대한 채팅기록 생성
if session_id not in chat_storage:
chat_storage[session_id] = ChatMessageHistory()
return chat_storage[session_id]
# RAG 체인을 생성
def get_rag_chain():
llm = get_llm()
# few-shot을 위한 프롬프트 설정
example_prompt = ChatPromptTemplate.from_messages(
[
("human", "{input}"),
("ai", "{answer}")
]
)
# few-shot을 위한 프롬프트 설정
few_shot_prompt = FewShotChatMessagePromptTemplate(
example_prompt=example_prompt,
examples=answer_examples
)
# 모델의 역할을 정의해주기 위한 시스템 프롬프트
system_prompt = ("""
당신은 최고의 KBO 리그 야구 규칙 전문가입니다.
아래의 문서를 참고해서 사용자의 질문에 대한 답변을 해주세요.
먄약에 야구 규칙과 관련이 없다면 "🚫야구 규칙에 대한 질문만 답변이 가능합니다."라고 답변해주세요.
만약에 답변할 수 없다면 모른다고 답변해주세요.
답변 시에 출처에 대해서도 명확하게 밝혀주시고, 사용자가 이해하기 쉽게 설명해주세요.
답변의 길이는 2-3줄 정도로 제한해주세요.
\n\n
문서: {context}
""")
# 채팅 기록 반영하여 QA를 위한 프롬프트
qa_prompt = ChatPromptTemplate.from_messages(
[
("system", system_prompt),
few_shot_prompt, # few-shot을 참고하도록 전달
MessagesPlaceholder("chat_history"), # 메시지 리스트(채팅 기록)를 전달
("human", "{input}")
]
)
history_aware_retriever = get_history_retriever() # 기록 기반 검색을 위한 retriever
qa_chain = create_stuff_documents_chain(llm, qa_prompt) # 모델에게 검색한 문서 리스트를 전달
rag_chain = create_retrieval_chain(history_aware_retriever, qa_chain) # RAG 체인 생성
conversational_rag_chain = RunnableWithMessageHistory(
rag_chain, # 실행할 RAG 체인이나 LLM 체인
get_session_history, # 세션_ID 기반의 채팅 기록
input_messages_key="input", # 유저의 입력이 담긴 키
history_messages_key="chat_history", # 채팅 기록을 전달할 키
output_messages_key="answer" # 출력 결과를 저장할 키
).pick("answer")
# 채팅 history를 포함한 retriever를 활용하여 rag_chain 반환
return conversational_rag_chain
def get_ai_response1():
dictionary_chain = get_dictionary_chain()
rag_chain = get_rag_chain()
baseball_chain = {"input": dictionary_chain} | rag_chain
ai_response = baseball_chain.invoke(
{"question": "인필드 플라이에 대해 설명해줘."},
config={"configurable": {"session_id": "abc123"}}
)
return ai_response
def get_ai_response2():
dictionary_chain = get_dictionary_chain()
rag_chain = get_rag_chain()
baseball_chain = {"input": dictionary_chain} | rag_chain
ai_response = baseball_chain.invoke(
{"question": "해당 상황에서 주자는 어떻게 행동하는지 설명해줘."},
config={"configurable": {"session_id": "abc123"}}
)
return ai_response
ai_response = get_ai_response1()
print(f"ai_response1:\n {ai_response}\n")
ai_response = get_ai_response2()
print(f"ai_response2:\n {ai_response}\n")# fewshot_doc.py
answer_examples = [
{
"input": "주자가 두 명이 동시에 도루하는 것을 뭐라고 해?",
"answer": "주자 두 명이 동시에 도루하는 것은 '더블 스틸'이라고 합니다!\n\n"
"더블 스틸(Double Steal)은 두 명 이상의 주자가 동시에 다음 베이스로 도루하는 전략적 플레이입니다. "
"특히, 수비 팀이 한 주자에게 집중할 때 다른 주자가 성공적으로 진루할 가능성이 높아지는 전술입니다.\n"
"[출처: KBO 규칙 문서]"
},
{
"input": "투수가 반칙 동작을 하면 뭐라고 해?",
"answer": "투수가 반칙 동작을 하면 '보크'라고 합니다!\n\n"
"보크(Balk)는 투수가 주자가 있는 상황에서 규칙에 어긋나는 불법 동작을 했을 때 선언됩니다. "
"보크가 선언되면 모든 주자는 한 베이스씩 진루하게 되며, 이는 투수의 페이크 동작 방지를 위한 규정입니다.\n"
"[출처: KBO 규칙 문서]"
},
{
"input": "경기 중 타자가 삼진을 당했지만 1루로 달릴 수 있는 상황은?",
"answer": "타자가 삼진을 당했지만 1루로 달릴 수 있는 상황은 '낫아웃 삼진'입니다!\n\n"
"낫아웃 삼진(Dropped Third Strike)은 스트라이크 아웃이 선언되었지만 포수가 공을 놓쳐서 잡지 못한 경우를 의미합니다. "
"이때, 1루에 주자가 없거나 2아웃 상황이라면 타자는 1루로 달려 출루를 시도할 수 있습니다.\n"
"[출처: KBO 규칙 문서]"
},
{
"input": "주자가 1루와 2루 사이에서 태그 아웃되지 않으려고 왔다 갔다 하는 것은?",
"answer": "주자가 1루와 2루 사이에서 태그 아웃되지 않으려고 왔다 갔다 하는 것은 '런다운'이라고 합니다!\n\n"
"런다운(Rundown)은 주자가 야수들의 송구 플레이 사이에서 아웃을 피하기 위해 좌우로 움직이며 도망가는 상황을 의미합니다. "
"수비팀은 주자를 협력하여 빠르게 태그 아웃하는 것이 중요하며, 이를 '런다운 플레이'라고도 합니다.\n"
"[출처: KBO 규칙 문서]"
},
# 답변할 수 없는 질문 예시
{
"input": "오늘 KBO 리그 경기 결과 알려줘.",
"answer": "🚫야구 규칙에 대한 질문만 답변이 가능합니다."
},
{
"input": "2024년 KBO 리그에서 가장 많은 홈런을 친 선수는 누구야?",
"answer": "🚫야구 규칙에 대한 질문만 답변이 가능합니다."
},
{
"input": "KBO 리그에서 가장 인기 있는 팀은 어디야?",
"answer": "🚫야구 규칙에 대한 질문만 답변이 가능합니다."
},
{
"input": "야구장에서 가장 맛있는 음식 추천해줘.",
"answer": "🚫야구 규칙에 대한 질문만 답변이 가능합니다."
}
]References
https://python.langchain.com/v0.2/docs/how_to/qa_chat_history_how_to/#chains
https://python.langchain.com/v0.2/docs/how_to/few_shot_examples_chat/#create-prompt-template
728x90
'AI' 카테고리의 다른 글
| [AI] RAG와 LangChain을 활용한 Chatbot 구현(3) (0) | 2025.02.21 |
|---|---|
| [AI] RAG와 LangChain을 활용한 Chatbot 구현(1) (0) | 2025.02.04 |
