これは、Python SDK を使用した OpenAI の Assistant API によるエージェントの構築に関する複数部構成のシリーズの最初の部分です。

エージェントとは何ですか?

私の見方では、エージェントは実際には LLM (大規模言語モデル) を活用し、人間の動作を模倣しようとする単なるソフトウェアです。つまり、言語を会話して理解できるだけでなく、現実世界に影響を与える行動も実行できるということです。これらのアクションは通常、ツールと呼ばれます。

このブログ投稿では、Python SDK を使用して OpenAI の Assistant API を使用してエージェントを構築する方法を検討します。パート 1 はアシスタントの骨組みだけになります。つまり会話部分だけです。

私はフレームワークに依存しないように意図的に CLI アプリを構築することにしました。 2 つを簡単に区別できるように、実装を意図的にエージェントと呼び、OpenAI SDK 実装をアシスタントと呼びます。

エージェントが呼び出すことができる機能に関しては、ツールと機能という用語を同じ意味で使用します。パート 2 では、関数呼び出しについて詳しく説明します。

前提条件

このチュートリアルを進めるには、次のものが必要です。

• マシンにインストールされている Python3
• OpenAI API キー
• Python プログラミングの基礎知識

OpenAI アシスタントの概念

アシスタント: アシスタント API のアシスタントは、ユーザー メッセージに応答するように構成されたエンティティです。命令、選択したモデル、ツールを使用して機能を操作し、答えを提供します。

スレッド: スレッドは、アシスタント API の会話または対話を表します。これはユーザー インタラクションごとに作成され、複数のメッセージを含めることができ、進行中の会話のコンテナとして機能します。

メッセージ: メッセージは、スレッド内の通信の単位です。これにはテキスト (および将来的にはファイル) が含まれており、スレッド内でユーザーのクエリやアシスタントの応答を伝えるために使用されます。

Run : Run は、スレッドを処理するアシスタントのインスタンスです。これには、スレッドを読み取り、ツールを呼び出すかどうかを決定し、スレッドのメッセージのモデルの解釈に基づいて応答を生成することが含まれます。

開発環境のセットアップ

最初のステップは、 venvを使用して仮想環境を作成し、アクティブ化することです。これにより、依存関係がシステムの Python インストールから確実に分離されます。

 python3 -m venv venv source venv/bin/activate

唯一の依存関係であるopenaiパッケージをインストールしましょう。

 pip install openai

main.pyファイルを作成します。 CLI アプリの基本的なランタイム ロジックをいくつか入力してみましょう。

 while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the assistant...") break print(f"Assistant: You said {user_input}")

python3 main.pyを実行して試してください。

 python3 main.py User: hi Assistant: You said hi

ご覧のとおり、CLI はユーザー メッセージを入力として受け入れますが、天才アシスタントにはまだ頭脳 🧠 が備わっていないため、メッセージをすぐに繰り返すだけです。まだそれほど賢くはありません。

エージェント

さあ、楽しい😁（または頭痛🤕）が始まります。最終クラスに必要なインポートをすべてここで提供します。簡潔にするためにコード サンプルからインポートを省いたため、何がどこから来ているかについて頭を悩ませる必要はありません。新しいファイルagent.pyにAgentクラスを構築することから始めましょう。

 import time import openai from openai.types.beta.threads.run import Run class Agent: def __init__(self, name: str, personality: str): self.name = name self.personality = personality self.client = openai.OpenAI(api_key="sk-*****") self.assistant = self.client.beta.assistants.create( name=self.name, model="gpt-4-turbo-preview" )

クラス コンストラクターでは、OpenAI API キーを渡すことによって、OpenAI クライアントをクラス プロパティとして初期化します。次に、新しく作成したアシスタントにマップするassistantクラス プロパティを作成します。後で使用できるように、 nameとpersonalityクラス プロパティとして保存します。

create メソッドに渡しているname引数は、OpenAI ダッシュボードでアシスタントを識別するためだけのもので、この時点では AI は実際にはそれを認識していません。実際には、後で説明するinstructionsに名前を渡す必要があります。

アシスタントの作成時にすでにinstructionsを設定することもできますが、実際にはアシスタントの動的な変更に対する柔軟性が低くなります。

