Zenn
EGGHEAD テックブログPublicationへの投稿
🐡

LangGraphで学ぶAgentic RAG解説

酒井 駿
に公開
LangChain
LLM
RAG
LangGraph
tech

こんにちは、酒井です！
株式会社 EGGHEAD（エッグヘッド）という「製造業で生成 AI を活用したシステム開発」をしている会社の代表をしております。

https://egghead.co.jp?utm_source=zenn&utm_medium=social&utm_campaign=normal

今回は LangGraph の Agentic RAG についてのチュートリアルをもとに Agentic RAG とは？やサンプルコードの補足解説をしました。

本記事を読むことで、以下のポイントが理解できるようになります。

👉 従来のRAGの限界と、Agentic RAGのメリットが分かる
👉 LangGraphを使ったAgentic RAGの基本的な実装方法が学べる

本記事が、LangGraph を活用した AI エージェント開発の初学者の方の参考になれば幸いです。

https://langchain-ai.github.io/langgraph/tutorials/rag/langgraph_agentic_rag/

1. はじめに

従来の RAG とその問題点

検索拡張生成（Retrieval-Augmented Generation、RAG）は、ドキュメントをエンべディングを行ってベクトルデータベースに保存して、ユーザーのクエリから近しい文脈を取り出し、LLM に関連するコンテキストとして提供し外部知識に基づいた回答を行うという技術です。

しかし、従来の RAG には以下の 2 つの大きな制約があります。

問題点 1：単一の知識ソースに限定される

基本的な RAG パイプラインは一つの外部知識のソースのみを使って検索を行うことです。多くの場合、複数の外部知識がないと答えられないケースがあります。

問題点 2：検証なしのワンショット検索

基本的に検索はワンショットで行うため、正しい文脈が検索できていない可能性があることです。従来の RAG では取得されたコンテキストが正しいかの検証は行いません。

これらを解決するのが Agentic RAG と呼ばれるアーキテクチャーです。

2. Agentic RAG とは

Agentic RAG とは、AI エージェントベースの RAG 実装を指します。従来の RAG の問題点を解決するために、エージェントは状態を持ち、自律的に判断をして行動します。具体的には、「検索するかどうか」「取得した文書は関連性があるか」「クエリを書き換えるべきか」といった判断を自律的に行い、複数の知識ソースを活用したり、検索結果の検証を行ったりすることで、より精度の高い回答を生成することができます。

図が示すように、Agentic RAGの最大の特徴は「考えて行動する」能力です。従来のRAGが単純に「検索して回答する」だけなのに対し、Agentic RAGは以下のような判断と行動(ReAct)を自律的に行います。

  1. 質問分析: ユーザーの質問を深く理解し、本当に知りたいことは何かを把握します
  2. 検索判断: 外部情報が必要かどうかを自分で判断します（既知の情報なら検索せずに回答）
  3. 情報源選択: 質問に応じて最適な情報源（データベース、API、ウェブなど）を選びます
  4. 結果評価: 検索結果が質問に関連しているか、十分かを評価します
  5. クエリ改善: 検索結果が不十分なら、より良いキーワードで再検索します
  6. ユーザー連携: 必要に応じて、ユーザーに追加情報を求めたり確認したりします

また、Agentic RAG と従来の RAG と比較した図がこちらになります。

機能 従来の RAG Agentic RAG
外部ツールへのアクセス いいえ はい
必要に応じたクエリ前処理 いいえ はい
複数ステップの検索 いいえ はい
取得した情報の検証 いいえ はい

上記の例や比較表が示すように、Agentic RAG は従来の RAG の限界を超え、複数の情報源の活用や検索結果の検証など、より高度な機能を備えています。これにより、ユーザーの質問に対してより正確で関連性の高い回答を提供することが可能になります。

Agentic RAGについての詳細はWeaviateのこちらの記事が分かりやすかったのでおすすめです！

https://weaviate.io/blog/what-is-agentic-rag

3. AI エージェント実装のアーキテクチャ

まずは、タスクを自動的に行うためのアプローチを整理したいと思います。主に二種類のアプローチに分けられます。

シーケンシャルアプローチ

最も単純な方法で、一連の処理を順番に実行します。Dify のワークフローにイメージが近いです。

グラフ構造アプローチ

グラフ構造アプローチでは、各処理ステップを「ノード」として定義し、それらの間の関係性や遷移を「エッジ」として接続することでグラフを構築します。このグラフ構造によって、処理の流れを柔軟に制御でき、条件に応じて異なる経路を選択することが可能になります。例えば、検索結果が不十分な場合は再検索ノードに進むなど、状況に応じた判断ができるエージェントを実装できます。

AI エージェントとは「ゴールに対して自律的にタスクを遂行できる AI システム」です。この AI エージェントを実装する効果的な方法がグラフ構造アプローチで、LangGraph はこのグラフ構造を簡単に実装できるフレームワークです。

グラフ構造について

