유튜브 영상 기반 RAG 챗봇 구현

들어가며,

유튜브 영상을 대상으로 스크립트를 추출하고, Pinecone + LangChain + Upstage Embeddings를 활용한 RAG 챗봇을 직접 구현했다. 이번 글에서는 코드와 함께 설계 과정을 공유하려고 한다.
✅ 개발 능력 향상을 위해 되도록 공식문서를 보며 개발했다.

1. 스크립트 분할
2. 벡터 스토어 구축
3. Retriever 생성
4. RAG chain 구성

구현과정

1. 스크립트 분할

def document_list(text):
  text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=100)
  document_list = text_splitter.create_documents([text])

  return document_list

청크 사이즈를 500으로 정했다. 더 작게도 해봤는데 아무래도 유튜브 영상 특성상 구어체이다 보니 불필요한 내용이 많았다. 문맥 손실도 줄이고, 세부 정보도 담아낼 수 있는 크기로 500이 적절하다고 판단했다.

2. 벡터 스토어 구축

def vector_store(video_id, document_list):
  embeddings = UpstageEmbeddings(api_key= os.getenv("UPSTAGE_API_KEY"), model= "solar-embedding-1-large")
  index_name = 'script-index'
  pinecone_api_key = os.environ.get("PINECONE_API_KEY")
  pc = Pinecone(api_key=pinecone_api_key)
  index = pc.Index(index_name)

    # 네임스페이스 존재 여부 확인
  stats = index.describe_index_stats()
  namespaces = stats.get("namespaces", {})
  if video_id in namespaces:
      print(f"Namespace '{video_id}' already exists.")
      return

  database = PineconeVectorStore(index=index, embedding=embeddings, namespace=video_id)
  uuids = [str(uuid4()) for _ in range(len(document_list))]
  database.add_documents(documents=document_list, ids=uuids)

임베딩은 업스테이지 임베딩을 사용했다. 무료라서 사용하는 중이다. 아직 모델별 차이는 모르겠다.
벡터 스토어는 파인콘의 벡터스토어를 사용했고, 하나의 인덱스에 유튜브 video ID를 네임스페이스로 사용하여 분리하였다. 추가적으로 video ID로 중복 방지까지 해놨다.

중복된 데이터가 들어가게 되면, 리트리브할때 같은 문장이 높은 유사도로 측정돼서 검색된 문장에 같은 문장이 들어있다.

3. Retriever 생성

def get_retriever(video_id):
  embeddings = UpstageEmbeddings(api_key= os.getenv("UPSTAGE_API_KEY"), model= "solar-embedding-1-large")
  index_name = 'script-index'
  pinecone_api_key = os.environ.get("PINECONE_API_KEY")
  pc = Pinecone(api_key=pinecone_api_key)
  database = PineconeVectorStore.from_existing_index(index_name=index_name, embedding= embeddings, namespace=video_id)

  retriever = database.as_retriever(search_kwargs={'k':2, "namespace": video_id})

  return retriever

리트리버 역시 video ID를 활용해서 생성하도록 했고, K 값은 일단 2로 정해놓았다. 청크 사이즈를 500으로 하기도 했고, 테스트에서 충분히 답변을 할 수 있는 정도였기에 k=2 로도 충분하다고 판단했다.
추후에 RAG 평가에 대해 공부 좀 하고 다시 조절하든지 할 것이다.

4. RAG chain 구성

few-shot 프롬프트 기법을 사용하고, 대화 히스토리 관리를 포함했다.
4.1 세션 대화 기록 관리

store = {}
def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

세션 아이디를 통해 딕셔너리에 히스토리를 저장한다. 인메모리라 재시작하면 사라진다. 비휘발적으로 저장하고 싶으면 레디스 사용하면 된다.

4.2 히스토리 기반 리트리버

def get_history_retriever(video_id):
  llm = get_llm()
  retriever = get_retriever(video_id)

  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."
  )

  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

https://python.langchain.com/v0.2/docs/how_to/qa_chat_history_how_to/

4.3 RAG 체인과 히스토리 결합

def get_rag_chain(video_id):
  llm = get_llm()
  example_prompt = ChatPromptTemplate.from_messages(
    [
      ("human", "{input}"),
      ("ai", "{answer}")
    ]
  )
  few_shot_prompt = FewShotChatMessagePromptTemplate(
      example_prompt=  example_prompt,
      examples= answer_examples,
  )
  system_prompt = (
    "당신은 유튜브 영상에 관련된 전문가입니다. 유튜브 영상에 관련된 모든 것을 답변해드립니다."
    "아래에 제공된 문서를 활용해서 답변해주세요."
    "답변을 알 수 없다면 모른다고 답변해주세요."
    "2-3 문장 정도의 한국어 답변을 원합니다."
    "{context}"
  )
  qa_prompt = ChatPromptTemplate.from_messages(
    [
      ("system", system_prompt),
      few_shot_prompt,
      MessagesPlaceholder("chat_history"),
      ("human", "{input}"),
    ]
  )
  question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)
  history_aware_retriever = get_history_retriever(video_id)
  rag_chain = create_retrieval_chain(history_aware_retriever, question_answer_chain)

  conversation_rag_chain = RunnableWithMessageHistory(
    rag_chain,
    get_session_history,
    input_messages_key="input",
    output_messages_key="answer",
    history_messages_key="chat_history",
  ).pick("answer")

  return conversation_rag_chain

few-shot 프롬프트 기법으로 예시 3개를 넣어줬다.

RAG chain의 흐름을 보자면

1. 히스토리 리트리버는 현재 질문에 대한 검색에 쓰이지 않고, 작성해준 프롬프트를 기반으로 하여 대화 기록을 포함해 현재 맥락 의존적인 질문을 독립적인 형태로 재작성한다.
   (예를 들어 사용자가 "그 사람은 뭐라고 했어?" 라고 물으면, "저자 A가 영상에서 말한 방법은 무엇인가?"처럼 명확한 질문으로 변환한다.)

2. 이렇게 추출된 문맥은 QA 체인 단계에서 시스템 프롬프트, few-shot 예시, 이전 대화 기록, 그리고 사용자의 질문과 함께 LLM에 주입된다.

따라서 최종적으로 모델은 검색된 문서 + 대화 맥락 + 규칙을 모두 반영한 답변을 생성하게 되고, 이 답변은 다시 히스토리에 저장되어 후속 질문에도 활용된다.

def get_ai_response(user_input, video_id):
  rag_chain = get_rag_chain(video_id)

  ai_response = rag_chain.stream({
    "input": user_input},
    config= {
      "configurable": {"session_id": "abc123"}
    })
  
  return ai_response

video ID로 rag chain을 생성하고 사용자에 질문에 맞는 답변을 스트림으로 생성해준다.

마무리

유튜브 영상 요약, 영상 기반 챗봇을 모두 완성했다. 예외처리를 추가하고, streamlit Cloud에서 제공하는 배포 기능을 통해 배포할 예정이다.
배포 후에는 RAG 평가, 개선 과정을 거칠 예정이다.