LangGraphでCRAGエージェントを構築！『ヒロアカ』について聞く

Ubuntuのコマンドラインで、Dockerfileを作成します。

mkdir crag
cd crag
nano Dockerfile

Dockerfileに以下の記述を貼り付けます。

# ベースイメージ(CUDA)の指定
FROM nvidia/cuda:12.1.0-cudnn8-devel-ubuntu22.04

# 必要なパッケージをインストール
RUN apt-get update && apt-get install -y python3-pip python3-venv git nano curl pciutils lshw python3-dev graphviz libgraphviz-dev pkg-config

RUN curl -fsSL https://ollama.com/install.sh | sh

# 作業ディレクトリを設定
WORKDIR /app

# アプリケーションコードをコピー
COPY . /app

# Python仮想環境の作成
RUN python3 -m venv /app/.venv

# 仮想環境をアクティベートするコマンドを.bashrcに追加
RUN echo "source /app/.venv/bin/activate" >> /root/.bashrc

# JupyterLabのインストール
RUN /app/.venv/bin/pip install Jupyter jupyterlab

# LangChain関連のインストール
RUN /app/.venv/bin/pip install ollama langchain-ollama langchain langsmith langgraph langchain-chroma faiss-gpu langchain-community langchain_huggingface langchain_core tiktoken pygraphviz

# コンテナの起動時にbashを実行
CMD ["/bin/bash"]

[Ctrl + S]キーで変更内容を保存し、[Ctrl + X]キーで編集モードから抜けます。

docker-compose.ymlでDockerコンテナの設定をします。

docker-compose.ymlのYAMLファイルを作成して開きます。

nano docker-compose.yml

以下のコードをコピーして、YAMLファイルに貼り付けます。

services:
  crag:
    build:
      context: .
      dockerfile: Dockerfile
    image: crag
    runtime: nvidia
    container_name: crag
    ports:
      - "8888:8888"
    volumes:
      - .:/app/crag
    deploy:
      resources:
        reservations:
          devices:
            - driver: nvidia
              count: 1
              capabilities: [gpu]
    command: >
      bash -c '/usr/local/bin/ollama serve & /app/.venv/bin/jupyter lab --ip="*" --port=8888 --NotebookApp.token="" --NotebookApp.password="" --no-browser --allow-root'

[Ctrl + S]キーで変更内容を保存し、[Ctrl + X]キーで編集モードから抜けます。

Dockerfileからビルドしてコンテナを起動します。   

docker compose up

Dockerの起動後にブラウザの検索窓に”localhost:8888″を入力すると、Jupyter Labをブラウザで表示できます。

localhost:8888

LangChainとTavilyのAPIに関する環境変数を設定します。

import os
from uuid import uuid4

unique_id = uuid4().hex[0:8]

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = f"crag - {unique_id}"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "***************"
os.environ["TAVILY_API_KEY"] = "***************"

日本語LLMモデル「Llama-3-ELYZA-JP-8B-q4_k_m.gguf」をダウンロードします。

!curl -L -o Llama-3-ELYZA-JP-8B-q4_k_m.gguf "https://huggingface.co/elyza/Llama-3-ELYZA-JP-8B-GGUF/resolve/main/Llama-3-ELYZA-JP-8B-q4_k_m.gguf?download=true"

Llama-3-ELYZA-JPについては、別記事で詳しく解説しています。

あわせて読みたい

GPT-4超えの日本語LLM「Llama-3-ELYZA-JP」の性能・使い方！ Llama-3-ELYZA-JPは、東大・松尾研発のスタートアップELYZAが公開した日本語LLMです。 Llama-3-ELYZA-JP-70Bは、「GPT-4」や「Claude 3 Sonnet」を上回る性能があります…

LLMの実行にはOllamaを使用します。

LLMのモデルがOllama使えるようにプロンプトテンプレートを指定して、モデルを作成します。

import ollama
from langchain_ollama.chat_models import ChatOllama

modelfile='''
FROM ./Llama-3-ELYZA-JP-8B-q4_k_m.gguf
TEMPLATE """{{ if .System }}<|start_header_id|>system<|end_header_id|>

{{ .System }}<|eot_id|>{{ end }}{{ if .Prompt }}<|start_header_id|>user<|end_header_id|>

{{ .Prompt }}<|eot_id|>{{ end }}<|start_header_id|>assistant<|end_header_id|>

{{ .Response }}<|eot_id|>"""
PARAMETER stop "<|start_header_id|>"
PARAMETER stop "<|end_header_id|>"
PARAMETER stop "<|eot_id|>"
PARAMETER stop "<|reserved_special_token"
'''

