Code Interpreter とは何ですか？#
以前に ChatGPT Code Interpreter が発表された後、皆さんはそれについてある程度の理解を持っていると思います。何であるか、何ができるかを知っているので、ここではこれらの基本的な問題を繰り返し説明することはありません。今回は、プロセスと要素の観点から Code Interpreter を理解する方法を見てみましょう。
通常の ChatGPT のインタラクションでは、プロセスと要素は次のようになります：プロンプト => 出力テキスト。これが、ChatGPT が発表されたときにすぐにプロンプトエンジニアリングの概念が登場し、プロンプトを構築するための埋め込み作業が行われた理由です。プロセスが短く、要素がシンプルであるため、良いプロンプトを構築することがこのプロセスの鍵となります。Code Interpreter に関しては、プロセスと要素は次のようになります：プロンプト => コード ==Interpreter==> 出力（説明、画像、ファイル...）。これにより、いくつかの変化がもたらされます：
- プロンプトの構築は出力に直接向けられるのではなく、中間コードを生成することになります。
- セッションを区別し、コードを実行し、中間変数を保存できるコードインタープリタが必要です。
- 出力がより多様になり、画像やファイルなどが含まれる可能性があります。
なぜ Code Interpreter を実現する必要があるのか#
ChatGPT はすでに Code Interpreter を実現しており、OpenAI の GPT モデルに基づいており、能力も非常に強力です。それでもなぜ独自の Code Interpreter を実現する必要があるのでしょうか？業界のリーダーに追いつくことや、社内のモデル能力と接続することを除いて、自分で実現することでどのような増分があるかを考えてみましょう。見える典型的な増分は次のとおりです：
- リアルタイムデータとインタラクションできる：ChatGPT の Code Interpreter はプラグインのネットワーク機能を持っておらず、Code Interpreter を有効にするとプラグインを選択できなくなります。これにより、ChatGPT Code Interpreter のデータのリアルタイム性が不足し、「2023 年のアップル社の株式パフォーマンスをグラフに描く」といったことができなくなります。
- より多くの環境とインタラクションできる：ローカルまたはクラウドにデプロイされた後、より自由な環境が得られます。ファイルシステムを操作したり、API を呼び出したり、ChatGPT Code Interpreter がサポートしていないパッケージをインストールしたりすることが可能になります。
Code Interpreter を実現するための考え方#
Code Interpreter を実現するためには、2 つの核心的な関心事があります。1 つはモデルの能力を活用すること、たとえば OpenAI API の Function Calling 機能を使用して呼び出すコードを生成することです。もう 1 つは、Python コードを実行できる環境を持つことです。たとえば、ユーザーが正弦関数のグラフを出力するように指定した場合、正弦関数のグラフを描画するコードを取得し、それを Python インタープリタに送信して実行し、画像を出力してユーザーに表示する必要があります。このプロセスでは、LLM エージェントが結果に対していくつかの説明や詳細を補足する必要があるかもしれません。また、ファイル IO やセッション中の変数保存の能力も考慮する必要があります。
LangChain を使用して実現する場合、これがさらに便利になります。ここでは、Python インタープリタ、ファイル IO、変数保存はすべて LangChain のツールと見なすことができ、それを LangChain のエージェントエグゼキュータに組み込んで呼び出すことができます。この考え方に基づいて、コミュニティにはオープンソースの実装があります：codebox-api 、これを LangChain ツールとして登録できます。コード実行のコア機能を提供するツールに加えて、次のような周辺機能も実現する必要があります：セッション管理、カーネル呼び出しの開始、ファイル IO。新しいセッションを作成するたびに、新しい Jupyter カーネルチャネルを作成してコードを実行し、その後、実行結果を出力のタイプに応じて分類してユーザーにフィードバックします。この部分については、上記の codebox-api の作者も解決策としてまとめています：codeinterpreter-api。
Code Interpreter の設計#
次に、codeinterpreter-api プロジェクトの整理と分解を行い、上記の考え方を具体的にどのように設計し実現するかを見ていきます。このプロジェクトは主に LangChain を使用して全体のプロセスを編成しているため、ここではプロジェクトで使用される LangChain の部分の基本概念を補足します。
LangChain の一部の基本概念：#
- LangChain エージェント：LangChain の基本モジュールの 1 つで、コアの考え方は LLM を使用して取るべき一連のアクションを選択することです。チェーン（chains）にハードコーディングされたアクションシーケンスとは異なり、エージェント（agents）は推論エンジンとして言語モデルを使用して、取るべきアクションとその順序を決定します。
- LangChain ツール：ツールはエージェントが呼び出す能力です。主に考慮すべき 2 つの側面があります：エージェントに正しいツールを提供し、エージェントに最も役立つ方法でこれらのツールを説明することです。Code Interpreter 向けにカスタム StructuredTool を作成する際には、name、description、func（同期呼び出しの関数）、coroutine（非同期呼び出しの関数）、args_schema（入力のスキーマ）を定義する必要があります。
- LangChain エージェントエグゼキュータ：エージェントエグゼキュータはエージェントの実行時です。実際にはエージェントを呼び出し、選択したアクションを実行します。このエグゼキュータは、エージェントが存在しないツールを選択した場合、ツールがエラーを出した場合、エージェントがツール呼び出しとして解釈できない出力を生成した場合など、複雑さを軽減するための追加の作業も行います。
プロセス設計と実装#
上記の基本概念を持った後、LangChain エージェントに基づいて Code Interpreter を実現する方法を見ていきます。以下のコードを通じて具体的な実行プロセスを見てみましょう：
from codeinterpreterapi import CodeInterpreterSession, File
async def main():
# セッションの開始/停止のためのコンテキストマネージャ
async with CodeInterpreterSession(model="gpt-3.5-turbo") as session:
# ユーザーリクエストを定義
user_request = "このデータセットを分析し、興味深いことをプロットしてください。"
files = [
File.from_path("examples/assets/iris.csv"),
]
# 応答を生成
response = await session.generate_response(user_request, files=files)
# 応答を出力（テキスト + 画像）
response.show()
if __name__ == "__main__":
import asyncio
# 非同期関数を実行
asyncio.run(main())
効果は以下の通りです：
実行環境とツールのインスタンス化#
with を使用してセッションを作成する際に、Jupyter カーネルとエージェントエグゼキュータのインスタンス化を開始します。以下は重要なステップのいくつかです：
jupyter-kernel-gatewayを使用して Jupyter カーネルと通信するサービスを作成し、起動成功の状態を検出します。
self.jupyter = await asyncio.create_subprocess_exec(
python,
"-m",
"jupyter",
"kernelgateway",
"--KernelGatewayApp.ip='0.0.0.0'",
f"--KernelGatewayApp.port={self.port}",
stdout=out,
stderr=out,
cwd=".codebox",
)
self._jupyter_pids.append(self.jupyter.pid)
# ...
while True:
try:
response = await self.aiohttp_session.get(self.kernel_url)
if response.status == 200:
break
except aiohttp.ClientConnectorError:
pass
except aiohttp.ServerDisconnectedError:
pass
if settings.VERBOSE:
print("カーネルの起動を待っています...")
await asyncio.sleep(1)
await self._aconnect()
stdout と stderr を指定し、プロセスの pid を記録し、このカーネルインスタンスをセッションに関連付けます。カーネルが作成された後、HTTP リクエストを送信してカーネルとの websocket 接続を確立します。
- エージェントエグゼキュータを作成します。
def _agent_executor(self) -> AgentExecutor:
return AgentExecutor.from_agent_and_tools(
agent=self._choose_agent(),
max_iterations=9,
tools=self.tools,
verbose=self.verbose,
memory=ConversationBufferMemory(
memory_key="chat_history",
return_messages=True,
chat_memory=self._history_backend(),
),
)
def _choose_agent(self) -> BaseSingleActionAgent:
return (
OpenAIFunctionsAgent.from_llm_and_tools(
llm=self.llm,
tools=self.tools,
system_message=code_interpreter_system_message,
extra_prompt_messages=[
MessagesPlaceholder(variable_name="chat_history")
],
)
# ...
)
def _tools(self, additional_tools: list[BaseTool]) -> list[BaseTool]:
return additional_tools + [
StructuredTool(
name="python",
description="IPython インタープリタにコードの文字列を入力します。"
"コード全体を単一の文字列で書きます。この文字列は非常に長くなる可能性があるため、"
"`;` 文字を使用して行を分割できます。"
"変数は実行間で保持されます。",
func=self._run_handler, # CodeBox を同期実行
coroutine=self._arun_handler, # CodeBox を非同期実行
args_schema=CodeInput,
),
]
ここでは OpenAIFunctionsAgent を使用することを定義していますが、必要に応じて独自のエージェントに変更できます。ただし、現在は OpenAI の API のみが便利で強力な Function Calling 機能を持っているため、これを例に挙げています。また、エージェントとエージェントエグゼキュータが使用するツールも指定しており、ツールの定義には名前や説明、Python コードを実行するための他のパラメータが含まれています。前のステップで作成した Jupyter カーネルインスタンスは CodeBox によってさらにラップされ、同期および非同期呼び出しメソッドとしてツールに渡されます。
入力テキストとファイルの処理#
プロンプトエンジニアリングは外部のステップとして行うことができるため、ユーザーが使用する際にはプロンプトを構築してから渡すべきです。このため、このステップではフレームワーク自体はあまり多くの作業を行わず、単に渡されたテキストとファイルを簡単にプロンプトに追加します（たとえば、LLM にユーザーがどのファイルを使用するかを指定した場合）、同時にファイルを CodeBox インスタンスに記録して後の実行を容易にします。
class UserRequest(HumanMessage):
files: list[File] = []
def __str__(self):
return self.content
def __repr__(self):
return f"UserRequest(content={self.content}, files={self.files})"
def _input_handler(self, request: UserRequest) -> None:
"""ユーザー入力を処理するためのコールバック関数。"""
if not request.files:
return
if not request.content:
request.content = (
"アップロードしましたので、ファイルを受け取ったことを確認してください。"
)
request.content += "\n**ユーザーがアップロードしたファイル：**\n"
for file in request.files:
self.input_files.append(file)
request.content += f"[添付ファイル: {file.name}]\n"
self.codebox.upload(file.name, file.content)
request.content += "**ファイルは現在 cwd に利用可能です。**\n"
実行と結果処理#
エージェントエグゼキュータを通じて、プロンプトからコードへの自動変換が実現できるようになりました。次に、このコードが具体的にどのように実行されるかを見てみましょう：
def _connect(self) -> None:
response = requests.post(
f"{self.kernel_url}/kernels",
headers={"Content-Type": "application/json"},
timeout=90,
)
self.kernel_id = response.json()["id"]
if self.kernel_id is None:
raise Exception("カーネルを起動できませんでした")
self.ws = ws_connect_sync(f"{self.ws_url}/kernels/{self.kernel_id}/channels")
まず、特定のカーネルと websocket を介して接続する必要があります。
self.ws.send(
json.dumps(
{
"header": {
"msg_id": (msg_id := uuid4().hex),
"msg_type": "execute_request",
},
"content": {
"code": code,
# ...
},
# ...
}
)
)
その後、websocket を介してコードをカーネルに送信して実行します。
while True:
# ...
if (
received_msg["header"]["msg_type"] == "stream"
and received_msg["parent_header"]["msg_id"] == msg_id
):
msg = received_msg["content"]["text"].strip()
if "Requirement already satisfied:" in msg:
continue
result += msg + "\n"
if settings.VERBOSE:
print("出力:\n", result)
elif (
received_msg["header"]["msg_type"] == "execute_result"
and received_msg["parent_header"]["msg_id"] == msg_id
):
result += received_msg["content"]["data"]["text/plain"].strip() + "\n"
if settings.VERBOSE:
print("出力:\n", result)
elif received_msg["header"]["msg_type"] == "display_data":
if "image/png" in received_msg["content"]["data"]:
return CodeBoxOutput(
type="image/png",
content=received_msg["content"]["data"]["image/png"],
)
if "text/plain" in received_msg["content"]["data"]:
return CodeBoxOutput(
type="text",
content=received_msg["content"]["data"]["text/plain"],
)
return CodeBoxOutput(
type="error",
content="出力を解析できませんでした",
)
その後、チャネルメッセージの一連の戻り値を処理します：
- msg_type: stream、msg に "Requirement already satisfied:" が含まれている場合、出力内容を追加し、ws の戻りを待ち続けます。
- msg_type: execute_result、
msg["content"]["data"]["text/plain"]を出力内容に追加し、引き続き待機します。 - msg_type: display_data、
msg["content"]["data"]を取得し、image/png があれば CodeBoxOutput に png を包んで返し、text/plain の場合も同様に返します。それ以外の場合はエラータイプの出力を返し、出力結果を解析できないことを示します。 - msg_type: status、execution_state: idle：コードが成功裏に実行されたが出力結果がない。
- msg_type: error：直接エラーを報告します。
出力結果#
上記の CodeBoxOutput 出力を得た後、出力を処理できます。テキストタイプの出力については、追加の処理は必要ありません。実行中に stdout を通じて出力されます。ファイルシステムの操作については、Jupyter を通じて直接行われるため、フレームワーク内で追加の処理は必要なく、ローカルファイルに自動的に保存されます。出力ファイルの説明や解釈が必要な場合にのみ、追加の処理が必要です。画像タイプの出力については、実行中に返された画像の base64 をセッションの out_files に保存し、最終的な出力処理時に Python の標準の Image タイプに変換し、その後 IPython の display メソッドを使用して表示します。
def get_image(self):
# ...
img_io = BytesIO(self.content)
img = Image.open(img_io)
# RGB に変換
if img.mode not in ("RGB", "L"): # L はグレースケール画像用
img = img.convert("RGB")
return img
def show_image(self):
img = self.get_image()
# 画像を表示
try:
# IPython シェルを取得しようとする
shell = get_ipython().__class__.__name__ # type: ignore
# シェルが Jupyter ノートブックまたはそれに類似している場合
if shell == "ZMQInteractiveShell" or shell == "Shell":
from IPython.display import display # type: ignore
display(img)
else:
img.show()
except NameError:
img.show()
社内での Code Interpreter の実現に向けた考え#
社内で Code Interpreter を実現する場合、以下の点に注目できます：
- サービス化またはソリューション化
- プラットフォームの既存モジュールに基礎的な実行能力やコンポーネントを提供
- 社内で Code Interpreter を必要とするチームに関連するソリューションを提供
- 社内モデルとの接続
- 社内システムや環境との接続、自動 API 呼び出しの実現
- 非 LangChain のオープン技術スタックのサポート
最後に#
今回は Code Interpreter の実現方法について大まかなプロセスと設計を見てきましたが、その中には多くの小さくても重要な詳細が議論されていません。たとえば、ウェブページで出力を行う方法、実行エラーが発生した場合にモデルにフィードバックして再生成を促す方法、依存パッケージがインストールされていない場合に自動でインストールして再実行する方法など、これらは堅牢な Code Interpreter が考慮し実現すべき要素です。コミュニティのソリューションは現在 MVP バージョンであり、エッジケースの処理や考慮は実際の生産アプリケーションにはまだいくつかのギャップがあります。完全で生産可能な Code Interpreter を実現するには、まだ多くの道のりがあり、手をもっと汚す必要があります。