【LangChain】GPTにYoutube動画を扱える能力を注入

はじめに

こんにちは。スペースマーケットでインターンしています、dumbled0reです。
一ヶ月前に大学を卒業して韓国から帰って来ましたが、韓国はマイナス10℃とかの世界なので日本は暖かいなと感じています。韓国は寒い。

僕は普段からYoutubeで動画を見ていますが、動画が長いと見たい箇所に辿り着くまで少しずつ飛ばしながら動画を見ていることがあります。これ結構めんどくさいなと思っています。
なので、LangChainを用いて〇〇については動画の何分くらいからなのか尋ねたら時間を教えてくれたり、要約してくれるなどの能力をGPTに与えてみました。

事前準備

OpenAI API Keyの発行

OpenAI API Keyの発行をお願いします。取得したAPIキーは.envファイルに以下のように設定しておいてください。アカウント作成時から最初の3ヶ月間は5ドルの無料クレジットが提供されています。（※2024年1月の情報になります）

.env

OPENAI_API_KEY="APIキー"

環境

今回は以下のライブラリを使用しました。インストールしてください。

• Python 3.11.6
• python-dotenv 1.0.0
• openai 1.8.0
• langchain 0.0.353
• streamlit 1.30.0
• youtube-transcript-api 0.6.2
• pytube 15.0.0

準備が整いました。

全体のコード

全体のコードは以下のようになります。(展開してご覧ください / このままコピペで動くはず)

全体のコード

import json
import math
import os
import re

import streamlit as st
from dotenv import load_dotenv
from langchain.chat_models import ChatOpenAI
from langchain.document_loaders import YoutubeLoader
from langchain.prompts import (ChatPromptTemplate, HumanMessagePromptTemplate,
                               SystemMessagePromptTemplate)
from langchain.schema import AIMessage, HumanMessage, SystemMessage
from youtube_transcript_api import YouTubeTranscriptApi

load_dotenv()

functions = [
    {
        "name": "generate_video_response",
        "description": "質問を抽出する",
        "parameters": {
            "type": "object",
            "properties": {
                "question": {
                    "type": "string",
                    "description": "質問事項 e.g. 動画内容を要約して",
                },
            },
            "required": ["question"],
        },
    },
    {
        "name": "generate_video_time_response",
        "description": "キーワードを抽出する",
        "parameters": {
            "type": "object",
            "properties": {
                "keyword": {
                    "type": "string",
                    "description": "キーワード e.g. 〇〇について、〇〇に関して",
                },
            },
            "required": ["keyword"],
        }
    },
]


def set_up_page() -> None:
    """
    ページの設定とヘッダーを表示
    """

    # ページ上部の部分の設定
    st.set_page_config(
        page_title="Youtube Chatbot",
        page_icon="🤖",
    )

    # ヘッダーの表示
    st.header("Youtube Chatbot 🤖")


def init_session_state(session_state: dict) -> dict[str, list[str]]:
    """セッションステートを初期化

    Args:
        session_state (dict): セッションステート
    Returns:
        dict: 初期化されたセッションステート
    """

    if not session_state:
        session_state.messages = [
            SystemMessage(content="You are a helpful assistant.")
        ]

    return session_state


def display_chat_history(messages: list) -> None:
    """チャット履歴を表示

    Args:
        messages (list): チャットメッセージのリスト
    """

    for message in messages:
        if isinstance(message, AIMessage):
            with st.chat_message('assistant'):
                st.markdown(message.content)
        elif isinstance(message, HumanMessage):
            with st.chat_message('user'):
                st.markdown(message.content)


def extract_number_from_text(text: str) -> int or None:
    """テキストから数字を抽出

    Args:
        text (str): 抽出対象のテキスト
    Returns:
        int|None: テキストから抽出された数字（見つからない場合は None）
    """

    match = re.search(r'\d+', text)

    return int(match.group()) if match else None


def convert_seconds(seconds: int) -> str:
    """秒を分や時間に換算

    Args:
        seconds (int): 換算対象の秒数
    Returns:
        str: 換算結果の数値または"時間:分"の文字列
    """

    minutes = 0
    hours = 0

    if seconds < 60:
        return f"{seconds}秒"

    elif seconds < 3600:
        minutes = math.floor(seconds / 60)
        remaining_seconds = seconds % 60
        return f"{minutes}分{remaining_seconds}秒"

    else:
        hours = math.floor(seconds / 3600)
        minutes = math.floor((seconds % 3600) / 60)
        return f"{hours}時間{minutes}分{seconds % 60}秒"