ollama.create(model='elyza8b', modelfile=modelfile)
llm = ChatOllama(model="elyza8b", temperature=0)
llm.invoke(("human","ヒロアカを知っていますか？")).content

確認のため、LLMからテキストを生成しています。

'「ヒロアカ」こと「僕のヒーローアカデミア」は、非常に人気のある日本の漫画作品です。作者は堀越耕平先生で、週刊少年ジャンプで2014年から連載されています。\n\n物語は、超常能力「個性」を持つ人々が当たり前のように存在する世界を舞台に、主人公の緑谷出久（みどりや いずく）がヒーローになるために通う雄英高校での学園生活と、様々な事件や敵との戦いを描いています。\n\n個性豊かなキャラクターたちが繰り広げるアクション、友情、成長の物語は多くのファンに支持され、TVアニメ化もされています。'

ベクトルストアに格納するデータをWikipediaから取得します。

Wikipediaは、『僕のヒーローアカデミア』のページを指定しています。

from langchain_community.document_loaders import WebBaseLoader
from langchain.text_splitter import CharacterTextSplitter

urls = [
    "https://ja.wikipedia.org/wiki/%E5%83%95%E3%81%AE%E3%83%92%E3%83%BC%E3%83%AD%E3%83%BC%E3%82%A2%E3%82%AB%E3%83%87%E3%83%9F%E3%82%A2",
]

docs = [WebBaseLoader(url).load() for url in urls]
docs_list = [item for sublist in docs for item in sublist]

text_splitter = CharacterTextSplitter.from_tiktoken_encoder(
    separator = "\n",
    chunk_size= 500,
    chunk_overlap=50,
)

doc_splits = text_splitter.split_documents(docs_list)
doc_splits[10]

確認のため、分割したチャンクを1つ表示しています。

Document(metadata={'source': 'https://ja.wikipedia.org/wiki/%E5%83%95%E3%81%AE%E3%83%92%E3%83%BC%E3%83%AD%E3%83%BC%E3%82%A2%E3%82%AB%E3%83%87%E3%83%9F%E3%82%A2', 'title': '僕のヒーローアカデミア - Wikipedia', 'language': 'ja'}, page_content='また、出久のキャラクターは『戦星のバルジ』2巻に描下ろされたあとがき漫画『獄宴編』の主人公が元になっている。それを見た堀越の担当編集者は「君は悟空やルフィのような主人公は描けなかったが、こういうオタクっぽい奴なら描ける」と評価していた[12]。\n連載当初は公式な略称は定まっておらず[13]『ヒロアカ[14]』『僕アカ[15]』などいくつか存在していたが、アニメのテレビシリーズ開始以降は公式な場での略称は『ヒロアカ』で統一されている[4]。\nテーマ・作風')

Wikipediaのデータをもとにベクトル検索が可能なRetrieverを構築します。

from langchain.vectorstores import FAISS
from langchain_huggingface import HuggingFaceEmbeddings

embedding = HuggingFaceEmbeddings(model_name="intfloat/multilingual-e5-large")
vectorstore = FAISS.from_documents(doc_splits, embedding)
retriever = vectorstore.as_retriever(k=4)
question = "デクとは"
retriever.invoke(question)

Retriverから取得したデータを確認しています。

