GLM API Python: Chat, Streaming und Tools-Beispiele

Installation und erste Anfrage

Erstelle eine virtuelle Umgebung und installiere das SDK. Die Version 1 oder höher des openai Pakets wird durchgehend vorausgesetzt.

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

Hole dir einen Schlüssel auf der API-Schlüssel-Seite; neue Nutzer können mit einem Klick ein $0,50 kostenloses Testguthaben beanspruchen, gültig für 7 Tage und ohne Kreditkarte. Speichere den Schlüssel in einer Umgebungsvariable und führe dann den ersten Aufruf aus:

Copyimport 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)

Das ist die gesamte Integration. Der Dienst ist ein unabhängiger, OpenAI-kompatibler Endpunkt für ein Open-Weight-Modell, also ist er nicht die offizielle API des Anbieters hinter GLM oder von sonst wem. Es gibt nur POST /v1/chat/completions und GET /v1/models. Die Schnellstart-Anleitung listet die unterstützten Anfragefelder auf.

Zwei kleine Gewohnheiten helfen von Anfang an. Rufe zuerst einmal client.models.list() in einem Smoke-Test auf: Er trifft GET /v1/models und beweist so, dass der Schlüssel, die Base URL und der Netzwerkpfad funktionieren, bevor du ein einziges Token für eine Completion ausgibst. Zweitens gib resp.usage während der Entwicklung aus. Prompt- und Completion-Token-Anzeigen zeigen dir frühzeitig, ob ein System-Prompt größer ist als gedacht, und bei $0,15 pro 1M Input und $0,50 pro 1M Output kannst du die Kosten einer Anfrage im Kopf berechnen.

Multi-Turn-Chat und Sampling-Einstellungen

Die API ist zustandslos, also verwaltest du die Konversationsliste selbst und sendest sie jedes Mal. Eine kleine Klasse reicht:

Copyclass 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

Der Slice self.history[-self.max_turns * 2:] behält die letzten zwölf Benutzer- und Assistenten-Paare. Eine echte Anwendung könnte auch einige frühe Nachrichten fixieren, wie ein Charakterbogen oder wichtige Fakten, und die Mitte zusammenfassen. Das Beschneiden auf die letzten 12 Gesprächsrunden hält die Input-Kosten konstant, während ein Gespräch wächst. Das Kontextfenster umfasst 100.000 Token für Prompt und Completion zusammen, also ist dies eine Kostenentscheidung mehr als eine Notwendigkeit. Die Parameter temperature, top_p, stop, seed und die Penalties werden durchgereicht. Setze seed, wenn du wiederholbare Läufe beim Testen eines Prompts möchtest, und lasse max_tokens weg, es sei denn, du möchtest die Länge begrenzen.

Streaming-Antworten

Streaming nutzt Server-Sent Events und gibt dir Token, sobald sie generiert werden, was für jedes interaktive Tool wichtig ist. Übergebe stream=True und iteriere:

Copystream = 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)

Die Prüfung auf chunk.choices ist wichtig, weil der letzte Chunk die Nutzungsinformationen trägt und eine leere choices-Liste haben kann. Ein operationeller Punkt: Eine gestreamte Anfrage belegt ihren Concurrency-Slot, bis der Stream endet. Packe die Schleife in einen try-Block mit einer finally-Klausel, die den Stream schließt, wenn dein Code in der Mitte eine Exception auslöst, zum Beispiel bei einer Client-Trennung. Schließe den Stream immer ab, sonst trifft eine Schleife, die Streams verlässt, das Concurrency-Limit, ohne nützliche Arbeit zu leisten.

JSON-Modus

Wenn die Ausgabe anderen Code speist, fordere ein JSON-Objekt an. Erwähne JSON im Prompt, damit das Modell die erwartete Struktur kennt:

Copyimport 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"])

Packe json.loads auch im Produktionscode in einen try-Block. Der JSON-Modus macht gültige Ausgaben wahrscheinlich, und ein Request, der durch eine Stop-Sequenz oder dein eigenes Limit abgeschnitten wird, kann immer noch ein abgeschnittenes Objekt hinterlassen. Das einmalige Wiederholen mit derselben Eingabe ist eine günstige Lösung, und da fehlgeschlagene Anfragen nicht berechnet werden, sind die Kosten für einen Retry nur die erfolgreiche Anfrage.

Function Calling

Function Calling folgt der OpenAI-Struktur: Du beschreibst Funktionen in tools, das Modell gibt tool_calls zurück, du führst sie aus und sendest Ergebnisse mit der Rolle tool zurück. Ein vollständiger Roundtrip:

Copyimport 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)

Die Schleife ist wichtiger als die Details des Würfels. Das Modell fordert einen Aufruf an, dein Code führt ihn aus, und du hängst sowohl die Assistenten-Nachricht als auch das Tool-Ergebnis an, bevor du erneut fragst. Überspringe die Assistenten-Nachricht, und die tool_call_id im Ergebnis hat nichts zum Abgleichen, was die übliche Ursache für eine verwirrte zweite Antwort ist. Bei tool_choice auf auto entscheidet das Modell, ob es überhaupt ein Tool aufruft, also behandle immer den Zweig, in dem es direkt antwortet, wie der obige Code es tut. Gestreamte Antworten können auch tool_calls tragen, die in Stücken geliefert werden, die du per Index verknüpfst, bevor du die Argumente parst.

