はじめに

「LangChainとLangGraphによるRAG・AIエージェント［実践］入門」の第6章で私がつまずいたことのメモです。

（このメモのほかの章へ：1章 / 2章 / 3章 / 4章 / 5章 / 6章）

この記事は個人で作成したものであり、内容や意見は所属企業・部門見解を代表するものではありません。

第6章 Advanced RAG

前章に続き大嶋さんが担当された章です。

6.1 Advaned RAGの概要

RAG界隈は本当にたくさんの工夫が編み出されていますね。昔ながらの手法も組み合わせられていたりして、温故知新を感じる部分も多いです。

6.2 ハンズオンの準備

本のコードを実行すると、なぜかドキュメントをベクトル化するところで失敗してしまいました。

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
db = Chroma.from_documents(documents, embeddings)

エラー内容

---------------------------------------------------------------------------
TypeError                                 Traceback (most recent call last)
<ipython-input-4-1b5f75d6ec00> in <cell line: 4>()
      2 from langchain_openai import OpenAIEmbeddings
      3 
----> 4 embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
      5 db = Chroma.from_documents(documents, embeddings)

    [... skipping hidden 1 frame]

3 frames
/usr/local/lib/python3.10/dist-packages/langchain_openai/embeddings/base.py in validate_environment(self)
    336                 self.http_client = httpx.Client(proxy=self.openai_proxy)
    337             sync_specific = {"http_client": self.http_client}
--> 338             self.client = openai.OpenAI(**client_params, **sync_specific).embeddings  # type: ignore[arg-type]
    339         if not self.async_client:
    340             if self.openai_proxy and not self.http_async_client:

/usr/local/lib/python3.10/dist-packages/openai/_client.py in __init__(self, api_key, organization, project, base_url, timeout, max_retries, default_headers, default_query, http_client, _strict_response_validation)
    121             base_url = f"https://api.openai.com/v1"
    122 
--> 123         super().__init__(
    124             version=__version__,
    125             base_url=base_url,

/usr/local/lib/python3.10/dist-packages/openai/_base_client.py in __init__(self, version, base_url, max_retries, timeout, transport, proxies, limits, http_client, custom_headers, custom_query, _strict_response_validation)
    855             _strict_response_validation=_strict_response_validation,
    856         )
