JA ▾
API キーを取得

GLM API Python: チャット、ストリーミング、ツール例

GLM API Python の開発には標準的な openai パッケージのみが必要です。https://api.glmapi.cc/v1 を指定し、モデル ID に uncensored を使用すれば、通常のチャット呼び出しが機能します。このページでは、最初のリクエストから非同期バッチ、レート制限処理まで、実行可能なコード付きで各パターンを解説します。

主要ポイント

  • openai をインストールし、base_url を https://api.glmapi.cc/v1 に設定して、モデル uncensored を使用します。
  • ストリーミング、JSON モード、関数呼び出しは、OpenAI チャット API と同じパラメータを使用します。
  • フルキーは同時リクエスト 8 件、1 分あたり 300 件まで許可するため、非同期ワーカーを 8 に制限してください。
  • 429 エラーは、エラーコードと Retry-After ヘッダーを読み取ることで処理します。
ラップトップのターミナルからクラウドサーバーへストリーミングされる光の帯状データブロックは、Python API 呼び出しを表しています

インストールと最初のリクエスト

仮想環境を作成し、SDK をインストールします。openai パッケージのバージョン 1 以降を前提としています。

python -m venv .venv
source .venv/bin/activate
pip install --upgrade openai

API キーページからキーを取得してください。新規ユーザーはワンクリックで 7 日間有効、クレジットカード不要の $0.50 無料トライアルクレジットを取得できます。キーを環境変数に保存し、最初の呼び出しを行います:

import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.glmapi.cc/v1",
    api_key=os.environ["GLM_API_KEY"],
)

resp = client.chat.completions.create(
    model="uncensored",
    messages=[
        {"role": "system", "content": "You are a concise writing assistant."},
        {"role": "user", "content": "Suggest five titles for a space-western novel."},
    ],
    temperature=0.8,
)

print(resp.choices[0].message.content)
print(resp.usage)

これで統合は完了です。このサービスは 1 つのオープンウェイトモデル向けの独立した OpenAI 互換エンドポイントであり、GLM の提供元ベンダーやその他のベンダーの公式 API ではありません。存在するのは POST /v1/chat/completions と GET /v1/models のみです。サポートされるリクエストフィールドは クイックスタート にリストされています。

小さな習慣を最初から身につけることで、効率が上がります。まず、スモークテストでclient.models.list()を1回呼び出します。これはGET /v1/modelsにアクセスするため、完了処理にトークンを1つ消費する前に、APIキー、ベースURL、ネットワークパスがすべて機能することを確認できます。次に、開発中にresp.usageを出力します。プロンプトと完了のトークン数を表示することで、システムプロンプトが想定より大きすぎないかを早期に確認でき、入力が1Mトークンあたり$0.15、出力が1Mトークンあたり$0.50の場合、呼び出しのコストを頭の中で計算できます。

多ターンチャットとサンプリング設定

API はステートレスなので、会話リストを自分で保持し、毎回送信します。小さなクラスで十分です:

class Chat:
    def __init__(self, client, system, max_turns=12):
        self.client = client
        self.system = system
        self.max_turns = max_turns
        self.history = []

    def say(self, text):
        self.history.append({"role": "user", "content": text})
        window = self.history[-self.max_turns * 2:]
        resp = self.client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "system", "content": self.system}] + window,
            temperature=0.9,
            top_p=0.95,
            frequency_penalty=0.3,
        )
        reply = resp.choices[0].message.content
        self.history.append({"role": "assistant", "content": reply})
        return reply

スライス self.history[-self.max_turns * 2:] は最後の 12 組のユーザーとアシスタントの対を保持します。実際のアプリケーションでは、キャラクターシートや重要な事実など、初期のメッセージをいくつか固定し、中間を要約することもあります。最後の 12 ターンに切り捨てることで、会話の拡大に伴う入力コストを一定に保てます。コンテキストウィンドウはプロンプトと完了を合わせて 100,000 トークンなので、これは必要性よりもコスト選択です。パラメータ temperature、top_p、stop、seed およびペナルティは渡されます。プロンプトのテスト中に再現可能な実行を望む場合は seed を設定し、長さを制限したい場合のみ max_tokens を指定してください。

ストリーミングレスポンス

ストリーミングはサーバー送信イベントを使用し、生成されたトークンを逐次提供します。これは対話型ツールにとって重要です。stream=True を渡してイテレートします:

stream = client.chat.completions.create(
    model="uncensored",
    messages=[{"role": "user", "content": "Describe a harbour at dawn in 120 words."}],
    stream=True,
)

parts = []
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta.content
    if delta:
        parts.append(delta)
        print(delta, end="", flush=True)

text = "".join(parts)

