ChatGPT+FastAPIでチャットボット（の原形）を作ってみる

# 概要
「API経由でユーザから投げられたプロンプトをChatGPTで回答し、そのプロンプトと回答をDBに保存させる」アプリケーションを作ってみます。

[コード全文はGithubにあります。](https://github.com/KENTAROSZK/ChatGPT_with_FastAPI_and_SQLite/tree/main)

完成形のイメージは↓の画像です。

![図1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/aaebd788-158d-f1dd-6391-ece17f422ff0.png)


今回の記事で作成したものをベースにすることで世の中であふれているChatGPTを使ったLINEボットなどを作ることができます。

# この記事の解説の流れ

シンプルなものを作成し、それをベースにDB機能の追加、ChatGPTの追加という3段階でアプリケーションを作ります。

## ①FastAPIでプロンプトを投げるアプリケーションを作る
まずはシンプルに、FastAPIを使って、API経由でプロンプトを投げて、そのプロンプトをそのままユーザへ返すものを作ります。
注：この段階では、ChatGPTを追加しません。

![図2.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/57116b11-6d96-5a6d-de64-e33064aec26c.png)

## ②DBを追加する
次に①で作ったものに対して、DBを追加します。DBの中にユーザIDとユーザが投げているプロンプト、そしてプロンプトを加工した文字列のデータをDBへ格納します。こうすることで、ユーザとの過去のやり取りをDBに記録させておくことができます。
注：まだChatGPTを追加しません。

![図4.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/cb01cdd9-2a8a-695d-742f-6ab9d819ffa5.png)


## ③ChatGPTを追加する
最後にChatGPTを追加します。②まではユーザからのプロンプトに対して、シンプルな加工（文字列の追加）をしただけでしたが、ここでようやくChatGPTを追加します。
②でDBを追加しているので、ChatGPTとのやりとりデータをDBに記録させることができるようになります。

![図6.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/541b1134-b3d1-447a-22b8-9ba03e780765.png)

# 実装

## 実装の基本方針
[コード全文はGithubにあります。](https://github.com/KENTAROSZK/ChatGPT_with_FastAPI_and_SQLite/tree/main)

- 今回作成するコードは次の2つのコードを1セットとして作成していきます。
    - APIにリクエストを投げるパート
    - FastAPIを使ってAPIの実装とその処理を担当するパート

## ①実装：FastAPIでプロンプトを投げるアプリケーションを作る

作成するもののイメージを先に記載すると次の図のようになります。

![図3.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/c62c6da6-0740-d91f-29bb-b1e882a00a2d.png)

### APIにリクエストを投げるパート
まずはAPIにリクエストを投げるパートを実装していきます。こちらは非常にシンプルな実装となっています。

``` python:simple_post.py
import requests
import json


url = "http://localhost:8000/process/"  # POSTリクエストを送信するエンドポイントのURL

data = {
    "user_id": "example_user", # ユーザID
    "query"  : "example_query" # ユーザが投げるクエリ(=プロンプト)
}

response = requests.post(url, json=data) # エンドポイントにクエリを投げる

# レスポンスの内容を表示
print(response.json())
```

これ単体ではまだ動かすことができません。次に実装する```api_simple_post.py```を実装して、初めて動かすことができます。


### APIと受け取ったクエリを処理させるパート
次のように実装します。

``` python:api_simple_post.py
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Input(BaseModel):
    user_id: str
    query: str

class Output(BaseModel):
    user_id: str
    query: str
    output: str

@app.post("/process/")
def main(input_data: Input):
    # 入力データを取得
    user_id = input_data.user_id
    query = input_data.query

    # 文字列の加工処理
    processed_text = f"Processed _ User ID: {user_id}, Query: {query}"

    # 処理結果と入力データをまとめて返す
    output_data = Output(user_id=user_id, query=query, output=processed_text)
    return output_data

```

コードを解説します。

``` python

class Input(BaseModel):
    user_id: str
    query: str


class Output(BaseModel):
    user_id: str
    query: str
    output: str
```

ここは、ユーザとのやりとりをするデータの定義をしています。ここで定義したオブジェクトをAPI経由で送受信します。
- ユーザからのインプット：```Input```
    - ```user_id``` ユーザID
    - ```query``` クエリ
- ユーザへのアウトプット：```Output```
    - ```user_id``` ユーザID
    - ```query``` クエリ
    - ```output``` アウトプット


``` python
@app.post("/process/")
def main(input_data: Input):
    # 入力データを取得
    user_id = input_data.user_id
    query = input_data.query

    # 文字列の加工処理
    processed_text = f"Processed _ User ID: {user_id}, Query: {query}"

    # 処理結果と入力データをまとめて返す
    output_data = Output(user_id=user_id, query=query, output=processed_text)
    return output_data

```

```simple_post.py```で定義したエンドポイント「/process/」へユーザリクエストがPOST（送信）されるので、それを処理するための関数が「main」です。

今回は単純な文字列の処理をするだけ（Processed_という文字列をqueryに追加するだけ）で、その結果をさきほど定義していた```Output```オブジェクトに代入し、```output_data```を作っています。

最後にこの```output_data```をリターンする、という処理です。こうすることで、```simple_post.py```に```output_data```データが返されることになります。

### 動かしてみる
ここまでで作ったものを動かしてみましょう。

まず、```api_simple_post.py```を置いているディレクトリ(フォルダ)へ移動し、次のコマンドを打ちます。

``` powershell
uvicorn api_simple_post:app --reload
```
実行して次のような画面が出てきたらOKです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/70da9bd5-a883-3877-d80c-0b5cff57c39f.png)

次にまた、別のコマンドプロンプト(MacやLinuxであればターミナル)を起動し、```simple_post.py```を置いているディレクトリへ移動します。

そしたら次のコマンドで```simple_post.py```を実行させます。

``` powershell
python simple_post.py
```
次のような実行画面が出てきたら成功です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/7a104415-8f47-b05e-9a9d-8611b960aa66.png)