--> 857         self._client = http_client or SyncHttpxClientWrapper(
    858             base_url=base_url,
    859             # cast to a valid type because mypy doesn't understand our type narrowing

/usr/local/lib/python3.10/dist-packages/openai/_base_client.py in __init__(self, **kwargs)
    753         kwargs.setdefault("limits", DEFAULT_CONNECTION_LIMITS)
    754         kwargs.setdefault("follow_redirects", True)
--> 755         super().__init__(**kwargs)
    756 
    757 

TypeError: Client.__init__() got an unexpected keyword argument 'proxies'

どうやら、HTTPクライアントのライブラリであるhttpxが0.28.0でproxiesという引数がなくなった¹のに、openaiの古いライブラリがそれを使ってしまっていることが原因らしいです。openaiのライブラリもv1.55.3ではhttpxの0.28.0に対応している²ので、openaiのライブラリを上げると解消します。

正しい対応方法なのか自信はないのですが、パッケージのインストール部分で以下のようにopenai==1.55.3を付け加えてあげるとエラーが解消します。うまくいかない場合は、Google Colabの「ランタイム」メニューで「ランタイムを接続解除して削除」を選び、環境をリセットしてからやり直してみてください。

!pip install langchain-core==0.3.0 langchain-openai==0.2.0 \
    langchain-community==0.3.0 GitPython==3.1.43 \
    langchain-chroma==0.1.4 tavily-python==0.5.0 openai==1.55.3

あと、コラムの「インデクシングの工夫」がホントその通りで首がもげます。私は所属企業でRAGのサービスを提供しはじめて1年半ほど経ちますが、現在のインデクシングの技術は万能ではなく、まだまだ工夫が必要です。宣伝になってしまいますが、以下の記事でこの辺りにも触れていますのでご興味がありましたらぜひ！

6.3 検索クエリの工夫

紹介されているHyDEの手法は、昔ながらのチャットボット（事前登録したQ&Aのペアを使って自動応答するもの）に近い発想です。チャットボットではユーザーからの質問文をクエリとしてQ（質問）を検索します。A（回答）は検索しません。そして、質問文と関連が高いと思われるQを見つけたら、それに対応するAを回答します。質問では回答を探しにくいという前提がHyDEと同じなのです。なお、要件や対象データによってはHyDEの手法をひっくり返して、検索対象のデータに対する想定質問をLLMに生成させてインデクシングしておき、それを検索する形の方が有効なこともあるでしょう。

また、複数の検索クエリを併用する手法もいいですね。たとえば、買おうかどうか悩んでいる商品をネットで調べる時、肯定的なレビュー記事と否定的なレビュー記事の両方をチェックしたくなるものです。「5.3 RunnableParallelー複数のRunnableを並列につなげる」でも、楽観的な意見と悲観的な意見から回答を生成する例がありましたが、いろいろな観点で情報を調べるというのは回答精度の向上に役立ちそうです。

そういえば、複数クエリの実装コードは、「4.5 ChainーLangChain Expression Language（LCEL）の概要」の最後のコラムで紹介されていたwith_structured_outputを利用する形になっていました。シンプルに書けてうれしい限りです。

6.4 検索後の工夫

リランクとしてRRFとCohereのリランクモデルが紹介されていますが、実は小難しいリランクのアルゴリズムだけでなく、単純な仕組みも有効だったりします。たとえば、対象データの更新日時で降順ソートするだけでも、最新情報を知りたがっているような質問には有効でしょう。また、ECサイトの商品検索などでは、リコメンドのデータを使ってソートする方が売上に貢献するかもしれません。万能なものはないので、目的に合わせた選択がポイントになるかと思います。

6.5 複数のRetrieverを使う工夫

本ではLLMによるルーティングの例で天気を聞いていますが、「5.4 RunnablePassthroughー入力をそのまま出力する」のメモにも書いたように、残念ながらTavilySearchAPIRetrieverでは最新の情報が取得できません。たとえば今日は12月7日(土)なのですが、古い情報を答えてしまいます。

route_rag_chain.invoke("東京の今日の天気は？")

東京の今日、2024年11月19日(火)の天気は晴時々曇で、最高気温は13℃、最低気温は8℃です。降水確率は0％です。

そこで、「2.6 Function calling」のメモで作ったコードをを持ってきて、天気予報のRetrieverを作ってみました。

from langchain_core.retrievers import BaseRetriever
from typing import List
import json
import requests

class WeatherRetriever(BaseRetriever):
    def _get_relevant_documents(self, query: str) -> List[Document]:
        # 気象庁の予報区コードの一覧を取得
        url = "https://www.jma.go.jp/bosai/common/const/area.json"
        response = requests.get(url)
        data = json.loads(response.text)

        # クエリに地名が含まれていたらその予報区コードを取得
        for code, info in data["offices"].items():
            if info["name"] in query:
                # 概観取得
                url = f"https://www.jma.go.jp/bosai/forecast/data/overview_forecast/{code}.json"
                response = requests.get(url)
                data = response.json()
                return [data['text']]

        # 見つからなかった
        return ["その地域の予報はわかりませんでした。"]

クエリの意味などはまったく無視して、クエリ中に気象庁の予報区コードの都道府県名があったら該当する概観の情報を返す、という超やっつけコードです。手前でLLMがルーティングしてくれて天気に関するクエリしかやってこないので、なんとなくいい感じになるでしょう
なお、気象庁ホームページに関するコードの詳細については、2章のメモを参照してください。

このコードは気象庁ホームページにアクセスして情報を取得する流れになっています。利用する際は気象庁ホームページ利用規約を守ってください。

エラー処理やテストは省きまくっています。動かす時は自己責任でお願いします。

さっそく組み込んでみましょう。まず、インスタンスを作ります。

weather_retriever = WeatherRetriever().with_config(
    {"run_name": "weather_retriever"}
)

そして、Retrieverを選択するChainに追加します。

from enum import Enum


class Route(str, Enum):
    langchain_document = "langchain_document"
    web = "web"
    weather = "weather"  # 追加


class RouteOutput(BaseModel):
    route: Route


route_prompt = ChatPromptTemplate.from_template("""\
質問に回答するために適切なRetrieverを選択してください。

質問: {question}
""")

route_chain = (
    route_prompt
    | model.with_structured_output(RouteOutput)
    | (lambda x: x.route)
)

最後にルーティングの結果を踏まえて検索する部分の修正です。

def routed_retriever(inp: dict[str, Any]) -> list[Document]:
    question = inp["question"]
    route = inp["route"]

    if route == Route.langchain_document:
        return langchain_document_retriever.invoke(question)
    elif route == Route.web:
        return web_retriever.invoke(question)
    elif route == Route.weather:  # 追加
        return weather_retriever.invoke(question)

    raise ValueError(f"Unknown retriever: {retriever}")


route_rag_chain = (
    {
        "question": RunnablePassthrough(),
        "route": route_chain,
    }
    | RunnablePassthrough.assign(context=routed_retriever)
    | prompt | model | StrOutputParser()
)

質問してみます。

route_rag_chain.invoke("千葉県の今日の天気は？")

千葉県の今日の天気は曇り時々晴れとなるでしょう。

この回答ではいつの情報を使っているのかわからないのでLangSmithで確認します。

お、いい感じ！きちんと作ったWeatherRetrieverが動いています。
念のため、天気に関係ない質問もしてみましょう。

route_rag_chain.invoke("千葉県の面積は？")

千葉県の面積は5,157平方キロメートルです。

正しくTavilySearchAPIRetrieverにルーティングされていますね、よしよし。

本では続いてハイブリッド検索の例が解説されています。最近、単純なベクトル検索ではなくハイブリッドが主流になってきました。「複数のRetrieverを使う工夫のまとめ」にもあるように、検索対象はさまざまで、それぞれに合った検索手段があります。また、要件によっても検索手段は変わります。万能な検索手段はまだないので、これからもいろいろな工夫が編み出されていくかと思います。

6.6 まとめ

今の仕事に絡んでいるので大変興味深い章でした。次回はRAGの評価のお話です。

（このメモのほかの章へ：1章 / 2章 / 3章 / 4章 / 5章 / 6章）