最終チャンクには使用情報が含まれ、空のchoicesリストを持つ可能性があるため、chunk.choicesのガードが重要です。運用上の注意点として、ストリーミングリクエストはストリームが終了するまで同時実行のスロットを保持します。クライアントの切断など、コードが途中で例外を発生させた場合に備え、ストリームを閉じるfinally節を持つtryブロックでループを囲んでください。ストリームは必ず終了またはクローズしてください。ストリームを放棄するループは、有用な処理を行わずに同時実行制限に達する可能性があります。

JSON モード

出力が他のコードにフィードされる場合は、JSON オブジェクトを要求してください。プロンプトに JSON を明記し、モデルに期待する形状を知らせます:

import json

resp = client.chat.completions.create(
    model="uncensored",
    response_format={"type": "json_object"},
    messages=[
        {"role": "system", "content": "Reply with a JSON object only."},
        {"role": "user", "content": "Give a character with keys name, age, motto."},
    ],
)

data = json.loads(resp.choices[0].message.content)
print(data["name"])

本番コードでは、json.loads を try ブロックで囲むことを忘れないでください。JSON モードは有効な出力を生成しやすくしますが、停止シーケンスや独自の制限によってリクエストが中断されると、切り捨てられたオブジェクトが残る場合があります。同じ入力でもう 1 回リトライするのは低コストな修正であり、失敗したリクエストには課金されないため、リトライのコストは成功した呼び出し分のみです。

関数呼び出し

ツール呼び出しは OpenAI の形式に従います:tools で関数を記述し、モデルは tool_calls を返します。それらを実行し、role tool で結果を送り返します。完全な往復:

import json

tools = [{
    "type": "function",
    "function": {
        "name": "roll_dice",
        "description": "Roll a die with the given number of sides.",
        "parameters": {
            "type": "object",
            "properties": {"sides": {"type": "integer"}},
            "required": ["sides"],
        },
    },
}]

messages = [{"role": "user", "content": "Roll a twenty-sided die for me."}]
first = client.chat.completions.create(
    model="uncensored", messages=messages, tools=tools, tool_choice="auto"
)
msg = first.choices[0].message

if msg.tool_calls:
    messages.append(msg)
    for call in msg.tool_calls:
        args = json.loads(call.function.arguments)
        result = {"value": 17, "sides": args["sides"]}  # replace with real logic
        messages.append({
            "role": "tool",
            "tool_call_id": call.id,
            "content": json.dumps(result),
        })
    final = client.chat.completions.create(
        model="uncensored", messages=messages, tools=tools
    )
    print(final.choices[0].message.content)
else:
    print(msg.content)

ループはサイコロの詳細よりも重要です。モデルが呼び出しを要求し、あなたのコードがそれを実行し、アシスタントメッセージとツール結果の両方を追加してから再度要求します。アシスタントメッセージをスキップすると、結果の tool_call_id が一致する対象を失い、これが混乱した 2 回目の応答の主な原因です。tool_choice を auto に設定すると、モデルはツールを呼び出すかどうかを判断するため、上記のコードのように、直接応答する分岐を常に処理してください。ストリーミングレスポンスには tool_calls も含まれる可能性があり、引数を解析する前にインデックスで結合する必要があります。

実行する前に引数を検証してください。モデルは不正な呼び出しを生成する可能性があり、あなたのコードは引数をユーザーからの入力と同様に信頼できない入力として扱うべきです。

制限内での非同期バッチ

複数のプロンプトを同時に処理するには、同時実行制限に一致するセマフォ付きの AsyncOpenAI を使用します。フルキーは同時リクエスト 8 件、トライアルキーは 2 件まで許可します:

import asyncio, os
from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.glmapi.cc/v1",
    api_key=os.environ["GLM_API_KEY"],
)
gate = asyncio.Semaphore(8)  # use 2 on a trial key

async def one(prompt):
    async with gate:
        r = await client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "user", "content": prompt}],
        )
        return r.choices[0].message.content

async def main():
    prompts = [f"Write a one-line tagline for product {i}." for i in range(40)]
    results = await asyncio.gather(*(one(p) for p in prompts))
    print(len(results))

asyncio.run(main())

セマフォは同時実行数を制御しますが、1分あたりの300リクエストという上限は制御しません。長時間のバッチ処理では、小さな遅延またはトークンバケット方式の制限子を設けてください。入力トークン100万あたり$0.15、出力トークン100万あたり$0.50の場合、約60トークンの入力と25トークンの出力を持つ40件のキャッチコピー生成リクエストの合計コストは$0.00036 + $0.0005 = $0.00086となり、バッチ設計を左右するのは価格ではなく制限です。チャージについては価格ページをご覧ください。

レート制限、リトライ、ヘッダー