これは、
1. ```simple_post.py```から```user_id```と```query```をAPI経由でデータを送信（POST）する
2. ```api_simple_post.py```側で```user_id```と```query```を受け取り、```query```の文字列に「Processed_」を追加して、```Output```へ格納する。
3. ```Output```データが```output_data```としてAPI呼び出し元の```simple_post.py```に返し、JSON形式でprintする
という処理が行われています。


## ②実装：DBを追加する
作成するもののイメージは次の通りです。

![図5.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/3dfcf581-c204-97e7-8382-8941570b533d.png)


### APIにリクエストを投げるパート

``` python:post_database.py
import requests
import json
import os
import sqlite3
import pandas as pd


url = "http://localhost:8000/process/"  # POSTリクエストを送信するエンドポイントのURL

data = {
    "user_id": "example_user",
    "query"  : "example_query"
}

response = requests.post(url, json=data)

# レスポンスの内容を表示
#print(response.json())

# ステータスコードを取得
status_code = response.status_code
print("status : " + str(status_code))


# データベースファイルのパス
DB_PATH = 'simple_database.db'

# データベースに接続
conn   = sqlite3.connect(DB_PATH)

# SQLクエリを実行し、データを取得
query = "SELECT * FROM data"
df    = pd.read_sql_query(query, conn)


# データベース接続を閉じる
conn.close()


# データフレームを表示
print(df)

```

コードを解説します。

``` python
# ステータスコードを取得
status_code = response.status_code
print("status : " + str(status_code))
```
ここは、APIにリクエストを投げる(POST)し、リクエストのステータスを表示するためのパートです。APIリクエストに成功していれば、200と出ます。

``` python
# データベースファイルのパス
DB_PATH = 'simple_database.db'

# データベースに接続
conn   = sqlite3.connect(DB_PATH)

# SQLクエリを実行し、データを取得
query = "SELECT * FROM data"
df    = pd.read_sql_query(query, conn)

# データベース接続を閉じる
conn.close()
```
ここは次に作る```api_post_database.py```側で生成したDBへ接続し、そのDBの中身を読み取るためのコードです。読み取った内容をpandasのDataFrame形式に変換し、そのあと、DBとの接続を閉じています。


### APIと受け取ったクエリを処理させるパート

