GLM API Python: примеры чата, потоковой передачи и инструментов

Установка и первый запрос

Создайте виртуальную среду и установите SDK. Предполагается версия 1 или выше пакета openai.

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

Получите ключ на странице API ключ; новые пользователи могут получить пробный баланс $0.50 в один клик, действительный в течение 7 дней без привязки карты. Сохраните ключ в переменной окружения, затем выполните первый вызов:

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)

Это вся интеграция. Сервис — независимый эндпоинт, совместимый с OpenAI, для одной модели с открытыми весами, поэтому это не официальный API поставщика GLM или кого-либо ещё. Доступны только POST /v1/chat/completions и GET /v1/models. В кратком руководстве указаны поддерживаемые поля запроса.

Две небольшие привычки помогут с самого начала. Во-первых, вызовите client.models.list() один раз в smoke-тесте: он обращается к GET /v1/models, так что это доказывает, что ключ, base URL и сетевой путь работают, прежде чем вы потратите хотя бы один токен на завершение. Во-вторых, выводите resp.usage во время разработки. Количество токенов промпта и завершения покажет вам заранее, не больше ли системный промпт, чем вы думали, и при $0.15 за 1M входных и $0.50 за 1M выходных токенов вы можете оценить стоимость вызова в уме.

Многоходовой чат и настройки сэмплирования

API не имеет состояния, поэтому вы храните список диалога самостоятельно и отправляете его каждый раз. Достаточно небольшого класса:

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

Срез self.history[-self.max_turns * 2:] сохраняет последние двенадцать пар пользовательских и ассистентских сообщений. В реальном приложении можно также зафиксировать несколько ранних сообщений, таких как описание персонажа или ключевые факты, и сжать середину до краткого пересказа. Обрезка до последних 12 раундов держит стоимость ввода постоянной по мере роста разговора. Контекстное окно составляет 100 000 токенов для промпта и завершения вместе, так что это выбор по стоимости, а не необходимость. Параметры temperature, top_p, stop, seed и штрафы передаются. Установите seed, когда хотите повторяемые запуски при тестировании промпта, и оставьте max_tokens вне запроса, если не хотите ограничивать длину.

Потоковые ответы

Потоковая передача использует серверные события (SSE) и выдает вам токены по мере их генерации, что важно для любого интерактивного инструмента. Передайте stream=True и итерируйте:

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)

Проверка на chunk.choices важна, потому что последний чанк содержит использование и может иметь пустой список choices. Один операционный момент: потоковый запрос удерживает слот параллельности до конца потока. Оберните цикл в блок try с finally, который закрывает поток, если ваш код прерывается на полпути, например, при отключении клиента. Всегда завершайте или закрывайте поток, иначе цикл, который оставляет потоки, достигнет лимита параллельности, не выполняя никакой полезной работы.

JSON-режим

Когда вывод питает другой код, запросите JSON-объект. Упомяните JSON в промпте также, чтобы модель знала, какую форму вы ожидаете:

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

В продакшн-коде все равно оберните json.loads в блок try. JSON-режим делает вероятным корректный вывод, а запрос, обрезанный последовательностью остановки или вашим собственным лимитом, все еще может оставить усеченный объект. Повторная попытка один раз с тем же входом — дешевое исправление, и поскольку неудачные запросы не оплачиваются, стоимость повторной попытки — только успешный вызов.

Вызов функций

Вызов инструментов следует формату OpenAI: вы описываете функции в tools, модель возвращает tool_calls, вы выполняете их и отправляете результаты обратно с ролью tool. Полный цикл:

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)

Цикл важнее деталей с кубиком. Модель запрашивает вызов, ваш код выполняет его, и вы добавляете сообщение ассистента и результат инструмента перед повторным запросом. Если пропустить сообщение ассистента, то для tool_call_id в результате не найдётся соответствия, что является обычной причиной сбивчивого второго ответа. При установке tool_choice в значение auto модель решает, вызывать ли инструмент вообще, поэтому всегда обрабатывайте ветку, где она отвечает напрямую, как в коде выше. Потоковые ответы могут также содержать tool_calls, доставляемые частями, которые вы объединяете по индексу перед разбором аргументов.

