Ejemplos de Chat, Streaming y Herramientas con GLM API en Python

Instalación y primera petición

Crea un entorno virtual e instala el SDK. Se asume la versión 1 o posterior del paquete openai en todo el texto.

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

Obtén una clave en la página de claves de API; los nuevos usuarios pueden reclamar un crédito de prueba de $0.50 con un clic, válido por 7 días y sin necesidad de tarjeta. Guarda la clave en una variable de entorno y luego haz la primera llamada:

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)

Esa es toda la integración. El servicio es un endpoint compatible con OpenAI para un único modelo de pesos abiertos, por lo que no es la API oficial del proveedor detrás de GLM ni de nadie más. Solo existen POST /v1/chat/completions y GET /v1/models. La guía de inicio rápido enumera los campos de petición admitidos.

Dos pequeños hábitos ayudan desde el inicio. Primero, llama a client.models.list() una vez en una prueba de humo: llama a GET /v1/models, así que demuestra que la clave, la base URL y la ruta de red funcionan antes de gastar un solo token en una finalización. Segundo, imprime resp.usage durante el desarrollo. Los recuentos de tokens de prompt y de finalización te muestran temprano si un prompt del sistema es más grande de lo que pensabas, y a $0.15 por 1M de entrada y $0.50 por 1M de salida puedes calcular el costo de una llamada mentalmente.

Chat multi-turno y ajustes de muestreo

La API es sin estado, así que mantienes la lista de conversación tú mismo y la envías cada vez. Una pequeña clase es 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

La porción self.history[-self.max_turns * 2:] conserva los últimos doce pares de usuario y asistente. Una aplicación real también podría fijar algunos mensajes iniciales, como una hoja de personaje o datos clave, y resumir el medio. Recortar a los últimos 12 turnos mantiene el coste de entrada plano a medida que crece la conversación. La ventana de contexto es de 100,000 tokens para prompt y finalización juntos, por lo que esta es una elección de coste más que una necesidad. Los parámetros temperature, top_p, stop, seed y las penalizaciones se pasan a través. Establece seed cuando quieras ejecuciones repetibles mientras pruebas un prompt, y deja max_tokens fuera a menos que quieras limitar la longitud.

Respuestas en streaming

El streaming usa eventos enviados por el servidor y te da tokens a medida que se generan, lo cual importa para cualquier herramienta interactiva. Pasa 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)

La comprobación de chunk.choices es importante porque el último fragmento lleva la información de uso y puede tener una lista de elecciones vacía. Un punto operativo: una petición en streaming mantiene su ranura de concurrencia hasta que termina el streaming. Envuelve el bucle en un bloque try con una cláusula finally que cierre el streaming si tu código se interrumpe a mitad de camino, por ejemplo cuando un cliente se desconecta. Siempre termina o cierra el streaming, de lo contrario un bucle que abandone streams alcanzará el límite de concurrencia sin hacer trabajo útil.

Modo JSON

Cuando la salida alimenta otro código, pide un objeto JSON. Menciona JSON en el prompt también, para que el modelo sepa qué forma esperas:

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

Aun así, envuelve json.loads en un bloque try en código de producción. El modo JSON hace probable una salida válida, y una petición cortada por una secuencia de parada o tu propio límite aún puede dejar un objeto truncado. Reintentar una vez con la misma entrada es una solución barata, y dado que las peticiones fallidas no se cobran, el coste de un reintento es solo la llamada que tiene éxito.

Llamadas a funciones

Las llamadas a funciones siguen la forma de OpenAI: describes funciones en tools, el modelo devuelve tool_calls, las ejecutas y envías los resultados de nuevo con el rol tool. Un viaje de ida y vuelta 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)

El bucle es más importante que los detalles del dado. El modelo pide una llamada, tu código la ejecuta y añades tanto el mensaje del asistente como el resultado de la herramienta antes de preguntar de nuevo. Si omites el mensaje del asistente, el tool_call_id del resultado no tendrá nada con qué coincidir, que es la causa habitual de una segunda respuesta confundida. Con tool_choice establecido en auto el modelo decide si llama a una herramienta en absoluto, así que siempre maneja la rama donde responde directamente, como hace el código anterior. Las respuestas en streaming pueden llevar tool_calls también, entregadas en piezas que unes por índice antes de analizar los argumentos.

