GLM API Python: czat, strumieniowanie i narzędzia

Instalacja i pierwsze zapytanie

Stwórz środowisko wirtualne i zainstaluj SDK. W całym tekście zakłada się wersję 1 lub nowszą pakietu openai.

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

Pobierz klucz ze strony klucza API; nowi użytkownicy mogą jednym kliknięciem uzyskać darmowy kredyt próbny o wartości $0.50, ważny przez 7 dni bez podawania karty. Przechowuj klucz w zmiennej środowiskowej, a następnie wykonaj pierwsze zapytanie:

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)

To cała integracja. Usługa to niezależny endpoint kompatybilny z OpenAI dla jednego modelu o otwartych wagach, więc nie jest to oficjalne API dostawcy stojącego za GLM ani nikogo innego. Istnieją tylko POST /v1/chat/completions i GET /v1/models. Szybki start wymienia obsługiwane pola zapytania.

Dwa małe nawyki pomogą od początku. Po pierwsze, wywołaj client.models.list() raz w smoke teście: wywołuje GET /v1/models, więc dowodzi, że klucz, base URL i ścieżka sieciowa działają, zanim wydasz choćby jeden token na uzupełnienie. Po drugie, wypisz resp.usage podczas rozwoju. Liczba tokenów promptu i uzupełnienia pokaże Ci wcześnie, czy prompt systemowy jest większy, niż sądziłeś, a przy cenie $0.15 za 1M tokenów wejściowych i $0.50 za 1M tokenów wyjściowych możesz wycenić wywołanie w głowie.

Wieloetapowy czat i ustawienia próbkowania

API jest bezstanowe, więc sam zarządzasz listą rozmów i wysyłasz ją za każdym razem. Wystarczy mała klasa:

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

Kawałek self.history[-self.max_turns * 2:] zachowuje ostatnie dwanaście par wiadomości użytkownika i asystenta. Rzeczywista aplikacja może również przypiąć kilka wczesnych wiadomości, takich jak karta postaci lub kluczowe fakty, i podsumować środek. Przycinanie do ostatnich 12 tur utrzymuje koszt wejścia płaskim, gdy rozmowa rośnie. Okno kontekstu ma 100 000 tokenów dla promptu i uzupełnienia razem, więc jest to wybór kosztowy, a nie konieczność. Parametry temperature, top_p, stop, seed i kary są przekazywane. Ustaw seed, gdy chcesz powtarzalne uruchomienia podczas testowania promptu, i zostaw max_tokens bez ustawień, chyba że chcesz ograniczyć długość.

Odpowiedzi strumieniowe

Strumieniowanie używa zdarzeń wysyłanych przez serwer i dostarcza Ci tokeny w miarę ich generowania, co ma znaczenie dla dowolnego narzędzia interaktywnego. Przekaż stream=True i iteruj:

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)

Sprawdzenie chunk.choices ma znaczenie, ponieważ ostatni chunk zawiera informacje o użyciu i może mieć pustą listę choices. Jeden punkt operacyjny: strumieniowane zapytanie trzyma swój slot konkurencji, aż strumień się zakończy. Zawiń pętlę w blok try z klauzulą finally, która zamyka strumień, jeśli Twój kod rzuci wyjątek w połowie, na przykład gdy klient się rozłączy. Zawsze kończ lub zamykaj strumień, inaczej pętla, która porzuca strumienie, trafi w limit konkurencji, nie wykonując żadnej użytecznej pracy.

Tryb JSON

Gdy wyjście zasila inny kod, poproś o obiekt JSON. Wspomnij o JSON w promptcie, więc model wie, jakiego kształtu oczekujesz:

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

Nadal owiń json.loads blokiem try w kodzie produkcyjnym. Tryb JSON sprawia, że poprawne wyjście jest prawdopodobne, a zapytanie ucięte przez sekwencję stop lub Twój własny limit może nadal pozostawić obcięty obiekt. Ponowne wywołanie raz z tym samym wejściem jest tanią poprawką, a ponieważ nieudane zapytania nie są rozliczane, koszt ponownego wywołania to tylko wywołanie, które się powiedzie.

Wywoływanie funkcji

Wywoływanie funkcji podąża za kształtem OpenAI: opisujesz funkcje w tools, model zwraca tool_calls, uruchamiasz je i wysyłasz wyniki z powrotem z rolą tool. Pełny przebieg:

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)

Pętla ma większe znaczenie niż szczegóły rzutu kością. Model prosi o wywołanie, Twój kod je wykonuje, a Ty dodajesz zarówno wiadomość asystenta, jak i wynik narzędzia przed ponownym pytaniem. Pomiń wiadomość asystenta, a tool_call_id w wyniku nie będzie miał niczego do dopasowania, co jest zwykłą przyczyną zagubionej drugiej odpowiedzi. Z tool_choice ustawionym na auto model decyduje, czy w ogóle wywołać narzędzie, więc zawsze obsługuj gałąź, w której odpowiada bezpośrednio, jak robi to powyższy kod. Odpowiedzi strumieniowe mogą również zawierać tool_calls, dostarczane w kawałkach, które łączysz według indeksu przed parsowaniem argumentów.