``` python:api_post_database.py
from fastapi import FastAPI
from pydantic import BaseModel

import sqlite3
import os

app = FastAPI()

class Input(BaseModel):
    user_id: str
    query: str

class Output(BaseModel):
    user_id: str
    query: str
    output: str

# データベースファイルのパス
DB_PATH = 'simple_database.db'


@app.post("/process/")
def main(input_data: Input):
    # 入力データを取得
    user_id = input_data.user_id
    query   = input_data.query


    # 文字列の加工処理
    processed_text = f"Processed_ User ID: {user_id}, Query: {query}"


    # データベースの存在を判定
    if os.path.exists(DB_PATH):
		# データベースが存在する場合
        conn   = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
    else:
        # データベースが存在しない場合、新しく作成する
        conn   = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
	    
	    # テーブルの作成
        cursor.execute('''CREATE TABLE data
                      (USER_ID TEXT, QUERY TEXT, OUTPUT TEXT)''')

	# データの挿入
    cursor.execute("INSERT INTO data VALUES (?, ?, ?)" ,(user_id, query, processed_text))

	# 変更を確定させる
    conn.commit()

	# データベース接続を閉じる
    conn.close()

    # 処理結果と入力データをまとめて返す
    output_data = Output(user_id=user_id, query=query, output=processed_text)
    return output_data
```

①で作成した```api_simple_post.py```との差異は次の箇所です。

``` python
# データベースの存在を判定
    # データベースの存在を判定
    if os.path.exists(DB_PATH):
		# データベースが存在する場合
        conn   = sqlite3.connect(DB_PATH) # DBとの接続を確立
        cursor = conn.cursor()
    else:
        # データベースが存在しない場合、新しく作成する
        conn   = sqlite3.connect(DB_PATH) # DBとの接続を確立
        cursor = conn.cursor()
	    
	    # テーブルの作成
        cursor.execute('''CREATE TABLE data
                      (USER_ID TEXT, QUERY TEXT, OUTPUT TEXT)''')

	# データの挿入
    cursor.execute("INSERT INTO data VALUES (?, ?, ?)" ,(user_id, query, processed_text))

	# 変更を確定させる
    conn.commit()

	# データベース接続を閉じる
    conn.close()
```

まず、```DB_PATH = 'simple_database.db'```というデータベースのパスの有無を判断し、
- DBが無い場合はDBとの接続を作成/確立してからDBを新規作成します
- 有る場合は、そのDBとの接続を確立します

そのあと、```cursor.execute("INSERT INTO data VALUES (?, ?, ?)" ,(user_id, query, processed_text))```で、user_idとquery、processed_textをDBに格納します。

上記の処理ではまだDBへの変更が確定されていないので（このあたりはDBの処理の細かい話になるので、詳しくは解説しないですが気になる方は、「DB コミットメント」「DB ロールバック」などで調べてみてください）、```conn.commit()```で処理を確定させます。

最後に```conn.close()```でDBとの接続を閉じます。


### 動かしてみる
ここまでで作ったものを動かしてみましょう。

まず、```api_post_database.py```を置いているディレクトリ(フォルダ)へ移動し、次のコマンドを打ちます。

``` powershell
uvicorn api_post_database:app --reload
```
実行して次のような画面が出てきたらOKです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/887ffc11-bbb6-8ae7-b922-b56898deb012.png)


次にまた、別のコマンドプロンプト(MacやLinuxであればターミナル)を起動し、```post_database.py```を置いているディレクトリへ移動します。

そしたら次のコマンドで```post_database.py```を実行させます。

``` powershell
python post_database.py
```
次のような実行画面が出てきたら成功です。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/f47335a5-1c44-c883-c8de-6885c6061c00.png)

今回は、ユーザからのクエリを受け付けると、DBに1行追加していく処理を実装しているので、もう一度```python post_database.py```を実行すると、DBが1行増えて2行になっていることがわかります。（1回実行するたびに1行追加されていきます。追加したものを削除する処理を実装することもできますが今回は解説しません）

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/9c311657-3593-0f9d-0d7e-cd2043b46233.png)


また、```api_post_database.py```と同じディレクトリに```simple_database.db```というファイルができていることを確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/a1e975de-befd-5acc-934a-b1b429375510.png)



これは、
1. ```post_database.py```から```user_id```と```query```をAPI経由でデータを送信（POST）する
2. ```api_post_database.py```側で```user_id```と```query```を受け取り、```query```の文字列に「Processed_」を追加する
3. ```api_post_database.py```は、DBの有無を確認し、DBが無ければDBを新規作成します。
4. DBの中に```user_id```と```query```、```processed_text```をDBに（1行追加して）格納します
5. ```Output```オブジェクトにも```user_id```と```query```、```processed_text```を格納します。
6. ```Output```データが```output_data```としてAPI呼び出し元の```post_database.py```に返す。
7. ```post_database.py```側でDBへ接続し、DBの中身をPandasのDataFrameに入れて、その中身をprintで表示します

