lmstudio_api_sample.py ダウンロード/コピー

lmstudio_api_sample.py をダウンロード

lmstudio_api_sample.py
lmstudio_api_sample.py
  1#!/usr/bin/env python3
  2"""
  3概要:
  4    LM Studio API sample (Python standard library only).
  5詳細説明:
  6    LM Studioのモデル一覧取得とチャットAPIのサンプルスクリプトです。
  7    python lmstudio_api_sample.py --list のように実行してモデル一覧を取得したり、
  8    python lmstudio_api_sample.py --model モデルID --prompt 質問 のようにチャットを行います。
  9関連リンク:
 10    lmstudio_api_sample_usage
 11"""
 12import argparse
 13import json
 14import sys
 15from urllib.error import HTTPError, URLError
 16from urllib.request import Request, ProxyHandler, build_opener
 17
 18
 19def request_json(base_url, endpoint, timeout, payload=None):
 20    """
 21    概要:
 22        JSON APIにHTTPリクエストを送信し、応答をパースして返します。
 23    詳細説明:
 24        システムプロキシをバイパスして直接サーバーに接続します。
 25        payloadが指定された場合はJSONとしてエンコードし、POSTリクエストを送信します。
 26    引数:
 27        :param base_url: APIのベースURL
 28        :type base_url: str
 29        :param endpoint: リクエスト先のエンドポイント
 30        :type endpoint: str
 31        :param timeout: 通信タイムアウト秒
 32        :type timeout: float
 33        :param payload: リクエストボディとして送信するデータオブジェクト
 34        :type payload: dict or None
 35    戻り値:
 36        :returns: 応答のJSON文字列をパースしたオブジェクト
 37        :rtype: dict or list
 38    """
 39    data = None if payload is None else json.dumps(payload, ensure_ascii=False).encode("utf-8")
 40    request = Request(
 41        base_url.rstrip("/") + endpoint,
 42        data=data,
 43        headers={"Content-Type": "application/json", "Accept": "application/json"},
 44    )
 45    # Connect directly to the private server, bypassing system HTTP proxies.
 46    with build_opener(ProxyHandler({})).open(request, timeout=timeout) as response:
 47        return json.load(response)
 48
 49
 50def main():
 51    """
 52    概要:
 53        コマンドライン引数を解析し、APIへのリクエストを実行します。
 54    詳細説明:
 55        引数に応じてモデル一覧の表示、またはチャットAPIへの問い合わせを行います。
 56        ネットワークエラーやパースエラーが発生した場合はエラーメッセージを標準エラー出力に書き出します。
 57    戻り値:
 58        :returns: プロセスの終了コード
 59        :rtype: int
 60    """
 61    parser = argparse.ArgumentParser(description="LM Studioのモデル一覧取得とチャットAPIのサンプル")
 62    parser.add_argument("--list", action="store_true", help="利用可能なモデルIDを表示（embeddingモデルも含む）")
 63    parser.add_argument("--model", help="問い合わせるモデルID")
 64    parser.add_argument("--prompt", help="モデルへの質問")
 65    parser.add_argument("--base-url", default="http://192.168.27.18:1234/v1", help="APIのベースURL")
 66    parser.add_argument("--timeout", type=float, default=300, help="通信タイムアウト秒（既定: 300）")
 67    parser.add_argument("--max-tokens", type=int, default=2048, help="最大生成トークン数（推論トークンを含む）")
 68    parser.add_argument("--show-reasoning", action="store_true", help="reasoning_contentも表示")
 69    args = parser.parse_args()
 70    if (args.model is None) != (args.prompt is None):
 71        parser.error("--modelと--promptは両方指定してください")
 72    if not args.list and args.model is None:
 73        parser.error("--list、または--modelと--promptを指定してください")
 74    if args.timeout <= 0 or args.max_tokens <= 0:
 75        parser.error("--timeoutと--max-tokensは正の値を指定してください")
 76    try:
 77        if args.list:
 78            result = request_json(args.base_url, "/models", args.timeout)
 79            models = result["data"]
 80            if not models:
 81                print("利用可能なモデルはありません。")
 82            for model in models:
 83                print(model["id"])
 84        if args.model is not None:
 85            result = request_json(args.base_url, "/chat/completions", args.timeout, {
 86                "model": args.model,
 87                "messages": [{"role": "user", "content": args.prompt}],
 88                "max_tokens": args.max_tokens,
 89                "stream": False,
 90            })
 91            choice = result["choices"][0]
 92            message = choice["message"]
 93            if args.show_reasoning and message.get("reasoning_content"):
 94                print("[reasoning]\n" + message["reasoning_content"] + "\n[answer]")
 95            print(message.get("content") or "")
 96            if choice.get("finish_reason") == "length":
 97                print("生成上限に達しました。必要なら--max-tokensを増やしてください。", file=sys.stderr)
 98        return 0
 99    except HTTPError as exc:
100        print(f"HTTPエラー {exc.code}: {exc.read().decode('utf-8', errors='replace')}", file=sys.stderr)
101    except (URLError, TimeoutError, OSError) as exc:
102        print(f"接続エラー: {exc}", file=sys.stderr)
103    except (ValueError, KeyError, IndexError, TypeError) as exc:
104        print(f"API応答を解釈できません: {exc}", file=sys.stderr)
105    return 1
106
107
108if __name__ == "__main__":
109    # Preserve Japanese output in Windows terminals and redirected output.
110    for stream in (sys.stdout, sys.stderr):
111        if hasattr(stream, "reconfigure"):
112            stream.reconfigure(encoding="utf-8")
113    sys.exit(main())