はじめに　ー　週末研究ノートとは？

個人的に研究的な活動をやるにあたり、オープンにしてみたら面白いかもと思い、自分が興味を持っているざっくりテーマについて、これから、ゆるい週末研究を公開していこうと思います。（有識者の方のアドバイスも、ちょっとというかかなり期待してます！笑）

どこかの権威的な学会やジャーナルなどで発表する予定はないため、万が一、私の記事を利用する際には自己責任でお願いします。そんな人はいないと思いますが、念のため。

今回のサマリ（TL; DR）

Large Language Models as Tool Makers の Notebook の内容を、より実験をしやすいようにクラス化しました。

主に以下を対応しました。

tool_maker.py にて、LlmToolMaker クラスを作成
tool_user.py にて、LlmToolUser クラスを作成
eval.py にて、LlmToolEvaluator クラスを作成
- LlmToolEvaluator のスレッド処理は、元の Notebook のスレッド処理を正しく動作するように修正
- 特に、TPM　の Rate Limit　に引っかかり、Accuracy が落ちる問題に対応

コードは、こちら (llm_toolmaker ブランチです)

環境

CPU: Intel(R) Core(TM) i7-7700K CPU @ 4.20GHz
- cpu cores: 4
- processors: 8
Memory: 47GiB available
OS: Ubuntu 22.04.2 LTS
Docker:
- Docker version 24.0.2
- Docker Compose version v2.18.1
Python: 3.10.6
- openai: 0.27.7 （GPT-4 を使える状態にしておく必要があります）

今回の週末研究ノート

クラス化の概要

toolmaker.ipynb の内容を、tool_maker.py に移植しました
- LlmToolMaker::buildup_tool_from_datasets() メソッドは、Notebook と極力互換性を持つようにしました
- 一方で、LlmToolMaker::buildup_tool() メソッドは、データを使わずにプロンプトを指定できるようにしています
tooluser.ipynb の内容を、tool_user.py と eval.py に分割して移植しました
- LlmToolEvaluator::eval(), run() メソッドは、Notebook と極力互換性を持つようにしました
  - また、Thread 管理方法を変更しました
  - OpenAI API の Rate Limit を超えることでエラーにならないように、リトライ機能を追加しました
- LlmToolUser::make_answer_from_sample() メソッドが Notebook と極力互換性を持つようにしました
- 一方で、LlmToolUser::make_answer() メソッドは、データサンプルを使わずにプロンプトを指定できるようにしています

クラス化したコード

実際に作成したコードを紹介しておきます。

tool_maker.py
- LlmToolMaker
tool_user.py
- LlmToolUser
eval.py
- LlmToolEvaluator

一つひとつ紹介していきます。

tool_maker.py

LlmToolMaker