という処理が行われています。





## ③実装：③ChatGPTを追加する
ようやくChatGPTを導入します。ここで実装するもののイメージは次の通りです。

![図7.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/f97f014d-6047-8f3c-2ed9-3e35943fc406.png)

### APIにリクエストを投げるパート

``` python:post_main.py
import requests
import json
import os
import sqlite3
import pandas as pd


url = "http://localhost:8000/process/"  # POSTリクエストを送信するエンドポイントのURL

data = {
    "user_id": "example_user",
    "query"  : "最初に学ぶプログラミング言語でおすすめを理由と合わせて3つ教えてください"
}

response = requests.post(url, json=data)

# レスポンスの内容を表示
#print(response.json())

# ステータスコードを取得
status_code = response.status_code
print("status : " + str(status_code))


# データベースファイルのパス
DB_PATH = 'chat_history.db'

# データベースに接続
conn   = sqlite3.connect(DB_PATH)

# SQLクエリを実行し、データを取得
query = "SELECT * FROM data"
df    = pd.read_sql_query(query, conn)




# データベース接続を閉じる
conn.close()



# データフレームを表示
print(df)



# chatGPTの回答を一覧で確認する
for _ in range(3):
	print()

for output in df["OUTPUT"].values.tolist():
	print(output)
```

基本的に②で作成した```post_database.py```と同じ実装です。ここでは差異の部分を解説します。

``` python
data = {
    "user_id": "example_user",
    "query"  : "最初に学ぶプログラミング言語でおすすめを理由と合わせて3つ教えてください"
}
```

ここの```query```にChatGPTに聞いてみたいことを書きます。今回はおすすめのプログラミング言語を聞いてみることにします。


``` python
# chatGPTの回答を一覧で確認する
for _ in range(3):
	print()

for output in df["OUTPUT"].values.tolist():
	print(output)
```
ここは、DBに格納されたChatGPTの回答 ```[OUTPUT]```カラムの中身だけを表示させるための処理です。特殊なことは何もしていないです。



### APIと受け取ったクエリを処理させるパート

``` python:api_main.py
from fastapi import FastAPI
from pydantic import BaseModel

import sqlite3
import os

app = FastAPI()

class Input(BaseModel):
    user_id: str
    query: str

class Output(BaseModel):
    user_id: str
    query: str
    output: str

# データベースファイルのパス
DB_PATH = 'chat_history.db' # データベース名をchat_historyへ変更


#################################################
############## ↓ChatGPTの処理↓ ###################
#################################################
def chatgpt(query):
    # chatGPTを使って回答を生成するための処理
    if not query:
        # クエリが空だった時は、NULLを文字列として返す
        return "NULL"
    else:
        # クエリに入力されているときは、ChatGPTで返答を作る
        import openai
        openai.api_key = 'OpenAI APIキー'

        response=openai.ChatCompletion.create(
        model = "gpt-3.5-turbo",
          messages = [{"role": "user", "content": query}]
        )

        return response['choices'][0]['message']['content']


#################################################
############## ↓ここからメインの処理↓ #################
#################################################
@app.post("/process/")
def main(input_data: Input):
    # 入力データを取得
    user_id = input_data.user_id
    query   = input_data.query


    # 文字列の加工処理
    processed_text = chatgpt(query)


    # データベースの存在を判定
    if os.path.exists(DB_PATH):
		# データベースが存在する場合
        conn   = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
    else:
        # データベースが存在しない場合、新しく作成する
        conn   = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()
	    
	    # テーブルの作成
        cursor.execute('''CREATE TABLE data
                      (USER_ID TEXT, QUERY TEXT, OUTPUT TEXT)''')

	# データの挿入
    cursor.execute("INSERT INTO data VALUES (?, ?, ?)" ,(user_id, query, processed_text))

	# 変更を確定させる
    conn.commit()

	# データベース接続を閉じる
    conn.close()

    # 処理結果と入力データをまとめて返す
    output_data = Output(user_id=user_id, query=query, output=processed_text)
    return output_data
```

