이야기 챗봇 구현-3

이야기 챗봇 구현-2에 이어서 작성

 

def initialize_rag_chain(model, splits):

RAG를 초기화하는 함수

모델과 텍스트(splits)를 주면 faiss 인덱스를 로드하고 검색기능을 설정하고 llm과 결합하는 일련의 과정을 자동화함

    try:
        vectorstore = FAISS.load_local(
            "vector_index",
            model,
            allow_dangerous_deserialization=True
        )

저장된 faiss index("vector_index")를 로드

이 때  model은 임베딩할 때 사용한 모델인 sentence transformer를 의미한다.

 

allow_dangerous_deserialization=True 

데이터 로드 중 보안경고를 무시하는 코드인데 이걸 설정안하면

ValueError: The de-serialization relies loading a pickle file. Pickle files can be modified to deliver a malicious payload that results in execution of arbitrary code on your machine.You will need to set allow_dangerous_deserialization to True to enable deserialization. If you do this, make sure that you trust the source of the data. For example, if you are loading a file that you created, and know that no one else has modified the file, then this is safe to do. Do not set this to True if you are loading a file from an untrusted source (e.g., some random site on the internet.).

이런 에러가 난다. 이걸 챗봇구동할 때 그대로 적용해도 되는지는..?

    except RuntimeError:
        print("FAISS 인덱스를 찾을 수 없습니다. 새로 생성합니다...")
        create_index(splits)  
        vectorstore = FAISS.load_local(
            "vector_index",
            model,
            allow_dangerous_deserialization=True
        )

만약 runtimerror가 나면 create_index로 인덱스를 새로 만든 후 faiss.local_load를 다시 호출

 

Q faiss.local_load 가 정확히 뭔가요?

A 디스크에 저장된 faiss 인덱스를 로드함 faiss인덱스를 생성하는 데 시간이 걸리기 때문에, 한 번 생성하면 저장을 해두고 필요할 때 불러오는 방식이 좋다. 그래서 local에 저장된 index를 로드하는건데, 만약 한 번도 인덱스를 생성해서 저장한 적이 없으면 creat_index로 인덱스를 만든 다음 불러온다.

 

    retriever = vectorstore.as_retriever(search_type="similarity", search_kwargs={"k": 3})
    llm = ChatOpenAI(model="gpt-4")
    return RetrievalQA.from_chain_type(llm=llm, retriever=retriever, return_source_documents=False)

vectorstore는 임베딩 벡터를 저장하고 유사도를 계산하여 유사한 것을 빠르게 계산하기 쉽게 설계됨

리트리버는 스토어 내에서 실질적으로 검색을 수행할 수 있다.

 

vectorstore를 리트리버로 설정하여 유사도 기반으로 가장 유사한 것 3개를 가져오게 하고

llm모델로는 gpt-4를 사용

최종적으로 리트리버와 llm모델을 결합한 rag chain을 형성한다. (RetrievalQA)

 

return_source_documents=False: 검색한 데이터 문서는 반환하지 않고 모델의 최종 답변만 반환하는 옵션

이걸 설정하지 않으면 나중에 이야기를 가져올 때

run not supported when there is not exactly one output key. Got ['result', 'source_documents'].

이렇게 result와 source_document 두 개를 받았다는 오류가 난다.

 

 

def get_story_parts(user_query=None, json_texts=None, rag_chain=None):
    if json_texts is None:
        raise ValueError("json_texts가 제공되지 않았습니다. JSON 데이터를 먼저 로드하세요.")

get_story_part() 함수는 사용자한테 어떤 이야기를 듣고싶은지 물어봤을 때 응답에 따라 story를 가져오는 함수로, 사용자 입력을 받아서 입력에 따라 이야기를 가져온다.

우선 texts가 로드되지 않았을 경우에는 에러메시지를 내도록 했고(디버깅용)

 

참고로 나중에 chatbot() 함수에서 json_texts를 다음과 같이 정의할 예정이다

    texts = process_json_data(json_files)

    json_texts = texts

 

    if user_query is None or user_query.strip().lower() in ["재미", "아무거나", "재밌", "암꺼나"]:
        print("재미난 이야기를 가져오는 중...")
        try:
            random_story = random.choice(json_texts)  
            print("랜덤 이야기 선택 완료!")
            return dividing_story(random_story)
        except Exception as e:
            print(f"랜덤 이야기를 선택하는 중 오류 발생: {e}")
            return None