class LlmToolMaker(object):
    def __init__(
        self,
        task_name: str = "example_task",
        model_name: str = "gpt-4",
        temperature: float = 0.3,
        max_tokens: int = 2048,
        n_retry: int = 3,
    ) -> None:
        self.task_name: str = task_name
        self.model_name: str = model_name
        self.temperature: float = temperature
        self.max_tokens: int = max_tokens
        self.n_retry: int = n_retry

        self._llm_messages: list = []

        assert self.model_name[: len("gpt")] == "gpt"

    def _params(self):
        params = {
            "model": self.model_name,
            "max_tokens": self.max_tokens,
            "temperature": self.temperature,
            "messages": self._llm_messages,
        }
        return params

    def _llm(self, **params):
        response = openai.ChatCompletion.create(**params)["choices"][0]["message"][
            "content"
        ]
        return response

    def buildup_tool(
        self,
        toolmaker_prompt: str = tool_maker_prompt,
        tooltest_prompt: str = tool_test_prompt,
        toolwrapper_prompt: str = tool_wrapper_prompt,
    ) -> LlmTool:
        toolcode = self.make_toolcode(toolmaker_prompt)
        g_logger.info(f"{toolcode=}")

        testcode = self.make_testcode(
            toolcode=toolcode,
            tooltest_prompt=tooltest_prompt,
        )
        g_logger.info(f"{testcode=}")

        wrapper = self.make_wrappercode(toolwrapper_prompt=toolwrapper_prompt)
        self.dump()

        tool = LlmTool(toolcode=toolcode, testcode=testcode, wrapper=wrapper)
        return tool

    def buildup_tool_from_datasets(
        self,
        trainset: list,  # using for making tool code
        validset: list,  # using for making test code
        n_train_samples: int = None,
        n_valid_samples: int = None,
        toolmaker_prompt: str = tool_maker_prompt,
        tooltest_prompt: str = tool_test_prompt,
        toolwrapper_prompt: str = tool_wrapper_prompt,
    ) -> LlmTool:
        _toolmaker_prompt = (
            "\n\n".join(
                [
                    f"Question: {sample['question']}\nAnswer: {sample['answer']}"
                    for sample in trainset[:n_train_samples]
                ]
            )
            + "\n\n"
            + toolmaker_prompt
        )
        _tooltest_prompt = (
            "\n\n".join(
                [
                    f"Question: {sample['question']}\nAnswer: {sample['answer']}"
                    for sample in validset[:n_valid_samples]
                ]
            )
            + "\n\n"
            + tooltest_prompt
        )
        return self.buildup_tool(
            toolmaker_prompt=_toolmaker_prompt,
            tooltest_prompt=_tooltest_prompt,
            toolwrapper_prompt=toolwrapper_prompt,
        )

    def make_toolcode(self, toolmaker_prompt: str, max_tokens: int = 2048) -> str:
        self._llm_messages = [{"role": "user", "content": toolmaker_prompt}]
        params = self._params()  # using latest self._llm_messages
        for _ in range(self.n_retry):
            try:
                response: str = self._llm(**params)
                self._llm_messages.append({"role": "assistant", "content": response})
                toolcode: str = "\n\n".join(
                    re.findall(r"```python\n(.*?)```", response, re.DOTALL)
                )
                _ = exec(toolcode, globals())  # output: printed texts
                break
            except Exception as e:
                g_logger.warning("Failed to generate tool", e)
                self._llm_messages.append(
                    {
                        "role": "user",
                        "content": f"Failed to execute the function due to the error: {type(e).__name__} {e}. "
                        "Please fix it and try again.",
                    }
                )
        return toolcode

    def make_testcode(self, toolcode: str, tooltest_prompt: str) -> str:
        self._llm_messages.append({"role": "user", "content": tooltest_prompt})
        params = self._params()  # using latest self._llm_messages

        success = False
        for _ in range(self.n_retry):
            try:
                response = self._llm(**params)
                self._llm_messages.append({"role": "assistant", "content": response})
                testcode = "\n\n".join(
                    re.findall(r"```python\n(.*?)```", response, re.DOTALL)
                )
                unittest = toolcode + "\n" + testcode
                _ = exec(unittest, globals())  # output: printed texts
                success = True
                break
            except Exception as e:
                g_logger.warning("Failed to the simple tooltest", e)
                self._llm_messages.append(
                    {
                        "role": "user",
                        "content": f"Failed to verify the function due to the error: {type(e).__name__} {e}. "
                        "Please fix it and try again.",
                    }
                )
            if not success:
                raise Exception(
                    f"Failed to make tooltest code for the toolcode [{toolcode}], "
                    f"the last testcode: {testcode}"
                )
        return testcode

    def make_wrappercode(self, toolwrapper_prompt: str) -> None:
        self._llm_messages.append({"role": "user", "content": toolwrapper_prompt})
        params = self._params()  # using latest self._llm_messages
        try:
            wrapper = self._llm(**params)
            self._llm_messages.append({"role": "assistant", "content": wrapper})
            g_logger.info("Wrapper:", wrapper)
        except Exception as e:
            g_logger.error("Failed to generate wrapper", e)
            raise e
        return wrapper

    def dump(self, tooldir="./llm_tools") -> Self:
        pathlib.Path(tooldir).mkdir(parents=True, exist_ok=True)
        json_file = f"{tooldir}/{self.task_name}.json"
        with open(json_file, "w") as f:
            json.dump(self._llm_messages, f)
        return self

tool_user.py

LlmToolUser

