Exemples Python de l'API GLM : chat, streaming et outils

Installation et première requête

Créez un environnement virtuel et installez le SDK. La version 1 ou supérieure du package openai est supposée tout au long de ce guide.

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

Obtenez une clé depuis la page clé API ; les nouveaux utilisateurs peuvent activer un crédit d'essai de 0,50 $ en un clic, valable 7 jours et sans carte requise. Conservez la clé dans une variable d'environnement, puis effectuez le premier appel :

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)

Voici toute l'intégration. Le service est un endpoint indépendant compatible OpenAI pour un seul modèle à poids ouverts, ce n'est donc pas l'API officielle du fournisseur derrière GLM ou d'un autre. Seuls POST /v1/chat/completions et GET /v1/models existent. Le quickstart liste les champs de requête pris en charge.

Deux petites habitudes aident dès le début. Premièrement, appelez client.models.list() une fois dans un test de fumée : il atteint GET /v1/models, ce qui prouve que la clé, l'URL de base et le chemin réseau fonctionnent avant de dépenser un seul token sur une complétion. Deuxièmement, affichez resp.usage pendant le développement. Les comptes de tokens de prompt et de complétion vous indiquent tôt si un prompt système est plus volumineux que prévu, et à 0,15 $ par 1M d'input et 0,50 $ par 1M d'output, vous pouvez estimer le coût d'un appel mentalement.

Chat multi-tours et paramètres d'échantillonnage

L'API est sans état, donc vous gérez vous-même la liste des conversations et l'envoyez à chaque fois. Une petite classe suffit :

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

La tranche self.history[-self.max_turns * 2:] conserve les douze dernières paires utilisateur et assistant. Une application réelle pourrait aussi épingler quelques messages initiaux, comme une fiche de personnage ou des faits clés, et résumer le milieu. Limiter aux 12 derniers tours maintient le coût d'entrée constant à mesure que la conversation s'allonge. La fenêtre de contexte est de 100 000 tokens pour le prompt et la complétion ensemble, c'est donc un choix de coût plus qu'une nécessité. Les paramètres temperature, top_p, stop, seed et les pénalités sont transmis. Définissez seed pour des exécutions reproductibles lors des tests d'un prompt, et laissez max_tokens absent sauf si vous voulez limiter la longueur.

Réponses en streaming

Le streaming utilise les événements envoyés par le serveur et vous donne des tokens au fur et à mesure de leur génération, ce qui est important pour tout outil interactif. Passez stream=True et itérez :

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)

La garde sur chunk.choices est importante car le dernier chunk contient les informations d'utilisation et peut avoir une liste de choix vide. Un point opérationnel : une requête en streaming occupe son slot de concurrence jusqu'à la fin du flux. Enveloppez la boucle dans un bloc try avec une clause finally qui ferme le flux si votre code lève une exception en cours de route, par exemple lors d'une déconnexion du client. Terminez ou fermez toujours le flux, sinon une boucle qui abandonne des flux atteindra la limite de concurrence sans faire de travail utile.

Mode JSON

Lorsque la sortie alimente du code, demandez un objet JSON. Mentionnez JSON dans le prompt, afin que le modèle connaisse la forme attendue :

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

Enveloppez toujours json.loads dans un bloc try dans le code de production. Le mode JSON rend une sortie valide probable, et une requête coupée par une séquence d'arrêt ou votre propre limite peut encore laisser un objet tronqué. Réessayer une fois avec la même entrée est une correction peu coûteuse, et comme les requêtes échouées ne sont pas facturées, le coût d'un retry n'est que l'appel qui réussit.

Appel de fonctions

L'appel de fonctions suit la forme OpenAI : vous décrivez les fonctions dans tools, le modèle retourne tool_calls, vous les exécutez et renvoyez les résultats avec le rôle tool. Un aller-retour complet :

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)

La boucle est plus importante que les détails du dé. Le modèle demande un appel, votre code l'exécute, et vous ajoutez à la fois le message de l'assistant et le résultat de l'outil avant de demander à nouveau. Si vous omettez le message de l'assistant, le tool_call_id du résultat n'aura rien à quoi correspondre, ce qui est la cause habituelle d'une deuxième réponse confuse. Avec tool_choice défini sur auto, le modèle décide s'il doit appeler un outil ou non, donc gérez toujours la branche où il répond directement, comme le fait le code ci-dessus. Les réponses en streaming peuvent également transporter tool_calls, livrées par morceaux que vous joignez par index avant d'analyser les arguments.

