LangChainのTool "DuckDuckGo Search" を拡張

LangChainのTool "DuckDuckGo Search"を拡張しました。

拡張内容

単純な検索ではなく、以下の5ステップにしています。

1. 検索: DuckDuck Go で検索
2. ページ読込: WebBaseLoaderで検索結果のページ読込
3. Text Split: RecursiveCharacterTextSplitterで読み込んだページを分割
4. Vector Store格納: Chromaに入れる(EmbeddingはOpenAI)
5. Vector Store検索: Chromaから単純に類似検索

動機

標準だと検索結果ページの概要部分(ページの上部？)しか取得してくれないので、単純なクエリ(例: 「今日の東京の天気は？」)には強いのです(LangSmithの画面参照)。

検索結果のサマリだけだと得られない情報の場合には、RAGがうまくいかないです。

LLM(OpenAIに渡す)DuckDuckGoの結果を見るとこんな感じ。そもそものクエリもあまり良くないですが。

拡張前のプログラム

Python 3.12.2で以下のパッケージ使っています。

Package	Version
duckduckgo_search	6.1.5
langchain	0.2.5
langchain-community	0.2.5
langchain-openai	0.1.9
python-dotenv	1.0.1

シンプルなFunction CallingでTool使ったプログラムです。明示的に実行していませんが、python-dotenvでファイル.envから環境変数を読み込んでいます。

拡張前

from langchain import hub
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchResults
from langchain_community.utilities import DuckDuckGoSearchAPIWrapper

wrapper = DuckDuckGoSearchAPIWrapper(region="jp-jp", max_results=3)
search = DuckDuckGoSearchResults(api_wrapper=wrapper, source="text")
model = ChatOpenAI(model_name='gpt-4-turbo')
tools = [search]

instructions = """You are an assistant for question-answering tasks. 
You are an agent designed to get information from DuckDuckGo Search answer questions.
"""

base_prompt = hub.pull("langchain-ai/openai-functions-template")
prompt = base_prompt.partial(instructions=instructions)

agent = create_openai_functions_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
await agent_executor.ainvoke({"input": "漫画の「火の鳥」には何編がある？"})

拡張後

Python Package

Python3.12.2で以下のパッケージ使っています

Package	Version	備考
duckduckgo_search	6.1.5
fake-useragent	1.5.1
langchain	0.2.5
langchain-community	0.2.5
langchain-openai	0.1.9
langchain-chroma	0.1.1
nest-asyncio	1.6.0	非同期実行する場合
python-dotenv	1.0.1

明示的に実行していませんが、python-dotenvでファイル.envから環境変数を読み込んでいます。

プログラム

1. 拡張クラス定義に必要な Package Import

拡張クラス定義に必要な Package を Import

from typing import Optional

from duckduckgo_search import DDGS
from fake_useragent import UserAgent
from langchain_chroma import Chroma
from langchain_community.tools import DuckDuckGoSearchResults
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.callbacks import CallbackManagerForToolRun
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

2. クラス拡張

DuckDuckGoSearchResultsのサブクラスを定義します。
_runと_arunを定義しています。パラメータrun_managerは使用していません。

class DuckDuckGoSearchExtend(DuckDuckGoSearchResults):
    def _search(self,
                query: str, 
                sync: bool=True) -> list:
        urls = [result['href'] for result in DDGS().text(
                keywords=query,      # 検索ワード
                region='jp-jp',       # リージョン 日本は"jp-jp",指定なしの場合は"wt-wt"
                safesearch='off',     # セーフサーチOFF->"off",ON->"on",標準->"moderate"
                timelimit=None,       # 期間指定 指定なし->None,過去1日->"d",過去1週間->"w", 過去1か月->"m",過去1年->"y"
                max_results=3)]
        loader = WebBaseLoader(urls, header_template={'User-Agent': UserAgent().chrome})
        if sync:
            docs = loader.load()
        else:       
            loader.requests_per_second = 10
            docs = loader.aload()
        text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
        splits = text_splitter.split_documents(docs)
        db = Chroma.from_documents(documents=splits, embedding=OpenAIEmbeddings())

        context = [content.page_content for content in db.similarity_search(query, k=3)]
        return context


    def _run(
            self,
            query: str,
            run_manager: Optional[CallbackManagerForToolRun]=None,
    ) -> str:
        return self._search(query)


    async def _arun(
            self,
            query: str,
            run_manager: Optional[CallbackManagerForToolRun]=None,
    ) -> str:
        return self._search(query, False)

試しにインスタンス化。

search = DuckDuckGoSearchExtend()

print(f"{search.name=}")
print(f"{search.description=}")
print(f"{search.args=}")

親クラスのDuckDuckGoSearchResultsの内容がそのまま出ています。

search.name='duckduckgo_results_json'
search.description='A wrapper around Duck Duck Go Search. Useful for when you need to answer questions about current events. Input should be a search query. Output is a JSON array of the query results'
search.args={'query': {'title': 'Query', 'description': 'search query to look up', 'type': 'string'}}

検索をしてみます。

同期

context = search.run("水原一平")

きちんと検索できています。

contextの中身抜粋

['水原一平 - Wikipedia\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\nホーム\n\n\n\n\n\nおまかせ表示\n\n\n\n\n\n付近\n\n\n\n\n\n\n\nログイン\n\n\n\n\n\n\n\n設定\n\n\n\n\n\n\n\n寄付\n\n\n\n\n\n\nウィキペディアについて\n\n\n\n\n免責事項\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n検索\n\n\n\n\n\n\n\n\n\n\n\n\n水原一平\n日本の元通訳者 (1984-)\n\n\n\n\n\n\n言語\n\n\n\n\n\nウォッチリストに追加\n\n\n\n\n\nソースを閲覧\n\n\n\n\n\n\n\n\n\n\nこの記事は最新の出来事を扱っています。記載される内容は出来事の進行によって急速に変更される可能性があります。（2024年4月）\n水原 一平（みずはら いっぺい、1984年12月31日 - ）は、日本の元通訳者及びアメリカの元通訳者である。大谷翔平の専属通訳として、2017年か

後略

非同期だとこんな呼び方(あまり自信なし)。同期だと5秒くらいだったのが、非同期だと4.7秒。WebBaseLoaderを並列で読み込んでいるので少し速い。

非同期

import asyncio
import nest_asyncio

nest_asyncio.apply()
loop = asyncio.get_event_loop()
context = loop.run_until_complete(search.arun("水原一平"))

３． Chain作成

Function CallingをするAgentのChainを作成。

from langchain import hub
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model_name='gpt-4-turbo')
tools = [search]

instructions = """You are an assistant for question-answering tasks. 
You are an agent designed to get information from DuckDuckGo Search answer questions.
"""

base_prompt = hub.pull("langchain-ai/openai-functions-template")
prompt = base_prompt.partial(instructions=instructions)

agent = create_openai_functions_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

4. 実行

同期で実行

agent_executor.invoke({"input": "漫画の「火の鳥」には何編がある？"})

拡張前の以下の回答と違い、だいたい合っていると思います。

漫画「火の鳥」は、全12章から構成されています。

結果

{'input': '漫画の「火の鳥」には何編がある？',
 'output': '手塚治虫の漫画「火の鳥」は、以下の編から構成されています：\n\n1. 黎明編\n2. エジプト編\n3. ギリシャ編\n4. ローマ編\n5. 未来編\n6. ヤマト編\n7. 宇宙編\n8. 鳳凰編\n9. 復活編\n10. 羽衣編\n11. 望郷編\n12. 乱世編\n13. 生命編\n14. 異形編\n15. 太陽編\n\nこれらの編が連載され、それぞれ異なる時代や場所を舞台にして、火の鳥と人間の関わりを描いています。'}

LangSmithで見た拡張クラスの実行結果。これがRAGとしてLLMに渡っています。

参考までに非同期呼出の場合。

await agent_executor.ainvoke({"input": "漫画の「火の鳥」には何編がある？"})

弱点

今回の拡張で結果が良くなる場合もあるのですが、以下の弱点もあります。

１. 遅い

いちいち、ページ読み込みして、Embeddingして再検索するので遅いです。

2. 課金コスト

有償LLM使っているとEmbedding分のコストが追加でかかります(Vector Storeも有償ならそのコストも)。

3. 質が低下する場合あり

例えば、質問が「キン肉マンのレオパルドンは誰に負けた？」の場合は拡張しない方が回答は正しいです。
ただ、単純なTextSplitやSimilaritySearchをするのではなく、もう少しチューニングすれば質が向上する可能性は高いです。

拡張前後	結果
拡張前	キン肉マンのレオパルドンはマンモスマンに負けました。彼は『キン肉マン』シリーズで最短時間で倒されたキャラクターとして知られています。
拡張後	キン肉マンのキャラクター、レオパルドンはキン肉マンビッグボディチームの一員として登場しましたが、具体的に誰に負けたかの情報は検索結果からは明確には得られませんでした。ただし、彼が強い姿を見せたエピソードが挙げられています。詳細な戦歴についてはキン肉マンのマンガや関連資料を参照するのが良いでしょう。