Exemplos de Chat, Streaming e Ferramentas da API GLM em Python

Instalação e primeira requisição

Crie um ambiente virtual e instale o SDK. Assume-se o uso da versão 1 ou posterior do pacote openai.

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

Obtenha uma chave na página chave de API; novos usuários podem reivindicar um trial de $0,50 com um clique, válido por 7 dias e sem necessidade de cartão. Mantenha a chave em uma variável de ambiente e faça a primeira chamada:

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)

Essa é a integração completa. O serviço é um endpoint independente e compatível com OpenAI para um único modelo de pesos abertos, portanto não é a API oficial do fornecedor por trás do GLM ou de qualquer outro. Apenas POST /v1/chat/completions e GET /v1/models existem. O guia de início rápido lista os campos de requisição suportados.

Dois pequenos hábitos ajudam desde o início. Primeiro, chame client.models.list() uma vez em um teste de fumaça: ele aciona GET /v1/models, provando que a chave, a base URL e o caminho de rede funcionam antes que você gaste um único token em uma conclusão. Segundo, imprima resp.usage durante o desenvolvimento. As contagens de tokens do prompt e da conclusão mostram antecipadamente se um prompt do sistema é maior do que você pensava, e a $0,15 por 1M de entrada e $0,50 por 1M de saída, você pode calcular o custo de uma chamada mentalmente.

Chat multi-turno e configurações de amostragem

A API é stateless, então você mantém a lista de conversas e a envia a cada vez. Uma classe pequena é suficiente:

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

A fatia self.history[-self.max_turns * 2:] mantém os últimos doze pares de usuário e assistente. Uma aplicação real pode também fixar algumas mensagens iniciais, como uma ficha de personagem ou fatos-chave, e resumir o meio. Limitar aos últimos 12 turnos mantém o custo de entrada plano conforme a conversa cresce. A janela de contexto é de 100.000 tokens para prompt e conclusão juntos, então esta é uma escolha de custo mais do que uma necessidade. Os parâmetros temperature, top_p, stop, seed e as penalidades são passados. Defina seed quando quiser execuções repetíveis ao testar um prompt, e deixe max_tokens de fora a menos que queira limitar o comprimento.

Respostas em streaming

O streaming usa eventos enviados pelo servidor e fornece tokens conforme são gerados, o que é importante para qualquer ferramenta interativa. Passe stream=True e itere:

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)

A verificação em chunk.choices é importante porque o último chunk carrega o uso e pode ter uma lista de choices vazia. Um ponto operacional: uma requisição em streaming ocupa seu slot de concorrência até o fim do stream. Envolva o loop em um bloco try com uma cláusula finally que feche o stream se seu código levantar uma exceção no meio, por exemplo, quando um cliente se desconecta. Sempre finalize ou feche o stream, caso contrário, um loop que abandona streams atingirá o limite de concorrência sem realizar trabalho útil.

Modo JSON

Quando a saída alimenta outro código, solicite um objeto JSON. Mencione JSON no prompt também, para que o modelo saiba que formato você espera:

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

Ainda assim, envolva json.loads em um bloco try no código de produção. O modo JSON torna a saída válida provável, e uma requisição cortada por uma sequência de parada ou seu próprio limite ainda pode deixar um objeto truncado. Retentar uma vez com a mesma entrada é uma correção barata, e como requisições com falha não são cobradas, o custo de uma retentativa é apenas a chamada que tem sucesso.

Chamada de funções

A chamada de funções segue o formato da OpenAI: você descreve funções em tools, o modelo retorna tool_calls, você as executa e envia os resultados de volta com o papel tool. Uma viagem completa de ida e volta:

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)

O loop é mais importante do que os detalhes do dado. O modelo pede uma chamada, seu código a executa, e você anexa tanto a mensagem do assistente quanto o resultado da ferramenta antes de pedir novamente. Se você pular a mensagem do assistente, o tool_call_id no resultado não terá com o que corresponder, que é a causa comum de uma segunda resposta confusa. Com tool_choice definido como auto, o modelo decide se chama uma ferramenta ou não, então sempre trate o ramo onde ele responde diretamente, como o código acima faz. Respostas em streaming podem carregar tool_calls também, entregues em pedaços que você une por índice antes de analisar os argumentos.

