Esempi Python GLM API: chat, streaming e strumenti

Installazione e prima richiesta

Crea un ambiente virtuale e installa l'SDK. La versione 1 o successiva del pacchetto openai è assunta in tutto il testo.

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

Ottieni una chiave dalla pagina delle chiavi API; i nuovi utenti possono richiedere un credito di prova da $0,50 con un clic, valido per 7 giorni e senza bisogno di carta. Tieni la chiave in una variabile d'ambiente, poi fai la prima chiamata:

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)

Questa è l'integrazione completa. Il servizio è un endpoint indipendente e compatibile con OpenAI per un modello a pesi aperti, quindi non è l'API ufficiale del fornitore dietro GLM o di chiunque altro. Esistono solo POST /v1/chat/completions e GET /v1/models. La guida rapida elenca i campi della richiesta supportati.

Due piccole abitudini aiutano fin da subito. Primo, chiama client.models.list() una volta in un smoke test: chiama GET /v1/models, quindi dimostra che la chiave, l'URL di base e il percorso di rete funzionano prima di spendere un solo token per un completamento. Secondo, stampa resp.usage durante lo sviluppo. I conteggi dei token del prompt e del completamento ti mostrano subito se un prompt di sistema è più grande del previsto, e a $0,15 per 1M di input e $0,50 per 1M di output puoi calcolare a mente il costo di una chiamata.

Chat multi-turno e impostazioni di campionamento

L'API è senza stato, quindi gestisci tu l'elenco delle conversazioni e invialo ogni volta. Una piccola classe è sufficiente:

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 fetta self.history[-self.max_turns * 2:] mantiene le ultime dodici coppie utente-assistente. Un'applicazione reale potrebbe anche fissare alcuni messaggi iniziali, come un foglio personaggio o fatti chiave, e riassumere la parte centrale. Il taglio agli ultimi 12 turni mantiene il costo dell'input costante mentre la conversazione cresce. La finestra di contesto è di 100.000 token per prompt e completamento insieme, quindi questa è una scelta di costo più che una necessità. I parametri temperature, top_p, stop, seed e le penalità vengono passati. Imposta seed quando vuoi esecuzioni ripetibili durante il test di un prompt, e lascia max_tokens fuori a meno che tu non voglia limitare la lunghezza.

Risposte in streaming

Lo streaming usa eventi inviati dal server e ti fornisce i token man mano che vengono generati, il che è importante per qualsiasi strumento interattivo. Passa stream=True e itera:

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)

Il controllo su chunk.choices è importante perché l'ultimo blocco contiene l'utilizzo e può avere un elenco choices vuoto. Un punto operativo: una richiesta in streaming mantiene il suo slot di concorrenza fino alla fine dello streaming. Avvolgi il ciclo in un blocco try con una clausola finally che chiude lo streaming se il tuo codice genera un'eccezione a metà, ad esempio quando un client si disconnette. Termina o chiudi sempre lo streaming, altrimenti un ciclo che abbandona gli streaming colpirà il limite di concorrenza senza fare lavoro utile.

Modalità JSON

Quando l'output alimenta altro codice, chiedi un oggetto JSON. Menziona JSON nel prompt, così il modello sa che forma ti aspetti:

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

Avvolgi comunque json.loads in un blocco try nel codice di produzione. La modalità JSON rende probabile un output valido, e una richiesta interrotta da una sequenza di stop o dal tuo limite può comunque lasciare un oggetto tronco. Ritentare una volta con lo stesso input è una soluzione economica, e poiché le richieste fallite non vengono addebitate, il costo di un nuovo tentativo è solo la chiamata che ha successo.

Chiamata di funzioni

La chiamata di funzioni segue la forma OpenAI: descrivi le funzioni in tools, il modello restituisce tool_calls, le esegui e invii i risultati indietro con il ruolo tool. Un giro completo:

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)

Il ciclo è più importante dei dettagli del dado. Il modello chiede una chiamata, il tuo codice la esegue, e tu aggiungi sia il messaggio dell'assistente che il risultato della chiamata prima di chiedere di nuovo. Se salti il messaggio dell'assistente, il tool_call_id nel risultato non avrà nulla con cui abbinarsi, che è la causa solita di una seconda risposta confusa. Con tool_choice impostato su auto il modello decide se chiamare uno strumento o meno, quindi gestisci sempre il ramo in cui risponde direttamente, come fa il codice sopra. Le risposte in streaming possono trasportare anche tool_calls, consegnate a pezzi che unisci per indice prima di analizzare gli argomenti.