Valida los argumentos antes de actuar sobre ellos. El modelo puede producir una llamada malformada, y tu código debe tratar los argumentos como entrada no confiable, igual que cualquier cosa de un usuario.

Lotes asíncronos dentro de los límites

Para muchos prompts a la vez, usa AsyncOpenAI con un semáforo que coincida con tu límite de concurrencia. Una clave completa permite 8 peticiones simultáneas, una clave de prueba 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())

El semáforo maneja la concurrencia pero no el límite por minuto de 300 peticiones. Para lotes largos añade un pequeño retraso o un limitador de cubeta de tokens. A $0.15 por 1M de tokens de entrada y $0.50 por 1M de tokens de salida, 40 peticiones de eslogan de unos 60 tokens de entrada y 25 de salida cuestan $0.00036 + $0.0005 = $0.00086 en total, así que los límites, no el precio, moldean el diseño de lotes. Consulta la página de precios para recargas.

Límites de peticiones, reintentos y encabezados

Al superar un límite, la API devuelve HTTP 429. El código de error es rate_limited para demasiadas peticiones por minuto y concurrency para demasiadas a la vez. La sobrecarga temporal también devuelve 429 con un encabezado Retry-After: 5. El SDK reintenta algunos fallos por su cuenta y puedes establecer max_retries en el cliente, pero el manejo explícito es más 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")

Aquí with_raw_response da acceso a X-Balance-USD, X-Request-Id y los encabezados de límite de tasa. Registra el ID de petición con cualquier error que quieras reportar. Nunca pongas la clave en un notebook que subas a un repositorio o en código de front-end.

Pruebas sin gastar crédito y errores comunes

La mayor parte de una integración de Python es fontanería, y la fontanería no necesita un modelo en vivo. Mantén la llamada de red detrás de una función, y tus pruebas pueden reemplazarla. Solo con la biblioteca estándar:

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"

Ejecuta un puñado de llamadas reales solo en los bordes: una prueba de humo por despliegue, más una ejecución ocasional de tu conjunto de prompts guardado. Una sola prueba de humo con unos 100 tokens de entrada y 20 de salida cuesta $0.000015 + $0.00001 = $0.000025, que es nada, y las peticiones fallidas no se cobran en absoluto. El crédito de prueba de $0.50 es mucho más de lo que necesita un conjunto de pruebas.

Para el trabajo de prompts, fija seed y temperature para que puedas comparar dos versiones de un prompt del sistema en igualdad de condiciones. Cambiar tanto el prompt como la configuración de muestreo entre ejecuciones hace que los resultados sean imposibles de leer. Almacena la salida de cada ejecución con el ID de petición del encabezado X-Request-Id, y podrás rastrear una respuesta sorprendente más tarde.

Errores comunes de Python

Estos causan la mayoría de las preguntas de soporte que vienen de usuarios de Python:

• Pasar la clave solo como OPENAI_API_KEY. Si configuras base_url en el código pero olvidas la variable de clave, el SDK envía una clave vacía o incorrecta. Pasa api_key explícitamente.
• Olvidar /v1. La base URL termina en /v1. El SDK añade /chat/completions por sí mismo.
• Usar un nombre de modelo de otro servicio. Solo uncensored es válido aquí.
• Concurrencia descontrolada. asyncio.gather sobre mil corutinas sin semáforo produce un muro de errores concurrency.
• Tratar una respuesta en streaming como una sola cadena. Une los deltas y salta los fragmentos con una lista de choices vacía.
• Ignorar el límite de contexto. El historial más la respuesta deben caber en 100,000 tokens. Recorta turnos antiguos antes de alcanzarlo.

Para recargas y la lista de tarifas actual, consulta la página de precios. El servicio es solo de texto, así que no hay llamadas de imagen, audio o incrustaciones que conectar, y los prompts no se usan para entrenamiento.

Más guías

• URL base de GLM API: Conecta cualquier cliente OpenAI
• Precios de GLM API: $0.15 Entrada, $0.50 Salida por 1M de Tokens
• API Gemma: Mitos vs. Realidad