* 겪고 있는 문제 상황을 최대한 자세하게 작성해주세요.
* 문제 해결을 위해 어떤 시도를 해보았는지 구체적으로 함께 알려주세요.
챕터 5 숙제 문제 풀이중에서 발생한 오류 관련 문의 합니다.
코드 수정후 pdf파일 업로더 후 Submit을 누를때 첨부된 파일 처럼 계속 오류가 나고 있는데 기본코드에서 수정하지 않는 부분에서 오류가 발생하는 것 같아 문의 드립니다.
작성한 코드는 아래와 같습니다.
# 기본 뼈대 코드
import gradio as gr
from sentence_transformers import SentenceTransformer
import openai
import os
import pdfplumber
import numpy as np
import faiss
import tiktoken
# Import necessary libraries
from pdf2image import convert_from_path
from azure.cognitiveservices.vision.computervision import ComputerVisionClient
from azure.cognitiveservices.vision.computervision.models import OperationStatusCodes
from msrest.authentication import CognitiveServicesCredentials
import cv2
import io
import time
# Azure API 인증 정보 설정
SUBSCRIPTION_KEY = ""
ENDPOINT_URL = ""
computervision_client = ComputerVisionClient(ENDPOINT_URL, CognitiveServicesCredentials(SUBSCRIPTION_KEY))
# OpenAI 인증 정보 설정
API_KEY = ""
openai_client = openai.Client(api_key=API_KEY)
embedder = SentenceTransformer('sentence-transformers/all-MiniLM-L6-v2')
chat_history = []
chunks = []
index = None
# OCR
def detect_text_from_image(image_path):
image_data = open(image_path, "rb").read()
sbuf = io.BytesIO(image_data)
# Call the API to process the image
response = computervision_client.read_in_stream(sbuf, raw=True)
operation_location = response.headers["Operation-Location"]
operation_id = operation_location.split("/")[-1]
# Polling the API until the operation completes
while True:
read_result = computervision_client.get_read_result(operation_id)
if read_result.status not in ['notStarted', 'running']:
break
time.sleep(1)
raw_text = ""
if read_result.status == OperationStatusCodes.succeeded:
for result in read_result.analyze_result.read_results:
for line in result.lines:
raw_text += line.text + "\n"
return raw_text
def extract_text_from_pdf(file_path):
text = ""
# First, try to extract text-based content using pdfplumber
try:
with pdfplumber.open(file_path) as pdf:
for page in pdf.pages:
page_text = page.extract_text(layout=True)
if page_text: # Check if any text was extracted
text += page_text
except Exception as e:
print(f"Error using pdfplumber: {e}")
# If no text is extracted (indicating it might be image-based), use OCR
if not text.strip(): # If no text was extracted, fall back to OCR
print("No text found with pdfplumber, switching to OCR...")
images = convert_from_path(file_path)
for i, image in enumerate(images):
image_path = f"page_{i + 1}.jpg"
image.save(image_path, "JPEG")
text += detect_text_from_image(image_path)
os.remove(image_path) # Clean up the saved image after processing
return text
def chunk_text(text, chunk_size = 128, overlap = 64):
words = text.split()
local_chunks = []
for i in range(0, len(words), chunk_size - overlap):
chunk = ' '.join(words[i:i+chunk_size])
local_chunks.append(chunk)
return local_chunks
# 임베딩
def create_embeddings(chunks):
embeddings = embedder.encode(chunks, convert_to_tensor = False)
return np.array(embeddings)
# -- FAISS 알고리즘 구현 함수에서 빈칸을 채워주세요 -- #
def create_faiss_index(embeddings, base_algorithm="Flat"):
dimension = embeddings.shape[1] # 임베딩의 차원 수
num_embeddings = embeddings.shape[0] # 임베딩의 개수
nlist = min(num_embeddings // 10, 100) # nlist 값 동적으로 설정, 최대 100으로 제한
global index # 전역 변수로 인덱스 선언
if base_algorithm in ("", None): # 알고리즘이 지정되지 않았을 경우 기본값 설정
base_algorithm = "Flat"
# 코사인 유사도를 위해 임베딩을 정규화 (IndexFlatIP에만 필요)
if base_algorithm in ["Flat", "IVF"]:
faiss.normalize_L2(embeddings)
if base_algorithm == "Flat":
# 코사인 유사도를 위한 내적 방식의 Flat 인덱스 사용
index = faiss.IndexFlatIP(dimension) # 내적 기반 (코사인 유사도) Flat 인덱스 생성
elif base_algorithm == "PQ":
# 대규모 데이터셋을 위한 Product Quantization (PQ)
# m (서브벡터 수) 값 동적 설정, 최대 64로 제한
m = 8
nbits = 8 # 서브벡터 당 비트 수
quantizer = faiss.IndexFlatL2(dimension) # L2 기반 PQ를 위한 양자화기 생성
index = faiss.IndexIVFPQ(quantizer, dimension, nlist, m, nbits) # IVF + PQ, 서브벡터당 8비트
index.train(embeddings) # PQ는 데이터셋을 기반으로 훈련 필요
elif base_algorithm == "IVF":
# 대규모 데이터셋에서 빠른 검색을 위한 Inverted File Index (IVF) 사용
quantizer = faiss.IndexFlatIP(dimension) # IVF 검색을 위한 내적 방식 사용
index = faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_INNER_PRODUCT)
index.train(embeddings) # IVF는 데이터셋을 기반으로 훈련 필요
elif base_algorithm == "HNSW":
# 계층적 탐색 가능한 소규모 세계 그래프(HNSW) 기반 인덱스
index = faiss.IndexHNSWFlat(dimension, 32) # 그래프에서 각 노드당 32개의 이웃 사용
else:
# 지원되지 않는 알고리즘 이름에 대한 예외 처리
raise ValueError(f"Unknown base_algorithm: {base_algorithm}. Choose from 'Flat', 'PQ', 'IVF', 'HNSW'.")
# 인덱스에 임베딩 추가
index.add(embeddings)
print(f"{base_algorithm}를 사용한 FAISS index가 만들어졌습니다.")
print(index)
print(f"인덱스에 저장된 벡터의 갯수: {index.ntotal}") # 저장된 벡터 개수
print(f"인덱스 벡터의 차원: {dimension}") # 벡터 차원
def retrieve_relevant_chunks(query, top_k = 5):
query_embedding = embedder.encode(query, convert_to_tensor = False)
query_embedding = np.array([query_embedding])
global index
faiss.normalize_L2(query_embedding)
distances, indices = index.search(query_embedding, top_k)
relevant_chunks = [chunks[i] for i in indices[0]]
return '\n\n'.join(relevant_chunks)
# 챗봇 페르소나 설정
def generate_answer(relevant_text, query, chat_history):
persona = """ 제품에 대한 고객의 질문에 친절하게 대답해주는 챗봇입니다."""
system_prompt = f"""You are knowledge based chatbot.
#follow these instructions.
1. Answer any asked questions based on given knowledge text.
2. Ask a question related to the currently on-going conversation.
3. Able to utilize outside information, however priority is knowledge based on given text.
4. generate answer that aligns with given persona of chatbot, in Korean
#Persona of chatbot
{persona}
"""
chat_history.append({'role':'system', 'content':system_prompt})
chat_history.append({'role':'user', 'content':f"Based on the following text:\n\n{relevant_text}\n\nAnswer the question: {query}"})
response = openai_client.chat.completions.create(
model = 'gpt-4o',
messages = chat_history,
max_tokens = 1024
)
answer = response.choices[0].message.content
return answer
def process_pdf(pdf_file):
global chunks
text = extract_text_from_pdf(pdf_file)
chunks = chunk_text(text)
embeddings = create_embeddings(chunks)
create_faiss_index(embeddings)
return f"PDF processed and FAISS index created using"
def answer_question(query, history):
relevant_text = retrieve_relevant_chunks(query)
global chat_history
for human, assistant in history:
chat_history.append({"role":"user", "content":human})
chat_history.append({"role":"assistant", "content":assistant})
answer = generate_answer(relevant_text, query, chat_history)
return answer
# Gradio UI 부분
pdf_input = gr.File(label = 'upload your pdf', type = 'filepath')
with gr.Blocks() as demo:
gr.Markdown("## 제품 문의 챗봇")
with gr.Tab("Step 1: Upload PDF and Extract Text"):
pdf_interface = gr.Interface(
fn=process_pdf,
inputs=[pdf_input],
outputs="text",
description="Upload a PDF and extract its text chunks and embeddings."
)
with gr.Tab("Step 2: Chatbot based on pdf"):
question_interface = gr.ChatInterface(
fn=answer_question,
textbox = gr.Textbox(placeholder = "질문이 있으신가요?",
container = False,
scale = 7)
).queue()
demo.launch(share = True, debug=True)
오류 나는 위치가
def create_embeddings(chunks):
embeddings = embedder.encode(chunks, convert_to_tensor = False)
return np.array(embeddings)
def retrieve_relevant_chunks(query, top_k = 5):
query_embedding = embedder.encode(query, convert_to_tensor = False)
query_embedding = np.array([query_embedding])
위의 2군데로 보이는데 제가 잘 파악하고 있는지 모르겠네요
확인 부탁드립니다.
