はじめに
前回の記事で自宅のパソコンにgpt-oss-20をインストールしましたが、せっかくならチャットだけでなく何か開発周りもやりたいと思い、まずはローカル環境でAPIとして利用できるようにしました。
LM StudioでローカルAPIを起動する
まずはLM Studioを起動します。
LM Studioを起動したら、緑の「Developer」ボタンをクリックします。
画面上側のモデル選択で「gpt-oss-20b」を選択します。
画面左上の「Status」をONにして、APIを起動します。
こちらで起動できました！
APIを叩いてみる
今回はPOSTMANから叩いて、正しく起動できたかを確認します。
bodyはjson形式で、プロンプトを含めます。準備ができたらURI右横の「Send」ボタンを押します。
無事にレスポンスが返ってきて、起動していることが確認できました！
Pythonプログラムから利用する
Pythonで入力されたプロンプトをAPIへ送り、レスポンスのコンテンツだけを表示するようにしました。
実行例
コード
gpt_oss_local_api.py
import os
import sys
import logging
from logging.handlers import RotatingFileHandler
import requests
import json
# ------------------------------------------------------------------
# 設定の読み込み
# ------------------------------------------------------------------
def load_config(file_path: str):
if not os.path.exists(file_path):
logging.error(f"設定ファイルが見つかりません: {file_path}")
raise FileNotFoundError(f"設定ファイルが見つかりません: {file_path}")
try:
with open(file_path, 'r', encoding='utf-8') as f:
return json.load(f)
except json.JSONDecodeError as e:
logging.error(f"設定ファイルのパースエラー: {e}")
raise
except Exception as e:
logging.error(f"設定ファイルを読み込む際にエラーが発生しました: {e}")
raise
# 設定をグローバル変数として読み込む
config = load_config('config/setting.json')
# ------------------------------------------------------------------
# ログ設定
# ------------------------------------------------------------------
def setup_logging():
log_max_bytes = config['log_max_bytes_mb'] * 1024 * 1024 # MBをバイトに変換
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)-7s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
handlers=[
RotatingFileHandler(config['log_file'], maxBytes=log_max_bytes, backupCount=config['log_backup_count'], encoding='utf-8')
]
)
# ------------------------------------------------------------------
# 入力の検証関数
# ------------------------------------------------------------------
def validate_input(user_input: str) -> bool:
if not user_input:
logging.warning("入力が空です。")
return False
if len(user_input) > 500: # 例: 最大500文字まで
logging.warning("入力が長すぎます。最大500文字までです。")
return False
return True
# ------------------------------------------------------------------
# LMStudio APIを利用して応答を生成する関数
# ------------------------------------------------------------------
def get_lmstudio_response(prompt: str) -> str:
url = config['api_url']
headers = {"Content-Type": "application/json"}
data = {"messages": [{"role": "user", "content": prompt}]}
try:
response = requests.post(url, json=data, headers=headers)
response.raise_for_status() # HTTPエラーが発生した場合は例外をスロー
logging.info(f"APIレスポンス: {response.json()}")
choices = response.json().get("choices", [])
if choices:
return choices[0].get("message", {}).get("content", "").strip()
logging.error("APIレスポンスに'choices'が含まれていません。")
return "（エラー: 応答が得られませんでした）"
except requests.RequestException as e:
logging.error(f"LMStudio API呼び出し中にエラーが発生しました: {e}")
return ""
# ------------------------------------------------------------------
# メイン処理
# ------------------------------------------------------------------
def main():
setup_logging()
if len(sys.argv) > 1: # 外部プログラムから呼ばれた場合
user_input = sys.argv[1]
if validate_input(user_input):
reply = get_lmstudio_response(user_input)
return reply
else:
return "無効な入力です。"
else: # 単体実行の場合
print("\nメッセージを入力してください")
print("終了するには Ctrl+C か 'exit' と入力してください。\n")
while True:
try:
user_input = input(">>> ").strip()
if not validate_input(user_input):
print("無効な入力です。再度入力してください。")
continue
if user_input.lower() == "exit":
logging.info("'exit' が入力されたため終了します。")
break
reply = get_lmstudio_response(user_input)
if reply:
print(reply)
logging.info(f"レスポンス: {reply}")
else:
print("（エラーまたは応答なし）")
logging.info("レスポンス: （エラーまたは応答なし）")
except KeyboardInterrupt:
logging.info("ユーザーによる中断（Ctrl+C）を検知しました。終了します。")
break
except Exception as e:
logging.error(f"予期しないエラーが発生しました: {e}")
if __name__ == "__main__":
try:
result = main()
if result is not None:
print(result) # 外部から呼ばれた場合の応答を出力
except Exception as e:
logging.error(f"致命的なエラーが発生しました: {e}")
sys.exit(1)
設定は別途jsonファイルに記述しています。
setting.json
{
"log_file": "gpt_oss_local_api.log",
"log_max_bytes_mb": 10,
"log_backup_count": 5,
"api_url": "http://localhost:1234/v1/chat/completions"
}
テスト
pytestモジュールを使って動作確認をしていきます。
他のプログラムからモジュールとして利用する場合の挙動確認を兼ねています。
ユニットテスト
import json
import gpt_oss_local_api # 本番コードをインポート
import logging
import pytest
from unittest.mock import patch, mock_open
# 試験用定数の定義
LOG_MAX_BYTES_MB = 10
LOG_FILE = "gpt_oss_local_api.log"
LOG_BACKUP_COUNT = 5
API_URL = "http://localhost:1234/v1/chat/completions"
# モックデータの定義
MOCK_CONFIG_DATA = json.dumps({
"log_file": LOG_FILE,
"log_max_bytes_mb": LOG_MAX_BYTES_MB,
"log_backup_count": LOG_BACKUP_COUNT,
"api_url": API_URL
})
# load_configのテスト
@patch("builtins.open", new_callable=mock_open, read_data=MOCK_CONFIG_DATA)
def test_load_config_success(mock_file):
config = gpt_oss_local_api.load_config("./config/setting.json")
# 設定内容をログ出力
logging.info(f"読み込まれた設定: {config}")
assert config['log_max_bytes_mb'] == LOG_MAX_BYTES_MB
assert config['log_file'] == LOG_FILE
assert config['log_backup_count'] == LOG_BACKUP_COUNT
assert config['api_url'] == API_URL
@patch("builtins.open", new_callable=mock_open, read_data=MOCK_CONFIG_DATA)
def test_load_config_json_decode_error(mock_file):
# 最後の閉じ括弧を削除して不正にする
mock_file.return_value.read.side_effect = MOCK_CONFIG_DATA[:-1]
with pytest.raises(json.JSONDecodeError):
gpt_oss_local_api.load_config("./config/setting.json")
# validate_inputのテスト
def test_validate_input():
assert gpt_oss_local_api.validate_input("有効な入力")
assert not gpt_oss_local_api.validate_input("") # 空の入力
assert not gpt_oss_local_api.validate_input("a" * 501) # 長すぎる入力
# get_lmstudio_responseのテスト
@patch('requests.post') # requests.postをモック
def test_get_lmstudio_response_success(mock_post):
mock_post.return_value.status_code = 200
mock_post.return_value.json.return_value = {
"id": "chatcmpl-jcgzm5sr5ydv2nf6s7pb",
"object": "chat.completion",
"created": 1754827971,
"model": "openai/gpt-oss-20b",
"choices": [
{
"index": 0,
"logprobs": None,
"finish_reason": "stop",
"message": {
"role": "assistant",
"content": "テスト応答" # 期待される応答
}
}
],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 10,
"total_tokens": 20
},
"stats": {},
"system_fingerprint": "openai/gpt-oss-20b"
}
response = gpt_oss_local_api.get_lmstudio_response("テストプロンプト")
assert response == "テスト応答" # 期待される応答と比較
@patch('requests.post') # requests.postをモック
def test_get_lmstudio_response_failure(mock_post):
mock_post.return_value.status_code = 500 # ステータスコードを500に設定
mock_post.return_value.json.return_value = {} # 空のJSONを返す
response = gpt_oss_local_api.get_lmstudio_response("テストプロンプト")
assert response == "（エラー: 応答が得られませんでした）"
以下のようにテストが全てPASSEDになればOKです。
結合テスト
ユニットテストが完了したら、LM StudioのAPIを叩いて動作するかテストします。
import sys
import os
# モジュールのパスを追加
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../')))
import gpt_oss_local_api # 本番コードをインポート
# 本番APIのURL
API_URL = "http://localhost:1234/v1/chat/completions"
# get_lmstudio_responseのテスト
def test_get_lmstudio_response():
# テスト対象のプロンプト
prompt = "テストプロンプト"
# API呼び出し
response = gpt_oss_local_api.get_lmstudio_response(prompt)
# レスポンスの型と内容を確認
print("レスポンスの型:", type(response)) # レスポンスの型を出力
print("レスポンスの内容:", response) # レスポンスの内容を出力
# レスポンスが空でないことを確認
assert response, "レスポンスが空です。APIが正しく応答していない可能性があります。"
# レスポンスの中のメッセージの検証
assert isinstance(response, str) # レスポンスのコンテンツが文字列であることを確認
assert len(response) > 0, "レスポンスのコンテンツは1文字以上である必要があります。"
こちらもテストがPASSEDになればOKです。
おわりに
APIの準備ができたので、独自RAGや株価分析など、やってみたいことの想像が膨らんでいます。
何か作ったらまた記事にしたいと思います！