James Tsang

Code Interpreter とは何ですか？#

以前、ChatGPT Code Interpreter が発表された後、皆さんはそれについてある程度の理解を持っていると思います。何であるか、何ができるかを知っているので、ここではこれらの基本的な問題を繰り返し説明することはしません。今回は、プロセスと要素の観点から Code Interpreter を理解する方法を見ていきましょう。

通常の ChatGPT のインタラクションでは、プロセスと要素は次のようになります：Prompt => Output Text。これが、ChatGPT が発表されたときにすぐに Prompt Engineering の概念とそれに関連する Prompt 構築の埋め込み作業が登場した理由です。プロセスが短く、要素がシンプルであるため、良い Prompt を構築することがこのプロセスの鍵となります。Code Interpreter に関しては、そのプロセスと要素は次のようになります：Prompt => Code ==Interpreter==> Output (Description, Image, File...)。これにより、いくつかの変化がもたらされます：

1. Prompt の構築は直接的な出力を目指すのではなく、中間コードを生成することになります。
2. セッションを区別し、コードを実行し、中間変数を保存できるコードインタープリターが必要です。
3. 出力がより多様になり、画像やファイルなどが含まれるようになります。

なぜ Code Interpreter を実現する必要があるのか#

ChatGPT はすでに Code Interpreter を実現しており、OpenAI の GPT モデルに基づいて強力な能力を持っていますが、なぜ独自の Code Interpreter を実現する必要があるのでしょうか？業界のリーダーに追いつくことや、社内モデルの能力と接続することを除いて、自分で実現することによる増分を考えてみることができます。見える典型的な増分は次の通りです：

1. リアルタイムデータとインタラクションできる：ChatGPT の Code Interpreter は Plugins のネットワーク能力を持っておらず、Code Interpreter を起動すると plugin を選択できなくなります。これにより、ChatGPT Code Interpreter のデータのリアルタイム性が不足し、「2023 年のアップル社の株式パフォーマンスをグラフに描く」といったことができなくなります。
2. より多くの環境とインタラクションできる：ローカルまたはクラウドにデプロイされた後、ファイルシステムの操作や 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 の基本モジュールで、核心的な考え方は LLM を使用して取るべき一連のアクションを選択することです。チェーンにハードコーディングされたアクションシーケンスとは異なり、エージェントは推論エンジンとして言語モデルを使用して取るべきアクションとその順序を決定します。
• 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 カーネルとエージェントエグゼキューターのインスタンス化を開始します。以下は重要なステップのいくつかです：

1. 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 接続を確立します。

2. エージェントエグゼキューターを作成します。

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 によってさらにラップされ、同期および非同期呼び出しメソッドとしてツールに渡されます。

入力テキストとファイルの処理#

Prompt Engineering は外部のステップとして行うことができ、ユーザーは使用時に Prompt を構築してから渡すべきです。そのため、このステップではフレームワーク自体はあまり作業を行わず、渡されたテキストとファイルを簡単に Prompt に追加するだけです（たとえば、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 を実装する場合、以下の点に注目できます：

1. サービス化またはソリューション化
   1. プラットフォームの既存モジュールに基礎的な実行能力やコンポーネントを提供
   2. 社内で Code Interpreter を必要とするチームに関連するソリューションを提供
2. 社内モデルとの接続
3. 社内システムや環境との接続、API 呼び出しの自動化を実現
4. 非 LangChain のオープン技術スタックをサポート

最後に#

今回は Code Interpreter の実装案を大まかなプロセスと設計の観点から見てきましたが、その中には多くの小さくても重要な詳細が議論されていません。たとえば、ウェブページで出力を行う方法、実行エラーが発生した場合にモデルにフィードバックして再生成を促す方法、依存パッケージが未インストールの場合に自動的にインストールして再実行する方法など、これらは堅牢な Code Interpreter が考慮し実現すべき要素です。コミュニティのソリューションは現在も MVP バージョンであり、エッジケースの処理や考慮が実際の生産アプリケーションにはまだいくつかのギャップがあります。完璧で生産可能な Code Interpreter を実現するには、まだ多くの道のりがあり、手をさらに汚す必要があります。