실전 RAG로 역량강화하기 - 멀티모달 프로토타입

실전 RAG으로 역량 강화하기

실전 RAG 역량 강화하기 스터디는 이제 본격적으로 개발하는 단계에 들어갔습니다. 스터디원 분들이 다양한 주제로 여러 가지 RAG use-case를 준비해 주셔서 마지막 회고날이 기대가 되는 스터디입니다. 매주 저희는 발표를 하고 있기 때문에 RAG로 발표를 할걸 찾다가 일단.. 프로토타입이나 만들어두고 떠들자 생각이 들었고, 한번 만들어봤습니다.

 

 

저는 이런 프로젝트를 기획하고 있었습니다. 이미 이전에 LangChain 과 gemini로 가능성을 확인해 봤기 때문에 바로 구현만 하면 된다고 판단을 했었는데요. 그러나 이 프로젝트가 어렵다고 느낀 건 데이터구조, 파싱이었습니다. 어떤 동영상을 어떻게 데이터구조를 형상화하고 파싱을 할지가 아직도 조금 어렵고 낯섭니다. 그래서 그냥 간단하게 유튜브 먹는 영상을 가지고 CrewAI를 썼던 경험을 활용해서 만들어보자 생각이 들었습니다.

 

 

뭔가 먹는 모습을 관련해서 저장을하고 그 음식에 대해 메타데이터를 저장하는 비교적 간단한 구조를 통해 먼저 만들려고 했습니다. 아마 나중에 저런 복합시스템을 만들 때는, 다양한 피처와, 데이터를 넣어서 만들어야 할 것 같습니다. 유튜브에서 테스트를 하기 위해 먹는 장면, 음식이 많이 바뀌는 장면의 숏츠를 파이썬코드로 크롤링해서 원본데이터를 수집하였습니다.(학습용) 

 

이전에 스터디에서 상윤님께서 소개해주신 업스테이지의 예시를 생각해서 그냥 실시간 임베딩처리해서 벡터디비 넣고 쿼리에 대해 질의를 시켜야겠다 생각이 들었고 그냥 일반적이게 코드를 구현했습니다.

 

import google.generativeai as genai
from IPython.display import Markdown
import os
import time


# .env에서 API 키 가져오기
GOOGLE_API_KEY = os.getenv("GOOGLE_API_KEY")

if not GOOGLE_API_KEY:
    raise ValueError("Google API key not found. Please set it in your .env file.")

# API 키 설정
genai.configure(api_key=GOOGLE_API_KEY)

# 비디오 파일 경로


# 파일 업로드 함수
def upload_and_process_file(file_path):
    print(f"Uploading file: {file_path}...")
    video_file = genai.upload_file(path=file_path)
    
    # 파일 처리 상태 확인
    while video_file.state.name == "PROCESSING":
        print('.', end='', flush=True)
        time.sleep(10)
        video_file = genai.get_file(video_file.name)
    
    if video_file.state.name == "FAILED":
        raise ValueError(f"File processing failed: {video_file.state.name}")
    
    print(f"\nCompleted upload: {video_file.uri}")
    return video_file

# LLM 요청 함수
def generate_content_from_video(video_file, prompt, model_name="gemini-1.5-flash-001", timeout=600):
    print("Making LLM inference request...")
    model = genai.GenerativeModel(model_name=model_name)
    response = model.generate_content([video_file, prompt], request_options={"timeout": timeout})
    return response

# 메인 로직
if __name__ == "__main__":
    video_file_name = "video/내가 엘든링 7번 깬 비법.f614.mp4"
    video_file = upload_and_process_file(video_file_name)
    # 프롬프트 설정
    prompt = '''
    해당영상에서 당신은 먹는 시간의 시작과 끝을 출력하고, 먹는 메뉴도 출력해야합니다.
    [출력예시]
    [1:01,1:31], 불고기
    [2:01,2:31], 비빔밥
    [3:01,3:31], 김치찌개
    '''
    
    # 컨텐츠 생성 요청
    response = generate_content_from_video(video_file, prompt)
    
    # 마크다운으로 출력
    display(Markdown(response.text))

 

위와 같이 few-shot 예제를 추고 아웃풋을 저렇게 뽑도록 일단 시도를 했습니다 늘 그렇듯 부연설명하면서 뽑는 멍청이 LLM 한 번에 해줄리는 없었기 때문에 파서를 동원했습니다.

 

 

처음에 이렇게 생각을 했었는데, 물 흐르듯 프로세스가 진행됬습니다. GPT가 아웃풋 파서를 지원해 주기 때문에 그걸 활용하려 했으나 몇 번을 시도해도 자꾸 beta 관련에러가 떠서 그냥 LangChain json parser를 사용했습니다.

class Seconds(BaseModel):
    start: int = Field(description="The start time of the food in seconds")
    end: int = Field(description="The end time of the food in seconds")

