الاستجابة المتدفقة
تتيح لك الاستجابة المتدفقة (Streaming) استقبال المخرجات في الوقت الفعلي أثناء توليد النموذج للمحتوى، مما يحسّن تجربة المستخدم والسرعة المُدرَكة.
آلية العمل#
تستخدم 诺玛AI بروتوكول Server-Sent Events (SSE) لتنفيذ الاستجابة المتدفقة:
- يضبط العميل `stream: true` عند إرسال الطلب
- يُعيد الخادم أجزاء المحتوى المُولَّد تدريجيًا (chunk)
- يُرسَل كل chunk عبر SSE ببادئة `data: `
- يُرسَل `data: [DONE]` عند انتهاء التوليد
التدفق ببروتوكول OpenAI#
cURL
الكود
curl https://as.apinoma.com/v1/chat/completions \
-H "Authorization: Bearer $APINOMA_API_KEY" \
-H "Content-Type: application/json" \
-N \
-d '{
"model": "openai/gpt-4o",
"messages": [{"role": "user", "content": "اكتب قصيدة عن البرمجة"}],
"stream": true
}'Python
TypeScript
التدفق ببروتوكول Anthropic#
Python
الكود
import anthropic
client = anthropic.Anthropic(
base_url="https://as.apinoma.com/anthropic",
api_key="<APINOMA_API_KEY_الخاص_بك>"
)
with client.messages.stream(
model="anthropic/claude-sonnet-4.6",
max_tokens=1024,
messages=[{"role": "user", "content": "اكتب قصيدة عن البرمجة"}]
) as stream:
for text in stream.text_stream:
print(text, end="", flush=True)TypeScript
التدفق + Function Calling#
تدعم الاستجابة المتدفقة أيضًا سيناريوهات استدعاء الدوال. يُخرِج النموذج أولًا طلب استدعاء الأداة بشكل متدفق، ثم تتابع المحادثة بعد انتهائك من معالجته:
الكود
stream = client.chat.completions.create(
model="openai/gpt-4o",
messages=[{"role": "user", "content": "كيف حال الطقس في الرياض اليوم؟"}],
tools=[{
"type": "function",
"function": {
"name": "get_weather",
"description": "جلب الطقس للمدينة المحددة",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "اسم المدينة"}
},
"required": ["city"]
}
}
}],
stream=True
)
for chunk in stream:
delta = chunk.choices[0].delta
if delta.tool_calls:
# معالجة استدعاء الأداة
print(f"استدعاء الأداة: {delta.tool_calls[0].function}")
elif delta.content:
print(delta.content, end="", flush=True)معالجة الأخطاء وإعادة الاتصال#
قد ينقطع الاتصال المتدفق بسبب مشكلات الشبكة. يُنصح بتنفيذ منطق إعادة الاتصال.
الكود
import time
def stream_with_retry(client, max_retries=3, **kwargs):
for attempt in range(max_retries):
try:
stream = client.chat.completions.create(stream=True, **kwargs)
for chunk in stream:
yield chunk
return # اكتمل بنجاح
except Exception as e:
if attempt < max_retries - 1:
wait = 2 ** attempt # تراجع أُسّي
print(f"\nانقطع الاتصال، {wait}s ثم أعد المحاولة...")
time.sleep(wait)
else:
raise eأفضل الممارسات#
- اضبط دائمًا مهلة زمنية — لتجنّب الانتظار اللانهائي
- عالِج الـ chunk غير المكتمل — قد لا يحتوي بعض الـ chunk على content
- نفّذ آلية إعادة الاتصال — باستخدام استراتيجية التراجع الأُسّي
- استخدم `flush` في الواجهة الأمامية — لضمان عرض المحتوى فورًا
آخر تحديث في 23 يونيو 2026