LLM과 RAG를 활용한 챗봇 구현-2
과제 내용
도전과제
- **LangSmith의 Prompt Library를 참고하여 prompt engineering을 수행**
- RAG의 성능은 prompt 의 품질에도 많은 영향을 받습니다. 이번 도전 구현과제는 prompt engineering을 해보고, prompt 실험 결과를 외부에서 잘 비교 정리할 수 있도록 코드를 고쳐보겠습니다!
i) LangSmith의 Prompt Library 를 참고하여 프롬프트를 3개 이상 아래와 같은 파일 구조로 저장해주세요.
예시)
.
├── main.jupynb
└── Prompts/
├── prompt1.txt
├── prompt2.txt
└── prompt3.txt
ii) 각 프롬프트를 외부에서 불러와서 실행할 수 있도록 코드를 고쳐주세요.
iii) 실행 결과는 자동으로 Result 디렉토리에 저장되어야 합니다. 이때, 실험 결과 파일 이름은 실험에 쓰인 프롬프트의 이름과 timestamp을 포함해야합니다.
예시)
.
├── main.jupynb
└── Prompts/
├── prompt1.txt
├── prompt2.txt
└── prompt3.txt
└── Results/
├── prompt1_result_1731314042.txt
├── prompt2_result_1731314050.txt
└── prompt3_result_1731314050.txt
전체코드
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain.document_loaders import PyPDFLoader
import pdfplumber
import pandas as pd
import os
from datetime import datetime
from langchain.text_splitter import RecursiveCharacterTextSplitter
import faiss
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_core.runnables import RunnablePassthrough
model = ChatOpenAI(model="gpt-4o-mini")
loader = PyPDFLoader("ai_industry.pdf")
docs = loader.load()
recursive_text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200,
length_function=len,
is_separator_regex=False,
)
splits = recursive_text_splitter.split_documents(docs)
with pdfplumber.open("ai_industry.pdf") as pdf:
all_tables = []
for page_num, page in enumerate(pdf.pages):
tables = page.extract_tables()
for table in tables:
df = pd.DataFrame(table[1:], columns=table[0])
all_tables.append((page_num, df))
embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
vectorstore = FAISS.from_documents(documents=splits, embedding=embeddings)
retriever = vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": 3})
# 프롬프트 파일 로드
prompts = {}
prompt_folder = "/Prompts"
for filename in os.listdir(prompt_folder):
if filename.endswith(".txt"):
with open(os.path.join(prompt_folder, filename), "r", encoding="utf-8") as file:
prompts[filename] = file.read()
class DebugPassThrough(RunnablePassthrough):
def invoke(self, *args, **kwargs):
output = super().invoke(*args, **kwargs)
print("Debug Output:", output)
return output
class ContextToText(RunnablePassthrough):
def invoke(self, inputs, config=None, **kwargs):
# context의 각 문서를 문자열로 결합
context_text = "\n".join([doc.page_content for doc in inputs["context"]])
# 전체 프롬프트를 단일 문자열로 생성
full_prompt = f"Context: {context_text}\n\nQuestion: {inputs['question']}"
return full_prompt
# 결과 저장 디렉토리
result_folder = "/Results"
os.makedirs(result_folder, exist_ok=True)
rag_chain_debug = DebugPassThrough() | ContextToText() | model
while True:
print("========================")
query = input("질문을 입력하세요 (종료하려면 'exit' 입력): ")
if query == "exit":
break
for i, (prompt_name, prompt_text) in enumerate(prompts.items(), start=1):
print(f"\nPrompt {i}: {prompt_name}")
context_documents = retriever.get_relevant_documents(query)
inputs = {"context": context_documents, "question": prompt_text + "\n" + query}
response = rag_chain_debug.invoke(inputs)
print(f"Response for {prompt_name}:")
print(response.content)
timestamp = int(datetime.now().timestamp())
result_filename = f"{prompt_name.replace('.txt', '')}_result_{timestamp}.txt"
with open(os.path.join(result_folder, result_filename), "w", encoding="utf-8") as result_file:
result_file.write(response.content)
이 코드에서 이전과 달라진 부분은 다음과 같다.
1. 프롬프트 파일 로드
prompts = {}
prompt_folder = "/Prompts"
for filename in os.listdir(prompt_folder):
if filename.endswith(".txt"):
with open(os.path.join(prompt_folder, filename), "r", encoding="utf-8") as file:
prompts[filename] = file.read()
os.listdir(폴더명) 폴더에 있는 모든 파일과 디렉토리를 리스트로 반환
endswith(".txt") 파일이름이 .txt로 끝나면
prompts[filename]=file.read()
파일을 읽어오고 prompts={} 딕셔너리에 filename을 키로 내용을 저장
-> prompts= {"prompt1": "content", "prompt2": "content", "prompt3": "content" }
2. 결과 저장 디렉토리
result_folder = "/Results"
os.makedirs(result_folder, exist_ok=True)
os.makedirs() 는 지정된 디렉토리 중 없는 디렉토리는 모두 생성
만약 C:/users/name/prmopt/results 가 경로인데, prompts 파일이 없다면 results파일 뿐 아니라 prompts 파일까지 생성함
exist_ok=True
디렉토리가 이미 존재할 경우 에러 없이 그냥 넘어감
이 옵션이 없으면 filexistserror 발생
3. rag chain, query
rag_chain_debug = DebugPassThrough() | ContextToText() | model
while True:
print("========================")
query = input("질문을 입력하세요 (종료하려면 'exit' 입력): ")
if query == "exit":
break
for i, (prompt_name, prompt_text) in enumerate(prompts.items(), start=1):
print(f"\nPrompt {i}: {prompt_name}")
context_documents = retriever.get_relevant_documents(query)
inputs = {"context": context_documents, "question": prompt_text + "\n" + query}
response = rag_chain_debug.invoke(inputs)
print(f"Response for {prompt_name}:")
print(response.content)
rag_chain_debug = DebugPassThrough() | ContextToText() | model
이전의 rag chain에는 model로 가기 전에 prompt 가 있었다.
하지만 이번에는 prompts= {"prompt1": "content", "prompt2": "content", "prompt3": "content" } 이런 딕셔너리이기 때문에, 체인에 prompts를 넣으면 에러가 난다. 그래서 prompts는 우선 체인에서 없애주었다.
for i, (prompt_name, prompt_text) in enumerate(prompts.items(), start=1):
print(f"\nPrompt {i}: {prompt_name}")
준비한 prompts에서 prompt를 하나씩 꺼내 쓸텐데, 알아보기 쉽게 하기 위해 start=1과 print()를 추가했다.
.items() 는 딕셔너리에서 (key,value)쌍을 반환
참고) 리스트 뿐 아니라 딕셔너리에서도 .items() .values() .keys()를 붙여 enumerate를 쓸 수 있다.
context_documents = retriever.get_relevant_documents(query)
retreiver는 검색작업을 수행하는 객체
retriever가 동작을 하려면 체인 내부에 결합해서 자동으로 메서드가 호출되게 하거나,
체인 밖에서 쓰려면 직접 메서드를 호출해야한다. (이번 코드는 체인 밖에서 쓰고 있다)
get_relevant_documents가 바로 리트리버를 동작시키는 메서드이다.
inputs = {"context": context_documents, "question": prompt_text + "\n" + query}
response = rag_chain_debug.invoke(inputs)
inputs 를 context, question을 키로 하는 딕셔너리로 만들어 rag chain에 전달하고
답변을 받아서 response에 저장한다.
4. timestamp
timestamp = int(datetime.now().timestamp())
result_filename = f"{prompt_name.replace('.txt', '')}_result_{timestamp}.txt"
with open(os.path.join(result_folder, result_filename), "w", encoding="utf-8") as result_file:
result_file.write(response.content)
timestamp = int(datetime.now().timestamp())
datetime.now: 현재시간 반환
.timestamp() 는 현재시간을 유닉스 타임스탬프로 표기
유닉스 타임스탬프: 1970.1.1 기준으로 몇 초가 지났는지
int변환으로 소수점까지 제거
result_filename = f"{prompt_name.replace('.txt', '')}_result_{timestamp}.txt"
문자열 포맷팅으로 파일이름 생성
with open(os.path.join(result_folder, result_filename), "w", encoding="utf-8") as result_file:
with open()함수로 파일을 열고 작업이 끝나면 자동으로 닫음
"w"이므로 쓰기모드 -> 파일이 있으면 덮어쓰기, 없으면 새로 생성
열린 파일을 result_file이라는 변수에 저장하여 파일을 씀
result_file.write(response.content)
result_file에 응답의 내용을 저장
timestamp만드는 이유?
파일이름 충돌 방지
모델간 응답 품질의 비교 용이
+ full prompt ??
class ContextToText(RunnablePassthrough):
def invoke(self, inputs, config=None, **kwargs):
# context의 각 문서를 문자열로 결합
context_text = "\n".join([doc.page_content for doc in inputs["context"]])
# 전체 프롬프트를 단일 문자열로 생성
full_prompt = f"Context: {context_text}\n\nQuestion: {inputs['question']}"
return full_prompt
여기서 마지막에 full prompt를 굳이 만든 이유는
단순히 context와 question을 줄글로 전달하면
ex
context: 고양이는 삶은 닭가슴살을 좋아하고, 거북이는 감마루스를 안먹는다.
question: 감마루스를 좋아하는 동물은
이렇게 전달되어야 할 내용이
고양이는 삶은 닭가슴살을 좋아하고, 거북이는 감마루스를 안먹는다.
감마루스를 좋아하는 동물은
-> 답변: 질문하고자 하는 바를 정확히 알 수 없습니다.
이렇게만 전달되어서 모델에 혼란을 줄 수 있다.
물론 나중의 코드에서 inputs = {"context": context_documents, "question": prompt_text + "\n" + query} 를 통해
문맥과 질문을 명확히 해주고 있지만, 설계의 일관성(+추후 가공, 확장 가능성)을 위해 full prompt를 만들어주었다.
다음 게시물
프롬프트 별 실행결과물