はじめに

やりたかった。
適用するのはおなじみ、ポートフォリオ（llm_chat）だ

TODO: githubのアドレスがsoil_analysisになっているのでなおして

参考

OpenAI APIキーの作成

クレジットカード支払いの設定

非公開情報を.envに移す（GitGuardian対策）

llm_service.py

lib/llm/llm_service.py

    :
from langchain.chains.qa_with_sources.retrieval import RetrievalQAWithSourcesChain
from langchain.prompts import (
    ChatPromptTemplate,
)
from langchain_chroma import Chroma
from langchain_openai import ChatOpenAI
from langchain_openai import OpenAIEmbeddings
    :
from lib.llm.valueobject.rag import PdfDataloader
from llm_chat.domain.valueobject.chat import MessageDTO
    :
class OpenAILlmRagService(LlmService):
    def __init__(
        self, config: OpenAIGptConfig, dataloader: PdfDataloader, n_results: int = 3
    ):
        """
        See Also: https://python.langchain.com/docs/how_to/qa_sources/
        """
        super().__init__()
        self.config = config
        self.dataloader = dataloader
        self.n_results = n_results
        self.system_template = """
            以下の資料の注意点を念頭に置いて回答してください
            ・ユーザの質問に対して、できる限り根拠を示してください
            ・箇条書きで簡潔に回答してください。
            ---下記は資料の内容です---
            {summaries}
            Answer in Japanese:
        """
        messages = [
            ("system", self.system_template),
            ("human", "{question}"),
        ]
        self.prompt_template = ChatPromptTemplate.from_messages(messages)

    @staticmethod
    def _create_vectorstore(dataloader: PdfDataloader) -> Chroma:
        """
        Note: OpenAIEmbeddings runs on "text-embedding-ada-002"
        """
        embeddings = OpenAIEmbeddings()

        return Chroma.from_documents(
            documents=dataloader.data,
            embedding=embeddings,
            persist_directory=".",
        )

    def retrieve_answer(self, message: MessageDTO) -> dict:
        embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
        docsearch = Chroma.from_texts(
            texts=[x.page_content for x in self.dataloader.data],
            embedding=embeddings,
            metadatas=[x.metadata for x in self.dataloader.data],
        )
        chain = RetrievalQAWithSourcesChain.from_chain_type(
            llm=ChatOpenAI(temperature=0, model_name=self.config.model),
            chain_type="stuff",
            reduce_k_below_max_tokens=True,
            return_source_documents=True,
            retriever=docsearch.as_retriever(),
            chain_type_kwargs={"prompt": self.prompt_template},
        )

        return chain.invoke({"question": message.content})

test_rag_service.py

lib/llm/test_rag_service.py

from pathlib import Path
from unittest import TestCase

from config.settings import BASE_DIR
from lib.llm.valueobject.rag import PdfDataloader


class TestPdfDataloader(TestCase):
    def test_this_pdf_has_pages_en(self):
        file_path = (
            Path(BASE_DIR)
            / "lib/llm/pdf_sample/doj_cloud_act_white_paper_2019_04_10.pdf"
        )

        dataloader = PdfDataloader(str(file_path))
        self.assertEqual(18, len(dataloader.pages))
        print(dataloader.data)

    def test_this_pdf_has_pages_jp(self):
        file_path = (
            Path(BASE_DIR)
            / "lib/llm/pdf_sample/令和4年版少子化社会対策白書全体版（PDF版）.pdf"
        )
        dataloader = PdfDataloader(str(file_path))
        self.assertEqual(6, len(dataloader.pages))
        print(dataloader.data)

valueobject/rag.py

lib/llm/valueobject/rag.py

import os
from abc import ABC, abstractmethod

from langchain.schema import Document
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter


class Dataloader(ABC):
    pages: list[Document] = []

    @property
    @abstractmethod
    def data(self) -> list[Document]:
        pass

    @abstractmethod
    def __init__(self, chunk_size: int = 1000, chunk_overlap: int = 200):
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size, chunk_overlap=chunk_overlap
        )

    @abstractmethod
    def _load(self):
        """
        See Also: https://python.langchain.com/docs/how_to/document_loader_pdf/
        """
        pass

    @abstractmethod
    def _split(self):
        """
        各ページに出典（ページ数）をつけます
        """
        pass

    def _shredder(self, source: str, attr: str) -> tuple:
        """
        日本語PDFでトークンを多く消費するような場合、ページ単位ではAPIが処理できないので
        さらに千切りにする
        """
        all_splits = self.text_splitter.split_documents(self.pages)
        all_text, all_metadata = [], []
        for text_fragment in all_splits:
            all_text.extend(text_fragment)
            all_metadata.extend({"source": source, "attr": attr})

        return all_text, all_metadata