class LlmToolUser(object):
    def __init__(
        self,
        wrapper: str,
        task_name: str = "example_task",
        model_name: str = "gpt-4",
        temperature: float = 0.3,
        max_tokens: int = 2048,
        n_retry: int = 3,
    ) -> None:
        self.wrapper: str = wrapper
        self.task_name: str = task_name
        self.model_name: str = model_name
        self.temperature: float = temperature
        self.max_tokens: int = max_tokens
        self.n_retry: int = n_retry

        assert self.model_name[: len("gpt")] == "gpt"

    def _params(self, messages: list[dict]):
        params = {
            "model": self.model_name,
            "max_tokens": self.max_tokens,
            "temperature": self.temperature,
            "messages": messages,
        }
        return params

    def _llm(self, **params):
        response = openai.ChatCompletion.create(**params)["choices"][0]["message"][
            "content"
        ]
        return response

    def generate(self, prompt):
        params = self._params(messages=[{"role": "user", "content": prompt}])
        for n_retry in range(self.n_retry):
            try:
                return self._llm(**params)
            except Exception as e:
                if "Rate limit" in " ".join(e.args):
                    sleep_seconds = 15 + 2**n_retry + random.random()
                    errmsg = re.sub(
                        r"org-\w+", "org-" + ("x" * 24), f"{e}"
                    )  # masking for secure
                    g_logger.warning(f"{errmsg} ... try to retry [{sleep_seconds=}]")
                    time.sleep(sleep_seconds)
                else:
                    g_logger.warning(f"{e} ... try to retry")
        raise Exception("Failed to generate")

    def make_answer_from_sample(self, task: str, sample: dict):
        prompt = self.wrapper + "\n\nQuestion: " + sample["question"]
        ans = self.make_answer(prompt=prompt)

        return ans

    def make_answer(self, prompt: str):
        caller = self.generate(prompt)
        func_call = pickup_func(caller)
        func_def = pickup_func(self.wrapper)

        exec_code = func_def + "\n" + func_call
        _ = exec(exec_code, globals())  # output: printed texts
        answer_variable = re.findall(r"(ans.*?) =", func_call, re.DOTALL)[-1]
        ans = globals()[answer_variable]

        return ans

eval.py

LlmToolEvaluator

class LlmToolEvaluator(object):
    def __init__(
        self,
        wrapper: str,
        task_name: str = "example_task",
        model_name: str = "gpt-3.5-turbo",
        max_threads: int = 8,
    ) -> None:
        self.wrapper: str = wrapper
        self.task_name: str = task_name
        self.model_name: str = model_name

        self.max_threads: int = max_threads
        self.pool = BoundedSemaphore(self.max_threads)
        self.lock = Lock()

        self.tool_user = LlmToolUser(
            wrapper=wrapper, model_name=model_name, task_name=task_name
        )

        self.n_totals: int = 0
        self.n_corrects: int = 0

    def _adjust(self, ans: str, sample):
        if is_option_selection(ans, sample):
            options = (
                re.findall(r"Options:(.*)", sample["question"], re.DOTALL)[0]
                .strip()
                .split("\n")
            )
            for option in options:
                if ans in option:
                    ans = option.split(" ")[0]
                    break

        if self.task_name == "schedule_meeting":
            if ans is None:
                ans = "No time slot works."
            elif isinstance(ans, list) or isinstance(ans, tuple):
                ans = f"{ans[0]} - {ans[1]}"

        ans = get_option(ans)
        return ans

    def run(self, sample: dict):
        with self.pool:
            try:
                ans = self.tool_user.make_answer_from_sample(
                    task=self.task_name, sample=sample
                )
                ans = self._adjust(ans, sample)
            except Exception as e:
                ans = f"Error: {e}"
            with self.lock:
                self.n_totals += 1
                if str(ans) == str(sample["answer"]):
                    self.n_corrects += 1
                else:
                    g_logger.info(f"incorrect: {ans=} / {sample['answer']=}")
                acc = self.n_corrects / self.n_totals
                g_logger.info(f"Thread Accuracy: {acc:.4f}")

    def eval(self, testset: list) -> Self:
        threads = []
        for sample in tqdm(testset, desc="creating threads"):
            thr = Thread(target=self.run, args=(sample,))
            threads.append(thr)
            thr.start()
            # self.run(sample)  # for debugging
            # break

        thr_bar = tqdm(threads, desc="waiting threads: ")
        for thr in thr_bar:
            thr.join()

        acc = self.n_corrects / self.n_totals
        g_logger.info(f"Last Accuracy: {acc:.4f}")
        return self