def split_text_by_time_intervals(json_data, split_duration=60) -> dict[str, dict[str, str]]:
    """与えられた JSON データを指定した時間間隔でテキストを分割し、各チャンクの情報を返す

    Args:
        json_data (list): JSON データのリスト
        split_duration (int): 区切りの秒数
    Returns:
        dict: 各チャンクと時間情報を含む辞書。キーはチャンク番号、値はチャンクと時間情報を含む辞書
    """

    split_texts = list()
    current_text = ""
    current_start = 0
    current_end = split_duration

    for entry in json_data:
        start_time = entry["start"]
        text = entry["text"]

        # テキストが現在の範囲内に収まっているかを確認
        if start_time >= current_start + split_duration:
            split_texts.append({"text": current_text.strip(), "start": current_start, "end": current_end})
            current_text = ""
            current_start += split_duration
            current_end += split_duration

        # テキストを追加
        current_text += text

    # 最後の部分を追加
    if current_text:
        split_texts.append({"text": current_text.strip(), "start": current_start, "end": current_end})

    chunk_dict = {i: chunk_info for i, chunk_info in enumerate(split_texts)}

    return chunk_dict


def call_chatbot_function(llm: ChatOpenAI, question: str) -> dict[str, dict[str, str]]:
　　"""function callingを行うかどうかを判定し、行う場合は関数名と引数を返す
  
  　Args:
    　  llm (ChatOpenAI): ChatOpenAIのインスタンス
  　    question (str): 質問内容
 　 Returns:
　      dict: 関数名と引数を含む辞書。関数を呼び出さない場合は空の辞書
　　"""
    
    messages = llm.predict_messages(
        [HumanMessage(content=question)],
        functions=functions,
    )

    return messages.additional_kwargs


def generate_video_response(llm: ChatOpenAI, question: str, content: str) -> str:
    system_template = "あなたは、質問者からの質問を回答するAIです。"
    human_template = """
        以下のテキストを元に「{question}」についての質問に答えてください。

        {content}
    """

    system_message_prompt = SystemMessagePromptTemplate.from_template(system_template)
    human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)
    chat_prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])
    prompt_message_list = chat_prompt.format_prompt(
        question=question,
        content=content).to_messages()
    response = llm(prompt_message_list)

    return response


def generate_video_time_response(llm: ChatOpenAI, keyword: str, chunk_dict: dict[str, dict]) -> str:
    system_template = "あなたは、質問者からの質問を回答するAIです。"
    human_template = """
    キーワード: {keyword}

    jsonデータ:
    --------------------
    {chunk_dict}
    --------------------

    上記のjsonデータの中から、キーワード「{keyword}」と最も関連性が高いtextのインデックスを答えてください。

    回答の形式は
    「{keyword}の説明は{index}番です。」
    としてください。
    もしも、{keyword}の説明がない場合は「{keyword}の説明は動画内にありません。」としてください。
    """

    system_message_prompt = SystemMessagePromptTemplate.from_template(system_template)
    human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)
    chat_prompt = ChatPromptTemplate.from_messages(
        [system_message_prompt, human_message_prompt])
    prompt_message_list = chat_prompt.format_prompt(
        keyword=keyword,
        chunk_dict=chunk_dict,
        index="インデックス").to_messages()
    response = llm(prompt_message_list)

    return response


def main() -> None:
    llm = ChatOpenAI(openai_api_key=os.environ.get("OPENAI_API_KEY"),
                     model_name="gpt-3.5-turbo-16k",
                     temperature=0)
    set_up_page()
    session_state = init_session_state(st.session_state)

    url = st.text_input("Youtube URL: ", key="input")
    if url:
        with st.spinner("Fetching Content ..."):
            loader = YoutubeLoader.from_youtube_url(
                url,
                add_video_info=True,
                language=['ja']
            )
            document = loader.load()
            video_id = document[0].metadata["source"]
            content = document[0].page_content
            transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=["ja"])
            chunk_dict = split_text_by_time_intervals(transcript)

    if user_input := st.chat_input("聞きたいことを入力してね！"):
        session_state.messages.append(HumanMessage(content=user_input))
        with st.spinner("Chatbot is typing ..."):
            additional_kwargs = call_chatbot_function(llm, user_input)
            if additional_kwargs:
                if additional_kwargs["function_call"]["name"] == "generate_video_response":
                    question = json.loads(additional_kwargs["function_call"]["arguments"]).get("question")
                    response = generate_video_response(llm, question, content)
                    session_state.messages.append(AIMessage(content=response.content))

                elif additional_kwargs["function_call"]["name"] == "generate_video_time_response":
                    keyword = json.loads(additional_kwargs["function_call"]["arguments"]).get("keyword")
                    response = generate_video_time_response(llm, keyword, chunk_dict)
                    index = extract_number_from_text(response.content)
                    if index is not None:
                        start = convert_seconds(chunk_dict[index]["start"])
                        end = convert_seconds(chunk_dict[index]["end"])
                        answer = f"{keyword}の説明は動画の{start}から{end}です。"
                        session_state.messages.append(AIMessage(content=answer))
                    else:
                        session_state.messages.append(AIMessage(content=response.content))
            else:
                response = llm(session_state.messages)
                session_state.messages.append(AIMessage(content=response.content))

    messages = session_state.get('messages', [])
    display_chat_history(messages)