class VideoParser(BaseModel):
    time: list[Seconds] = Field(description="The time of the food in seconds")
    food_name: str = Field(description="The name of the food")

 

시간을 추출하고 상속받아서 비디오 파서에서의 시간과, 이름을 반환하도록하였습니다.

 

{'time': [{'start': 8, 'end': 57}], 'food_name': '딸기 크림 빙수'}

import cv2
import os

def extract_video_segment(input_video, start_time, end_time, output_folder, food_name):
    # 입력 비디오 열기
    cap = cv2.VideoCapture(input_video)
    
    # 비디오 속성 가져오기
    fps = int(cap.get(cv2.CAP_PROP_FPS))
    width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    
    # 출력 비디오 설정
    fourcc = cv2.VideoWriter_fourcc(*'mp4v')
    output_filename = f"{food_name}_{start_time}_{end_time}.mp4"
    output_path = os.path.join(output_folder, output_filename)
    out = cv2.VideoWriter(output_path, fourcc, fps, (width, height))
    
    # 시작 프레임으로 이동
    cap.set(cv2.CAP_PROP_POS_FRAMES, start_time * fps)
    
    # 프레임 추출 및 저장
    for _ in range((end_time - start_time) * fps):
        ret, frame = cap.read()
        if not ret:
            break
        out.write(frame)
    
    # 리소스 해제
    cap.release()
    out.release()
    
    return output_path

# 사용 예시
input_video = "video path"
output_folder = "extract_video"
os.makedirs(output_folder, exist_ok=True)

for segment in response['time']:
    extracted_path = extract_video_segment(input_video, segment['start'], segment['end'], output_folder, response['food_name'])
    print(f"추출된 비디오: {extracted_path}")

 

그다음은 비디오를 세그먼트로 정리해 주는 단계입니다. main 코드에서는 조금 다르긴 한데 영상을 파싱해온 값을 활용해서 자르고 저장하는 방식입니다. 이 과정에서 1시간을 날렸는데 Streamlit에서 자꾸 동영상이 실행이 안 되길래 왜 그러지 하고 api 만 여러 번 호출시키고 테스트를 해본 결과 코덱의 문제였습니다..

꼭 수정해서 쓰시길 바랍니다. 메인코드에 공유는 해드리겠지만 이것 때문에 진작 끝날게 엄청 길어졌습니다.

import pandas as pd

def create_metadata_table(extracted_videos, food_name):
    metadata = []
    for video in extracted_videos:
        metadata.append({
            'video_path': video,
            'food_name': food_name
        })
    
    df = pd.DataFrame(metadata)
    df.to_csv('metadata.csv', index=False)
    print("메타데이터가 저장되었습니다.")
    return df

# 사용 예시
extracted_videos = [f for f in os.listdir(output_folder) if f.endswith('.mp4')]
metadata_df = create_metadata_table([os.path.join(output_folder, v) for v in extracted_videos], response['food_name'])

 

Document를 만들어도 되지만 또 활용할 일이 있지 않을까 싶어서 df로 만들고 csv 형태로 저장해서 메타데이터를 만들었습니다.

 

from langchain_community.document_loaders.csv_loader import CSVLoader
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import CharacterTextSplitter
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA

# CSV 파일 로드
loader = CSVLoader(file_path='metadata.csv', encoding='utf-8')
documents = loader.load()

# 텍스트 분할
text_splitter = CharacterTextSplitter(chunk_size=1000, chunk_overlap=0)
docs = text_splitter.split_documents(documents)

# 임베딩 및 벡터 저장소 생성
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(docs, embeddings)

# 검색 기반 QA 체인 생성
qa = RetrievalQA.from_chain_type(llm=OpenAI(), chain_type="stuff", retriever=vectorstore.as_retriever())

# 사용 예시
query = "어떤 음식이 영상에 나오나요?"
result = qa.invoke({"query": query})
print(result)

 

csv 로더를 통해로드를 해주고 일반적으로 기본적인 navie RAG를 구성해서 처리를 하였습니다.

 

{'query': '어떤 음식이 영상에 나오나요?', 'result': '  딸기 크림 빙수'}

 

이렇게 결과가 나오는걸 잘 보고 흠칫하고 이거 내가 잘만 구성하면 진짜 프로젝트 가능하겠다 생각이 들었습니다.

 

그 후에 Router LLM을 붙이려고 시도해 봤는데 잘 안 붙어서 그건 나중에 추가하기로 했습니다. 그 후 유저가 결국 동영상을 보고 싶거나 안 보고 싶은걸 판단을 해야 하는데 보고 싶을 때 저 관련영상을 출력시키는 게 목표였습니다. 그래서 Agent를 도입해서 분기판단을 해주는 로직을 구성을 했습니다. CrewAI를 통해 쉽게 구현을 하였고 랭그래프 썼으면 하루종일 걸렸을 텐데 라는 생각과 함께.. 빠른 구현을 하였습니다.

