2주차 스터디까지 뭘 할지 갈피를 못 잡고 있는데, 스터디장 박정기님과 브랜든님의 조언 중, “툴 자체에 매몰되지 말고, 목적이 무엇인지를 분명히 해서 그에 필요한 툴을 배워보세요” 라는 코멘트에 꽂혔습니다.
육아 대디로서 요즘 최대 관심사인 육아 관련 챗봇을 만들어 보기로 결정!
“삐뽀삐뽀 119 소아과”라는 소아과 의사가 집필한 육아 관련 유명 도서가 있습니다.(간혹 육아 바이블로 여겨지기도...)
집필자인 의사 쌤은 유튜브도 운영 중인데, 이 유트브의 모든 내용을 학습시켜서 육아와 관련된 궁금한 점이 있으면 영상을 일일히 뒤져보지 않아도 챗봇에게 답을 얻을 수 있는 육아도우미 AI챗봇을 만들어보려고 합니다.
(진행 요약) 1. ChatGPT 4o canvas에 전체 프로세스를 물어보고, 이에 대한 코드 생성 진행 2. 정리된 코드를 cursor에 입력하고 코드리뷰, 수정 및 보완을 요청 3. 로컬 환경에서 코드를 실행하며 오류를 cursor와 GPT를 이용해 다시 수정 4. 프로세스나 코드 로직에 대해 궁금한 점들은 GPT에게 설명을 요청
(결과를 우선 공개하고, 세부 진행방법은 결과 밑에 있습니다.)
실행 결과
필요한 API key와 유튜브 채널ID를 입력하고 코드를 실행하니, 결과가 잘 나온 듯 합니다.
메시지가 표시된 컴퓨터 화면의 스크린샷
(위에 나온 창에 질문을 입력하면 아래와 같이 답변을 생성해줍니다.)
질의/응답 테스트 결과 1 (몇 가지 직접 봤던 영상들인데, 알맞은 결과가 나온 듯 합니다.)
검정색 배경에 한국어 텍스트
한국어로 된 문자 메시지의 스크린샷
검정색 배경에 한국어 텍스트
질의/응답 테스트 결과 2 (만약 관련 없는 이상한 질문을 한다면?)
노란색 원이 있는 검은색 화면
채널의 내용과 관련 없는 질문을 하니, 아주 일반적이고 전형적인 답변이 나왔습니다. 하지만, 출처의 결과를 보니... 할루시네이션 같네요. (차후 수정해야할 부분ㅠ)
세부 진행방법
1-1. ChatGPT 4o canvas에 나의 목적을 전달하고,
한국어 텍스트가 있는 검은 화면
롱차트 AP
1-2. 캔바스를 열어, 위 절차 전체에 대해 파이썬 코드 생성을 요청하여 아래 결과를 얻었습니다.
좋아, 이 프로젝트를 쉽게 따라갈 수 있도록 각 단계를 상세히 설명할게. 아래는 각 단계별로 필요한 지식과 진행 방법이야. 차근차근 따라가면 충분히 완성할 수 있을 거야!
### 1. 유튜브 데이터 수집 및 자막 추출 - Python 코드 예시
```python
import os
import googleapiclient.discovery
import whisper
# 1.1 YouTube API를 사용해 데이터 수집
API_KEY = 'YOUR_YOUTUBE_API_KEY'
youtube = googleapiclient.discovery.build("youtube", "v3", developerKey=API_KEY)
# 채널의 모든 영상 리스트 가져오기
def get_video_list(channel_id):
request = youtube.search().list(
part="snippet",
channelId=channel_id,
maxResults=50
)
response = request.execute()
return response.get("items", [])
# 1.2 Whisper를 사용해 자막 생성하기
def transcribe_video(video_path):
model = whisper.load_model("base")
result = model.transcribe(video_path)
return result["text"]
# 예시 사용
youtube_videos = get_video_list('YOUR_CHANNEL_ID')
for video in youtube_videos:
print(f"Title: {video['snippet']['title']}")
print(f"Description: {video['snippet']['description']}")
```
### 2. 텍스트 전처리 - Python 코드 예시
```python
from transformers import pipeline
# 2.1 텍스트 요약
summarizer = pipeline("summarization")
def summarize_text(text):
summary = summarizer(text, max_length=100, min_length=30, do_sample=False)
return summary[0]['summary_text']
# 자막 요약 예시
transcribed_text = "이곳에 Whisper로 추출된 자막을 넣으세요."
summarized_text = summarize_text(transcribed_text)
print("Summarized Text: ", summarized_text)
# 2.2 토픽 분류 - Python 코드 예시
classifier = pipeline("zero-shot-classification")
def classify_topic(text):
labels = ["육아", "건강", "교육", "가족"]
result = classifier(text, candidate_labels=labels)
return result['labels'][0]
# 주제 분류 예시
topic = classify_topic(summarized_text)
print("Topic: ", topic)
```
### 3. LangChain과 ChatGPT API 활용 - Python 코드 예시
```python
import openai
from langchain import ConversationChain
# 3.1 LangChain 설정 및 사용
conversation = ConversationChain()
conversation.add_input("사용자의 질문을 여기 넣으세요.")
response = conversation.run()
print("Response from LangChain: ", response)
# 3.2 ChatGPT API 사용
openai.api_key = "YOUR_OPENAI_API_KEY"
def chat_with_gpt(prompt):
response = openai.Completion.create(
engine="text-davinci-003",
prompt=prompt,
max_tokens=150
)
return response.choices[0].text.strip()
# 예시 사용
user_question = "유아의 성장 발달 단계에 대해 알려주세요."
gpt_response = chat_with_gpt(user_question)
print("GPT Response: ", gpt_response)
```
### 4. 데이터베이스 구축 - Python 코드 예시
```python
import pinecone
import openai
# 4.1 Pinecone 설정
pinecone.init(api_key="YOUR_PINECONE_API_KEY", environment="us-west1-gcp")
index = pinecone.Index("youtube-video-index")
# 벡터 생성 및 저장
def store_embeddings(text):
embedding = openai.Embedding.create(input=text, model="text-embedding-ada-002")
index.upsert([(text, embedding['data'][0]['embedding'])])
# 예시 사용
store_embeddings(summarized_text)
# 4.2 질의응답 최적화 - 벡터 검색
def search_similar_text(query):
query_embedding = openai.Embedding.create(input=query, model="text-embedding-ada-002")
results = index.query(query_embedding['data'][0]['embedding'], top_k=5)
return results
search_results = search_similar_text(user_question)
print("Search Results: ", search_results)
```
### 5. 챗봇 개발 및 배포 - Python 코드 예시
```python
from flask import Flask, request, jsonify
import openai
app = Flask(__name__)
openai.api_key = "YOUR_OPENAI_API_KEY"
# 5.1 Flask를 사용해 챗봇 서버 개발
@app.route("/chat", methods=["POST"])
def chat():
user_input = request.json.get("message")
response = chat_with_gpt(user_input)
return jsonify({"response": response})
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000)
```
### 6. 추가 기능 구현 - Python 코드 예시
```python
# 6.1 사용자 맞춤형 추천 시스템
user_interests = ["건강", "육아"]
def recommend_videos(user_interests):
recommended_videos = []
for video in youtube_videos:
topic = classify_topic(video['snippet']['description'])
if topic in user_interests:
recommended_videos.append(video['snippet']['title'])
return recommended_videos
print("Recommended Videos: ", recommend_videos(user_interests))
# 6.2 감정 분석 - Python 코드 예시
from transformers import pipeline
sentiment_analyzer = pipeline("sentiment-analysis")
def analyze_sentiment(feedback):
result = sentiment_analyzer(feedback)
return result[0]
feedback = "이 영상은 정말 유익했어요!"
sentiment = analyze_sentiment(feedback)
print("Sentiment: ", sentiment)
```
---
이제 각 단계별로 필요한 부분을 하나씩 진행해보면서 실습해보자. 진행하면서 막히는 부분이 있으면 언제든지 질문해줘!
2-1. 위에서 얻은 GPT Canvas의 답변을 복사해서 cursor에 그대로 넣고, 원하는 목적과 함께 코드의 리뷰, 수정 및 보완을 해달라고 요청
검은 화면에 한국어 텍스트 스크린샷
위와 같이 답변과 함께 수정된 코드를 받은 것... 같지만, 파이썬 초보자에 VScode도 아니고 주피터 노트북을 쓰는 저에겐 활용하기 어려운 결과였습니다.
그래서 작성된 코드를 전부 순차적으로 정리하여 주피터 노트북에서도 실행할 수 있도록 작성해달라 요청해보았습니다.
한국어 프로그래밍 언어의 스크린샷
2-2. 생성된 코드를 한번에 쭉 작성하도록 요청하고(cursor), 이를 주피터 노트북에서 실행하면서 오류나 추가 요구사항 등을 cursor와 GPT에 수정 요청해가며 코드를 완성
(오류가 발생한 경우와 추가 요구사항이 있는 경우엔 코드 수정을 요청했고, 진행 과정 확인이 어려운 경우엔 진행 코드 전체를 입력하여 추후 절차에 대해 추가 코드 작성을 요청함)
<최종 완성 코드>
import os
from dotenv import load_dotenv
from googleapiclient.discovery import build
from youtube_transcript_api import YouTubeTranscriptApi
import openai
import numpy as np
import faiss
import pandas as pd
from tqdm import tqdm
import time
import logging
# 로깅 설정
logging.basicConfig(level=logging.INFO)
# .env 파일 로드
load_dotenv()
# 환경변수 설정
YOUTUBE_API_KEY = os.getenv('YOUTUBE_API_KEY')
OPENAI_API_KEY = os.getenv('OPENAI_API_KEY')
# OpenAI API 키 설정
openai.api_key = OPENAI_API_KEY
class YouTubeService:
def __init__(self):
self.youtube = build('youtube', 'v3', developerKey=YOUTUBE_API_KEY)
def get_video_list(self, channel_id):
videos = []
next_page_token = None
try:
while True:
request = self.youtube.search().list(
part="snippet",
channelId=channel_id,
maxResults=50,
pageToken=next_page_token,
type="video"
)
response = request.execute()
videos.extend(response.get("items", []))
next_page_token = response.get("nextPageToken")
if not next_page_token:
break
return videos
except Exception as e:
logging.error(f"비디오 목록 가져오기 실패: {str(e)}")
return []
def get_video_transcript(video_id):
try:
transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=['ko'])
return ' '.join([entry['text'] for entry in transcript])
except Exception as e:
logging.error(f"자막 가져오기 실패 (비디오 ID: {video_id}): {str(e)}")
return None
def create_embedding(text):
try:
# 텍스트가 너무 길 경우 잘라내기 (토큰 제한)
max_tokens = 8000
if len(text.split()) > max_tokens:
text = ' '.join(text.split()[:max_tokens])
response = openai.Embedding.create(
input=text,
model="text-embedding-ada-002"
)
return response['data'][0]['embedding']
except Exception as e:
logging.error(f"임베딩 생성 실패: {str(e)}")
return None
def process_videos_batch(videos, batch_size=10):
"""비디오를 배치로 처리"""
all_data = []
for i in range(0, len(videos), batch_size):
batch = videos[i:i + batch_size]
batch_data = []
for video in tqdm(batch, desc=f"배치 처리 중 ({i+1}-{i+len(batch)}/{len(videos)})"):
video_id = video['id']['videoId']
title = video['snippet']['title']
description = video['snippet']['description']
# 자막 가져오기
transcript = get_video_transcript(video_id)
if not transcript:
continue
# 임베딩 생성
embedding = create_embedding(transcript)
if not embedding:
continue
batch_data.append({
"video_id": video_id,
"title": title,
"description": description,
"transcript": transcript,
"embedding": embedding
})
# API 제한 방지를 위한 짧은 딜레이
time.sleep(0.5)
all_data.extend(batch_data)
# 배치 간 딜레이
time.sleep(2)
# 중간 결과 저장
save_progress(all_data)
return all_data
def save_progress(data):
"""진행 상황 저장"""
df = pd.DataFrame([{
'video_id': item['video_id'],
'title': item['title'],
'description': item['description'],
'transcript': item['transcript']
} for item in data])
df.to_csv("youtube_videos_progress.csv", index=False)
# 임베딩 별도 저장
embeddings = np.array([item['embedding'] for item in data])
np.save("embeddings_progress.npy", embeddings)
def main():
# YouTube 서비스 초기화
youtube_service = YouTubeService()
# 채널 ID 설정
channel_id = "UC6t0ees15Lp0gyrLrAyLeJQ"
# 비디오 목록 가져오기
logging.info("비디오 목록을 가져오는 중...")
videos = youtube_service.get_video_list(channel_id)
# 배치 처리
all_data = process_videos_batch(videos, batch_size=10)
logging.info(f"총 {len(all_data)}개의 비디오가 처리되었습니다.")
return all_data
if __name__ == "__main__":
processed_data = main()
# 저장된 데이터 불러오기
df = pd.read_csv("youtube_videos_progress.csv")
embeddings = np.load("embeddings_progress.npy")
# FAISS 인덱스 생성 및 데이터 로드
dimension = len(embeddings[0])
index = faiss.IndexFlatL2(dimension)
index.add(embeddings.astype('float32'))
def create_embedding(text):
"""텍스트의 임베딩 생성"""
try:
response = openai.Embedding.create(
input=text,
model="text-embedding-ada-002"
)
return response['data'][0]['embedding']
except Exception as e:
print(f"임베딩 생성 실패: {str(e)}")
return None
def get_relevant_context(question, k=3):
"""질문과 관련된 컨텍스트 검색"""
# 질문 임베딩 생성
question_embedding = create_embedding(question)
if question_embedding is None:
return []
# 유사한 컨텍스트 검색
D, I = index.search(np.array([question_embedding]).astype('float32'), k)
# 결과 구성
results = []
for idx, distance in zip(I[0], D[0]):
results.append({
'text': df.iloc[idx]['transcript'],
'title': df.iloc[idx]['title'],
'distance': float(distance)
})
return results
def chat_with_gpt(prompt):
"""GPT와 대화"""
try:
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": "육아 관련 전문가입니다."},
{"role": "user", "content": prompt}
],
max_tokens=500, # 응답 길이 증가
temperature=0.7
)
return response.choices[0].message['content']
except Exception as e:
print(f"GPT 응답 생성 실패: {str(e)}")
return "죄송합니다. 일시적인 오류가 발생했습니다."
def ask_question(question):
"""질문에 대한 답변 생성"""
# 관련 컨텍스트 검색
contexts = get_relevant_context(question)
if not contexts:
return "죄송합니다. 관련 정보를 찾을 수 없습니다."
# 프롬프트 구성
context_text = "\n\n".join([
f"영상: {ctx['title']}\n내용: {ctx['text'][:500]}..." # 컨텍스트 길이 제한
for ctx in contexts
])
prompt = f"""다음 컨텍스트를 바탕으로 질문에 답변해주세요.
답변 후에는 반드시 참고한 영상의 제목을 [출처] 형태로 표시해주세요.
컨텍스트:
{context_text}
질문: {question}
답변 형식:
[답변]
(여기에 답변 내용)
[출처]
- (영상 제목 1)
- (영상 제목 2) (있는 경우)
- (영상 제목 3) (있는 경우)"""
return chat_with_gpt(prompt)
# 대화형 인터페이스
print("\n=== 육아 관련 질문답변 시스템 ===")
print("종료하려면 'q' 또는 'quit'를 입력하세요.")
print("질문을 입력해주세요:")
while True:
question = input("\n질문: ").strip()
if question.lower() in ['q', 'quit', '종료']:
print("\n대화를 종료합니다. 감사합니다!")
break
if not question:
print("질문을 입력해주세요!")
continue
print("\n답변을 생성하는 중입니다...")
response = ask_question(question)
print("\n" + response)
(참고사항)
중간에 유튜브 특정 채널의 정보 수집을 위해 해당 채널 고유의 채널ID를 입력해야합니다. 이것 역시 ChatGPT에게 물어봐서 아래의 방법으로 채널ID를 추출했습니다.
추후 보완 사항들...
채널 내 영상 정보를 전부, 잘 수집했는지 확인 필요 (채널 내 전체 영상에 대해 관련 정보와 스크립트가 잘 저장되었는지 확인)
필요시, 챗봇의 답변 생성에 대한 세부 프롬프트 수정 (출처 영상 정보 제공시, 게시 날짜나 관련 URL 등을 함께 제공하기 등)