Validez les arguments avant d'agir dessus. Le modèle peut produire un appel malformé, et votre code doit traiter les arguments comme une entrée non fiable, comme n'importe quoi venant d'un utilisateur.

Lots asynchrones dans les limites

Pour de nombreux prompts à la fois, utilisez AsyncOpenAI avec un sémaphore correspondant à votre limite de concurrence. Une clé complète permet 8 requêtes simultanées, une clé d'essai en permet 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())

Le sémaphore gère la concurrence mais pas le plafond de 300 requêtes par minute. Pour des lots longs, ajoutez un petit délai ou un limiteur à seau de jetons. À 0,15 $ par 1 M de tokens d'entrée et 0,50 $ par 1 M de tokens de sortie, 40 demandes de slogan d'environ 60 tokens en entrée et 25 en sortie coûtent 0,00036 $ + 0,0005 $ = 0,00086 $ au total, donc les limites, et non le prix, façonnent la conception des lots. Consultez la page tarifs pour les recharges.

Limites de débit, retries et en-têtes

Au-delà d'une limite, l'API retourne HTTP 429. Le code d'erreur est rate_limited pour trop de requêtes par minute et concurrency pour trop en même temps. Une surcharge temporaire retourne également 429 avec un en-tête Retry-After: 5. Le SDK réessaie certaines erreurs lui-même et vous pouvez définir max_retries sur le client, mais une gestion explicite est plus claire :

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

Ici, with_raw_response donne accès à X-Balance-USD, X-Request-Id et aux en-têtes de limite de débit. Journalisez l'ID de requête avec toute erreur que vous souhaitez signaler. Ne mettez jamais la clé dans un notebook que vous commitez ou dans du code front-end.

Tests sans dépenser de crédit et erreurs courantes

La plupart d'une intégration Python est du câblage, et le câblage n'a pas besoin d'un modèle en direct. Gardez l'appel réseau derrière une fonction, et vos tests peuvent le remplacer. Avec uniquement la bibliothèque standard :

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"

Exécutez quelques appels réels uniquement aux bords : un test de fumée par déploiement, plus un lancement occasionnel de votre ensemble de prompts enregistré. Un seul test de fumée avec environ 100 tokens en entrée et 20 en sortie coûte 0,000015 $ + 0,00001 $ = 0,000025 $, ce qui est négligeable, et les requêtes échouées ne sont pas facturées du tout. Le crédit d'essai de 0,50 $ est bien supérieur à ce dont a besoin une suite de tests.

Pour le travail de prompt, figez seed et temperature afin que vous puissiez comparer deux versions d'un prompt système à conditions égales. Changer à la fois le prompt et les paramètres d'échantillonnage entre les runs rend les résultats impossibles à lire. Stockez la sortie de chaque run avec l'ID de requête de l'en-tête X-Request-Id, et vous pourrez retracer une réponse surprenante plus tard.

Erreurs Python courantes

Celles-ci causent la plupart des questions de support venant des utilisateurs Python :

• Passer la clé uniquement comme OPENAI_API_KEY. Si vous définissez base_url dans le code mais oubliez la variable de clé, le SDK envoie une clé vide ou incorrecte. Passez api_key explicitement.
• Oublier /v1. L'URL de base se termine par /v1. Le SDK ajoute /chat/completions lui-même.
• Utiliser un nom de modèle d'un autre service. Seul uncensored est valide ici.
• Concurrence non bornée. asyncio.gather sur mille coroutines sans sémaphore produit un mur d'erreurs concurrency.
• Traiter une réponse en streaming comme une seule chaîne. Joignez les deltas, et sautez les chunks avec une liste choices vide.
• Ignorer la limite de contexte. L'historique plus la complétion doivent tenir dans 100 000 tokens. Coupez les anciens tours avant de l'atteindre.

Pour les recharges et la grille tarifaire actuelle, consultez la page tarifs. Le service est textuel uniquement, il n'y a donc pas d'appels image, audio ou embedding à configurer, et les prompts ne sont pas utilisés pour l'entraînement.

Plus de guides

• URL de base de l'API GLM : Connectez n'importe quel client OpenAI
• Tarifs API GLM : 0,15 $ Entrée, 0,50 $ Sortie par 1M de Tokens
• API Gemma : Mythes et réalités