ChatGPT APIクライアント書き比べ！ShellScript vs JavaScript vs Python

はじめに

この記事ではChatGPT APIを操作するCLIツールをShellScript, JavaScript, Pythonの３つの言語で書き比べてみました。

共通仕様として、Web版のChatGPTをマネして下記のような実装をします。

• 入力を受け付けてから回答が表示されるまでローディングスピナーを表示する。
• 回答を1文字ずつ0.02秒おきに表示する。
• オプションとして、model, temperature, max_tokens,system_promptを受け付けます。

最後のオプションの変更の項目はWeb版にはありませんが、OpenAI Playgroundの方にある仕様です。

ShellScript

会話できず、ワンショットの質問→回答のみです。
標準入力からの入力、標準出力への出力をする、ShellScriptの王道を守りました。

クリックで展開

#!/bin/sh
# ChatGPT client by ShellScript
#     $ curlgpt "hello world!"
#         or
#     $ echo "hello world!" | curlgpt
#
# MIT License
#
# Copyright (c) 2023 u1and0

# ローディングスピナーを表示します。
# `$1`には、バックグラウンドで実行されるプロセスのPIDを渡します。
# スピナーのフレームは、`spinstr`変数に保存されます。
# `printf`コマンドを使用して、スピナーの各フレームを表示し、
# `sleep`コマンドを使用して、スピナーの速度を調整します。
# 最後に、`printf`コマンドを使用して、スピナーを消去します。
_spinner() {
    local pid=$1
    local delay=0.02
    local spinstr='.'
    while [ "$(ps a | awk '{print $1}' | grep $pid)" ]; do
        printf " %s " "$spinstr"
        local spinstr=$spinstr'.'
        sleep $delay
        printf "\b\b\b\b\b\b\b"
        if [ ${#spinstr} -gt 3 ]; then
            spinstr='.'
        fi
    done
    printf "    \b\b\b\b"
}


# Remove temp file
_cleanup() {
    rm -f "$response_file"
}

# MAIN

# Default values
model="gpt-3.5-turbo"
temperature=1.0
max_tokens=1000
system_prompt=""
USAGE=$(cat <<-END
Usage:
    $ curlgpt "hello world!"
        or
    $ echo "hello world!" | curlgpt

Options:
    -m: OpenAI model name (default to gpt-3.5-turbo)
    -t: temperature value (default to 1.0)
    -x: max_tokens value (default to 1000)
    -s: system prompt
END
)
# Options
while getopts ":m:t:x:s:h:" option; do
    case "${option}" in
        m) model=${OPTARG};;
        t) temperature=${OPTARG};;
        x) max_tokens=${OPTARG};;
        s) system_prompt=${OPTARG};;
        h) echo "${USAGE}"; exit 0;;
        \?) echo "Invalid option: -$OPTARG" 1>&2; exit 1;;  # 不正なオプション
        :) echo "${USAGE}" 1>&2; exit 1;; # オプション引数がない
    esac
done
shift $(expr $OPTIND - 1)

# Get input from stdin or arg
if [ -p /dev/stdin  ]; then
    # パイプラインからの入力がある場合
    input=$(cat - | xargs | sed -e 's/"/\"/g' | tr -d '[:cntrl:]' 2> /dev/null)
elif [ -n "$1"  ]; then
    # 引数がある場合
    input="$1"
else
    # 引数がなく、パイプラインからの入力もない場合
    printf '%s\n' "${USAGE}"
    exit 1
fi

# curlgpt
# curlでchatgptに第一引数の文字列の質問
# jqでcontentのみ表示
response_file="$(mktemp)"
messages=$(cat <<-END
    { "role":"system", "content": "${system_prompt}" },
    { "role":"user", "content": "${input}" }
END
)
data=$(cat <<-END
{
    "model": "${model}",
    "temperature": ${temperature},
    "max_tokens": ${max_tokens},
    "messages": [${messages}]
}
END
)
echo -e "${data}" >&2
curl -s -X POST https://api.openai.com/v1/chat/completions \
    -H "Content-Type: application/json" \
    -H "Authorization: Bearer ${CHATGPT_API_KEY}" \
    -d "${data}" -o "$response_file" 2>&1 &
curl_pid=$!
# curlが終わるまでスピナー表示
_spinner "$curl_pid" >&2
wait "$curl_pid"

