العودة إلى المدونة

API وSDKدليل

بث الاستجابات في Python

اطبع إجابة النموذج أثناء توليدها باستخدام حزمة openai الرسمية — أمثلة متزامنة وغير متزامنة، مخرجات الاستدلال، استخدام الرموز ومعالجة الأخطاء.

بواسطة CheapTokens Teamنُشر في 3 دقيقة قراءة

كود Python مع حلقة for على قطع البث والنص يظهر رمزًا برمز
في هذه الصفحة
  1. الإعداد
  2. مثال بسيط
  3. نماذج الاستدلال
  4. استخدام الرموز
  5. asyncio
  6. الإيقاف المبكر والمهلات الزمنية
  7. الأخطاء
  8. ما يحدث على الشبكة

مع البث، تصل الإجابة قطعة قطعة بينما لا يزال النموذج يولّدها، فيرى مستخدموك النص فورًا بدلًا من انتظار الاستجابة كاملة. تستخدم الواجهة البرمجية تنسيق البث القياسي من OpenAI (أحداث مرسلة من الخادم)، وتتولى حزمة openai الرسمية لـ Python ذلك عنك.

كل مثال أدناه كامل: انسخه إلى ملف وشغّله. تم التحقق من الأمثلة باستخدام Python 3.10.12 وopenai 3.31.0 مقابل خادم اختبار محلي يعيد إنتاج تنسيق البث الخاص بالواجهة البرمجية (الصياغة والتنفيذ).

الإعداد#

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

مثال بسيط#

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

تفصيلان مهمان:

  • تخطَّ القطع التي لا تحتوي على choices. بعض القطع لا تحمل نصًا — مثل القطعة الأخيرة التي تحتوي على استخدام الرموز.
  • قد يكون delta.content فارغًا. غالبًا ما تحمل القطعة الأولى الدور فقط، والأخيرة finish_reason فقط. اطبع النص فقط عندما يكون موجودًا.

نماذج الاستدلال#

zai-org/GLM-5.3-Flash يستدل قبل أن يجيب. يأتي الاستدلال في حقل منفصل من الدلتا، reasoning_content، والإجابة نفسها في content. تحتفظ SDK بالحقول غير المعروفة، لذا يمكنك قراءتها باستخدام getattr. هذا مثال كامل بعميله وبثه الخاص:

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

قد يستغرق الاستدلال وقتًا قبل ظهور الكلمة الأولى من الإجابة — وهذا سبب آخر للبث.

استخدام الرموز#

اطلب الاستخدام باستخدام stream_options؛ يأتي في القطعة الأخيرة، التي لا تحتوي على choices:

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

اضرب هذه الأرقام في أسعار النموذج للحصول على تكلفة الطلب — راجع كيفية تقدير تكلفة وكيل ذكاء اصطناعي.

asyncio#

في الكود غير المتزامن استخدم AsyncOpenAI وasync for. للعميل اسمه الخاص، async_client، حتى لا يتعارض مع العميل المتزامن client من الأمثلة الأخرى:

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

الإيقاف المبكر والمهلات الزمنية#

اترك الحلقة لإيقاف التوليد. فتح البث باستخدام with يغلق الاتصال عند انتهاء الكتلة. يتم تعيين مهلة العميل صراحةً؛ كما تعيد SDK محاولة بعض الطلبات الفاشلة من تلقاء نفسها (max_retries، 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)

الأخطاء#

الأخطاء التي تحدث قبل القطعة الأولى — مفتاح خاطئ، رصيد فارغ، نموذج غير معروف — تُرفع كاستثناءات عند إنشاء البث:

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)

قائمة كاملة برموز الحالة والإصلاحات موجودة في أخطاء الاتصال الشائعة وكيفية إصلاحها.

ما يحدث على الشبكة#

لتصحيح الأخطاء، انظر إلى البث الخام باستخدام curl (-N يوقف التخزين المؤقت):

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"}]}'

كل حدث هو سطر data: {…} مع قطعة JSON؛ ينتهي البث بـdata: [DONE]. الأسطر التي تبدأ بنقطتين هي تعليقات إبقاء الاتصال حيًا تُرسل خلال فترات التوقف الطويلة — يتجاهلها العملاء.