Проверяйте аргументы перед их использованием. Модель может создать некорректный вызов, и ваш код должен рассматривать аргументы как ненадежный ввод, как и любой ввод от пользователя.

Асинхронные пакеты в пределах лимитов

Для множества промптов одновременно используйте AsyncOpenAI с семафором, соответствующим вашему лимиту параллельных запросов. Полный ключ позволяет 8 параллельных запросов, пробный ключ — 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())

Семафор обрабатывает параллельность, но не лимит 300 запросов в минуту. Для длинных пакетов добавьте небольшую задержку или ограничитель на основе ведра токенов. По цене $0.15 за 1M входных и $0.50 за 1M выходных токенов, 40 запросов на слоганы с 60 токенами на входе и 25 на выходе стоят $0.00036 + $0.0005 = $0.00086 в общей сложности, так что лимиты, а не цена, определяют устройство пакетной обработки. См. страницу тарифов для пополнений.

Лимиты запросов, повторные попытки и заголовки

При превышении лимита API возвращает HTTP 429. Код ошибки — rate_limited при слишком большом числе запросов в минуту и concurrency при слишком большом числе одновременных. Временная перегрузка также возвращает 429 с заголовком Retry-After: 5. SDK сам повторяет некоторые ошибки, и вы можете установить max_retries на клиенте, но явная обработка понятнее:

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

Здесь with_raw_response дает доступ к X-Balance-USD, X-Request-Id и заголовкам лимитов запросов. Логируйте идентификатор запроса с любой ошибкой, о которой хотите сообщить. Никогда не помещайте ключ в Jupyter-ноутбук, который вы коммитите, или в код фронтенда.

Тестирование без траты баланса и распространенные ошибки

Большая часть интеграции Python — это обвязка, и обвязке не нужна живая модель. Держите сетевой вызов за одной функцией, и ваши тесты могут заменить его. Используя только стандартную библиотеку:

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"

Выполните несколько реальных вызовов только на краях: один дымовой тест на каждое развертывание, плюс периодический запуск вашего сохраненного набора промптов. Один дымовой тест с 100 токенами на входе и 20 на выходе стоит $0.000015 + $0.00001 = $0.000025, что ничтожно, а неудачные запросы вообще не оплачиваются. Кредита пробного баланса $0.50 гораздо больше, чем нужно для набора тестов.

Для работы с промптами зафиксируйте seed и temperature, чтобы сравнивать две версии системного промпта на равных условиях. Изменение как промпта, так и настроек сэмплирования между запусками делает результаты нечитаемыми. Сохраните вывод каждого запуска с идентификатором запроса из заголовка X-Request-Id, и вы сможете отследить неожиданный ответ позже.

Распространенные ошибки Python

Именно они вызывают большинство вопросов поддержки от пользователей Python:

• Передача ключа только как OPENAI_API_KEY. Если вы установите base_url в коде, но забудете переменную ключа, SDK отправит пустой или неверный ключ. Передайте api_key явно.
• Пропущенный /v1. Base URL заканчивается на /v1. SDK добавляет /chat/completions сам.
• Использование имени модели из другого сервиса. Здесь допустим только uncensored.
• Неограниченный параллелизм. asyncio.gather более тысячи сопрограмм без семафора приводит к лавине ошибок concurrency.
• Обработка потокового ответа как одной строки. Объедините дельты и пропускайте чанки с пустым списком choices.
• Игнорирование лимита контекста. История плюс завершение должны помещаться в 100 000 токенов. Обрезайте старые сообщения, прежде чем достигнете его.

Для пополнений и текущего прайс-листа см. страницу тарифов. Сервис работает только с текстом, поэтому нет вызовов изображений, аудио или эмбеддингов для подключения, и промпты не используются для обучения.

Дополнительные руководства

• Base URL GLM API: Подключите любой клиент OpenAI
• Тарифы GLM API: $0.15 вход, $0.50 выход за 1M токенов
• Gemma API: Мифы против Фактов