🐶
RAG開発の超入門【RaggleのQuickStart | Pythonのソースコードあり】

ますみ / 生成AIエンジニア
2025/01/02に公開
はじめまして、ますみです！
株式会社Galirage（ガリレージ）という「生成AIに特化して、システム開発・アドバイザリー支援・研修支援をしているIT企業」で、代表をしております^^
この記事では、入門者向けの「RAG」の開発手法を解説します！
もしもPythonを使ったことがない方は、下記のZenn本を参考にしてください。

https://zenn.dev/umi_mori/books/python-programming
また、RAGについての基礎知識を学びたい方は、下記のZenn本を参考にしてください。

https://zenn.dev/umi_mori/books/llm-rag-langchain-python
さらに、RaggleというRAGの精度を競うコンペを開催しているため、ご興味のある方は、こちらのコンペを通して、RAGのスキルアップにご活用ください！
なんと1位の人には、賞金30万円も付与されます🏆
https://raggle.jp/competition/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae
それでは、早速解説をしていきます！
この記事の内容を習得すれば、Raggleに応募できる状態になるため、ぜひ皆さんもRaggleのコンペに挑戦していただけたら幸いです^^

 全体の流れまず全体の流れとしては、以下のようになります。

データ準備：必要なデータを用意します。

API準備：生成AIのAPIを登録して、APIキーを取得します。

環境構築：必要なライブラリをインストールして、実行環境を整備します。

データの読み込み：データを読み込みます。

データのインデックス化：データをインデックス化して、検索をできる状態にします。

質問に対する類似情報の検索：質問に対して、ベクトル検索などを用いて、類似情報を検索します。

回答の生成：検索結果を元に、回答を生成します。
それぞれの詳細な流れについて、説明していきます。

 1. データ準備今回は、第2回のRaggleコンペのデータを用いて、解説していきます！
!通常の参加者については、コンペ期間中にコンペのソースコードを公開することを禁止しているため、ご注意ください！
本著者は、主催者であり、より参加の敷居を低くするために、このような記事を執筆しています。
そのため、本記事では、あえて、精度改善の施策は、全く実装しておらず、スタンダードな手法（ベースラインとなるような手法）で実装をしていきます。
Raggleのコンペページから、下記のように「データ」のタブをクリックすることで、生のデータをダウンロードすることができます。
実際には、各データをURLで読み込むためのコードを書いていくため、この操作は必要ありません。

 2. API準備次に、動作確認をしたり、精度検証をするために、APIキーを発行していきます。
ここでは、OpenAIのAPIキーの発行方法は、下記の記事をご参照ください。
https://zenn.dev/umi_mori/books/chatbot-chatgpt/viewer/how_to_use_openai_api#apiの発行方法
sk-proj-xxxxxのような文字列をコピーできたらOKです！

忘れないように、どこかにメモしておきましょう。

 3. 環境構築まずは、環境構築をしましょう。
Pythonの環境構築の方法については、下記の記事を参考にしてください。
https://zenn.dev/umi_mori/books/python-programming/viewer/python-setup-macos
https://zenn.dev/umi_mori/books/python-programming/viewer/python-setup-windows
ちなみに、第二回のRaggleコンペでは、Pythonのv3.11.9の使用を想定しています。
そのために、以下のように、venvを使った環境構築をすることを推奨します。
macOSの場合のコマンドの例
% python -m venv .venv
% source .venv/bin/activate
% pip install --upgrade pip
Pythonの環境構築が完了したら、以下のように、Raggleのコンペページから、以下の2つのファイルをダウンロードしてください。

main.py：メインのソースコード

requirements.txt：必要なライブラリのリスト
まず、ルールのタブをクリックして、下の方にスクロールします。
そして、「プログラム提出の流れ」という箇所から、ダウンロードします。
この2つのファイルを、空のフォルダの中に格納します。
次に、パッケージのインストールをしていきます。
以下のように、プロジェクトの格納されたディレクトリの中で、ターミナルを開いて、以下のコマンドを実行します。
pip install -r requirements.txt
最後に、環境変数を設定します。
以下のように、.envファイルを作成して、先ほど作成したAPIキーを用いて、以下のように記述します。
OPENAI_API_KEY=sk-proj-xxxxx

 4. データの読み込み次に、データの読み込みをしていきます。
ソースコード全体は下記の通りです。

（解説はソースコードの後に記述しています。）
main.py
import json
import sys
import time

from dotenv import load_dotenv
+ import requests
+ import pdfplumber

from langchain import callbacks
+ from langchain.schema import Document

