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

في هذه الصفحة
مع البث، تصل الإجابة قطعة قطعة بينما لا يزال النموذج يولّدها، فيرى مستخدموك النص فورًا بدلًا من انتظار الاستجابة كاملة. تستخدم الواجهة البرمجية تنسيق البث القياسي من OpenAI (أحداث مرسلة من الخادم)، وتتولى حزمة openai الرسمية لـ Python ذلك عنك.
كل مثال أدناه كامل: انسخه إلى ملف وشغّله. تم التحقق من الأمثلة باستخدام Python 3.10.12 وopenai 3.31.0 مقابل خادم اختبار محلي يعيد إنتاج تنسيق البث الخاص بالواجهة البرمجية (الصياغة والتنفيذ).
الإعداد#
pip install openai
export CHEAPTOKENS_API_KEY="paste-your-key-here"مثال بسيط#
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- يستدل قبل أن يجيب. يأتي الاستدلال في حقل منفصل من الدلتا، reasoning_، والإجابة نفسها في content. تحتفظ SDK بالحقول غير المعروفة، لذا يمكنك قراءتها باستخدام getattr. هذا مثال كامل بعميله وبثه الخاص:
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:
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 من الأمثلة الأخرى:
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 افتراضيًا).
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)الأخطاء#
الأخطاء التي تحدث قبل القطعة الأولى — مفتاح خاطئ، رصيد فارغ، نموذج غير معروف — تُرفع كاستثناءات عند إنشاء البث:
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 يوقف التخزين المؤقت):
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]. الأسطر التي تبدأ بنقطتين هي تعليقات إبقاء الاتصال حيًا تُرسل خلال فترات التوقف الطويلة — يتجاهلها العملاء.



