🥬

Assistant APIでConversation Agent作る

2023/11/08に公開

Python

OpenAI

tech

モチベーション

TRPGで動かすようなロングタームなAgentの記憶をどうしようか考えていたタイミングでAssistant APIが登場し、無限に長いスレッドの面倒を見てくれるので試してみました。

また、サンプルはtoolとの連携はあるものの、複数会話でのThreadとMessageの紐付きがよく分からなかったので実際に動かして確認してみたかったのもやってみた理由です。
なお、作成したのはツールを使わないシンプルな対話モデルになります。

A key change introduced by this API is persistent and infinitely long threads, which allow developers to hand off thread state management to OpenAI and work around context window constraints. With the Assistants API, you simply add each new message to an existing thread.
このAPIによって導入された重要な変更点は、持続的で無限に長いスレッドです。これにより、開発者はスレッドの状態管理をOpenAIに委ねることができ、コンテキスト・ウィンドウの制約を回避することができます。Assistants APIでは、新しいメッセージを既存のスレッドに追加するだけです。

実装

from openai import OpenAI


class Agent:
    def __init__(self, api_key: str, system_prompt: str, character_setting: str) -> None:
        self.client = OpenAI(api_key=api_key)
	# アシスタントの設定（ここでinstructionsを渡せば、以降は無視で良い）
        self.assistant = self.client.beta.assistants.create(
            name="Sample Agent",
            description=system_prompt,
            instructions=system_prompt,
            model="gpt-4-1106-preview",
        )
        # スレッド作成(以降のmessageは自動で紐づけられる)
        self.thread = self.client.beta.threads.create()
        # 初期メッセージ(キャラクリ)
        self.client.beta.threads.messages.create(
            thread_id=self.thread.id,
            role="user",
            content=character_setting,
        )
        # 初期化
        self.client.beta.threads.runs.create(
            thread_id=self.thread.id,
            assistant_id=self.assistant.id,
        )
        # run終了前に次のrunを開始するとエラーが出るので、一定時間待機
        self._wait()

    def action(self, user_message: str) -> str:
        """ユーザーのメッセージに対して、返答を生成する"""
	# スレッドIDを通して、コンテキストにメッセージを追加している（キャラクリ同様、これで完結する）
        self.client.beta.threads.messages.create(
            thread_id=self.thread.id,
            role="user",
            content=user_message,
        )
        self.client.beta.threads.runs.create(
            thread_id=self.thread.id,
            assistant_id=self.assistant.id,
        )
        # 返答を見たいので一定時間待機
        self._wait()
        self.show_message(show_num=1)

    def _wait(self) -> None:
        status = "queued"
        while status != "completed":
            time.sleep(0.5)
            run = self.client.beta.threads.runs.list(
                thread_id=self.thread.id,
            )
            for r in run:
                # 最新だけ見れれば良い
                status = r.status
                break

    def show_message(self, show_num: int = 5) -> None:
        """スレッドのメッセージを表示する"""
        messages = self.client.beta.threads.messages.list(
        thread_id=self.thread.id,
        )
        messages_list = [message for message in messages]
        for message in (messages_list[:show_num]):
            print(message.role, ":", message.content[0].text.value)

  
if __name__ == "__main__":
    agent = Agent(
	OPENAI_API_KEY,
	SYSTEM_PROMPT,
	CHARACTER_SETTING,
    )
    agent.action(f"{user_message}")

感想

こんな感じで会話ログが取れました。メモリーを使った会話の実装がこれだけで済むのは便利ですね。
これまでの実装経験的に複数回対話はuser_messageを何かのリストに追加すると思っていたのですが、thread.idベースでOpenAIがマネージドしてくれるので何も書くことないですね。
API Referenceやライブラリのコード上にmessageの受け渡しインターフェースがなくて、ちょっと使い方に迷いました。

余談ですが、最新のopenaiライブラリはclientクラスができて、モジュールを引き回さなくてよくなったので、DI出来るようになり、とてもありがたいです。

どこまでのコンテキスト長に対応するかはわかりませんが、LangchainのCombinedMemory[VectorStoreRetrieverMemory, ConversationSummaryBufferMemory]みたいな組み合わせよりも簡単で効果的な感じは見れるので、色々試していこうと思います。

モチベーション

実装

感想

Discussion