# ==============================================================================
# !!! 警告 !!!: 以下の変数を変更しないでください。
# ==============================================================================
model = "gpt-4o-mini"
pdf_file_urls = [
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Financial_Statements_2023.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Hada_Labo_Gokujun_Lotion_Overview.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Shibata_et_al_Research_Article.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/V_Rohto_Premium_Product_Information.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf",
]
# ==============================================================================


# ==============================================================================
# この関数を編集して、あなたの RAG パイプラインを実装してください。
# !!! 注意 !!!: デバッグ過程は標準出力に出力しないでください。
# ==============================================================================
def rag_implementation(question: str) -> str:
    """
    ロート製薬の製品・企業情報に関する質問に対して回答を生成する関数
    この関数は与えられた質問に対してRAGパイプラインを用いて回答を生成します。

    Args:
        question (str): ロート製薬の製品・企業情報に関する質問文字列

    Returns:
        answer (str): 質問に対する回答

    Note:
        - デバッグ出力は標準出力に出力しないでください
        - model 変数 と pdf_file_urls 変数は編集しないでください
        - 回答は日本語で生成してください
    """
    # 戻り値として質問に対する回答を返却してください。
+     def download_and_load_pdfs(urls: list) -> list:
+         """
+         PDFファイルをダウンロードして読み込む関数
+
+         Args:
+             urls (list): PDFファイルのURLリスト
+
+         Returns:
+             documents (list): PDFファイルのテキストデータを含むDocumentオブジェクトのリスト
+
+         Raises:
+             Exception: ダウンロードまたは読み込みに失敗した場合に発生する例外
+
+         Examples:
+             >>> urls = ["https://example.com/example.pdf"]
+             >>> download_and_load_pdfs(urls)
+             [Document(page_content="...", metadata={"source": "https://example.com/example.pdf"})]
+         """
+         try:
+             def download_pdf(url, save_path):
+                 response = requests.get(url)
+                 if response.status_code == 200:
+                     with open(save_path, 'wb') as f:
+                         f.write(response.content)
+                 else:
+                     raise Exception(f"Failed to download {url}")
+             documents = []
+
+             for i, url in enumerate(urls):
+                 tmp_path = f"pdf_{i}.pdf"
+                 download_pdf(url, tmp_path)
+
+                 with pdfplumber.open(tmp_path) as pdf:
+                     full_text = ""
+                     for page in pdf.pages:
+                         text = page.extract_text()
+                         if text:
+                             full_text += text + "\n"
+
+                     documents.append(
+                         Document(
+                             page_content=full_text,
+                             metadata={"source": url}
+                         )
+                     )
+             return documents
+         except Exception as e:
+             raise Exception(f"Error reading {url}: {e}")
+
+      docs = download_and_load_pdfs(pdf_file_urls)
+      print(docs)
        answer = "ここに生成した回答を入れます"
        return answer


# ==============================================================================