Convalida gli argomenti prima di agire su di essi. Il modello può produrre una chiamata non valida e il tuo codice dovrebbe trattare gli argomenti come input non fidato, come qualsiasi cosa proveniente da un utente.

Batch asincroni nei limiti

Per molti prompt contemporaneamente, usa AsyncOpenAI con un semaforo che corrisponda al tuo limite di concorrenza. Una chiave completa permette 8 richieste concorrenti, una chiave di prova 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())

Il semaforo gestisce la concorrenza ma non il limite per minuto di 300 richieste. Per batch lunghi aggiungi un piccolo ritardo o un limiter a secchio di token. A $0,15 per 1M di input e $0,50 per 1M di output token, 40 richieste di slogan di circa 60 token in ingresso e 25 in uscita costano $0,00036 + $0,0005 = $0,00086 in totale, quindi i limiti, non il prezzo, modellano il design del batch. Vedi la pagina dei prezzi per le ricariche.

Limiti di richiesta, retry e intestazioni

Superato un limite, l'API restituisce HTTP 429. Il codice di errore è rate_limited per troppe richieste al minuto e concurrency per troppe richieste simultanee. Un sovraccarico temporaneo restituisce anche 429 con un'intestazione Retry-After: 5. L'SDK ritenta alcuni errori da solo e puoi impostare max_retries sul client, ma la gestione esplicita è più chiara:

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

Qui with_raw_response dà accesso a X-Balance-USD, X-Request-Id e alle intestazioni del limite di richieste. Registra l'ID della richiesta con qualsiasi errore che vuoi segnalare. Non mettere mai la chiave in un notebook che committi o nel codice front-end.

Test senza spendere credito e errori comuni

La maggior parte di un'integrazione Python è cablaggio, e il cablaggio non ha bisogno di un modello reale. Tieni la chiamata di rete dietro una funzione, e i tuoi test possono sostituirla. Solo con la libreria 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"

Esegui una manciata di chiamate reali solo ai bordi: un smoke test per ogni deploy, più un'esecuzione occasionale del tuo set di prompt salvato. Un singolo smoke test con circa 100 token in ingresso e 20 in uscita costa $0,000015 + $0,00001 = $0,000025, che è nulla, e le richieste fallite non vengono addebitate affatto. Il credito di prova da $0,50 è ben più di quanto serva a una suite di test.

Per il lavoro sui prompt, fissa seed e temperature così puoi confrontare due versioni di un prompt di sistema in condizioni di parità. Cambiare sia il prompt che le impostazioni di campionamento tra le esecuzioni rende i risultati impossibili da leggere. Memorizza l'output di ogni esecuzione con l'ID della richiesta dall'intestazione X-Request-Id, e potrai tracciare una risposta sorprendente in seguito.

Errori Python comuni

Questi causano la maggior parte delle domande di supporto che arrivano dagli utenti Python:

• Passare la chiave solo come OPENAI_API_KEY. Se imposti base_url nel codice ma dimentichi la variabile della chiave, l'SDK invia una chiave vuota o errata. Passa api_key esplicitamente.
• Dimenticare /v1. L'URL di base termina con /v1. L'SDK aggiunge /chat/completions da solo.
• Usare un nome del modello di un altro servizio. Solo uncensored è valido qui.
• Concorrenza illimitata. asyncio.gather su mille coroutine senza un semaforo produce un muro di errori concurrency.
• Trattare una risposta in streaming come una stringa unica. Unisci i delta e salta i segmenti con una lista choices vuota.
• Ignorare il limite di contesto. Cronologia più completamento devono stare in 100.000 token. Taglia i turni vecchi prima di raggiungerlo.

Per le ricariche e il tariffario corrente, vedi la pagina dei prezzi. Il servizio è solo testo, quindi non ci sono chiamate per immagini, audio o embedding da collegare, e i prompt non sono usati per l'addestramento.

Altre guide

• URL di base API GLM: Connetti qualsiasi client OpenAI
• Prezzi API GLM: $0,15 In, $0,50 Out per 1M di Token
• Gemma API: Miti vs Fatti