②で作成した```api_post_database.py```をベースにChatGPTの処理を追加しています。

``` python
#################################################
############## ↓ChatGPTの処理↓ ###################
#################################################
def chatgpt(query):
    # chatGPTを使って回答を生成するための処理
    if not query:
        # クエリが空だった時は、NULLを文字列として返す
        return "NULL"
    else:
        # クエリに入力されているときは、ChatGPTで返答を作る
        import openai
        openai.api_key = 'OpenAI APIキー'

        response=openai.ChatCompletion.create(
        model = "gpt-3.5-turbo",
          messages = [{"role": "user", "content": query}]
        )

        return response['choices'][0]['message']['content']
```
if-elseがあるのでややこしいように見えますが、やっていることは単純です。
ユーザからの```query```が空の場合もあり得るため、その場合は、文字列として"NULL"を入れるようにしています。```query```が空ではない場合は、GPT3.5-turboにクエリをプロンプトとしてそのまま投入しています。
その回答をreturn しています。

あとは、DBの有無の判定をし、DBに対して、```user_id```、```query```、```processed_text```を格納し、```Output```オブジェクトとしてAPIの呼び出し元にリターンしてます。


### 動かしてみる
ここまでで作ったものを動かしてみましょう。





まず、```api_main.py```を置いているディレクトリ(フォルダ)へ移動し、次のコマンドを打ちます。

``` powershell
uvicorn api_main:app --reload
```
実行して次のような画面が出てきたらOKです。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/218a979d-d0b6-688e-bb5d-f73133e33670.png)




次にまた、別のコマンドプロンプト(MacやLinuxであればターミナル)を起動し、```post_main.py```を置いているディレクトリへ移動します。

そしたら次のコマンドで```post_main.py```を実行させます。

``` powershell
python post_main.py
```
次のような実行画面が出てきたら成功です。注意：①②とは異なり、少し時間がかかります。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/95a7113b-cbd3-632b-74c5-57b33c9eb467.png)




また、```api_main.py```と同じディレクトリに```chat_history.db```というファイルができていることを確認できます。


![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/687015/c54fe053-7022-d243-104d-9cf1dd4e869c.png)



これは、
1. ```post_database.py```から```user_id```と```query```をAPI経由でデータを送信（POST）する
2. ```api_post_database.py```側で```user_id```と```query```を受け取り、ChatGPTが回答を生成、```processed_text```へ格納する
3. ```api_post_database.py```は、DBの有無を確認し、DBが無ければDBを新規作成します。
4. DBの中に```user_id```と```query```、```processed_text```をDBに（1行追加して）格納します
5. ```Output```オブジェクトにも```user_id```と```query```、```processed_text```を格納します。
6. ```Output```データが```output_data```としてAPI呼び出し元の```post_database.py```に返す。
7. ```post_database.py```側でDBへ接続し、DBの中身をPandasのDataFrameに入れて、その中身をprintで表示します

という処理が行われています。




# まとめ

今回は、FastAPIを使って、API経由でChatGPTとユーザとのやりとりをDBに保存する仕組みを作りました。これ自体は非常にシンプルなアプリケーションですが、これを発展させていくことで、

- 過去の自分のやり取り内容をDBへ記憶させ、その記憶を基に自分用（自分の会社用）のチャットボットを作ることができる
    - マーケティングの領域であれば、お客様とのやりとりをDBに格納しておけば、過去のお客様とのやり取りを基にChatGPTが回答を生成してくれるので、よりパーソナライズされたチャットボットができあがり、顧客体験の向上などに活用することができます
- 専門知識などをDB化しておけば、その知識を基にChatGPTが回答を生成するFAQアプリケーションを作ることができる
    - 会社のルールなどがテキスト化されているのであれば、それをDBに格納しておけば、そのルールを基にChatGPTが回答を作ってくれるので、社内で不明点があったらChatGPTに聞いてみる、といった活用方法が考えられます
    - 保険や不動産、ケータイなどルールや契約形態が複雑で難しいような業種であれば、各種データをDB化しておくことで、ChatGPTがそのDBを読み取って、お客様からの問い合わせに回答してくれるFAQを実現することができます。

といった応用先に使うことができます。


個人的には、サーバとクライアント間でのやりとりをAPIでさせることを初めて実装したので、そういうものを作った経験として勉強になりました。