# Crew 생성
from crewai import Agent, Task, Crew

research_agent = Agent(
    role='Video Recommender',
    goal='Determine if the user wants to watch a video based on the query',
    backstory="""You are an AI agent responsible for analyzing the user's query
    and deciding whether or not they want to watch a video.""",  # 사용할 LLM 지정
    verbose=True
)
# Task 정의: 동영상 추천 여부 판단
task = Task(
    description='{query}에 대하여 동영상을 틀지 말지 결정하는 작업',
    expected_output='0 if the user does not want to watch a video, 1 if the user wants to watch a video',
    agent=research_agent,
)
crew = Crew(
    agents=[research_agent],
    tasks=[task],
    verbose=True
)

# Crew 실행
result = crew.kickoff(
    inputs=dict(
        query="아 그래 ? 나는 백종원이 먹는 동영상을 보고싶은데"
    )
)

result.raw

 

task는 1과 0으로 답변을 하는 간단한 Agent입니다. 예로 들어서 질문에 상관없는 얘기도 이런 로직을 써서 처리가 충분히 가능합니다. 일종의 moderation api를 범용적이게 구현을 한 느낌입니다. (물론 돈이 들지만)

 

[2024-09-05 15:33:35][DEBUG]: == Working Agent: Video Recommender

[2024-09-05 15:33:35][INFO]: == Starting Task: 아 그래? 나는 백종원이 먹는 동영상을 보고 싶은데 대하여 동영상을 틀지 말지 결정하는 작업
> Entering new CrewAgentExecutor chain... I now can give a great answer Final Answer: 1
> Finished chain. [2024-09-05 15:33:36][DEBUG]: == [Video Recommender] Task output: 1

 

위처럼 구현을 해주면 끝입니다. 그 후 깃허브에 굴러다니는 챗봇 코드를 복붙 해서 Cursor와 함께 간단한 챗봇형태로 구현을 해냈습니다.

 

streamlit

 

동영상을 업로드하면 분석을 하게 됩니다.

 실제로 호떡이 있어서, 잘 나오는 모습입니다.

파일시스템엔 위처럼 저장이 되어있습니다. 먹는 짤과 함께 이름들이 메타데이터로 저장이 되어있습니다.

 

이렇게 관련된 질문을 하면 유사도기반으로 벡터디비에서 관련 metadata와 주소를 가져와서 그 주소를 리턴해서 동영상이 틀어지는 모습입니다.

물론 이상한 쿼리에 따라 이상하게 답변도 하는 때가 있습니다 ㅋㅋ 데이터구조를 잘 짜야하는 부분이 여기서 느껴졌습니다.

 

그래도 엔간해선 연관 있게 답변을 잘 출력시켜 주고 해당영상도 같이 잘 출력시켜 줍니다. 

 

 

 

하나 추가로 아키텍처적으로 고민할게 있다면, 여러개의 동영상이 나오는 경우에는 어떻게 벡터디비에 넣고 처리할지도 고민해봐야하는 부분이 있을듯 합니다. metadata가 늘어나면 늘어날수록 벡터들간의 거리가 늘어나서 상관은 없겠지만 단순할때는 이상한게 튀어나오는 경우도 있었습니다. 처음에 얘가 고유하게 여러음식들의 정보는 없기 때문에 "영상에서 나오는 음식들은 뭐야?" 라는 쿼리에 답변이 한개밖에 안나왔습니다. 제가 일단 서치값을 1로 설정한 것도 있었고, 데이터가 너무적기도 한 영향이 있을텐데요. 그래서 나중에 프롬프트로 {foods}를 해서 context에 넣어서 처리를 했습니다. 그때 이제 무슨음식이 들어간지 알게 되었는데 대규모 / 소규모 효율적인 알고리즘을 고려해봐야 할 것 같았습니다.

 

결과적으로 많이 다듬어야 하겠지만 Louter LLM으로 쿼리에 따른 비용최적화도 고려해야 하고, 영상도 비용을 줄 일수 있게 용량을 줄여서 처리하던지, 다양한 방법들이 생각들이 났습니다. 그리고 내부의 json 구조화를 잘 정리하고 메타데이터를 잘 정리하면 진짜 의미 있는 멀티모달 챗봇이 충분히 가능성 있겠다 생각이 들었습니다. 물론 메타데이터만 구조화만 잘 시켜도 충분히 유사도기반으로 다른 리소스 없이 잘 나오기 때문에 결국 들어가는 데이터가 중요한 것 같습니다. 

비록 맛보기를 구현했지만 코드가 장황해져서 Github에 업로드를 해놓겠습니다. 하면서도 재밌어서 몰두를 할 수 있는 순기능의 프로젝트였습니다. 점차 고도화해서 스터디가 끝날때는 Solution 으로 판매할 정도로 좋은 산출물이 나오면 좋겠습니다