사용자가 아무거나 이야기해줘 재미있는 이야기 들려줘 등의 말을 했을 때는

random.choice로 json_texts중 아무 이야기나 선택하고

dividing_story() 함수에 넘겨줘서 쪼개진 이야기 형태로 받아온다.

 

except 이후는 오류가 발생했을 때 디버깅용으로 추가했다.

    else:
        print("요청한 이야기를 가져오는 중...")

사용자가 특별히 요청한 이야기가 있으면 ex.살인사건 방화사건 등

관련 내용이 있는 이야기를 검색해서 가져와야 한다. 

 

        if rag_chain is None:
            raise ValueError("RAG 체인이 초기화되지 않았습니다.")

rag_chain이 초기화되지 않았을 때를 위한 디버깅용 코드

        try:
            result = rag_chain.invoke({"query": user_query})

rag_chain.invoke로 rag chain에 사용자의 입력query를 전달하면, chain에서 관련된 정보를 검색해서 반환

result에 반환값이 저장되는데, 보통 텍스트 또는 json형식으로 반환되어야 함

그런데 코드를 실행시켜봤더니 rag chain에서 이야기를 검색하는 도중 오류가 생겼고 오류 내용은 string indices must be integers 라고 해서 result의 데이터 타입을 처리해줘야 했다.

            if isinstance(result, dict) and "result" in result:
                story_text = result["result"]

result가 dictionary형태이고 "result"가 키로 들어가 있으면 result키의 value를 가져오도록 처리

            elif isinstance(result, str):
                story_text = result  

result가 str이면 그대로 가져오도록 처리

            else:
                print("RAG 체인에서 반환된 데이터 구조를 처리할 수 없습니다.")
                return None

그 외의 예외에 대해서는 에러문구를 프린트하고 None 반환

            print("검색된 이야기:\n", story_text)
            return dividing_story(story_text)

 

최종적으로  story_text를 dividing story에 넘겨서 기승전결로 쪼갤수 있게 함

        except Exception as e:
            print(f"이야기를 검색하는 중 오류 발생: {e}")
            return None

기타 예외는 에러문구를 프린트하고 None 반환

 

여기까지 시행 후 발생한 문제점

이야기의 랜덤생성은 문제가 없는데, 특정 주제를 검색해서 가져오게 하면 string indices must be integers 오류가 남

result를 확인해보니 None으로 반환되는 것으로 보임

rag chain의 최초 동작 자체이 문제가 있는 것으로 보인다.

 

 

conversation_logs = {}

conversation_logs에 딕셔너리 형태로 모든 대화내용을 저장

# 대화 저장
def save_conversation(session_id, user_message, assistant_message):

인자로 session_id와 사용자, 챗봇의 대화내용을 받음

    if session_id not in conversation_logs:
        conversation_logs[session_id] = []

만약 session_id가 없으면 빈 리스트를 새로 만들고

 

    conversation_logs[session_id].append({
        "timestamp": datetime.now().isoformat(),
        "user": user_message,
        "assistant": assistant_message
    })

session_id 리스트에 timestamp, user, assistant를 키로 하는 딕셔너리를 생성해서 추가함

 

    with open("conversation_logs.json", "w", encoding="utf-8") as file:
        json.dump(conversation_logs, file, indent=4, ensure_ascii=False)

conversation_log 전체를 json 파일로 저장함

한글은 ascii문자가 아니라서 ascii=False 로 해야 문자 그대로 json파일에 저장된다.

디폴트가 True라서 그대로 놔두면 한글을 다른 문자열로 변환해버릴 수 있다.

# 대화 로드
def load_conversation(session_id):
    try:
        with open("conversation_logs.json", "r", encoding="utf-8") as file:
            data = json.load(file)
        return data.get(session_id, [])
    except FileNotFoundError:
        return []

저장한 파일을 불러오는 코드인데 주목할것은  data.get(session_id, [])

get은 session_id에 해당하는 대화기록을 가져오는데, 만약 session_id가 존재하지 않으면 두번째인자인 [] 를 기본값으로 반환함

파일이 없어도 []를 반환하게 해서 예외처리를 했다.

 

이야기 챗봇 구현-4 에서 계속