Validiere die Argumente, bevor du sie ausführst. Das Modell kann einen fehlerhaften Aufruf erzeugen, und dein Code sollte die Argumente als nicht vertrauenswürdige Eingabe behandeln, genau wie alles andere von einem Benutzer.

Asynchrone Batches innerhalb der Limits

Für viele Prompts gleichzeitig verwende AsyncOpenAI mit einem Semaphore, das deinem Concurrency-Limit entspricht. Ein vollständiger Schlüssel erlaubt 8 gleichzeitige Anfragen, ein Trial-Schlüssel 2:

Copyimport 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())

Das Semaphore regelt die Concurrency, nicht aber das pro Minute geltende Limit von 300 Anfragen. Für lange Batches füge eine kleine Verzögerung oder einen Token-Bucket-Limiter hinzu. Bei $0,15 pro 1M Input und $0,50 pro 1M Output-Token kosten 40 Slogan-Anfragen mit etwa 60 Token Input und 25 Output insgesamt $0,00036 + $0,0005 = $0,00086, also formen die Limits, nicht der Preis, das Batch-Design. Sieh dir die Preisseite für Aufladungen an.

Ratenlimits, Wiederholungen und Header

Über einem Limit gibt die API HTTP 429 zurück. Der Fehlercode ist rate_limited für zu viele Anfragen pro Minute und concurrency für zu viele gleichzeitig. Temporäre Überlastung gibt ebenfalls 429 mit einem Retry-After: 5 Header zurück. Das SDK wiederholt einige Fehler automatisch, und du kannst max_retries auf dem Client setzen, aber eine explizite Behandlung ist klarer:

Copyimport 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")

Hier gibt with_raw_response Zugriff auf X-Balance-USD, X-Request-Id und die Ratenlimit-Header. Logge die Request-ID bei jedem Fehler, den du melden möchtest. Schreibe den Schlüssel niemals in ein Notebook, das du committest, oder in Frontend-Code.

Testen ohne Guthabenverbrauch und häufige Fehler

Der Großteil einer Python-Integration ist Plumbing, und Plumbing benötigt kein Live-Modell. Halte den Netzwerkaufruf hinter einer Funktion, und deine Tests können ihn ersetzen. Mit nur der Standardbibliothek:

Copyfrom 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"

Führe nur an den Rändern eine Handvoll echter Aufrufe aus: einen Smoke-Test pro Deploy sowie gelegentliche Läufe mit deinem gespeicherten Prompt-Set. Ein einzelner Smoke-Test mit etwa 100 Token Input und 20 Output kostet $0,000015 + $0,00001 = $0,000025, was nichts ist, und fehlgeschlagene Anfragen werden überhaupt nicht berechnet. Das $0,50 Testguthaben ist weit mehr, als eine Testsuite benötigt.

Fixiere für die Prompt-Arbeit seed und temperature, damit du zwei Versionen eines System-Prompts unter gleichen Bedingungen vergleichen kannst. Das Ändern sowohl des Prompts als auch der Sampling-Einstellungen zwischen den Läufen macht die Ergebnisse unlesbar. Speichere die Ausgabe jedes Laufs mit der Request-ID aus dem X-Request-Id Header, und du kannst eine überraschende Antwort später nachverfolgen.

Häufige Python-Fehler

Diese verursachen die meisten Support-Fragen, die von Python-Nutzern kommen:

• Den Schlüssel nur als OPENAI_API_KEY zu übergeben. Wenn du base_url im Code setzt, aber die Schlüsselvariable vergisst, sendet das SDK einen leeren oder falschen Schlüssel. Übergebe api_key explizit.
• Das Vergessen von /v1. Die Base URL endet auf /v1. Das SDK fügt /chat/completions selbst hinzu.
• Die Verwendung eines Modellnamens aus einem anderen Dienst. Hier ist nur uncensored gültig.
• Unbegrenzte Concurrency. asyncio.gather über tausend Koroutinen ohne Semaphore erzeugt eine Wand aus concurrency-Fehlern.
• Eine gestreamte Antwort als einen String behandeln. Verknüpfe die Deltas und überspringe Chunks mit einer leeren choices Liste.
• Das Kontextlimit ignorieren. History plus Completion müssen in 100.000 Token passen. Schneide alte Nachrichten ab, bevor du es erreichst.

Für Aufladungen und die aktuelle Preisliste sieh dir die Preisseite an. Der Dienst ist nur Text, also gibt es keine Image-, Audio- oder Embedding-Aufrufe zum Einrichten, und Prompts werden nicht zum Training verwendet.

Weitere Anleitungen

• GLM API Base URL: Verbinde jeden OpenAI Client
• GLM API Preise: $0,15 rein, $0,50 raus pro 1M Token
• Gemma-API: Mythen vs. Fakten