client.beta.assistants.updateを呼び出すことでアシスタントを更新できますが、Runs に到達したときに表示される動的な値を渡すためのより良い場所があります。

Run の作成時に何度もinstructionsを渡すと、アシスタントのinstructions Run のinstructionsによって上書きされることに注意してください。これらは互いに補完するものではないため、ニーズに基づいていずれかを選択してください。静的命令の場合はアシスタント レベル、動的命令の場合は実行レベルです。

このシリーズのパート 2 で関数呼び出しを追加できるように、モデルとしてgpt-4-turbo-previewモデルを選択しました。ツールを実装する際に、将来純粋にイライラして偏頭痛が起きる一方で、ほんの数ペニーを節約したい場合は、 gpt-3.5-turboを使用できます。

GPT 3.5 はツールの呼び出しが苦手です。この問題に対処するために費やした時間のおかげで、そう言えます。 😝 それについてはこのくらいにして、これについては後で詳しく説明します。

スレッドの作成、メッセージの追加、最後のメッセージの取得

エージェントを作成したら、会話スレッドを開始する必要があります。

 class Agent: # ... (rest of code) def create_thread(self): self.thread = self.client.beta.threads.create()

そして、そのスレッドにメッセージを追加する方法が必要になります。

 class Agent: # ... (rest of code) def add_message(self, message): self.client.beta.threads.messages.create( thread_id=self.thread.id, role="user", content=message )

現時点では、ロールuserを持つメッセージのみを追加できることに注意してください。これはかなり制限的であるため、OpenAI は将来のリリースでこれを変更する予定だと思います。

これで、スレッド内の最後のメッセージを取得できます。

 class Agent: # ... (rest of code) def get_last_message(self): return self.client.beta.threads.messages.list( thread_id=self.thread.id ).data[0].content[0].text.value

次に、エントリ ポイントのrun_agentメソッドを作成して、これまでの内容をテストします。現在、 run_agentメソッドはスレッド内の最後のメッセージを返すだけです。実際には Run は実行されません。相変わらず無脳ですね。

 class Agent: # ... (rest of code) def run_agent(self): message = self.get_last_message() return message

main.pyに戻り、エージェントと最初のスレッドを作成します。スレッドにメッセージを追加します。次に、同じメッセージをユーザーに返しますが、今回はライブ スレッドからのものです。

 from agent import Agent agent = Agent(name="Bilbo Baggins", personality="You are the accomplished and renowned adventurer from The Hobbit. You act like you are a bit of a homebody, but you are always up for an adventure. You worry a bit too much about breakfast.") agent.create_thread() while True: user_input = input("User: ") if user_input.lower() == 'exit': print("Exiting the agent...") break agent.add_message(user_input) answer = agent.run_agent() print(f"Assistant: {answer}")

実行してみましょう:

 python3 main.py User: hi Assistant: hi

まだあまり賢くありません。ホビットよりもオウム🦜に近い。次のセクションから、本当のお楽しみが始まります。

実行の作成とポーリング

実行を作成するときは、定期的にRunオブジェクトを取得して実行のステータスを確認する必要があります。これはポーリングと呼ばれますが、これは最悪です。エージェントが次に何をすべきかを判断するには、ポーリングする必要があります。 OpenAI は、これを簡単にするためにストリーミングのサポートを追加する予定です。それまでの間、次のセクションでポーリングを設定する方法を説明します。

次のメソッド名の_に注意してください。これは、メソッドが内部使用を目的としており、外部コードから直接アクセスすべきでないことを示す Python の標準です。

まず、 Runを作成するためのヘルパー メソッド_create_runを作成し、このメソッドを呼び出すようにrun_agentを更新しましょう。

 class Agent: # ... (rest of code) def get_breakfast_count_from_db(self): return 1 def _create_run(self): count = self.get_breakfast_count_from_db() return self.client.beta.threads.runs.create( thread_id=self.thread.id, assistant_id=self.assistant.id, instructions=f""" Your name is: {self.name} Your personality is: {self.personality} Metadata related to this conversation: {{ "breakfast_count": {count} }} """, ) def run_agent(self): run = self._create_run() # add this line message = self.get_last_message() return message

thread.idとassistant.id渡して実行を作成する方法に注目してください。