グラフ構造は、以下の 3 つの要素で構成されています。

  1. ノード: 各処理（エージェント判断、検索、クエリ書き換え、回答生成など）を実行する場所
  2. エッジ: ノード間の接続線で、処理の流れを定義
  3. 条件付きエッジ: 状態に応じて次に進むノードを動的に決定する特殊なエッジ

これらの要素を組み合わせることで、柔軟で状況に応じた処理の流れを実現できます。

4. Agentic RAG の実装詳細

ここからは LangGraph のチュートリアルの例をもとに LangGraph を使った Agentic RAG の実装の詳細を見ていきます。
主要なノードは以下となっています。

  1. エージェントノード: LLM を使用して文書取得の判断を行う
  2. 文書取得ノード: ユーザーの質問に関連する文書を取得する
  3. クエリ再構築ノード: ユーザーの質問を改善する
  4. 回答生成ノード: 取得した文書を使用して回答を生成する

これらのノードで以下のようにワークフローが実行されます。ここでは下記のようなワークフローを LangGraph を使って実装していきます。

状態管理

LangGraph では、各ノードで状態（State）を共有します。各ノードは独立していますが、この状態を更新（メッセージを追加）することで情報を伝達します。
たとえば、エージェントノードはツール呼び出しのメッセージを追加し、文書取得ノードは取得した文書の内容をメッセージとして追加します。

class AgentState(TypedDict):
    # add_messages関数は更新方法を定義します
    # デフォルトは置換ですが、add_messagesは「追加」を意味します
    messages: Annotated[Sequence[BaseMessage], add_messages]

グラフ構造の構築

メインであるグラフ構造の構築は以下のように行います。StateGraphの引数に先ほどのAgentStateを指定して初期化し、そこにadd_node、add_edge、add_conditional_edgesを使ってグラフ構造を作成していきます。
add_nodeとadd_edgeの第二引数にはそれぞれ関数が指定されていて、これが呼び出されます。

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent)
retrieve = ToolNode([retriever_tool])
workflow.add_node("retrieve", retrieve)
workflow.add_node("rewrite", rewrite)
workflow.add_node("generate", generate)

workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", tools_condition, {"tools": "retrieve", END: END})
workflow.add_conditional_edges("retrieve", grade_documents)
workflow.add_edge("generate", END)
workflow.add_edge("rewrite", "agent")

graph = workflow.compile()

このコードは、起点となるエージェントから始まり、条件に応じて検索、評価、書き換え、生成といった一連のプロセスをグラフとして定義しています。compile()メソッドで実行可能なグラフが生成され、stream()メソッドで実際に処理を行います。

エージェントノードによるクエリ分析と検索判断

ユーザーの質問が入力されると、まずエージェントノード（agent関数）が実行されます。

  1. エージェントはユーザーの質問を分析し、検索ツールを使用するかどうかを判断します。
  2. この判断は、質問の性質（事実関連の質問か意見を求める質問かなど）や、モデルが既に知識を持っているかどうかに基づいて行われます。
  3. 検索が必要と判断した場合、エージェントはツール呼び出し（tool_calls）を生成し、状態に追加します。
def agent(state):
    messages = state["messages"]
    model = ChatOpenAI(temperature=0, streaming=True, model="gpt-4o")
    model = model.bind_tools(tools)
    response = model.invoke(messages)
    return {"messages": [response]}

文書の関連性評価と条件付きエッジの実装

検索ツールが呼び出されると、文書取得ノードが実行され、ベクトル DB から関連文書を取得します。

  1. 文書取得ノードはリトリーバーツールを使い、ユーザーの質問を使用してベクトルストアに問い合わせます。
  2. 取得された文書は、ツールメッセージとして状態に追加されます。
  3. 次に、grade_documents関数が文書の関連性を評価します。
def grade_documents(state) -> Literal["generate", "rewrite"]:
    model = ChatOpenAI(temperature=0, model="gpt-4o", streaming=True)
    llm_with_tool = model.with_structured_output(grade)

    messages = state["messages"]
    question = messages[0].content
    docs = last_message.content

    scored_result = chain.invoke({"question": question, "context": docs})
    score = scored_result.binary_score

    if score == "yes":
        return "generate"
    else:
        return "rewrite"

この関数は取得された文書がユーザーの質問に関連しているかどうかを LLM に判断させています。関連性があれば「generate」ノードへ、なければ「rewrite」ノードへと処理が進みます。

クエリ再構築ノードによる検索クエリの最適化

文書の関連性が低いと判断された場合、クエリ再構築ノードが実行されます。

  1. rewrite関数は、元の質問の意味を解析し、より効果的な検索クエリになるよう書き換えます。
  2. 書き換えられたクエリは、新しいメッセージとして状態に追加されます。
  3. 処理は再びエージェントノードに戻り、新しいクエリで検索を試みます。
def rewrite(state):
    messages = state["messages"]
    question = messages[0].content

    msg = [
        HumanMessage(
            content=f""" \n
    入力を見て、基本的な意味的意図/意味について考えてみてください。 \n
    これが最初の質問です:
    \n ------- \n
    {question}
    \n ------- \n
    改善された質問を作成してください: """,
        )
    ]

    model = ChatOpenAI(temperature=0, model="gpt-4o", streaming=True)
    response = model.invoke(msg)
    return {"messages": [response]}