Zweryfikuj argumenty przed ich użyciem. Model może wyprodukować niepoprawne wywołanie, a Twój kod powinien traktować argumenty jako niezaufane wejście, tak samo jak cokolwiek od użytkownika.

Wsady asynchroniczne w ramach limitów

Dla wielu promptów jednocześnie użyj AsyncOpenAI z semaforem dopasowanym do limitu równoległych zapytań. Pełny klucz pozwala na 8 równoległych zapytań, klucz próbny na 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())

Semafor obsługuje konkurencję, ale nie limit 300 zapytań na minutę. Dla długich wsadów dodaj małe opóźnienie lub ogranicznik wiadra tokenów. Przy cenie $0.15 za 1M tokenów wejściowych i $0.50 za 1M tokenów wyjściowych, 40 zapytań o slogany po około 60 tokenów wejściowych i 25 wyjściowych kosztuje $0.00036 + $0.0005 = $0.00086 łącznie, więc limity, a nie cena, kształtują projekt wsadów. Zobacz stronę cenową, aby doładować.

Limity zapytań, ponowne próby i nagłówki

Po przekroczeniu limitu API zwraca HTTP 429. Kod błędu to rate_limited przy zbyt wielu zapytaniach na minutę i concurrency przy zbyt wielu naraz. Tymczasowe przeciążenie również zwraca 429 z nagłówkiem Retry-After: 5. SDK ponawia niektóre błędy samodzielnie i możesz ustawić max_retries na kliencie, ale jawna obsługa jest jaśniejsza:

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

Tutaj with_raw_response daje dostęp do X-Balance-USD, X-Request-Id i nagłówków limitu zapytań. Zaloguj ID zapytania z każdym błędem, który chcesz zgłosić. Nigdy nie umieszczaj klucza w notatniku, który commitujesz, ani w kodzie front-end.

Testowanie bez wydawania kredytu i typowe błędy

Większość integracji Pythona to kod pomocniczy, który nie potrzebuje żywego modelu. Trzymaj wywołanie sieciowe za jedną funkcją, a Twoje testy mogą ją zastąpić. Używając wyłącznie biblioteki standardowej:

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"

Uruchom garść prawdziwych wywołań tylko na krawędziach: jeden test dymny na wdrożenie, plus okazjonalne uruchomienie zapisanego zestawu promptów. Pojedynczy test dymny z około 100 tokenami wejściowymi i 20 wyjściowymi kosztuje $0.000015 + $0.00001 = $0.000025, co jest niczym, a nieudane zapytania w ogóle nie są rozliczane. Kredyt próbny $0.50 jest znacznie większy niż potrzebuje zestaw testów.

Do pracy z promptami ustaw seed i temperature, aby móc porównać dwie wersje promptu systemowego na równych warunkach. Zmiana zarówno promptu, jak i ustawień próbkowania między uruchomieniami sprawia, że wyniki są nieczytelne. Zapisz wyjście każdego uruchomienia z ID zapytania z nagłówka X-Request-Id, a później będziesz mógł prześledzić zaskakującą odpowiedź.

Typowe błędy Pythona

Te powodują większość pytań wsparcia od użytkowników Pythona:

• Przekazywanie klucza tylko jako OPENAI_API_KEY. Jeśli ustawisz base_url w kodzie, ale zapomnisz o zmiennej klucza, SDK wyśle pusty lub zły klucz. Przekaż api_key jawnie.
• Zapominanie o /v1. Base URL kończy się na /v1. SDK dodaje /chat/completions samo.
• Używanie nazwy modelu z innej usługi. Tylko uncensored jest tu poprawny.
• Nieograniczona konkurencja. asyncio.gather na tysiącu korutyn bez semafora produkuje ścianę błędów concurrency.
• Traktowanie odpowiedzi strumieniowej jako jednego stringa. Połącz delty i pomiń chunki z pustą listą choices.
• Ignorowanie limitu kontekstu. Historia plus uzupełnienie muszą zmieścić się w 100 000 tokenach. Obcinaj stare tury, zanim go osiągniesz.

Informacje o doładowaniach i aktualny cennik znajdziesz na stronie cenowej. Usługa jest tylko tekstowa, więc nie ma wywołań obrazu, audio ani embeddingów do podpięcia, a prompty nie są używane do trenowania.

Więcej przewodników

• Bazowy URL GLM API: Połącz dowolnego klienta OpenAI
• Cennik GLM API: $0.15 Wejście, $0.50 Wyjście na 1 mln tokenów
• Gemma API: Mity vs fakty