冒頭で、動的な命令とデータを渡すのに適した場所があると述べたことを覚えていますか?これは、Run を作成するときのinstructionsパラメーターになります。私たちの場合、朝食のcountデータベースから取得できます。これにより、応答をトリガーするたびに、さまざまな関連する動的データを簡単に渡すことができます。

これで、エージェントは周囲の世界の変化を認識し、それに応じて行動できるようになります。私は、関連する動的コンテキストを保持するメタデータ JSON オブジェクトを指示に含めることを好みます。これにより、冗長性が低くなり、LLM がよく理解できる形式でデータを渡すことができます。

これはまだ実行しないでください。最後のメッセージを取得するときに実行が完了するのを待っていないため、これは機能しません。そのため、それが最後のユーザー メッセージのままになります。

ポーリング メカニズムを構築することでこれを解決しましょう。まず、実行を繰り返し簡単に取得する方法が必要なので、 _retrieve_runメソッドを追加しましょう。

 class Agent: # ... (rest of code) def _retrieve_run(self, run: Run): return self.client.beta.threads.runs.retrieve( run_id=run.id, thread_id=self.thread.id)

特定の実行を見つけるためにrun.idとthread.idの両方を渡す必要があることに注目してください。

_poll_runメソッドを Agent クラスに追加します。

 class Agent: # ... (rest of code) def _cancel_run(self, run: Run): self.client.beta.threads.runs.cancel( run_id=run.id, thread_id=self.thread.id) def _poll_run(self, run: Run): status = run.status start_time = time.time() while status != "completed": if status == 'failed': raise Exception(f"Run failed with error: {run.last_error}") if status == 'expired': raise Exception("Run expired.") time.sleep(1) run = self._retrieve_run(run) status = run.status elapsed_time = time.time() - start_time if elapsed_time > 120: # 2 minutes self._cancel_run(run) raise Exception("Run took longer than 2 minutes.")

🥵 ふー、多いですね…開梱してみましょう。

_poll_run引数としてRunオブジェクトを受け取り、現在の Run statusを抽出します。利用可能なすべてのステータスは、OpenAIドキュメントに記載されています。現在の目的に合ったものをいくつか使用します。

次に、while ループを実行して、いくつかのエラー シナリオを処理しながら完了ステータスを確認します。 Assistant API の実際の請求は少し曖昧なので、安全を期すために、2 分後に実行をキャンセルすることにしました。

OpenAI が 10 分後に実行をキャンセルすると、 expiredステータスが表示される場合でも。実行に 2 分以上かかる場合は、いずれにしても問題が発生している可能性があります。

また、数ミリ秒ごとにポーリングしたくないため、2 分マークに達して実行をキャンセルするまで、1 秒ごとにのみポーリングすることでリクエストを抑制します。これは好みに合わせて調整できます。

遅延後の反復ごとに、実行ステータスを再度取得します。

それでは、これらすべてをrun_agentメソッドに組み込んでみましょう。最初に_create_runで実行を作成し、次に応答が得られるかエラーがスローされるまで_poll_runでポーリングし、最後にポーリングが終了すると、スレッドから最後のメッセージ (現在はエージェントからのもの) を取得していることがわかります。

次に、メッセージをランタイム ループに返し、ユーザーに送り返すことができます。

 class Agent: # ... (rest of code) def run_agent(self): run = self._create_run() self._poll_run(run) # add this line message = self.get_last_message() return message

ほら、エージェントを再度実行すると、フレンドリーなエージェントから返信が届きます。

 python3 main.py User: hi Assistant: Hello there! What adventure can we embark on today? Or perhaps, before we set out, we should think about breakfast. Have you had yours yet? I've had mine, of course – can't start the day without a proper breakfast, you know. User: how many breakfasts have you had? Assistant: Ah, well, I've had just 1 breakfast today. But the day is still young, and there's always room for a second, isn't there? What about you? How can I assist you on this fine day?

パート 2 では、エージェントがツールを呼び出す機能を追加します。

完全なコードは私のGitHubにあります。

読んでいただきありがとうございます。コメントでご意見やフィードバックをいただければ幸いです。このようなコンテンツをもっと見るには、Linkedin でフォローしてください: https://www.linkedin.com/in/jean-marie-dalmasso-1b5473141/