Bloga dön

API ve SDKKılavuz

Python'da akış yanıtları

Modelin yanıtını üretildiği anda resmi openai paketiyle yazdırın — senkron ve asenkron örnekler, akıl yürütme çıktısı, token kullanımı ve hata yönetimi.

Yazan: CheapTokens TeamYayınlanma tarihi: 3 dk okuma

Akış parçaları üzerinde bir for döngüsü içeren ve metnin token token göründüğü Python kodu
Bu sayfada
  1. Kurulum
  2. Minimal örnek
  3. Akıl yürütme modelleri
  4. Token kullanımı
  5. asyncio
  6. Erken durdurma ve zaman aşımları
  7. Hatalar
  8. Hatta neler var

Akış (streaming) ile yanıt, model onu hâlâ üretirken parça parça gelir; böylece kullanıcılarınız tüm yanıtı beklemek yerine metni hemen görür. API, standart OpenAI akış formatını (sunucu tarafından gönderilen olaylar) kullanır ve resmi openai Python paketi bunu sizin için halleder.

Aşağıdaki her örnek eksiksizdir: bir dosyaya kopyalayın ve çalıştırın. Örnekler Python 3.10.12 ve openai 3.31.0 ile, API'nin akış formatını yeniden üreten yerel bir test sunucusuna karşı doğrulanmıştır (sözdizimi ve yürütme).

Kurulum#

bash
pip install openai
export CHEAPTOKENS_API_KEY="paste-your-key-here"

Minimal örnek#

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Write a haiku about latency"}],
    stream=True,
)
for chunk in stream:
    if not chunk.choices:  # service chunk without text, e.g. the final usage chunk
        continue
    text = chunk.choices[0].delta.content
    if text:
        print(text, end="", flush=True)
print()

İki ayrıntı önemlidir:

  • choices içermeyen parçaları atlayın. Bazı parçalar metin taşımaz — örneğin token kullanımını içeren son parça.
  • delta.content boş olabilir. İlk parça genellikle yalnızca rolü, son parça ise yalnızca finish_reason taşır. Metni yalnızca mevcut olduğunda yazdırın.

Akıl yürütme modelleri#

zai-org/GLM-5.3-Flash yanıt vermeden önce akıl yürütür. Akıl yürütme, delta'nın ayrı bir alanı olan reasoning_content içinde, yanıtın kendisi ise content içinde gelir. SDK bilinmeyen alanları korur, böylece bunu getattr ile okuyabilirsiniz. Bu, kendi istemcisi ve kendi akışıyla eksiksiz bir örnektir:

python
import os
import sys
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="zai-org/GLM-5.3-Flash",
    messages=[{"role": "user", "content": "Is 1013 a prime number?"}],
    stream=True,
)
for chunk in stream:
    if not chunk.choices:
        continue
    delta = chunk.choices[0].delta
    thinking = getattr(delta, "reasoning_content", None)
    if thinking:
        print(thinking, end="", flush=True, file=sys.stderr)  # reasoning → stderr
    if delta.content:
        print(delta.content, end="", flush=True)              # answer → stdout
print()

Yanıtın ilk kelimesi görünmeden önce akıl yürütme biraz zaman alabilir — akış kullanmak için bir neden daha.

Token kullanımı#

Kullanımı stream_options ile isteyin; bu bilgi, choices içermeyen son parçada gelir:

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

stream = client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Explain idempotency in one paragraph"}],
    stream=True,
    stream_options={"include_usage": True},
)
usage = None
for chunk in stream:
    if chunk.usage:
        usage = chunk.usage
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

if usage:
    print(f"\ninput: {usage.prompt_tokens}, output: {usage.completion_tokens}")

İsteğin maliyetini bulmak için bu sayıları modelin ücretleriyle çarpın — bkz. Bir AI ajanının maliyeti nasıl tahmin edilir.

asyncio#

Async kodda AsyncOpenAI ve async for kullanın. İstemcinin kendi adı, async_client, vardır, böylece diğer örneklerdeki senkron client ile çakışmaz:

python
import asyncio
import os
from openai import AsyncOpenAI

async_client = AsyncOpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

async def main() -> None:
    stream = await async_client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V4-Flash-0731",
        messages=[{"role": "user", "content": "Name three uses of a message queue"}],
        stream=True,
    )
    async for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()

asyncio.run(main())

Erken durdurma ve zaman aşımları#

Üretimi durdurmak için döngüden çıkın. Akışı with ile açmak, blok sona erdiğinde bağlantıyı kapatır. İstemci zaman aşımı açıkça ayarlanır; SDK ayrıca bazı başarısız istekleri kendi başına yeniden dener (max_retries, varsayılan olarak 2).

python
import os
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
    timeout=120,  # seconds; long answers take time
)

received = ""
with client.chat.completions.create(
    model="deepseek-ai/DeepSeek-V4-Flash-0731",
    messages=[{"role": "user", "content": "Write a long story about a lighthouse"}],
    stream=True,
) as stream:
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            received += chunk.choices[0].delta.content
        if len(received) > 300:  # stop early: enough text received
            break
print(received)

Hatalar#

İlk parçadan önce meydana gelen hatalar — yanlış anahtar, boş bakiye, bilinmeyen model — akışı oluşturduğunuzda istisna olarak yükseltilir:

python
import os
import openai
from openai import OpenAI

client = OpenAI(
    base_url="https://api.cheaptokens.dev/v1",
    api_key=os.environ["CHEAPTOKENS_API_KEY"],
)

try:
    stream = client.chat.completions.create(
        model="deepseek-ai/DeepSeek-V4-Flash-0731",
        messages=[{"role": "user", "content": "Hello"}],
        stream=True,
    )
    for chunk in stream:
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()
except openai.AuthenticationError:          # 401: wrong or rotated key
    print("Check the API key")
except openai.BadRequestError as e:         # 400: e.g. unknown model ID
    print("Bad request:", e.message)
except openai.APIStatusError as e:          # other statuses, e.g. 402 when the balance is used up
    print(e.status_code, e.message)

Durum kodlarının ve çözümlerinin tam listesi Yaygın bağlantı hataları ve bunları nasıl düzelteceğiniz sayfasındadır.

Hatta neler var#

Hata ayıklamak için ham akışa curl ile bakın (-N arabelleğe almayı kapatır):

bash
curl -N https://api.cheaptokens.dev/v1/chat/completions \
  -H "Authorization: Bearer $CHEAPTOKENS_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"model": "deepseek-ai/DeepSeek-V4-Flash-0731", "stream": true,
       "messages": [{"role": "user", "content": "Hi"}]}'

Her olay, bir JSON parçası içeren bir data: {…} satırıdır; akış data: [DONE] ile biter. İki nokta üst üste ile başlayan satırlar, uzun duraklamalar sırasında gönderilen canlı tutma yorumlarıdır — istemciler bunları yok sayar.

Etiketler: