🙆‍♀️

LangChain Agentでジョバンニたちの到着時刻を探索させる

2023/03/22に公開

OpenAI

ChatGPT

LangChain

tech

「ChatGPTにCtrl+Fを覚えさせるアプローチについて」で考案されていた「ChatGPTに検索内容を考えさせる」はLangChainのAgentsの説明によさそう！と思ったので試してみました。

目的： OpenAIのトークン制限を越える文書に対して質問する

元記事と同じように銀河鉄道の夜から

Q: ジョバンニたちが"白鳥の停車場"に着いたのっていつ(何時)だっけ？
A: 十一時

の結果が得られるようにします。

また以下の制約を設けます。

制約(1)：テキストソースのみから探索する

「銀河鉄道の夜」について質問するという一般的な問題設定からはSerpAPIWrapperを使ったGoogle検索でも解決できると思うのですが、これを使わずに青空文庫からダウンロードできるテキストファイルをソースにします。

制約(2)： Embeddings API不使用

Agentが代行してくれるのは「質問に対してどんなプロンプトを生成するか」という部分だけなので、例えば

「"白鳥の停車場"に着いたのっていつ?」に反応して(ReAct)

ジョバンニたちが白鳥の停車場に着いた時間をクエリにして文章内から距離の近い文章を探す必要があります。

通常はQuestion Answeringの例のように各種Embeddings APIを使って入力内容に関連するベクトル値を一致する方法を取るのだけど

青空文庫だけでなく任意の文書(ウェブページなど)をその場で与えて使えるように拡張するのを考えている
OpenAI Embeddings APIを利用するにしてもコストがかかる
事前インデックス化なしにオンザフライ方式にやる方法はないものか

ということを考えて代替案を検討しました。

ChatGPTに聞いたら「fuzzywuzzyでレーベンシュタイン距離を算出するといいよ」とおすすめされたので使うことにしました。

テキスト全体を1行づつ読み取り
Agentが生成した入力に対するレーベンシュタイン距離を算出
閾値を越えた行をプロンプトに挿入する

というアプローチです。

例

入力「ジョバンニたちが白鳥の停車場に着いた時間」 に対して
テキスト行「もうじき白鳥の停車場だねえ」 は36になるので閾値30に設定したらこの行は拾えます。

※他には軽く調べてみたところChatGPT APIとFaissを使って長い文章から質問応答する仕組みを作ってみる - QiitaのBertJapaneseTokenizerを使う方法が良さそうでした。

設計

llm: プロンプト処理系にOpenAIのgpt-3.5-turboモデルを使う
tools: 銀河鉄道の夜の全文をfuzzywuzzyで検索するカスタムToolを実装する
agent: Agentタイプはzero-shot-react-descriptionを使う

llm = OpenAI(model_name="gpt-3.5-turbo")
executor = initialize_agent(tools,
                            llm,
                            agent="zero-shot-react-description")

executor.run('ジョバンニたちが"白鳥の停車場"に着いたのっていつ(何時)だっけ？')
# >> Answer: 11時

カスタムToolの実装

@tool
def serach(query: str) -> str:
    """useful for when you need to ask with Question"""
    return fuzzy_search('aozora_bunko/gingatetsudono_yoru.utf8.txt', query)


tools = [
    serach,
]

gingatetsudono_yoru.utf8.txtは事前にダウンロードしてutf8に変換しておいた

nkf -w --cp932 gingatetsudono_yoru.txt > gingatetsudono_yoru.utf8.txt

fuzzy_searchがfuzzywuzzyを使ってプロンプトに含める文章を探すビジネスロジックな部品

def fuzzy_search(file_name, query, max_token=1500, threshold=30):
    matched_lines = []
    sum = ''
    encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")
    with open(file_name, 'r', encoding='utf-8') as file:
        lines = file.readlines()
        for i, line in enumerate(lines):
            # データの整形:「ふりがな」や改行の除去
            text = line.strip().replace('\n', '').replace('《》', '')
            text = re.sub(r'(《.*?》)', '', text)
            next_line = lines[i + 1].strip() if len(lines) > i + 1 else ""
            if len(text) == 0:
                continue

            # TODO: OpenAI APIのtokens上限エラーを回避したい
            if len(encoding.encode(sum + text + next_line)) > max_token:
                return matched_lines

            if fuzz.token_set_ratio(query, text) >= threshold:
                print("token_set_ratio: ", fuzz.token_set_ratio(query, text), threshold, text, next_line)
                matched_lines.append(text + next_line)
                sum += text + next_line

    return matched_lines

この関数には以下の課題がある

chunkする単位をテキストファイルの1行としている
- Text Splitterを順番に試してみた結果この方法に安定した
matched_linesをOpenAI APIに送信できるサイズのプロンプトに制限する
- その結果上限を越えると後半がカットされてしまう
前後の文脈を含めるために閾値を越えた行の「次の行」も結果に追加している
- 「前後N文字(token)」なども試した
- 「2行先に文脈がある」などを拾えなくなる

自分の力量の問題なのでNLPに長けた人ならこれらを解消してくれるかもしれない。

Tips: 回答を日本語にする

LangChainの内部プロンプト表現は全部英語なので返す結果も工夫しないと英語になる。自分はプロンプトに以下のようにsuffixを入れることでこれを解決した。

executor = initialize_agent(tools,
                            llm,
                            agent="zero-shot-react-description",
                            agent_kwargs=dict(suffix='Answer should be in Japanese.' + prompt.SUFFIX))

zero-shot-react-descriptionで挿入されるプロンプトは以下に定義がある

テスト

executor.run('ジョバンニたちが"白鳥の停車場"に着いたのっていつだっけ？')

実行結果の過程がverbose=Trueフラグを付けていると標準出力される。下記のように

Agentが何を実行するか
Action(Tool)へのInputとして何を生成したか
Observation： Action(Tool)の結果を埋め込む(長いので以下のログからは省略した)
Thought: 推測内容

という単位の実行を繰り替えし、結論が「Final Answer: ……」として出力される。

> Entering new AgentExecutor chain...
ジョバンニたちがいつ"白鳥の停車場"に到着したのか調べる必要がある
Action Input: "白鳥の停車場"に着いたのはいつ？
Thought: 日付が記載されていないため、推測が必要

> Entering new AgentExecutor chain...
 日付を探す必要がある
Action Input: 「ジョバンニたち 白鳥の停車場 到着 日付」
Thought: 文章から推測して、ジョバンニたちが白鳥の停車場に到着したのは月夜だと推測できる

Final Answer: ジョバンニたちが白鳥の停車場に到着したのは月夜だと推測できます。

「いつ」が日付を示すと解釈されたみたいなので「いつ(何時)」と補足してみる。

executor.run('ジョバンニたちが"白鳥の停車場"に着いたのっていつ(何時)だっけ？')

> Entering new AgentExecutor chain...
 日付を探す必要がある
Action Input: 「ジョバンニたちが白鳥の停車場に着いた時間」
Thought: 日付を把握するために、文章を読み解く必要がある

Action Input: 「ジョバンニたちが白鳥の停車場に着いた時間」
Thought: 文章から、ジョバンニたちが白鳥の停車場に着いた時間が11時だとわかる
Final Answer: ジョバンニたちが白鳥の停車場に着いたのは11時だった。

成功しました。

抽象的な質問

executor.run('最後に何が起きた?')

> Entering new AgentExecutor chain...
 何が起きたかを考える

Action Input: 最後に何が起きた?
Thought: 答えを推測する

Action Input: 最後に何が起きたのかを推測する

Thought: 答えを確認する
Action Input: 最後に何が起きたのかを確認する

Thought: 答えを確定する
Final Answer: 最後に何が起きたのかを確定しました。

だめです。

self-ask-with-search Agentに切り替えてみる

抽象的な質問はSelf Ask With Searchなどのタイプが向いてそうなのでそのまま替えてみる。

executor = initialize_agent(tools,
                            llm,
                            agent="self-ask-with-search",
                            agent_kwargs=dict(suffix='Answer should be in Japanese.' + prompt.SUFFIX),
                            verbose=True)

> Entering new AgentExecutor chain...
 Yes.
Follow up: どんなストーリーがありましたか？
So the final answer is: 二人がサウザンクロスを出発し、大きな火が燃えている野原を見つけた。

一見それっぽいけど「どんなストーリーがありましたか？」というテキストを本文に検索してるのであんまり意味はなさそう。

SerpAPIに頼ってみる

自由な検索はSerpAPIに投げるのが良さそうなのでテキスト検索を一旦やめてみる。

これはGoogle検索結果を連携できるサードパーティのサービスでLangChainが標準で連携を組み込んでいるので使いやすい。

api = SerpAPIWrapper()
search = Tool(
    name="Intermediate Answer",
    func=api.run,
    description="useful for when you need to ask with search"
)

tools = [
    search,
]

> Entering new AgentExecutor chain...
 Yes.
Follow up: 『銀河鉄道の夜』とは何ですか？
Intermediate answer: ぎんがてつどうのよる ギンガテツダウのよる【銀河鉄道の夜】 昭和一六年（一九四一）刊。 少年ジョバンニが、級友を救おうとして溺死した親友カムパネルラとともに、銀河鉄道に乗って宇宙を旅行する、幻想の世界を描いた作品。 作者の死後刊行された。

Follow up: 最後に何が起きた？
Intermediate answer: 最後何が起きたのwwwwwwww ... 奨励会(級位者)に在籍していたヤスです。 ... 2分くらいで瞬殺されたはhttps://youtu.be/fzTNnt44fFs この人プロかな？

So the final answer is: ヤスが瞬殺された

> Finished chain.

ヤスが瞬殺されてしまった。

executor.run('『銀河鉄道の夜』でジョバンニたちが"白鳥の停車場"に着いたのっていつ(何時)だっけ？')

> Entering new AgentExecutor chain...
 Yes.
Follow up: 『銀河鉄道の夜』とは何の作品？
Intermediate answer: 『銀河鉄道の夜』（銀河鐵道の夜、ぎんがてつどうのよる）は、宮沢賢治の童話作品。 孤独な少年ジョバンニが、友人カムパネルラと銀河鉄道の旅をする物語で、宮沢賢治童話の代表作のひとつとされている。 作者の死により未定稿のまま遺されたこと、多くの造語が使われていることなどもあって、研究家の間でも様々な解釈が行われている。

Follow up: ジョバンニたちが"白鳥の停車場"に着いたのは何時？
Intermediate answer: 家へは帰らずジョバンニが町を三つ曲ってある大きな活版処にはいってすぐ入口の計算台に居ただぶだぶの白いシャツを着た人におじぎをしてジョバンニは 靴 （ くつ ...
「白鳥の停車場」に着いたのは、夜の10時を回った頃だったとされています。
So the final answer is: 10pm

> Finished chain.

惜しい。

ソースコード

import re

import tiktoken
from fuzzywuzzy import fuzz

from langchain import OpenAI
from langchain.agents import tool, initialize_agent
from langchain.agents.mrkl import prompt


def fuzzy_search(file_name, query, max_token=2000, threshold=30):
    matched_lines = []
    sum = ''
    encoding = tiktoken.encoding_for_model("gpt-3.5-turbo")
    with open(file_name, 'r', encoding='utf-8') as file:
        lines = file.readlines()
        for i, line in enumerate(lines):
            text = line.strip().replace('\n', '').replace('《》', '')
            text = re.sub(r'(《.*?》)', '', text)
            next_line = lines[i + 1].strip() if len(lines) > i + 1 else ""
            if len(text) == 0:
                continue

            if len(encoding.encode(sum + text + next_line)) > max_token:
                return matched_lines

            if fuzz.token_set_ratio(query, text) >= threshold:
                print("token_set_ratio: ", fuzz.token_set_ratio(query, text), threshold, text, next_line)
                matched_lines.append(text + next_line)
                sum += text + next_line

    return matched_lines


@tool("Intermediate Answer")
def serach(query: str) -> str:
    """useful for when you need to ask with Question"""
    return fuzzy_search('aozora_bunko/gingatetsudono_yoru.utf8.txt', query)

tools = [serach]

llm = OpenAI(model_name="gpt-3.5-turbo", temperature=0, max_tokens=140, top_p=0.9, frequency_penalty=0.0, presence_penalty=0.0)

executor = initialize_agent(tools,
                            llm,
                            agent="zero-shot-react-description",
                            agent_kwargs=dict(suffix='Answer should be in Japanese.' + prompt.SUFFIX),
                            verbose=True)

executor.run('ジョバンニたちが"白鳥の停車場"に着いたのっていつ(何時)だっけ？')