# ==============================================================================
# !!! 警告 !!!: 以下の関数を編集しないでください。
# ==============================================================================
def main(question: str):
    with callbacks.collect_runs() as cb:
        result = rag_implementation(question)
        for attempt in range(2):  # 最大2回試行
            try:
                run_id = cb.traced_runs[0].id
                break
            except IndexError:
                if attempt == 0:  # 1回目の失敗時のみ
                    time.sleep(3)  # 3秒待機して再試行
                else:  # 2回目も失敗した場合
                    raise RuntimeError("Failed to get run_id after 2 attempts")

    output = {"result": result, "run_id": str(run_id)}
    print(json.dumps(output, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    load_dotenv()

    if len(sys.argv) > 1:
        question = sys.argv[1]
        main(question)
    else:
        print("Please provide a question as a command-line argument.")
        sys.exit(1)
# ==============================================================================

今回、URLでデータを読み込むため、以下のようにmain.pyの中で、データをダウンロードするためのコードを書いていきます。
一時的にダウンロードされたファイルに対して、データを読み込むためのpdfplumberというライブラリで読み込みをしています。
その結果を、この後、LangChainによって処理するためのDocumentという形式に格納しています。
この状態で、python main.py 存在意義（パーパス）は、なんですか？というコマンドで実行すると、まだLangChainのchainが実行されていないため、次のようなエラーが出ますが、それ問題ありません。
Traceback (most recent call last):
  File "/xxx/main.py", line 120, in main
    run_id = cb.traced_runs[0].id
             ~~~~~~~~~~~~~~^^^
IndexError: list index out of range

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/xxx/main.py", line 137, in <module>
    main(question)
  File "/xxx/main.py", line 126, in main
    raise RuntimeError("Failed to get run_id after 2 attempts")
RuntimeError: Failed to get run_id after 2 attempts
また、エラーは出ますが、次のように、print(docs)の部分で、データが取得できているかどうかを確認することができます。
以下のように読み込まれたDocumentのリストが取得できていることが確認できます。

（実際には、"..."の部分にテキストデータが入っています。）
[
    Document(
        metadata={'source': 'https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Financial_Statements_2023.pdf'},
        page_content='1/134\nEDIN...'
    ),
    Document(
        metadata={'source': 'https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Hada_Labo_Gokujun_Lotion_Overview.pdf'},
        page_content='2024/12/06...'
    ),
    Document(
        metadata={'source': 'https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Shibata_et_al_Research_Article.pdf'},
        page_content='Resource\nS...'
    ),
    Document(
        metadata={'source': 'https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/V_Rohto_Premium_Product_Information.pdf'},
        page_content=''
    ),
    Document(
        metadata={'source': 'https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf'},
        page_content='ROHTO Well...'
    )
]

 5. データのインデックス化次に、読み込まれたデータに対して、ベクトル検索をするためのインデックス化をしていきます。
ソースコード全体は下記の通りです。

（解説はソースコードの後に記述しています。）
main.py
import json
import sys
import time

from dotenv import load_dotenv
import requests
import pdfplumber

from langchain import callbacks
from langchain.schema import Document
+ from langchain_text_splitters import CharacterTextSplitter
+ from langchain_openai import OpenAIEmbeddings
+ from langchain_chroma import Chroma

# ==============================================================================
# !!! 警告 !!!: 以下の変数を変更しないでください。
# ==============================================================================
model = "gpt-4o-mini"
pdf_file_urls = [
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Financial_Statements_2023.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Hada_Labo_Gokujun_Lotion_Overview.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Shibata_et_al_Research_Article.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/V_Rohto_Premium_Product_Information.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf",
]
# ==============================================================================


# ==============================================================================
# この関数を編集して、あなたの RAG パイプラインを実装してください。
# !!! 注意 !!!: デバッグ過程は標準出力に出力しないでください。
# ==============================================================================
def rag_implementation(question: str) -> str:
    """
    ロート製薬の製品・企業情報に関する質問に対して回答を生成する関数
    この関数は与えられた質問に対してRAGパイプラインを用いて回答を生成します。

    Args:
        question (str): ロート製薬の製品・企業情報に関する質問文字列

    Returns:
        answer (str): 質問に対する回答

    Note:
        - デバッグ出力は標準出力に出力しないでください
        - model 変数 と pdf_file_urls 変数は編集しないでください
        - 回答は日本語で生成してください
    """
    # 戻り値として質問に対する回答を返却してください。
    def download_and_load_pdfs(urls: list) -> list:
        """
        PDFファイルをダウンロードして読み込む関数

        Args:
            urls (list): PDFファイルのURLリスト

        Returns:
            documents (list): PDFファイルのテキストデータを含むDocumentオブジェクトのリスト

        Raises:
            Exception: ダウンロードまたは読み込みに失敗した場合に発生する例外

        Examples:
            >>> urls = ["https://example.com/example.pdf"]
            >>> download_and_load_pdfs(urls)
            [Document(page_content="...", metadata={"source": "https://example.com/example.pdf"})]
        """
        try:
            def download_pdf(url, save_path):
                response = requests.get(url)
                if response.status_code == 200:
                    with open(save_path, 'wb') as f:
                        f.write(response.content)
                else:
                    raise Exception(f"Failed to download {url}")
            documents = []

            for i, url in enumerate(urls):
                tmp_path = f"pdf_{i}.pdf"
                download_pdf(url, tmp_path)

                with pdfplumber.open(tmp_path) as pdf:
                    full_text = ""
                    for page in pdf.pages:
                        text = page.extract_text()
                        if text:
                            full_text += text + "\n"

                    documents.append(
                        Document(
                            page_content=full_text,
                            metadata={"source": url}
                        )
                    )
            return documents
        except Exception as e:
            raise Exception(f"Error reading {url}: {e}")

+     def create_vectorstore(docs: list) -> Chroma:
+         """
+         テキストデータからベクトルストアを生成する関数
+
+         Args:
+             docs (list): Documentオブジェクトのリスト
+
+         Returns:
+             vectorstore (Chroma): ベクトルストア
+
+         Raises:
+             Exception: ベクトルストアの生成に失敗した場合に発生する例外
+
+         Examples:
+             >>> docs = [Document(page_content="...", metadata={"source": "https://example.com/example.pdf"})]
+             >>> create_vectorstore(docs)
+             Chroma(...)
+         """
+         try:
+             text_splitter = CharacterTextSplitter(
+                 separator="\n",
+                 chunk_size=1000,
+             )
+             splitted_docs = []
+             for doc in docs:
+                 chunks = text_splitter.split_text(doc.page_content)
+                 for chunk in chunks:
+                     splitted_docs.append(Document(page_content=chunk, metadata=doc.metadata))
+
+             embedding_function = OpenAIEmbeddings()
+
+             vectorstore = Chroma.from_documents(
+                 splitted_docs,
+                 embedding_function,
+             )
+             return vectorstore
+         except Exception as e:
+             raise Exception(f"Error creating vectorstore: {e}")

    docs = download_and_load_pdfs(pdf_file_urls)
-     print(docs)
+     db = create_vectorstore(docs)
+     retriever = db.as_retriever()

    answer = "ここに生成した回答を入れます"
    return answer


# ==============================================================================


# ==============================================================================
# !!! 警告 !!!: 以下の関数を編集しないでください。
# ==============================================================================
def main(question: str):
    with callbacks.collect_runs() as cb:
        result = rag_implementation(question)
        for attempt in range(2):  # 最大2回試行
            try:
                run_id = cb.traced_runs[0].id
                break
            except IndexError:
                if attempt == 0:  # 1回目の失敗時のみ
                    time.sleep(3)  # 3秒待機して再試行
                else:  # 2回目も失敗した場合
                    raise RuntimeError("Failed to get run_id after 2 attempts")

    output = {"result": result, "run_id": str(run_id)}
    print(json.dumps(output, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    load_dotenv()

    if len(sys.argv) > 1:
        question = sys.argv[1]
        main(question)
    else:
        print("Please provide a question as a command-line argument.")
        sys.exit(1)
# ==============================================================================

各ドキュメントに対して、特定の大きさの塊で分割をして、それぞれのテキストに対して、ベクトル化をしていきます。
その結果を、Chromaという形式に格納して、検索を行うためのretrieverを作成しています。
これにより、ベクトルストアが出来上がりました。

 6. 質問に対する類似情報の検索次に、質問に対して、ベクトル検索を用いて、類似情報を検索していきます。
ソースコード全体は下記の通りです。

（解説はソースコードの後に記述しています。）
main.py
import json
import sys
import time

from dotenv import load_dotenv
import requests
import pdfplumber

from langchain import callbacks
from langchain.schema import Document
from langchain_text_splitters import CharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma

# ==============================================================================
# !!! 警告 !!!: 以下の変数を変更しないでください。
# ==============================================================================
model = "gpt-4o-mini"
pdf_file_urls = [
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Financial_Statements_2023.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Hada_Labo_Gokujun_Lotion_Overview.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Shibata_et_al_Research_Article.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/V_Rohto_Premium_Product_Information.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf",
]
# ==============================================================================


# ==============================================================================
# この関数を編集して、あなたの RAG パイプラインを実装してください。
# !!! 注意 !!!: デバッグ過程は標準出力に出力しないでください。
# ==============================================================================
def rag_implementation(question: str) -> str:
    """
    ロート製薬の製品・企業情報に関する質問に対して回答を生成する関数
    この関数は与えられた質問に対してRAGパイプラインを用いて回答を生成します。

    Args:
        question (str): ロート製薬の製品・企業情報に関する質問文字列

    Returns:
        answer (str): 質問に対する回答

    Note:
        - デバッグ出力は標準出力に出力しないでください
        - model 変数 と pdf_file_urls 変数は編集しないでください
        - 回答は日本語で生成してください
    """
    # 戻り値として質問に対する回答を返却してください。
    def download_and_load_pdfs(urls: list) -> list:
        """
        PDFファイルをダウンロードして読み込む関数

        Args:
            urls (list): PDFファイルのURLリスト

        Returns:
            documents (list): PDFファイルのテキストデータを含むDocumentオブジェクトのリスト

        Raises:
            Exception: ダウンロードまたは読み込みに失敗した場合に発生する例外

        Examples:
            >>> urls = ["https://example.com/example.pdf"]
            >>> download_and_load_pdfs(urls)
            [Document(page_content="...", metadata={"source": "https://example.com/example.pdf"})]
        """
        try:
            def download_pdf(url, save_path):
                response = requests.get(url)
                if response.status_code == 200:
                    with open(save_path, 'wb') as f:
                        f.write(response.content)
                else:
                    raise Exception(f"Failed to download {url}")
            documents = []

            for i, url in enumerate(urls):
                tmp_path = f"pdf_{i}.pdf"
                download_pdf(url, tmp_path)

                with pdfplumber.open(tmp_path) as pdf:
                    full_text = ""
                    for page in pdf.pages:
                        text = page.extract_text()
                        if text:
                            full_text += text + "\n"

                    documents.append(
                        Document(
                            page_content=full_text,
                            metadata={"source": url}
                        )
                    )
            return documents
        except Exception as e:
            raise Exception(f"Error reading {url}: {e}")

    def create_vectorstore(docs: list) -> Chroma:
        """
        テキストデータからベクトルストアを生成する関数

        Args:
            docs (list): Documentオブジェクトのリスト

        Returns:
            vectorstore (Chroma): ベクトルストア

        Raises:
            Exception: ベクトルストアの生成に失敗した場合に発生する例外

        Examples:
            >>> docs = [Document(page_content="...", metadata={"source": "https://example.com/example.pdf"})]
            >>> create_vectorstore(docs)
            Chroma(...)
        """
        try:
            text_splitter = CharacterTextSplitter(
                separator="\n",
                chunk_size=1000,
            )
            splitted_docs = []
            for doc in docs:
                chunks = text_splitter.split_text(doc.page_content)
                for chunk in chunks:
                    splitted_docs.append(Document(page_content=chunk, metadata=doc.metadata))

            embedding_function = OpenAIEmbeddings()

            vectorstore = Chroma.from_documents(
                splitted_docs,
                embedding_function,
            )
            return vectorstore
        except Exception as e:
            raise Exception(f"Error creating vectorstore: {e}")

    docs = download_and_load_pdfs(pdf_file_urls)
    db = create_vectorstore(docs)
    retriever = db.as_retriever()

+     response = retriever.invoke(question)
+     print(response)

    answer = "ここに生成した回答を入れます"
    return answer


# ==============================================================================


# ==============================================================================
# !!! 警告 !!!: 以下の関数を編集しないでください。
# ==============================================================================
def main(question: str):
    with callbacks.collect_runs() as cb:
        result = rag_implementation(question)
        for attempt in range(2):  # 最大2回試行
            try:
                run_id = cb.traced_runs[0].id
                break
            except IndexError:
                if attempt == 0:  # 1回目の失敗時のみ
                    time.sleep(3)  # 3秒待機して再試行
                else:  # 2回目も失敗した場合
                    raise RuntimeError("Failed to get run_id after 2 attempts")

    output = {"result": result, "run_id": str(run_id)}
    print(json.dumps(output, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    load_dotenv()

    if len(sys.argv) > 1:
        question = sys.argv[1]
        main(question)
    else:
        print("Please provide a question as a command-line argument.")
        sys.exit(1)
# ==============================================================================

質問に対して、retriever.invoke(question)を実行することで、類似情報を検索しています。
これにより、質問に対して、類似情報を検索することができました。
python main.py 存在意義（パーパス）は、なんですか？というコマンドで実行すると、以下のような結果が出力されます。
質問に類似するDocumentのリストが取得できていることが確認できます。

（実際には、"..."の部分にテキストデータが入っています。）
[
	Document(
		metadata={
			'source': 'https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf'
		},
		page_content='ロイーエクスペリエンス...'
	),
	Document(
		metadata={
			'source': 'https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf'
		},
		page_content='に向けた歩みを...'
	),
	Document(
		metadata={
			'source': 'https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf'
		},
		page_content='パになりたいと願う...'),
	Document(
		metadata={
			'source': 'https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf'
		},
		page_content='生活者\nインプット...'
	)
]

 7. 回答の生成最後に、検索結果を元に、回答を生成していきます。
ソースコード全体は下記の通りです。

（解説はソースコードの後に記述しています。）
main.py
import json
import sys
import time

from dotenv import load_dotenv
import requests
import pdfplumber

from langchain import callbacks
from langchain.schema import Document
from langchain_text_splitters import CharacterTextSplitter
- from langchain_openai import OpenAIEmbeddings
+ from langchain_openai import ChatOpenAI, OpenAIEmbeddings
+ from langchain_core.output_parsers import StrOutputParser
+ from langchain_core.prompts import ChatPromptTemplate
+ from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_chroma import Chroma

# ==============================================================================
# !!! 警告 !!!: 以下の変数を変更しないでください。
# ==============================================================================
model = "gpt-4o-mini"
pdf_file_urls = [
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Financial_Statements_2023.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Hada_Labo_Gokujun_Lotion_Overview.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Shibata_et_al_Research_Article.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/V_Rohto_Premium_Product_Information.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf",
]
# ==============================================================================


# ==============================================================================
# この関数を編集して、あなたの RAG パイプラインを実装してください。
# !!! 注意 !!!: デバッグ過程は標準出力に出力しないでください。
# ==============================================================================
def rag_implementation(question: str) -> str:
    """
    ロート製薬の製品・企業情報に関する質問に対して回答を生成する関数
    この関数は与えられた質問に対してRAGパイプラインを用いて回答を生成します。

    Args:
        question (str): ロート製薬の製品・企業情報に関する質問文字列

    Returns:
        answer (str): 質問に対する回答

    Note:
        - デバッグ出力は標準出力に出力しないでください
        - model 変数 と pdf_file_urls 変数は編集しないでください
        - 回答は日本語で生成してください
    """
    # 戻り値として質問に対する回答を返却してください。
    def download_and_load_pdfs(urls: list) -> list:
        """
        PDFファイルをダウンロードして読み込む関数

        Args:
            urls (list): PDFファイルのURLリスト

        Returns:
            documents (list): PDFファイルのテキストデータを含むDocumentオブジェクトのリスト

        Raises:
            Exception: ダウンロードまたは読み込みに失敗した場合に発生する例外

        Examples:
            >>> urls = ["https://example.com/example.pdf"]
            >>> download_and_load_pdfs(urls)
            [Document(page_content="...", metadata={"source": "https://example.com/example.pdf"})]
        """
        try:
            def download_pdf(url, save_path):
                response = requests.get(url)
                if response.status_code == 200:
                    with open(save_path, 'wb') as f:
                        f.write(response.content)
                else:
                    raise Exception(f"Failed to download {url}")
            documents = []

            for i, url in enumerate(urls):
                tmp_path = f"pdf_{i}.pdf"
                download_pdf(url, tmp_path)

                with pdfplumber.open(tmp_path) as pdf:
                    full_text = ""
                    for page in pdf.pages:
                        text = page.extract_text()
                        if text:
                            full_text += text + "\n"

                    documents.append(
                        Document(
                            page_content=full_text,
                            metadata={"source": url}
                        )
                    )
            return documents
        except Exception as e:
            raise Exception(f"Error reading {url}: {e}")

    def create_vectorstore(docs: list) -> Chroma:
        """
        テキストデータからベクトルストアを生成する関数

        Args:
            docs (list): Documentオブジェクトのリスト

        Returns:
            vectorstore (Chroma): ベクトルストア

        Raises:
            Exception: ベクトルストアの生成に失敗した場合に発生する例外

        Examples:
            >>> docs = [Document(page_content="...", metadata={"source": "https://example.com/example.pdf"})]
            >>> create_vectorstore(docs)
            Chroma(...)
        """
        try:
            text_splitter = CharacterTextSplitter(
                separator="\n",
                chunk_size=1000,
            )
            splitted_docs = []
            for doc in docs:
                chunks = text_splitter.split_text(doc.page_content)
                for chunk in chunks:
                    splitted_docs.append(Document(page_content=chunk, metadata=doc.metadata))

            embedding_function = OpenAIEmbeddings()

            vectorstore = Chroma.from_documents(
                splitted_docs,
                embedding_function,
            )
            return vectorstore
        except Exception as e:
            raise Exception(f"Error creating vectorstore: {e}")

    docs = download_and_load_pdfs(pdf_file_urls)
    db = create_vectorstore(docs)
    retriever = db.as_retriever()

-     response = retriever.invoke(question)
-     print(response)
-     answer = "ここに生成した回答を入れます"

+     template = """
+     # ゴール
+     私は、参考文章と質問を提供します。
+     あなたは、参考文章に基づいて、質問に対する回答を生成してください。
+
+     # 質問
+     {question}
+
+     # 参考文章
+     {context}
+     """
+
+     prompt = ChatPromptTemplate.from_template(template)
+
+     chat = ChatOpenAI(model=model)
+
+     output_parser = StrOutputParser()
+
+     setup_and_retrieval = RunnableParallel(
+         {"context": retriever, "question": RunnablePassthrough()}
+     )
+
+     chain = setup_and_retrieval | prompt | chat | output_parser
+
+     answer = chain.invoke(question)

    return answer


# ==============================================================================


# ==============================================================================
# !!! 警告 !!!: 以下の関数を編集しないでください。
# ==============================================================================
def main(question: str):
    with callbacks.collect_runs() as cb:
        result = rag_implementation(question)
        for attempt in range(2):  # 最大2回試行
            try:
                run_id = cb.traced_runs[0].id
                break
            except IndexError:
                if attempt == 0:  # 1回目の失敗時のみ
                    time.sleep(3)  # 3秒待機して再試行
                else:  # 2回目も失敗した場合
                    raise RuntimeError("Failed to get run_id after 2 attempts")

    output = {"result": result, "run_id": str(run_id)}
    print(json.dumps(output, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    load_dotenv()

    if len(sys.argv) > 1:
        question = sys.argv[1]
        main(question)
    else:
        print("Please provide a question as a command-line argument.")
        sys.exit(1)
# ==============================================================================

ここでは、LangChainのLCELという記法を使って、回答を生成しています。
LCELについては、下記の記事をご参照ください。
https://zenn.dev/umi_mori/books/prompt-engineer/viewer/lcel
他の手法としては、LangGraphを用いる方法があります。
https://zenn.dev/umi_mori/books/prompt-engineer/viewer/langgraph

 全てのソースコードここまでの内容をまとめたソースコード全体は、下記の通りです。
main.py
import json
import sys
import time

from dotenv import load_dotenv
import requests
import pdfplumber

from langchain import callbacks
from langchain.schema import Document
from langchain_text_splitters import CharacterTextSplitter
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnableParallel, RunnablePassthrough
from langchain_chroma import Chroma

# ==============================================================================
# !!! 警告 !!!: 以下の変数を変更しないでください。
# ==============================================================================
model = "gpt-4o-mini"
pdf_file_urls = [
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Financial_Statements_2023.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Hada_Labo_Gokujun_Lotion_Overview.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Shibata_et_al_Research_Article.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/V_Rohto_Premium_Product_Information.pdf",
    "https://storage.googleapis.com/gg-raggle-public/competitions/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae/dataset/Well-Being_Report_2024.pdf",
]
# ==============================================================================


# ==============================================================================
# この関数を編集して、あなたの RAG パイプラインを実装してください。
# !!! 注意 !!!: デバッグ過程は標準出力に出力しないでください。
# ==============================================================================
def rag_implementation(question: str) -> str:
    """
    ロート製薬の製品・企業情報に関する質問に対して回答を生成する関数
    この関数は与えられた質問に対してRAGパイプラインを用いて回答を生成します。

    Args:
        question (str): ロート製薬の製品・企業情報に関する質問文字列

    Returns:
        answer (str): 質問に対する回答

    Note:
        - デバッグ出力は標準出力に出力しないでください
        - model 変数 と pdf_file_urls 変数は編集しないでください
        - 回答は日本語で生成してください
    """
    # 戻り値として質問に対する回答を返却してください。
    def download_and_load_pdfs(urls: list) -> list:
        """
        PDFファイルをダウンロードして読み込む関数

        Args:
            urls (list): PDFファイルのURLリスト

        Returns:
            documents (list): PDFファイルのテキストデータを含むDocumentオブジェクトのリスト

        Raises:
            Exception: ダウンロードまたは読み込みに失敗した場合に発生する例外

        Examples:
            >>> urls = ["https://example.com/example.pdf"]
            >>> download_and_load_pdfs(urls)
            [Document(page_content="...", metadata={"source": "https://example.com/example.pdf"})]
        """
        try:
            def download_pdf(url, save_path):
                response = requests.get(url)
                if response.status_code == 200:
                    with open(save_path, 'wb') as f:
                        f.write(response.content)
                else:
                    raise Exception(f"Failed to download {url}")
            documents = []

            for i, url in enumerate(urls):
                tmp_path = f"pdf_{i}.pdf"
                download_pdf(url, tmp_path)

                with pdfplumber.open(tmp_path) as pdf:
                    full_text = ""
                    for page in pdf.pages:
                        text = page.extract_text()
                        if text:
                            full_text += text + "\n"

                    documents.append(
                        Document(
                            page_content=full_text,
                            metadata={"source": url}
                        )
                    )
            return documents
        except Exception as e:
            raise Exception(f"Error reading {url}: {e}")

    def create_vectorstore(docs: list) -> Chroma:
        """
        テキストデータからベクトルストアを生成する関数

        Args:
            docs (list): Documentオブジェクトのリスト

        Returns:
            vectorstore (Chroma): ベクトルストア

        Raises:
            Exception: ベクトルストアの生成に失敗した場合に発生する例外

        Examples:
            >>> docs = [Document(page_content="...", metadata={"source": "https://example.com/example.pdf"})]
            >>> create_vectorstore(docs)
            Chroma(...)
        """
        try:
            text_splitter = CharacterTextSplitter(
                separator="\n",
                chunk_size=1000,
            )
            splitted_docs = []
            for doc in docs:
                chunks = text_splitter.split_text(doc.page_content)
                for chunk in chunks:
                    splitted_docs.append(Document(page_content=chunk, metadata=doc.metadata))

            embedding_function = OpenAIEmbeddings()

            vectorstore = Chroma.from_documents(
                splitted_docs,
                embedding_function,
            )
            return vectorstore
        except Exception as e:
            raise Exception(f"Error creating vectorstore: {e}")

    docs = download_and_load_pdfs(pdf_file_urls)
    db = create_vectorstore(docs)
    retriever = db.as_retriever()

    template = """
    # ゴール
    私は、参考文章と質問を提供します。
    あなたは、参考文章に基づいて、質問に対する回答を生成してください。

    # 質問
    {question}

    # 参考文章
    {context}
    """

    prompt = ChatPromptTemplate.from_template(template)

    chat = ChatOpenAI(model=model)

    output_parser = StrOutputParser()

    setup_and_retrieval = RunnableParallel(
        {"context": retriever, "question": RunnablePassthrough()}
    )

    chain = setup_and_retrieval | prompt | chat | output_parser

    answer = chain.invoke(question)

    return answer


# ==============================================================================


# ==============================================================================
# !!! 警告 !!!: 以下の関数を編集しないでください。
# ==============================================================================
def main(question: str):
    with callbacks.collect_runs() as cb:
        result = rag_implementation(question)
        for attempt in range(2):  # 最大2回試行
            try:
                run_id = cb.traced_runs[0].id
                break
            except IndexError:
                if attempt == 0:  # 1回目の失敗時のみ
                    time.sleep(3)  # 3秒待機して再試行
                else:  # 2回目も失敗した場合
                    raise RuntimeError("Failed to get run_id after 2 attempts")

    output = {"result": result, "run_id": str(run_id)}
    print(json.dumps(output, ensure_ascii=False, indent=2))


if __name__ == "__main__":
    load_dotenv()

    if len(sys.argv) > 1:
        question = sys.argv[1]
        main(question)
    else:
        print("Please provide a question as a command-line argument.")
        sys.exit(1)
# ==============================================================================


 Raggleへの投稿方法最後に、Raggleへの投稿方法について説明します。
まずRaggleのサイトにアクセスします。
https://raggle.jp/competition/617b10e9-a71b-4f2a-a9ee-ffe11d8d64ae
そして、右側の「投稿する」のボタンをクリックします。
そして、作成したmain.pyのファイルを選択して、投稿します。
この時、こちらに記載されいている注意事項を確認してから投稿してください。
また、「投稿についての説明」の中で、工夫したポイントなどを記載しておくと、ナイスアイデア賞を受賞できる可能性があるため、記載しておくことをおすすめします。
投稿が完了すると、以下のような画面になります。
すると、リーダーボードの中で、自分の投稿が表示されるようになります。
日次で評価が回るため、しばらく待って結果を確認してください。
評価が回ると、下記のようにスコアが表示されます。
今回は、トレーニングデータでは、304点になりました。
無料で参加できるイベントなため、ぜひ初めての方も挑戦していただけますと幸いです^^
賞金を目指すのもありですし、何よりもゲームとして、楽しんでいただけたら幸いです😊

 最後に最後まで読んでくださり、ありがとうございました！

この記事を通して、少しでもあなたの学びに役立てば幸いです！
!【📩 仕事の相談はこちら 📩】

お仕事の相談のある方は、下記のフォームよりお気軽にご相談ください。

https://forms.gle/G5g1SJ7BBZw7oXYA7
もしもメールでの問い合わせの方がよろしければ、下記のメールアドレスへご連絡ください。
info*galirage.com（*を@に変えてご送付ください）
宣伝：もしもよかったらご覧ください^^『AIとコミュニケーションする技術（インプレス出版）』という書籍を出版しました🎉
これからの未来において「変わらない知識」を見極めて、生成AIの業界において、読まれ続ける「バイブル」となる本をまとめ上げました。
かなり自信のある一冊なため、もしもよろしければ、ご一読いただけますと幸いです^^
Galirage Inc.
株式会社Galirageのテックブログです！ Galirageでは、生成AIのシステム開発・コンサルティング・研修を行なっております。 ▼ 問い合わせ先 ▼ info@galirage.com
全体の流れ

1. データ準備

2. API準備

3. 環境構築

4. データの読み込み

5. データのインデックス化

6. 質問に対する類似情報の検索

7. 回答の生成

全てのソースコード

Raggleへの投稿方法

最後に

Discussion