再現性テスト

オリジナル LATM github の Notebook のように、bbh※ のタスクを実行するには、以下のようにします。

※ bbh については元サイト google/BIG-bench を参照ください

tool_maker の実行例

cd backend
python -m app.llm_toolmaker.tool_maker --task-name=word_sorting

bbh の word_sorting タスクに対して、ツールコードを作成し、その後テストコードを作成し、結果を llm_tools ディレクトリに保存します。

tool_user の実行例

python -m app.llm_toolmaker.tool_user --task-name=word_sorting

bbh の word_sorting タスクについて、テストデータを使って精度評価をします。（Notebook との違いは、あまりAPI を使わないようにテストデータ数を５０個に制限しました。）

ログの確認

上記のようにスクリプトを実行すると backend/log/app.log にログが出力されます。
今回は、ログを抜粋して内容を確認してみます。

too_maker のログ（抜粋）

2023/06/27 23:54:17.231 app INFO params.task_name='word_sorting'
2023/06/27 23:54:33.126 app INFO toolcode='def sort_words_alphabetically(word_list):\n    return sorted(word_list)\n\n# Example usage:\nwords = ["syndrome", "therefrom", "thrill", "splutter", "panicking", "scorch", "same", "dot", "prod", "obstetric", "malton", "onus", "drumhead", "delmarva", "barn", "embezzle", "it&t", "damp", "guru", "subsist", "entirety", "greene"]\nsorted_words = sort_words_alphabetically(words)\nprint(sorted_words)\n'
2023/06/27 23:55:38.968 app INFO testcode='def test_sort_words_alphabetically():\n    # Test 1\n    words1 = ["conference", "apparition", "ignore", "dutton", "layperson", "coupe", "superstitious", "westward", "turnoff", "messenger", "copra", "floruit", "primitive", "implement"]\n    ret1 = sort_words_alphabetically(words1)\n    ans1 = " ".join(ret1)\n    assert ans1 == "apparition conference copra coupe dutton floruit ignore implement layperson messenger primitive superstitious turnoff westward"\n\n    # Test 2\n    words2 = ["covalent", "spiderwort", "horowitz", "divisive", "spiritual", "cheshire", "affluent", "gideon", "quadrature", "julio", "peanut", "epsilon", "diagnostician", "grover", "folklore", "gothic", "salient"]\n    ret2 = sort_words_alphabetically(words2)\n    ans2 = " ".join(ret2)\n    assert ans2 == "affluent cheshire covalent diagnostician divisive epsilon folklore gideon gothic grover horowitz julio peanut quadrature salient spiderwort spiritual"\n\n    # Test 3\n    words3 = ["euclidean", "stonehenge", "hobby", "cloudy", "winsome", "invite", "thrifty", "fight", "majestic", "citrus", "surge", "scene"]\n    ret3 = sort_words_alphabetically(words3)\n    ans3 = " ".join(ret3)\n    assert ans3 == "citrus cloudy euclidean fight hobby invite majestic scene stonehenge surge thrifty winsome"\n\n    # Test 4\n    words4 = ["thunderclap", "swab", "built", "poland"]\n    ret4 = sort_words_alphabetically(words4)\n    ans4 = " ".join(ret4)\n    assert ans4 == "built poland swab thunderclap"\n\n    # Test 5\n    words5 = ["regret", "starlight", "wallboard", "cotyledon", "more", "pepperoni"]\n    ret5 = sort_words_alphabetically(words5)\n    ans5 = " ".join(ret5)\n    assert ans5 == "cotyledon more pepperoni regret starlight wallboard"\n\n# Run the tests\ntest_sort_words_alphabetically()\n'
2023/06/27 23:57:02.722 app INFO Wrapper: Here is a function to solve a class of problems:
:
:
:

実際は最後まで見ましたが、特に、エラーなく終わったようです。

tool_user のログ（抜粋）