Valide os argumentos antes de agir sobre eles. O modelo pode produzir uma chamada malformada, e seu código deve tratar os argumentos como entrada não confiável, assim como qualquer coisa vinda de um usuário.

Lotes assíncronos dentro dos limites

Para muitos prompts de uma vez, use AsyncOpenAI com um semáforo correspondente ao seu limite de concorrência. Uma chave completa permite 8 requisições simultâneas, uma chave de teste 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())

O semáforo controla a concorrência, mas não o limite de 300 requisições por minuto. Para lotes longos, adicione um pequeno atraso ou um limitador de token-bucket. A $0,15 por 1M de tokens de entrada e $0,50 por 1M de tokens de saída, 40 requisições de slogan com cerca de 60 tokens de entrada e 25 de saída custam $0,00036 + $0,0005 = $0,00086 no total, então os limites, não o preço, moldam o design do batch. Veja a página de preços para recargas.

Limites de requisições, retentativas e cabeçalhos

Acima do limite, a API retorna HTTP 429. O código de erro é rate_limited para muitas requisições por minuto e concurrency para muitas de uma vez. Sobrecarga temporária também retorna 429 com um cabeçalho Retry-After: 5. O SDK retenta algumas falhas por conta própria e você pode definir max_retries no cliente, mas o tratamento explícito é mais claro:

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

Aqui with_raw_response dá acesso a X-Balance-USD, X-Request-Id e aos cabeçalhos de limite de requisições. Registre o ID da requisição com qualquer erro que você queira reportar. Nunca coloque a chave em um notebook que você faça commit ou em código front-end.

Testes sem gastar crédito e erros comuns

A maior parte de uma integração Python é encanamento, e encanamento não precisa de um modelo ativo. Mantenha a chamada de rede atrás de uma função, e seus testes podem substituí-la. Apenas com a biblioteca padrão:

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"

Execute algumas chamadas reais apenas nas bordas: um smoke test por deploy, além de uma execução ocasional do seu conjunto de prompts salvo. Um único smoke test com cerca de 100 tokens de entrada e 20 de saída custa $0,000015 + $0,00001 = $0,000025, que é nada, e requisições com falha não são cobradas. O crédito de trial de $0,50 é muito mais do que uma suíte de testes precisa.

Para trabalho de prompt, fixe seed e temperature para que você possa comparar duas versões de um prompt do sistema em condições iguais. Alterar tanto o prompt quanto as configurações de amostragem entre execuções torna os resultados impossíveis de ler. Armazene a saída de cada execução com o ID da requisição do cabeçalho X-Request-Id, e você pode rastrear uma resposta surpreendente mais tarde.

Erros comuns em Python

Estes causam a maioria das perguntas de suporte que vêm de usuários Python:

• Passar a chave apenas como OPENAI_API_KEY. Se você definir base_url no código mas esquecer a variável da chave, o SDK envia uma chave vazia ou errada. Passe api_key explicitamente.
• Esquecendo /v1. A base URL termina em /v1. O SDK adiciona /chat/completions sozinho.
• Usando um nome de modelo de outro serviço. Apenas uncensored é válido aqui.
• Concorrência descontrolada. asyncio.gather sobre mil coroutines sem um semáforo produz uma parede de erros de concurrency.
• Tratando uma resposta em streaming como uma única string. Junte os deltas e pule chunks com uma lista de choices vazia.
• Ignorando o limite de contexto. Histórico mais conclusão devem caber em 100.000 tokens. Corte turnos antigos antes de atingi-lo.

Para recargas e a tabela de preços atual, consulte a página de preços. O serviço é apenas de texto, então não há chamadas de imagem, áudio ou embedding para conectar, e os prompts não são usados para treinamento.

Mais guias

• Base URL da API GLM: Conecte Qualquer Cliente OpenAI
• Preços da API GLM: $0,15 In, $0,50 Out por 1M de Tokens
• API Gemma: Mitos vs Fatos