if __name__ == '__main__':
    main()
    

LangChainとは

https://js.langchain.com/docs/get_started/introduction/
LangChainとは大規模言語モデル（LLM）を使用したアプリケーション開発のためのフレームワークです。つまり、LangChain×ChatGPTでRAG(Retrieval-Augmented Generation)を構築できます。RAGは精度向上技術なので、既存のChatGPTに新しい機能を追加できるよ。ということになります。(多分)

実装

1. やること

GPTに何分から始まっているか回答してほしいので、動画のテキストとそのテキストが動画の何分から何分までのデータなのかを表す時間情報が必要なので取得していきます。あとはいい感じにプロンプトを作成してGPTに投げます。

2. 動画のテキストと時間情報を取得

langchainが提供しているYoutubeLoaderで動画のテキスト情報と動画のIDを取得します。テキスト情報は要約などのために使用し、動画のIDはYouTubeTranscriptApiでテキストとそのテキストに対する開始時間を取得するために使用します。これで動画のテキスト情報とテキストに対する開始時間をゲットです。

loader = YoutubeLoader.from_youtube_url(
                url,
                add_video_info=True,
                language=['ja']
            )
document = loader.load()
video_id = document[0].metadata["source"]
content = document[0].page_content
transcript = YouTubeTranscriptApi.get_transcript(video_id, languages=["ja"])

以下はtranscriptの中身でリストの中にディクショナリでテキストとそのテキストの開始時間(秒)が含まれています。

durationは字幕の表示時間を表していますが、今回は使用していません。
[{'text': 'はい皆様ご視聴ありがとうございます私は', 'start': 0.52, 'duration': 4.68}, {'text': '株式会社スペースマーケットイベント', 'start': 3.58, 'duration': 3.87}...]

3. 時間情報の整形

ただ、上記のtranscriptの中身をみると開始時間しか取得できないので、複数のデータを特定の秒や分で区切ってチャンクに分割します。今回はsplit_durationを60にしているので、1分区切りで分割していきます。

def split_text_by_time_intervals(json_data, split_duration=60) -> dict[str, dict[str, str]]:
    """与えられた JSON データを指定した時間間隔でテキストを分割し、各チャンクの情報を返す

    Args:
        json_data (list): JSON データのリスト
        split_duration (int): 区切りの秒数

    Returns:
        dict: 各チャンクと時間情報を含む辞書。キーはチャンク番号、値はチャンクと時間情報を含む辞書
    """

    split_texts = list()
    current_text = ""
    current_start = 0
    current_end = split_duration

    for entry in json_data:
        start_time = entry["start"]
        text = entry["text"]

        # テキストが現在の範囲内に収まっているかを確認
        if start_time >= current_start + split_duration:
            split_texts.append({"text": current_text.strip(), "start": current_start, "end": current_end})
            current_text = ""
            current_start += split_duration
            current_end += split_duration

        # テキストを追加
        current_text += text

    # 最後の部分を追加
    if current_text:
        split_texts.append({"text": current_text.strip(), "start": current_start, "end": current_end})

    chunk_dict = {i: chunk_info for i, chunk_info in enumerate(split_texts)}

    return chunk_dict

以下はsplit_text_by_time_intervalsを使用した時の返されるchunk_dictの中身です。動画のテキスト情報はYoutubeの字幕を取得しているので、若干おかしな文章が所々ありますがChatGPTに頑張ってもらいましょう。