:
:
:
2023/06/28 00:25:33.458 app INFO Thread Accuracy: 1.0000
2023/06/28 00:25:34.189 app INFO Thread Accuracy: 1.0000
2023/06/28 00:25:34.416 app WARNING Rate limit reached for default-gpt-3.5-turbo in organization org-xxxxxxxxxxxxxxxxxxxxxxxx on tokens per min. Limit: 90000 / min. Current: 88027 / min. Contact us through our help center at help.openai.com if you continue to have issues. ... try to retry [sleep_seconds=16.69230213424598]
2023/06/28 00:25:35.774 app INFO Thread Accuracy: 1.0000
2023/06/28 00:25:36.183 app INFO incorrect: ans='Error: list index out of range' / sample['answer']='aeneas colombo foothold fox garry glycerine inviolate lucre magnanimity nevada notoriety plebiscite pompey quagmire satanic scription softball spleenwort tennyson type'
2023/06/28 00:25:36.183 app INFO Thread Accuracy: 0.9773
2023/06/28 00:25:37.105 app INFO Thread Accuracy: 0.9778
2023/06/28 00:25:38.231 app INFO Thread Accuracy: 0.9783
2023/06/28 00:25:52.362 app WARNING The server is overloaded or not ready yet. ... try to retry
2023/06/28 00:25:54.820 app INFO Thread Accuracy: 0.9787
2023/06/28 00:25:55.331 app WARNING The server is overloaded or not ready yet. ... try to retry
2023/06/28 00:25:56.971 app INFO Thread Accuracy: 0.9792
2023/06/28 00:25:58.097 app INFO Thread Accuracy: 0.9796
2023/06/28 00:30:29.456 app WARNING Bad gateway. {"error":{"code":502,"message":"Bad gateway.","param":null,"type":"cf_bad_gateway"}} 502 {'error': {'code': 502, 'message': 'Bad gateway.', 'param': None, 'type': 'cf_bad_gateway'}} {'Date': 'Tue, 27 Jun 2023 15:30:29 GMT', 'Content-Type': 'application/json', 'Content-Length': '84', 'Connection': 'keep-alive', 'X-Frame-Options': 'SAMEORIGIN', 'Referrer-Policy': 'same-origin', 'Cache-Control': 'private, max-age=0, no-store, no-cache, must-revalidate, post-check=0, pre-check=0', 'Expires': 'Thu, 01 Jan 1970 00:00:01 GMT', 'Server': 'cloudflare', 'CF-RAY': '7ddeb39f4b3de005-NRT', 'alt-svc': 'h3=":443"; ma=86400'} ... try to retry
2023/06/28 00:30:32.330 app INFO Thread Accuracy: 0.9800
2023/06/28 00:30:32.333 app INFO Last Accuracy: 0.9800

今回は、途中で 1個だけ list index out of range でエラーになったことにより、Accuracy が 0.98 になりました。

一度だけ、tool_maker のテストコードをパスしたのに、tool_user の評価で全滅というケースがありました。実は、bbh のタスクでは、すべてツールの出力が文字列（ハッシュ化可能なもの）を前提としていますが、テストコードでは、結果が一致していればOKなので、出力を list で比較するようにテストコードが生成されるとこのような全滅ケースになるようです。なので、実用的な別タスクに利用する際には、ツールの出力の型を指定したり、テストコードで出力の型もチェックするようなプロンプトを作るのが良さそうとわかりました。

まとめ

今回は、LATM github の Notebook の内容を部品化・クラス化して別の実験をしやすいように準備しました
作成したスクリプトは、主に以下２つです
- tool_maker.py
- tool_user.py （ただし、評価用のテストデータ数を50に制限した）
今回追加した他の２ファイルは、以下の通りです
- bbh.py
  - bbh の再現性を確認するために用意したファイルです
  - 他のタスクで試す際には使わなくてOKな位置づけです
- prompt.py
  - 他のタスクでプロンプトエンジニアリングできるよう(prompt をカスタマイズできるよう)にしました
次は、今回作成したクラスを使って、（多少）実用的なタスク実行について何かしらの実験をする予定です
- ツール出力の型をプロンプトで指定したり、テストコードで型チェックするようにプロンプトをカスタマイズすると良さそうです
- 出力制御を考えるとまさにメタ言語を設計していくイメージですねー
【参考】課金額
- 10分〜15分後ぐらいに Usage に反映された金額で確認
- タスク： word_sorting で確認
  - tool-maker: 約$0.15
  - tool-user: 約$0.08

参考文献