[Document(metadata={'source': 'https://ja.wikipedia.org/wiki/%E5%83%95%E3%81%AE%E3%83%92%E3%83%BC%E3%83%AD%E3%83%BC%E3%82%A2%E3%82%AB%E3%83%87%E3%83%9F%E3%82%A2', 'title': '僕のヒーローアカデミア - Wikipedia', 'language': 'ja'}, page_content='幼馴染の爆豪からは蔑称として「デク」と呼ばれるが、麗日が好意的に受け取ったことで「頑張れって感じ」という意味に転回し、後に自身のヒーロー名として定める。\n私服は無地の物に「○○シャツ」（○○内には「T」や「Y」、「ポロ」、「ドレス」などが入る）と書かれた服をよく着る。深く考えるときは手を口にあて呟く癖があり、その際の台詞が入ったふきだしや背景は、「ブツブツ……」という文字の繰り返しで構成される。'),
以下省略

ユーザーの質問と取得したドキュメントが関連しているかを評価するChainを構築しています。

評価はバイナリースコア（yes または no）で表され、その結果はJSON形式で返されます。

# retrieval_chain
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.prompts import PromptTemplate

prompt = PromptTemplate(
    template="""<|begin_of_text|><|start_header_id|>system<|end_header_id|> 
    あなたは採点者です。以下の内容が与えられます：
    ・質問
    ・事実
    
    あなたは「関連性の再現」を評価しています：
    スコアが1の場合、事実が質問に関連していることを意味します。
    スコアが0の場合、事実が質問に関連していないことを意味します。
    1が最高（最良）のスコアです。0は付けられる最低のスコアです。
    
    理由付けをステップバイステップで説明してください。理由付けと結論が正しいことを確認してください。
    最初に正しい答えを単に述べるのは避けてください。
    
    ドキュメントが質問に関連しているかどうかを示すバイナリスコア「yes」または「いいえ」を提供してください。 \n
    バイナリスコアを単一のキー「score」を持つJSON形式で提供し、前置きや説明は不要です。
     <|eot_id|><|start_header_id|>user<|end_header_id|>
    質問: {question} \n
    事実: \n\n {documents} \n\n <|eot_id|><|start_header_id|>assistant<|end_header_id|>
    """,
    input_variables=["question", "documents"],
)

llm = ChatOllama(model="elyza8b",format="json",temperature=0)
retrieval_chain = prompt | llm | JsonOutputParser()
question = "デクとは？"
docs = retriever.invoke(question)
doc_txt = docs[0].page_content
print(retrieval_chain.invoke({"question": question, "documents": doc_txt}))
print("\nユーザーの質問：\n" + question)
print("\n取得したドキュメント：\n" + docs[0].page_content)

出力結果の確認

{'score': 'yes'}

ユーザーの質問：
デクとは？

取得したドキュメント：
幼馴染の爆豪からは蔑称として「デク」と呼ばれるが、麗日が好意的に受け取ったことで「頑張れって感じ」という意味に転回し、後に自身のヒーロー名として定める。
私服は無地の物に「○○シャツ」（○○内には「T」や「Y」、「ポロ」、「ドレス」などが入る）と書かれた服をよく着る。深く考えるときは手を口にあて呟く癖があり、その際の台詞が入ったふきだしや背景は、「ブツブツ……」という文字の繰り返しで構成される。

質問とドキュメントが関連する場合、{'score': 'yes'}とJSON形式で出力されます。

回答を生成するChainを構築します。

# generate_chain
from langchain_core.output_parsers import StrOutputParser

prompt = PromptTemplate(
    template="""<|begin_of_text|><|start_header_id|>system<|end_header_id|> 
    あなたは質問応答のためのアシスタントです。
    次のドキュメントを使って質問に答えてください。 
    答えがわからない場合は、正直に「わかりません」と言ってください。
    回答は最大で3文にまとめ、簡潔にしてください。
    <|eot_id|><|start_header_id|>user<|end_header_id|>
    質問: {question} 
    コンテキスト: {documents} 
    回答: <|eot_id|><|start_header_id|>assistant<|end_header_id|>""",
    input_variables=["question", "documents"],
)

llm = ChatOllama(model="elyza8b",temperature=0)
generate_chain = prompt | llm | StrOutputParser()

question = "デクとは？"
docs = retriever.invoke(question)
generation = generate_chain.invoke({"documents": docs, "question": question})
print("\nユーザーの質問：\n" + question)
print("\n生成した回答：\n" + generation)

出力結果の確認

ユーザーの質問：
デクとは？

生成した回答：
デクとは、緑谷出久のヒーロー名です。

Tavily APIを利用してWEB検索を行うToolの設定をします。

# web_search_tool
from langchain_community.tools.tavily_search import TavilySearchResults

web_search_tool = TavilySearchResults(max_results=3)
question = "デクとは？"
web_search_tool.invoke({"query": question})

デクについてWeb検索した結果

[{'url': 'https://dic.pixiv.net/a/デク',
  'content': 'デクがイラスト付きでわかる! デクには複数の意味があるが、その殆どは緑谷出久のヒーロー名として使われる事が多い。 曖昧さ回避 +『僕のヒーローアカデミア』の主人公。ピクシブ上では彼を描いた作品に付けられることがほとんど→ 緑谷出久 +『僕のヒーローアカデミア』の構想段階で ...'},
(以下省略)

必要なライブラリをインポートします。

from pprint import pprint
from typing import List
from langchain_core.documents import Document
from typing_extensions import TypedDict

Graphを初期化したときのノードやエッジの状態を定義します。

class GraphState(TypedDict):
    question: str
    generation: str
    search: str
    documents: List[str]
    steps: List[str]

Retriverを使用してベクトルストアからユーザーの質問に関連するドキュメントを取得します。

「question(質問)」と「documents(ドキュメント)」、「steps」を次のステップで使用するため、状態 (state) を更新します。

# Retriver関数
def retrieve(state):
    print("---RETRIEVE---")
    question = state["question"]
    documents = retriever.invoke(question)
    steps = state["steps"]
    steps.append("retrieve_documents")
    return {"documents": documents, "question": question, "steps": steps}

ユーザーの質問にもとづいた回答を生成します。

「documents(ドキュメント)」と「question(質問)」、「generation(回答)」、「steps」を次のステップで使用するため、状態 (state) を更新します。

# 回答を生成する関数
def generate(state):
    print("---GENERATE---")
    question = state["question"]
    documents = state["documents"]
    generation = generate_chain.invoke({"documents": documents, "question": question})
    steps = state["steps"]
    steps.append("generate_answer")
    return {
        "documents": documents,
        "question": question,
        "generation": generation,
        "steps": steps,
    }

取得したドキュメントが質問に関連しているかどうかを評価します。

関連性の高いドキュメントを選別し、リストに格納します。

関連性が低いドキュメントが存在する場合は、Web検索に進むフラグを立てます。

「filtered_docs(関連性のあるドキュメント)」と「question(質問)」、「search(検索)」、「steps」を次のステップで使用するため、状態(sate)を更新します。

# ドキュメントの関連性を評価する関数
def grade_documents(state):
    print("---CHECK DOCUMENT RELEVANCE TO QUESTION---")
    question = state["question"]
    documents = state["documents"]
    steps = state["steps"]
    steps.append("grade_document_retrieval")
    filtered_docs = []
    search = "No"
    for d in documents:
        score = retrieval_chain.invoke(
            {"question": question, "documents": d.page_content}
        )
        grade = score["score"]
        if grade == "yes":
            print("---GRADE: DOCUMENT RELEVANT---")
            filtered_docs.append(d)
        else:
            print("---GRADE: DOCUMENT NOT RELEVANT---")
            search = "Yes"
            continue
    return {
        "documents": filtered_docs,
        "question": question,
        "search": search,
        "steps": steps,
    }

質問に基づいてウェブ検索を実行します。検索結果を取得して、それをドキュメントに追加します。

「question(質問)」と「documents(ドキュメント)」、「steps」を次のステップで使用するため、状態(state)を更新します。

# Web検索の関数
def web_search(state):
    print("---WEB SEARCH---")
    question = state["question"]
    documents = state.get("documents", [])
    steps = state["steps"]
    steps.append("web_search")
    web_results = web_search_tool.invoke({"query": question})
    documents.extend(
        [
            Document(page_content=d["content"], metadata={"url": d["url"]})
            for d in web_results
        ]
    )
    return {"documents": documents, "question": question, "steps": steps}

WEB検索をするか、RAGによる生成をするかを判断します。

この関数は「search」または「generate」を返します。この返り値は、後続のステップで利用されます。

#Web検索を判断する関数
def decide_to_generate(state):
    print("---GENERATE OR WEB SEARCH---")
    search = state["search"]
    if search == "Yes":
        print("---TO WEB SEARCH---")
        return "search"
    else:
        print("---TO GENERATE---")
        return "generate"

必要なライブラリをインポートします。

from langgraph.graph import END, StateGraph, START
from IPython.display import Image, display
from langchain_core.runnables.graph import CurveStyle, MermaidDrawMethod, NodeStyles

Graphのワークフローを作成し、各ノードを追加します。

#Graphのワークフローを作成
workflow = StateGraph(GraphState)

#ノードの追加
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("web_search", web_search)

ワークフローにおけるエッジ（ノード間の接続）の設定を行います。

エッジは、各ノードをどの順序で実行するか、どのような条件で次のノードに進むかを定義します。

# Edgeの追加
workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
workflow.add_conditional_edges(
    "grade_documents",
    decide_to_generate,
    {
        "search": "web_search",
        "generate": "generate",
    },
)
workflow.add_edge("web_search", "generate")
workflow.add_edge("generate", END)

全体のワークフローをコンパイルして、実行可能な形にします。

# コンパイル
app = workflow.compile()

定義したLangGraphワークフローのグラフ構造を、Mermaid形式で可視化します。

# ワークフローの可視化
display(
    Image(
        app.get_graph().draw_mermaid_png(
            draw_method=MermaidDrawMethod.API,
        )
    )
)