الاستجابة المتدفقة

تتيح لك الاستجابة المتدفقة (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