制限を超えると、API は HTTP 429 を返します。エラーコードは、1 分あたりのリクエストが多すぎる場合は rate_limited、一度に多すぎる場合は concurrency です。一時的な過負荷も 429 を返し、Retry-After: 5 ヘッダーを付与します。SDK は一部の失敗を自動でリトライしますが、クライアントに max_retries を設定することもできます。ただし、明示的な処理の方が明確です:

import time
from openai import RateLimitError

def ask(prompt, tries=4):
    for attempt in range(tries):
        try:
            raw = client.chat.completions.with_raw_response.create(
                model="uncensored",
                messages=[{"role": "user", "content": prompt}],
            )
            print("balance:", raw.headers.get("X-Balance-USD"))
            return raw.parse().choices[0].message.content
        except RateLimitError as e:
            wait = float(e.response.headers.get("Retry-After", 2 ** attempt))
            time.sleep(wait)
    raise RuntimeError("still rate limited")

ここで with_raw_response は X-Balance-USD、X-Request-Id およびレート制限ヘッダーへのアクセスを提供します。報告したいエラーにはリクエスト ID をログに記録してください。キーをコミットされるノートブックやフロントエンドコードに入れないでください。

クレジットを使わずにテストし、よくある間違い

Python 統合の大部分は配線(plumbing)であり、ライブモデルは必要ありません。ネットワーク呼び出しを1つの関数の背後に保持し、テストでそれを置き換えることができます。標準ライブラリのみを使用して:

from unittest.mock import MagicMock

def reply_text(client, prompt):
    r = client.chat.completions.create(
        model="uncensored",
        messages=[{"role": "user", "content": prompt}],
    )
    return r.choices[0].message.content.strip()

def test_reply_text_strips_whitespace():
    fake = MagicMock()
    fake.chat.completions.create.return_value.choices = [
        MagicMock(message=MagicMock(content="  hello  "))
    ]
    assert reply_text(fake, "hi") == "hello"

エッジでのみ数回の実際の呼び出しを実行します。デプロイごとに 1 回のスモークテスト、および保存したプロンプトセットの定期的な実行です。入力が約 100 トークン、出力が 20 トークンのスモークテスト 1 回あたりのコストは $0.000015 + $0.00001 = $0.000025 で、これは微々たるものです。失敗したリクエストは課金されません。$0.50 の無料トライアルクレジットは、テストスイートに必要な金額をはるかに上回ります。

プロンプトの作業では、seed と temperature を固定し、システムプロンプトの2つのバージョンを同等の条件で比較できるようにします。実行間でプロンプトとサンプリング設定の両方を変更すると、結果を読み取れなくなります。各実行の出力を X-Request-Id ヘッダーからのリクエスト ID とともに保存し、後で驚くべき回答を追跡できるようにします。

よくある Python の間違い

Python ユーザーからのサポート質問の大部分を引き起こすものは次の通りです:

  • キーを OPENAI_API_KEY としてのみ渡すこと。 コード内で base_url を設定してもキー変数を忘れると、SDK は空または間違ったキーを送信します。api_key を明示的に渡してください。
  • /v1 を忘れること。 base URL は /v1 で終わります。SDK は /chat/completions を自動で追加します。
  • 他のサービスのモデル名を使用すること。 ここで有効なのは uncensored のみです。
  • 同時実行数に上限なし。 セマフォなしで1000以上のコルーチンに対してasyncio.gatherを実行すると、concurrencyエラーの山が発生します。
  • ストリーミング応答を 1 つの文字列として扱うこと。 デルタを結合し、空の choices リストを持つチャンクをスキップします。
  • コンテキスト制限を無視すること。 履歴と完了は 100,000 トークン以内に収まる必要があります。それに達する前に古いターンを切り捨ててください。

チャージと現在の料金表については、価格ページ を参照してください。このサービスはテキストのみなので、画像、音声、埋め込み呼び出しを接続する必要はなく、プロンプトはトレーニングに使用されません。

質問と回答

どの Python パッケージが必要ですか?

標準的なopenaiパッケージ(バージョン1以降)。ベースURLをhttps://api.glmapi.cc/v1に設定し、モデルをuncensoredに設定します。

ストリーミングは機能しますか?

はい。stream=True を渡してチャンクをイテレートしてください。ストリームは使用情報で終了し、完了するまで同時実行スロットを占有します。

並行して何件のリクエストを実行できますか?

フルキーは同時リクエスト 8 件、1 分あたり 300 件まで許可します。トライアルキーは 2 件と 60 件です。

代わりに LangChain や LiteLLM を使用できますか?

Base URL、API キー、モデル名「uncensored」を指定すれば、両方とも OpenAI 互換のエンドポイントと連携できます。正確なパラメータ名は各プロジェクトのドキュメントをご確認ください。

キーはフォーム 1 つで手に入ります

アカウントを作成し、キーをコピーし、ベースURLを変更します。これでセットアップは完了です。

APIキーを取得ドキュメントを読む