0: {
'text': 'はい皆様ご視聴ありがとうございます私は株式会社スペースマーケットイベントプロデュースチームの岩崎と申しますどうぞよろしくお願い致します皆様突然ですがこの中でですね急遽普及していったのが winner かと思いますこのロリな参加してみてなんかいかがでしょうかとお思いでしょうか集中してずっと見ているの月代だったり退屈に感じることはないでしょうかそんなみなさまのお悩みを解決するために最近リリースさせていただきましたスペースマーケットオンラインライトについて本日はご紹介させていただきたいと思います弊社ではですねスペースを貸したい人と借りたい人を繋げるプラットフォームの運営をなってますこのスペースを活用してイベントプロデュースも今までオフラインで行ってまいりましたただですねこの子の中でなかなかオフライン開催が難しいということでオンラインに移行してイベントのご提供してまいりましたはいそこでつながってくるんですけれども高等皆さんが聞きした日なぁを参加しているとちょっと海窟してしまうだったり', 
'start': 0, 
'end': 60}, 
1: {
'text': '開催されているかとを集中して観ていただいて離脱率を下げれば難しいだったりお悩みがたくさんあると思いますそんな皆様のメニューを解決するために俺はれスクリーンコミュニケーションという切り口で邦人の皆様にイベントのプロデュースをご提供してまいりましたスクリーンコミュニケーションとはですね従来年にただ本件つを流すだけではなく視聴者の方とのタッチポイントであるスクリーンに最適化した配信クリエイティブでコミュニケーションをとるは特性になっています画面レイアウトや絵作りテロップワークブーツ某所やムービーエースなどなど今日紹介するスペースマーケットオンラインライトなんですがスクリーンコミュニケーションで設定されたオンライン配信をより多くの方々に背コースとで体験していただくために弊社パートナーであるニューピーク者と強度0開発したスタジオパッケージプランになるんですねここ new ピークスタジオは', 
'start': 60, 
'end': 120}...

これでテキストとそのテキストが何分から何分までの時間情報が入ったデータを作成できました。このデータをもとにChatGPTに聞いてみます。

4. ChatGPTに回答してもらう

OpenAI APIのfunction callingを使用して入力された質問からキーワードを抽出してgenerate_video_time_responseにキーワードと先ほど作成したchunk_dictを渡します。function callingって何？って思った方のために簡単に説明します。function callingとは入力された質問に対して、特定の関数を実行するための引数をChatGPTが考えて渡してくれる機能のことです。以下のfunctionsのようにfucntion callingを使用したい関数名(name)と関数を実行するためのに必要な引数をparameters属性にセットしておきます。descriptionにe.g.などをつけておくとよりChatGPTが汲み取りやすくなります。

functions = [
    {
        "name": "generate_video_time_response",
        "description": "キーワードを抽出する",
        "parameters": {
            "type": "object",
            "properties": {
                "keyword": {
                    "type": "string",
                    "description": "キーワード e.g. 〇〇について、〇〇に関して",
                },
            },
            "required": ["keyword"],
        }
    },
]

果たしてこのプロンプトでほんとにいいのかは気になります。

def generate_video_time_response(llm: ChatOpenAI, keyword: str, chunk_dict: dict[str, dict]) -> str:
    system_template = "あなたは、質問者からの質問を回答するAIです。"
    human_template = """
    キーワード: {keyword}

    jsonデータ:
    --------------------
    {chunk_dict}
    --------------------

    上記のjsonデータの中から、キーワード「{keyword}」と最も関連性が高いtextのインデックスを答えてください。

    回答の形式は
    「{keyword}の説明は{index}番です。」
    としてください。
    もしも、{keyword}の説明がない場合は「{keyword}の説明は動画内にありません。」としてください。
    """

    system_message_prompt = SystemMessagePromptTemplate.from_template(system_template)
    human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)
    chat_prompt = ChatPromptTemplate.from_messages([system_message_prompt, human_message_prompt])
    prompt_message_list = chat_prompt.format_prompt(
        keyword=keyword,
        chunk_dict=chunk_dict,
        index="インデックス").to_messages()
    response = llm(prompt_message_list)

    return response

動作確認

今回はこちらのスペースマーケットオンラインライトについて説明されている動画を使用していきます。(チャンネル登録もお願いします)

スペースマーケットオンラインライトについて話されている箇所は何分くらい？と聞いてみます。

2分から動画を確認してみたら、スペースマーケットオンラインライトとはオンライン配信イベントに特化したものなのかなと所見でも感じれました。これで大体ではありますが、何分くらいからどんなことが話されているか知れるようになりました。動画の要約も頼んでみます。

動画の要約もいい感じにまとめてくれました。

動画に全く関係ないAIについて聞いてみます。

普通のChatGPTが答えてくれてますね。

おわりに

今回はLangChainを使ってGPTに新しい拡張機能を追加してみました。Youtubeの字幕からテキストを取得しているので誤字が多いなどの改善の余地がありますが、LangChainを使ってGPT越えの賢いモデルを作れる予感がしました。