Code Interpreterを自前環境で走らせて自宅のデジタルツインと連携してみた【ChatGPT API】
開発者向けの記事、と見せかけて実は生成AI好きな人全員が対象です。
（開発者でない方は前半は読み飛ばしてください）
はじめに
OpenAIからは「Assistants API」なるものが出ていて、これを使えばWeb版ChatGPTのように、OpenAIの環境でCodeInterpreterを使うことはできます。
でも、そろそろ自分で用意したローカル環境でCodeInterpreterを使い、ChatGPTに自らプログラミングをさせて課題解決させたいと思い始めている人もいるんじゃないですか？
今回はそれを実行できるスクリプトを作ってみようと思いました。
ちなみに高機能版は「OpenInterpreter」というものを有志が作ってくれています。今回はよりローレベルで制御したく、自作してみるという挑戦ですね。
ローカルCodeInterpreterのメリット
なんで、ローカル環境でこれをやりたいのかと言うと、それは色々と理由があります。
自由なライブラリを使用できる
既に自身が持っている別のシステムと連携しやすい
APIの費用をいろいろ節約できる
そもそも裏での処理をPythonに縛る必要すらない
僕の場合は２つ目の理由が大きかったんですが、今回試してみたところ３つ目のメリットにも気づいたという感じです。
ローカル環境でシステムを走らせるので、AssistantsAPIでは必要だったCodeInterpreterの費用が不要になることに加えて、FunctionCallingを使う時と比べてもコスパが良くなる可能性が出てきました。
内容としては、その実、シェルを実行するだけみたいなところがあるので、裏で走らせるものがPythonである必要はありません。
とはいえ今回は、ChatGPTのAPIをPythonから呼ぶつもりなので、そのまま裏でも、Pythonスクリプトを実行させようと思います。
ChatGPTにスクリプトを作成していただく
今回の作戦としては、以下の感じで行こうと思います。
ユーザーから指示を受け取る
指示を達成できるように、ChatGPTにスクリプトを作成していただく
スクリプトをファイル保存
指定する仮想環境（venv）を使って、保存したスクリプトを実行
実行結果をユーザーに返す
まずは、スクリプトの保存までを作るとこんな感じ。
user_message = "~~~~"
openai.api_key = "<OpenAI API Key>"
prompt_path = "<system_prompt.txt のパス>"
tmp_path = "<tmp.py の保存パス>"
import openai
# あらかじめ保存したシステムプロンプトを読み込む
with open(prompt_path, mode='r', encoding='UTF-8') as f:
system_prompt = f.read()
# ChatGPTに指示を送信
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_message}
]
response = openai.chat.completions.create(
model="gpt-4-turbo-preview",
messages=messages,
temperature=0
)
script = response.choices[0].message.content
script = script.replace("```python", "").replace("```", "")
# スクリプトを一旦保存
with open(tmp_path, mode='w', encoding='UTF-8') as f:
f.write(script)やってる内容はシンプルですね。
これで1～3のステップは完了です。
ちなみに、モデルですが、後述の「systemプロンプト」をしっかり描きたい人はgpt4を使う方がハルシネーションが抑えられて良いと思います。
作成してもらったスクリプトを実行する
続いて、4, 5の部分を作ります。
こちらも内容はとても単純で、Pythonの「subprocess」ライブラリを使うだけです。
import sys
import subprocess
tmp_path = "<tmp.py の保存パス>"
venv_activate_exec = "..\\venv\\Scripts\\activate"
# 実行コマンドを作成
script_exec = "python " + tmp_path
cmd = ' && '.join([venv_activate_exec, script_exec])
# コマンド実行
cp = subprocess.run(cmd, encoding='UTF-8', shell=True, stdout=subprocess.PIPE)
if cp.returncode != 0:
sys.exit(1)
result = cp.stdout
# 結果の出力
print("回答")
print(result)「venv_activate_exec」の仮想環境アクティベートや、「subprocess.run」の「shell=True」はWindows環境で動作する仕様なのでご注意ください。
いまのところ試した感じでは、標準出力を受け取るのが良さそうなので、「subprocess.run」の「stdout」を最終的なアウトプットとしています。
今回のスクリプトは以上です。とてもあっさりした内容ですね。
エラー時にChatGPTに修正させる機能も思いつきますが、今回は走りなのでまずはこの程度でいいでしょう。
試してみる
さっそくこれらのスクリプトを試してみます。
まず、system_prompt.txtには以下の通り指定します。
質問の要件を満たす「pythonスクリプト」のコードのみを回答してください。
ただし、回答は以下のスクリプト条件を満たしてください。
#スクリプト条件
・数値計算にはnumpyを使ってください。
・最終的に「print」を使って質問の回答を標準出力します。ただし、その時は必ず日本語を使用してください。そして、仮想環境には「numpy」を入れておき、user_messageには以下の通り入力してみます。
user_message = "sin(1/4*pi)は？"これを実行した結果がこちら。
回答
sin(1/4*pi)の値は 0.7071067811865476 です。ちなみに、保存されていたtmp.pyの内容がこちら。
import numpy as np
result = np.sin(np.pi / 4)
print(f'sin(1/4*pi)の値は {result} です。')完璧な結果ですね。
というか、これほどまでの回答ができるなら、FunctionCallingよりもトークン数を節約した上で、信頼できる回答を作成できるじゃないか。。。
（FunctionCallingは、「APIの呼び出しと回答の生成」の2回ChatGPTを呼び出す必要があるため、コストがかかる上に遅い。）
これは思いがけない副産物を得ることができました。
もっとがっつり試してみる
我が家では、現在、デジタルツインを稼働させている最中で、そのために様々なデータをIoT的に収集し、データベースへの蓄積を行っています。
デジタルツインの作成は、以前Youtubeにまとめたものがあります。
それで、せっかくなので、我が家のデータベース（influxdb）へアクセスし、知りたい情報が知れるような自動集計が可能か試してみます。
まあ色々と試行錯誤してみたので、
最終的にできあがったプロンプトからご覧ください。
質問の要件を満たす「pythonスクリプト」のコードのみを回答してください。
ただし、回答は以下のスクリプト条件を満たしてください。
#スクリプト条件
・influxdbからデータを取り出し、最終的に「print」を使って質問の回答を標準出力します。ただし、その時は必ず日本語を使用してください。
・スクリプト内にて「仮のデータを使用せず」、必ずinfluxdbのデータから取得した正確な情報を使用してください。したがって、printで標準出力する内容には、常にinfluxdbから取得した実際のデータが含まれます。
・influxdbからデータを取り出すときは、「influxdb_client」ライブラリの「InfluxDBClient」モジュールを使用し、FLUXクエリを用いてデータ検索を実行するようにしてください。
・FLUXクエリだけでは実現できない操作については、「pandas」「numpy」「scipy」「scikit-learn」などのようなライブラリを用いた数値計算をpythonスクリプトを用いて行うことができます。
・influxdbに蓄積されているテーブルは「#influxdbのテーブルデータ例」のような形式です。他にも、「#列の説明」「#_measurementに含まれる要素説明」「#_fieldに含まれる要素説明」を参考にしてクエリを作成してください。
・スクリプトの開始は、「#スクリプト開始文」を改変せず使ってください。
・また、データを得る時は「result = query_api.query(org=env.INFLUX_DB_ORGANIZATION, query=query)」でクエリを実行してください。
#スクリプト開始文
import sys
from pathlib import Path
sys.path.append(str(Path(__file__).parents[2]))
from env import env
url=env.INFLUX_DB_ENDPOINT
token=env.INFLUX_DB_TOKEN
org=env.INFLUX_DB_ORGANIZATION
bucket =env.INFLUX_DB_BUCKET
import numpy as np
import pandas as pd
from influxdb_client import InfluxDBClient
#influxdbのテーブルデータ例
table (last), _measurement (group string), _field (group string), _value (no group long), _time (no group dateTime:RFC3339), device (group string), room (group string)
2, AirConditioner2, status, 1, 2024-03-08T16:41:20.000Z , SwitchBotPlugMini, work_room
3, AirConditioner2, watts, 0.4, 2024-03-08T15:48:50.000Z , SwitchBotPlugMini, work_room
3, AirConditioner2, watts, 0.4, 2024-03-08T15:59:10.000Z, SwitchBotPlugMini, work_room
3, AirConditioner2, watts, 0.4, 2024-03-08T16:09:30.000Z, SwitchBotPlugMini, work_room
#列の説明
_measurement: 対象行が何についてのデータなのかを表す（エアコン、カメラ、天気、カレンダー）
_field: データの何の属性なのかを示す（電力値、位置情報、温度、予定、金額）
_value: データの値を示す
_time: その行のデータを取得した時刻を示す
_device: データをセンシングしているデバイスを示す
_room: データが紐づく部屋名を示す（kitchen, work_room, living_room）
#_measurementに含まれる要素説明
AirConditioner1: リビングのエアコン
AirConditioner2: 書斎のエアコン
DesktopPC: メインのPC
Refrigerator: 冷蔵庫
TV: テレビ
Camera: カメラ
Weather: 天気
KIT: KITという名の人物
#_fieldに含まれる要素説明、（）内は_valueが取り得る値の種類を説明
watts: 瞬間電力の値（数値、単位はワット）
door_status: ドアの開閉状態（0（開放）または1（閉鎖））
detection: カメラで検出されたもの（文字列）
temp: 気温（数値）
humid: 湿度（数値）
x: 位置座標（数値）
y: 位置座標（数値）
z: 位置座標（数値）
clothes: 検出された服（文字列）はい、すごくゴツいプロンプトができてしまった。
環境変数の指定とかも全部プロンプト内で指定すればよい感じです。
で、一昨日は休日。一日家に居たので、暖房の電気代が気になったのでこの質問を。
user_message = "一昨日、リビングで使ったエアコンの電気代を教えてください。ただし、1kWhで20円として計算してください。"その回答がこちら。
回答
一昨日、リビングで使ったエアコンの電気代は77.37円です。おお、なんだか良さそう。
この回答の信頼性をチェックするために、作成されたtmp.pyも見てみます。
import sys
from pathlib import Path
sys.path.append(str(Path(__file__).parents[2]))
from env import env
url=env.INFLUX_DB_ENDPOINT
token=env.INFLUX_DB_TOKEN
org=env.INFLUX_DB_ORGANIZATION
bucket =env.INFLUX_DB_BUCKET
import numpy as np
import pandas as pd
from influxdb_client import InfluxDBClient
from datetime import datetime, timedelta
# InfluxDBクライアントの設定
client = InfluxDBClient(url=url, token=token, org=org)
query_api = client.query_api()
# 一昨日の日付を計算
two_days_ago = (datetime.utcnow() - timedelta(days=2)).date()
# FLUXクエリの作成
query = f'''
from(bucket: "{bucket}")
|> range(start: {two_days_ago}T00:00:00Z, stop: {two_days_ago}T23:59:59Z)
|> filter(fn: (r) => r["_measurement"] == "AirConditioner1")
|> filter(fn: (r) => r["_field"] == "watts")
|> filter(fn: (r) => r["room"] == "living_room")
|> aggregateWindow(every: 1h, fn: mean, createEmpty: false)
|> sum()
'''
# クエリの実行
result = query_api.query(org=org, query=query)
# 結果の処理
total_watts = 0
for table in result:
for record in table.records:
total_watts += record.get_value()
# 電気代の計算 (1kWh = 1000Watts, 20円/kWh)
electricity_cost = (total_watts / 1000) * 20
print(f"一昨日、リビングで使ったエアコンの電気代は{electricity_cost:.2f}円です。")データベースから、一昨日のリビングのエアコンのデータにフィルタをかけ、１時間毎の平均集計を１日分合算し、1kWh20円の計算もできていますね。
たぶんこのスクリプトには、ほぼ問題はないかと思います。
ということで、データベースへのアクセスを指示する「systemプロンプト」さえ用意しておけば、１つの日本語指示だけでCodeInterpreterがあとはよろしくやってくれる仕組みが無事にできあがりました。
所感
ローカル版CodeInterpreterを使うことで、ChatGPTをもっとローレイヤーから操ることができ、既存の色々なシステムと連携できるほか、コスパも良くなりそう、という良い結果が得られました。
このレベルなら引き続き、我が家のデジタルツインでもどんどん使っていきたいと思います。とりあえずは、アレクサのようなボイスコントロールから、この仕組みを呼び出せるようにして、いつでも使えるようにし、このシステムの可能性の境界線をさらにクリアにしていきたいところです。
（その時はYoutubeの方でお届けします）
最後まで読んでいただきありがとうございました。