このプロセスにより、ユーザーの質問が不明瞭だったとしても、システムは自律的に改善して適切な文書を検索することができます。

回答生成ノードによる最終レスポンスの作成

関連性の高い文書が取得できた場合、生成ノードが実行されて最終的な回答を作成します。

  1. generate関数は、ユーザーの質問と取得された文書を組み合わせてプロンプトを作成します。
  2. このプロンプトは LLM に送られ、回答が生成されます。
  3. 生成された回答は、最終的なメッセージとして状態に追加され、ユーザーに返されます。
def generate(state):
    messages = state["messages"]
    question = messages[0].content
    last_message = messages[-1]
    docs = last_message.content

    prompt = hub.pull("rlm/rag-prompt")
    llm = ChatOpenAI(model_name="gpt-4o-mini", temperature=0, streaming=True)
    rag_chain = prompt | llm | StrOutputParser()

    response = rag_chain.invoke({"context": docs, "question": question})
    return {"messages": [response]}

この回答生成プロセスでは、RAG 専用に設計されたプロンプトを使用して、取得した文書の情報を正確に反映した回答を作成します。

実行結果が以下になります。回答が英語で表示されているのはrlm/rag-promptのプロンプトが英語なためなので、日本語にしたい場合は、プロンプトを自分で書く必要があります。

---エージェント呼び出し---
"Output from node 'agent':"
'---'
{ 'messages': [ AIMessage(content='', additional_kwargs={'tool_calls': [{'index': 0, 'id': 'call_hoAPuod1TD9fUBF7LIDwiuMj', 'function': {'arguments': '{"query":"Lilian Weng agent memory types"}', 'name': 'retrieve_blog_posts'}, 'type': 'function'}]}, response_metadata={'finish_reason': 'tool_calls', 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_eb9dce56a8'}, id='run-8ca9485e-ef3a-47a5-b588-3bc2001194e1-0', tool_calls=[{'name': 'retrieve_blog_posts', 'args': {'query': 'Lilian Weng agent memory types'}, 'id': 'call_hoAPuod1TD9fUBF7LIDwiuMj', 'type': 'tool_call'}])]}
'\n---\n'
---関連性チェック---
---判断: ドキュメントは関連あり---
"Output from node 'retrieve':"
'---'
{ 'messages': [ ToolMessage(content='The design of generative agents combines LLM with memory, planning and reflection mechanisms to enable agents to behave conditioned on past experience, as well as to interact with other agents.\n\nTable of Contents\n\n\n\nAgent System Overview\n\nComponent One: Planning\n\nTask Decomposition\n\nSelf-Reflection\n\n\nComponent Two: Memory\n\nTypes of Memory\n\nMaximum Inner Product Search (MIPS)\n\n\nComponent Three: Tool Use\n\nCase Studies\n\nScientific Discovery Agent\n\nGenerative Agents Simulation\n\nProof-of-Concept Examples\n\n\nChallenges\n\nCitation\n\nReferences\n\nCitation#\nCited as:\n\nWeng, Lilian. (Jun 2023). “LLM-powered Autonomous Agents”. Lil’Log. https://lilianweng.github.io/posts/2023-06-23-agent/.\n\nMemory\n\nShort-term memory: I would consider all the in-context learning (See Prompt Engineering) as utilizing short-term memory of the model to learn.\nLong-term memory: This provides the agent with the capability to retain and recall (infinite) information over extended periods, often by leveraging an external vector store and fast retrieval.\n\n\nTool use', name='retrieve_blog_posts', id='d0560d74-2dbf-4f76-82e7-01049cf68f83', tool_call_id='call_hoAPuod1TD9fUBF7LIDwiuMj')]}
'\n---\n'
---生成---
/usr/local/lib/python3.11/dist-packages/langsmith/client.py:253: LangSmithMissingAPIKeyWarning: API key must be provided when using hosted LangSmith API
  warnings.warn(
"Output from node 'generate':"
'---'
{ 'messages': [ 'Lilian Weng discusses two types of memory in generative '
                'agents: short-term memory and long-term memory. Short-term '
                'memory involves in-context learning, while long-term memory '
                'allows agents to retain and recall information over extended '
                'periods using an external vector store. This dual memory '
                "system enhances the agents' ability to learn from past "
                'experiences and interact effectively.']}
'\n---\n'

簡略化のためところどころ省略しているので、実際に動かしてみたい方はこちらでお試しください！
Google Colabノートブック

5. まとめ

ここまで読んでいただきありがとうございました！
今回は、LangGraph のサンプルコードから Agentic RAG の概念と実装方法について見ていきました。複数の情報源からの検索や検索結果の検証を行うことができる Agentic RAG は、より高度な質問応答システムを構築するための重要なアプローチです。

自分もこの実装を参考により実用的な RAG の開発をしたいと思いました。

GitHubで編集を提案
EGGHEAD テックブログ
EGGHEAD テックブログPublication

製造業を中心に生成AIを使った業務改善のシステム開発を行っています。

Discussion