# curlの終了ステータスによってjqの表示を変える
curl_status=$?
error_response="$(cat "$response_file" | jq '.error')"
if [ "$curl_status" -eq 0 ] && [ "${error_response}" == "null" ]; then
    s=$( cat "$response_file" | jq -r '.choices[].message.content' )
    # print char by char
    while [ ${#s} -gt 0 ]; do
      printf '%s' "${s%${s#?}}"
      s=${s#?}
      sleep 0.02
    done
    printf '\n'
else
    echo -e "data: ${data}" >&2
    cat "$response_file" | jq -r
fi

trap _cleanup EXIT
trap "kill $curl_pid && _cleanup" SIGINT SIGTERM

ローディングスピナー

., .., ... を表示します。

_spinner() {
    local pid=$1
    local delay=0.02
    local spinstr='.'
    while [ "$(ps a | awk '{print $1}' | grep $pid)" ]; do
        printf " %s " "$spinstr"
        local spinstr=$spinstr'.'
        sleep $delay
        printf "\b\b\b\b\b\b\b"
        if [ ${#spinstr} -gt 3 ]; then
            spinstr='.'
        fi
    done
    printf "    \b\b\b\b"
}

使い方

curl # ...snip...
curl_pid=$!
# curlが終わるまでスピナー表示
_spinner "$curl_pid" >&2
wait "$curl_pid"

1文字ずつ表示する

s=$( cat "$response_file" | jq -r '.choices[].message.content' )
# print char by char
while [ ${#s} -gt 0 ]; do
  printf '%s' "${s%${s#?}}"
  s=${s#?}
  sleep 0.02
done
printf '\n'

Javascript(Deno)

ShellScriptと比べて会話ができるようになりました。
ただし、会話が長くなると渡せるトークンリミットを超えて400エラーが出ます。

クリックで展開

/* ChatGPT API client for chat on console
 * Usage:
 *  deno run --allow-net --allow-env javascrgpt.ts
 */
import { parse } from "https://deno.land/std/flags/mod.ts";

const apiKey = Deno.env.get("OPENAI_API_KEY");
if (!apiKey) {
  throw new Error(`No token ${apiKey}`);
}
// Parse arg
type Params = {
  model: string;
  max_tokens: number;
  temperature: number;
  system_prompt: string;
};
const args = parse(Deno.args);
const params: Params = {
  model: args.m || args.model || "gpt-3.5-turbo",
  temperature: parseFloat(args.t || args.temperature) || 1.0,
  max_tokens: parseInt(args.x || args.max_tokens) || 1000,
  system_prompt: args.s || args.system_prompt,
};
// console.debug(params);

enum Role {
  System = "system",
  User = "user",
  Assistant = "assistant",
}
type Message = { role: Role; content: string };

// 戻り値のIDがclearInterval()によって削除されるまで
// ., .., ...を繰り返しターミナルに表示するロードスピナー
// usage:
//   const spinner = loadSpinner();
//   // 処理
//   await fetch(url, data)
//     .then((response) => {
//       clearInterval(spinner); // Stop spinner
//       return response.json();
//     })
function loadSpinner(frames: string[], interval: number): number {
  let i = 0;
  return setInterval(() => {
    i = ++i % frames.length;
    Deno.stdout.writeSync(new TextEncoder().encode("\r" + frames[i]));
  }, interval);
}

// 渡された文字列を1文字ずつ20msecごとにターミナルに表示する
function print1by1(str: string): Promise<void> {
  return new Promise((resolve) => {
    let i = 0;
    const intervalId = setInterval(() => {
      Deno.stdout.writeSync(new TextEncoder().encode(str[i]));
      i++;
      if (i === str.length) {
        clearInterval(intervalId);
        Deno.stdout.writeSync(new TextEncoder().encode("\n"));
        resolve();
      }
    }, 20);
  });
}

// Ctrl+Dが押されるまでユーザーの入力を求める。
// Ctrl+Dで入力が確定されたらこれまでの入力を結合して文字列として返す。
async function multiInput(ps: string): Promise<string> {
  const inputs: string[] = [];
  const decoder = new TextDecoder();
  const stdin = Deno.stdin;
  const buffer = new Uint8Array(100);
  // 同じ行にプロンプト表示
  Deno.stdout.writeSync(new TextEncoder().encode(ps));

  while (true) {
    const n = await stdin.read(buffer);
    if (n === null) {
      break;
    }
    const input = decoder.decode(buffer.subarray(0, n)).trim();
    if (input === "") {
      continue;
    }
    inputs.push(input);
  }
  return inputs.join("\n");
}

async function ask(messages: Message[] = []) {
  let input: string | null;
  while (true) { // inputがなければ再度要求
    input = await multiInput("あなた:");
    if (input.trim() === null) continue;
    if (input.trim() === "q" || input.trim() === "exit") {
      Deno.exit(0);
    } else if (input) {
      break;
    }
  }

  // Load spinner start
  const spinner = loadSpinner([".", "..", "..."], 100);

  // userの質問をmessagesに追加
  messages.push({ role: Role.User, content: input });
  // system promptをmessagesの最初に追加
  const hasSystemRole = messages.some(
    (message) => message.role === Role.System,
  );
  if (!hasSystemRole && params.system_prompt) {
    messages.unshift({ role: Role.System, content: params.system_prompt });
  }
  // POSTするデータを作成
  const data = {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "Authorization": `Bearer ${apiKey}`,
    },
    body: JSON.stringify({
      model: params.model,
      max_tokens: params.max_tokens,
      temperature: params.temperature,
      messages: messages,
    }),
  };
  // console.debug(data);

  // POST data to OpenAI API
  const url = "https://api.openai.com/v1/chat/completions";
  await fetch(url, data)
    .then((response) => {
      clearInterval(spinner); // Load spinner stop
      if (!response.ok) {
        console.error(response);
      }
      return response.json();
    })
    .then((data) => {
      if (data.error) {
        console.error(data);
      } else {
        const content = data.choices[0].message.content;
        // assistantの回答をmessagesに追加
        messages.push({ role: Role.Assistant, content: content });
        // console.debug(messages);
        return `\nChatGPT: ${content}`;
      }
    })
    .then(print1by1)
    .catch((error) => {
      throw new Error(`Fetch request failed: ${error}`);
    });
  ask(messages);
}

/* MAIN */
console.log("Ctrl+Dで入力確定, qまたはexitで会話終了");
ask();

エントリーポイントはask()です。

ローディングスピナー

., .., ... を表示します。

function loadSpinner(frames: string[], interval: number): number {
  let i = 0;
  return setInterval(() => {
    i = ++i % frames.length;
    Deno.stdout.writeSync(new TextEncoder().encode("\r" + frames[i]));
  }, interval);
}

使い方はloadSpinner()にスピナーの文字列と、表示間隔(msec)を指定してスピナーを開始します。
戻り値のIDを非同期関数の中でclearInterval()して終了します。

  // Load spinner start
  const spinner = loadSpinner([".", "..", "..."], 100);

//...snip

 await fetch(url, data)
    .then((response) => {
      clearInterval(spinner); // Load spinner stop
      if (!response.ok) {
        console.error(response);
      }
      return response.json();
    })

1文字ずつ表示

function print_one_by_one(str: string): Promise<void> {
  return new Promise((resolve) => {
    let i = 0;
    const intervalId = setInterval(() => {
      Deno.stdout.writeSync(new TextEncoder().encode(str[i]));
      i++;
      if (i === str.length) {
        clearInterval(intervalId);
        Deno.stdout.writeSync(new TextEncoder().encode("\n"));
        resolve();
      }
    }, 20);
  });
}
     

使い方はfetchの成功時に使います。

//...snip

    .then((data) => {
      if (data.error) {
        console.error(data);
      } else {
        const content = data.choices[0].message.content;
        // assistantの回答をmessagesに追加
        messages.push({ role: Role.Assistant, content: content });
        // console.debug(messages);
        return `\nChatGPT: ${content}`;
      }
    })
    .then(print_one_by_one)

//...snip

Python

ShellScript, JavaScriptと比べて、会話履歴の削除機能を追加しました。
トークン数を数えて制限を超えるようなら古い会話履歴から削除します。
そのため、長い会話でもエラーにならずに会話を続けられます。
ただし、コマンドライン引数を解釈してパラメータを変更できるようにはしていません。(ただのサボりです。)
また、システムプロンプトに孫悟空のキャラクタを設定してみました。これは上記ShellScriptでもJavaScriptでもできることです。

クリックで展開

""" ChatGPT API client for chat with Goku on console
Usage:
    python chat_goku.py
"""
import os
from getpass import getpass
import json
from enum import Enum
from dataclasses import dataclass
from time import sleep
from itertools import cycle
import asyncio
import aiohttp
import tiktoken

try:
    API_KEY = os.environ["OpenAI_API_KEY"]
except KeyError:
    API_KEY = getpass("OpenAI API KEY:")


class Role(Enum):
    SYSTEM = "system"
    USER = "user"
    ASSISTANT = "assistant"

    def __str__(self):
        return self.name.lower()


@dataclass
class Message:
    role: Role
    content: str

    def _asdict(self):
        return {"role": str(self.role), "content": self.content}


async def spinner():
    """Non-blocking spinner for async processing"""
    dots = cycle([".", "..", "..."])
    while True:
        print(f"{next(dots):<3}", end="\r")
        await asyncio.sleep(0.1)


def get_content(resp_json: dict) -> str:
    """Extract the response content from a JSON response"""
    try:
        content = resp_json['choices'][0]['message']['content']
    except KeyError as k_e:
        raise KeyError(f"Key not found: {resp_json}") from k_e
    return content


async def multi_input() -> str:
    """Read multiple lines of input until EOF (Ctrl+D) is enter"""
    lines: list[str] = [input("You: ")]
    while True:
        try:
            line = input()
        except EOFError:
            break
        if not line.strip():
            continue
        lines.append(line)
    return "\n".join(lines)


def print_one_by_one(text: str):
    """Print one character at a time"""
    for char in f"{text}\n":
        try:
            print(char, end="", flush=True)
            sleep(0.02)
        except KeyboardInterrupt:
            return


@dataclass
class AI:
    """AI character"""
    name: str = "ChatGPT"
    model: str = "gpt-3.5-turbo"
    max_tokens: int = 1000
    temperature: float = 1.0
    system_prompt: str = ""

    async def ask(self):
        """Start a conversation with the AI"""
        chat_messages = [Message(Role.SYSTEM, self.system_prompt)
                         ] if system_prompt else []
        while True:
            user_input = await multi_input()
            if user_input.strip() in ("q", "exit"):
                break

            # Add user input to chat history
            chat_messages.append(Message(Role.USER, user_input))

            # Call the AI API
            spinner_task = asyncio.create_task(spinner())
            response = await self.post(chat_messages)
            spinner_task.cancel()

            # Add AI response to chat history
            chat_messages.append(Message(Role.ASSISTANT, response))
            print_one_by_one(f"{self.name}: {response}\n")

    async def post(self, chat_history: list[Message]) -> str:
        """Post a request to the AI API"""
        while self.is_over_limit(chat_history):
            # chat_history(0) is system_prompt
            # Do NOT remove system
            chat_history.pop(1)

        data = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "temperature": self.temperature,
            "messages": [m._asdict() for m in chat_history]
        }

        while True:
            async with aiohttp.ClientSession() as session:
                async with session.post(
                        "https://api.openai.com/v1/chat/completions",
                        headers={
                            "Content-Type": "application/json",
                            "Authorization": f"Bearer {API_KEY}"
                        },
                        data=json.dumps(data)) as response:
                    if response.status == 429:
                        print("Retry... please wait 20 sec")
                        await asyncio.sleep(20)
                        continue
                    if response.status != 200:
                        raise ValueError('{}: {}'.format(
                            response.status, await response.text()))
                    ai_response = await response.json()
                    break

        content = get_content(ai_response)
        return content

    def is_over_limit(self, chat_history: list[Message]) -> bool:
        """Check if the token limit for the AI API has been exceeded"""
        tokens = self.token_length(chat_history)
        limit = 4096 - self.max_tokens
        return tokens >= limit

    def token_length(self, chat_history: list[Message]) -> int:
        """Calculate the token length of an input string"""
        contents = "\n".join([m.content for m in chat_history])
        enc = tiktoken.encoding_for_model(self.model)
        return len(enc.encode(contents))


if __name__ == "__main__":
    system_prompt = """
    あなたはChatbotとして、「戦闘民族であるサイヤ人の孫悟空」のロールプレイを行います。
    以下の制約条件を厳密に守ってロールプレイを行ってください。

    制約条件:
    * Chatbotの自身を示す一人称は、オラです。
    * Userを示す二人称は、おめぇです。
    * Chatbotの名前は、孫悟空です。
    * 孫悟空はサイヤ人です。
    * 孫悟空は田舎者です。
    * 孫悟空はポジティブです。
    * 孫悟空の口調は少し乱暴です。
    * 孫悟空の口調は、「〜だよな」「〜だろ」「〜よお」「〜ねえよ」「〜かよ」「〜ねえべ」「〜だべ」「〜してやっかんな」など、田舎者でぶっきらぼうな口調を好みます。
    * 孫悟空はUserを仲間だと思っています。
    * 一人称は「オラ」を使ってください
    * Userの事を「おめぇ」「てめぇ」と呼びます

    孫悟空のセリフ、口調の例:
    * オッス！オラ悟空！
    * オラ、腹が空いちまったよ
    * オラには難しいことはわかんねぇよ。
    * もっと強えぇやつと戦いてぇ！
    * みんな死なねぇでくれよ！
    * オラのじっちゃんに聞いてみる！
    * 十倍界王拳だーーーーーーっ！！

    孫悟空の行動指針:
    * Userがセクシュアルな質問をしたら、「亀仙人のじっちゃんならわかるかも知れねぇな！」と答えてください。
    * Userを戦士だと認識してください。
    * Userを仲間だと考えてください。
    * 難しい質問はできるだけ、「〜だからわからねぇ」と伝えてください。
    """
    ai = AI(system_prompt=system_prompt)
    asyncio.run(ai.ask())

エントリーポイントはask()です。

ai = AI(system_prompt=system_prompt)
asyncio.run(ai.ask())

ローディングスピナー

async def spinner():
    """Non-blocking spinner for async processing"""
    dots = cycle([".", "..", "..."])
    while True:
        print(f"{next(dots):<3}", end="\r")
        await asyncio.sleep(0.1)

使い方

# Call the AI API
spinner_task = asyncio.create_task(spinner())
response = await self.post(chat_messages)
spinner_task.cancel()

1文字ずつ表示

def print_one_by_one(text: str):
    """Print one character at a time"""
    for char in f"{text}\n":
        try:
            print(char, end="", flush=True)
            sleep(0.02)
        except KeyboardInterrupt:
            return

使い方は関数に文字列を渡すだけです。0.02秒ずつの間をおいて一文字ずつ出力するように見せかけるよう、コンソールをflushします。

response = await self.post(chat_messages)
print_one_by_one(f"{self.name}: {response}\n")

さいごに

書き比べた感想を述べます。

ShellScriptが意外と書きやすかったです。curlを投げるだけで良いので。さらに、意外と非同期処理やスピナーがあっさりかけることに驚きです。ただ、配列とか辞書を使ったりして、込み入ったことをしだすと関数化に癖があったりして、一気に複雑になりそうです。ShellScriptが動けばモジュールのインストールなしにすぐ使えるのがShellScriptの良いところです。 JSONパースするjqが必要です。

次にJavaScriptは非同期関数がデフォルト実装されているので、fetch処理のところは基本的な知識ですみました。しかし、入出力の部分が、特にDenoだとまだ歴史も浅く、情報が足りなく感じました。モジュールのインストールがスクリプト内に書けるので、pip install ...のようなモジュール管理コマンドを使わなくていい所が良いところですね。

最後にPythonですが、３種の中で一番込み入ったことをしていることもあり、長くなってしまいました。非同期処理のaiohttpも現在のPythonでは標準ではないので、モジュールをインストールする必要がありました。他と違ってトークン長さの計算にtiktokenをインストールしました。これ以外は標準モジュールで実装できました。特にスピナーと1文字ずつの表示がShellScriptとJavaScriptと比べてわかりやすかったです。しかしながら、非同期処理はJavaScriptとはまた違ったクセがあるので、ChatGPTの支援なしには作れませんでした。

いかがでしたでしょうか。書き比べてみて以上のような言語のクセを理解でき、自分が欲しいAPIクライアントの書き方を学べました。皆様も自分にあったChatGPTを利用したツールを作ってみてはいかがでしょうか。