class PdfDataloader(Dataloader):
    @property
    def data(self) -> list[Document]:
        return self.pages

    def __init__(self, file_path: str):
        super().__init__()
        self._file_path = file_path
        self._load()
        self._split()

    def _load(self):
        # TODO: 非同期化したいね https://python.langchain.com/docs/how_to/document_loader_pdf/#simple-and-fast-text-extraction
        self.pages = PyPDFLoader(self._file_path).load()

    def _split(self):
        filename = os.path.basename(self._file_path)
        for i, doc in enumerate(self.pages):
            doc.page_content = doc.page_content.replace("\n", " ")
            doc.metadata = {"source": f"{filename} {i + 1}ページ"}


class RetrievalQAWithSourcesChainAnswer:
    def __init__(
        self, question: str, answer: str, sources: str, source_documents: list[Document]
    ):
        self.question = question
        self.answer = answer
        self.sources = sources
        self.source_documents = source_documents

service/chat.py

llm_chat/domain/service/chat.py

    :
from config.settings import MEDIA_ROOT, BASE_DIR
from lib.llm.llm_service import (
        :
    OpenAILlmRagService,
)
    :
from lib.llm.valueobject.rag import PdfDataloader, RetrievalQAWithSourcesChainAnswer
    :
class OpenAIRagChatService(ChatService):
    def __init__(self):
        super().__init__()
        self.config = OpenAIGptConfig(
            api_key=os.getenv("OPENAI_API_KEY"),
            temperature=0.5,
            max_tokens=4000,
            model="gpt-4o-mini",
        )

    def generate(self, message: MessageDTO):
        # Step1: User の質問を保存
        self.save(message)

        # Step2: langchainからの回答を保存
        file_path = (
            Path(BASE_DIR)
            / "lib/llm/pdf_sample/令和4年版少子化社会対策白書全体版（PDF版）.pdf"
        )
        answer_dict = OpenAILlmRagService(
            config=self.config,
            dataloader=PdfDataloader(str(file_path)),
        ).retrieve_answer(message)
        message.role = RoleType.ASSISTANT
        message.content = RetrievalQAWithSourcesChainAnswer(**answer_dict).answer
        self.save(message)

    def save(self, message: MessageDTO) -> None:
        message.to_entity().save()

usecase/chat.py

llm_chat/domain/usecase/chat.py

    :
    OpenAIRagChatService,
)
    :
class OpenAIRagUseCase(UseCase):
    def execute(self, user: User, content: str | None):
        """
        RagServiceを利用し、Pdfをソースに。
        contentパラメータは必ずNoneであること。
        Args:
            user (User): DjangoのUserモデルのインスタンス
            content (str | None): この引数は現在利用されていません。
        Raises:
            ValueError: contentがNoneでない場合
        Returns:
            音声をテキストに変換した結果
        """
        if content is None:
            raise ValueError("content cannot be None for OpenAIRagUseCase")
        chat_service = OpenAIRagChatService()
        message = MessageDTO(
            user=user,
            role=RoleType.USER,
            content=content,
            invisible=False,
        )
        return chat_service.generate(message)

views.py

llm_chat/views.py

        :
    OpenAIRagUseCase,
)
from llm_chat.forms import UserTextForm
    :
        if use_case_type == "Gemini":
            use_case = GeminiUseCase()
            content = form_data["question"]
                :
        elif use_case_type == "OpenAIRag":
            use_case = OpenAIRagUseCase()
            content = form_data["question"]
                :

まとめ

大事なことは「ChatGPTは一度に扱えるテキストの量に限界がある」ということ。これがために、ページ単位にわけたドキュメントをさらに千切りにする。その千切りにしたデータは、自然言語1に対するmetadata1というかたちで、単に辞書で取り扱われていて、pdfのどこから引用したかというのを紐づけることができる chat/domain/valueobject/pdfdataloader.py のあたりの処理。今回はページでしか割っていないがうまくいったのは英文だったからなのかも。そしたらページ単位に割った後、さらに千切りする必要が出てくるが自然言語とmetadataをペアにして「文書のどこから考えたのか」を人間に示すことができる。

Next Action