ممارسات استقرار واجهة برمجة التطبيقات الوسيطة: المهلة الزمنية، إعادة المحاولة، طوابير حدّ المعدل وانقطاع الاتصال في البث

صنّف الأخطاء إلى ثلاث فئات

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

الفئة	المظهر النموذجي	طريقة المعالجة
قابل للاستعادة فورًا	حدّ المعدل 429، 503 upstream_busy، إعادة تعيين الاتصال، مهلة زمنية	إعادة المحاولة بعد التراجع الأسي، مع تحديد عدد المرات الإجمالي
مشكلة في الطلب نفسه	400 (الإدخال زائد 100,000 رمز أو حجم كبير)، 403 content_blocked	عدم إعادة المحاولة، تصحيح الطلب أو إعلام المستخدم مباشرة
مشكلة في حالة الحساب	401 مفتاح غير صالح، 402 no_credit	عدم إعادة المحاولة، تنبيه المسؤول، شحن الرصيد أو تغيير المفتاح

من الأفضل كتابة هذا الجدول في تعليقات الكود. حادث شائع هو: بعد نفاد الرصيد، تبدأ الواجهة بإرجاع 402 بشكل مستقر، وبما أن منطق إعادة المحاولة الخاص بك لم يميز بين أكواد الحالة، فقد أعاد كل طلب المحاولة خمس مرات، مما زاد حركة المرور بمقدار خمسة أضعاف دون داعٍ، وغطى السجلات باللون الأحمر.

يجب ضبط المهلة الزمنية على طبقات

ضبط مهلة زمنية كلية واحدة غير كافٍ. نوصي بتقسيم المهلة الزمنية إلى ثلاث طبقات:

• مهلة الاتصال: الوقت اللازم لإنشاء اتصال TCP و TLS، وعادةً ما تكون 5 إلى 10 ثوانٍ كافية، ويجب أن يفشل الاتصال بسرعة إذا لم يتحقق.
• وقت قراءة الاستجابة: الوقت المستغرق لانتظار بيانات الاستجابة. في الطلبات غير المتدفقة، يجب أن يغطي الوقت انتظار توليد النص بالكامل؛ أما في الطلبات المتدفقة، فيصبح «أطول صمت بين كتلتين بيانات متتاليتين»، حيث يُعدّ 10 إلى 30 ثانية أمراً أكثر معقولية.
• المهلة الزمنية الكلية للعملية: أطول انتظار يمكن لعملك تحمله، يتم ضمانه باستخدام asyncio.wait_for أو طبقة البوابة، ويشمل جميع عمليات إعادة المحاولة.

كلما كان الإخراج أطول، استغرق الطلب غير المتدفق وقتًا أطول. إذا قمت بضبط max_tokens إلى الحد الأقصى 32,000 وضبطت مهلة زمنية كلية تبلغ 30 ثانية فقط، فستواجه مهلات زمنية متكررة في سيناريوهات النصوص الطويلة. الطريقة الأكثر أمانًا هي استخدام البث المتدفق دائمًا للإطالة، مما يوفر للمستخدم ملاحظات فورية ويجعل معنى المهلة الزمنية واضحًا. يمكن لـ SDK تمرير رقم أو تكوين مقسم لمعلمة timeout، راجع التوثيق حسب الإصدار الذي تستخدمه.

إعادة المحاولة بالتراجع الأسي لأخطاء 429 و 503

ثلاث نقاط رئيسية للتراجع: تضاعف التأخير مع زيادة عدد مرات الفشل، تحديد حد أقصى، وإضافة اهتزاز عشوائي. بدون اهتزاز، ستعيد مجموعة المهام التي فشلت في نفس الوقت المحاولة في نفس اللحظة، مما يعيد ضغط الخدمة التي تعافت للتو. تقوم الدالة أدناه بإيقاف إعادة المحاولة المضمنة في SDK، ونتحكم فيها بشكل موحد، ونعيد المحاولة فقط لأخطاء 429 و 503 وأخطاء الاتصال:

复制import asyncio
import os
import random

from openai import AsyncOpenAI, APIConnectionError, APIStatusError, APITimeoutError

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=60.0,
    max_retries=0,  # 重试由下面的函数统一控制
)

RETRY_STATUS = {429, 503}


async def chat_with_retry(messages, max_attempts=5, base_delay=1.0, cap=20.0):
    for attempt in range(1, max_attempts + 1):
        try:
            resp = await client.chat.completions.create(
                model="uncensored", messages=messages, max_tokens=800
            )
            return resp
        except APIStatusError as e:
            if e.status_code not in RETRY_STATUS or attempt == max_attempts:
                raise  # 400/401/402/403 重试没有意义
            reason = f"HTTP {e.status_code}"
        except (APIConnectionError, APITimeoutError) as e:
            if attempt == max_attempts:
                raise
            reason = type(e).__name__
        delay = min(cap, base_delay * 2 ** (attempt - 1))
        delay = random.uniform(0, delay)  # 全抖动，避免所有任务同时醒来
        print(f"第 {attempt} 次失败（{reason}），{delay:.1f}s 后重试")
        await asyncio.sleep(delay)

يعني 503 أن الخادم مشغول مؤقتًا، نوصي بإعادة المحاولة بعد بضع ثوانٍ. التأخير الأساسي 1 ثانية، الحد الأقصى 20 ثانية، بحد أقصى 5 مرات، وهو كافٍ لتغطية هذا السيناريو. بالنسبة لـ 429، إذا لاحظت ظهوره بشكل متكرر، فهذا يعني أن المشكلة ليست في إعادة المحاولة، بل في عدم تحكم المرسل في معدل الإرسال، انظر القسم التالي.

طابور الطلبات المتزامنة تحت حدّ المعدل البالغ 300 طلب في الدقيقة

300 طلب في الدقيقة لكل مفتاح، أي ما يعادل 5 طلبات في الثانية. أسهل الأخطاء في مهام المعالجة الدفعية: إرسال آلاف الطلبات دفعة واحدة باستخدام asyncio.gather، حيث يستهلك العشرات الأوليون الحصة فورًا، ويتلقى الباقي خطأ 429.

الطريقة الموثوقة هي التحكم على مستويين: يحدّ المسمaphore من عدد «الطلبات في الطريق» لمنع انفجار اتصالات الذاكرة المحلية واستهلاك الذاكرة؛ ويحدّ مُعدّل الإرسال من «معدل الإرسال» لتوزيع الطلبات بالتساوي. كلاهما ضروري: عند استخدام المسمaphore فقط، قد تتجاوز الطلبات 300 طلب في الدقيقة إذا كانت سريعة؛ وعند استخدام مُعدّل الإرسال فقط، تتراكم الطلبات إذا كانت بطيئة. في المثال، تم تحديد المعدل بـ 270 طلباً في الدقيقة، مع ترك هامش 10% للتطبيقات المتعددة وإعادة المحاولة وأخطاء الساعة:

复制import asyncio
import os
import time

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=60.0,
    max_retries=0,
)

MAX_IN_FLIGHT = 8          # 同时在途的请求数
PER_MINUTE = 270           # 留 10% 余量，低于每分钟 300 次的上限


class Pacer:
    """把请求的发出时刻均匀铺开：每 60/PER_MINUTE 秒放行一个。"""

    def __init__(self, per_minute):
        self.interval = 60.0 / per_minute
        self.next_at = 0.0
        self.lock = asyncio.Lock()

    async def wait(self):
        async with self.lock:
            now = time.monotonic()
            delay = self.next_at - now
            if delay > 0:
                await asyncio.sleep(delay)
            self.next_at = max(now, self.next_at) + self.interval


sem = asyncio.Semaphore(MAX_IN_FLIGHT)
pacer = Pacer(PER_MINUTE)


async def summarize(idx, text):
    async with sem:            # 控制并发
        await pacer.wait()     # 控制速率
        resp = await client.chat.completions.create(
            model="uncensored",
            messages=[{"role": "user", "content": f"用两句话概括：{text}"}],
            max_tokens=120,
        )
        return idx, resp.choices[0].message.content


async def main():
    texts = [f"第 {i} 条工单的正文……" for i in range(1000)]
    tasks = [summarize(i, t) for i, t in enumerate(texts)]
    done = 0
    for coro in asyncio.as_completed(tasks):
        idx, out = await coro
        done += 1
        if done % 100 == 0:
            print(f"已完成 {done} 条")


asyncio.run(main())

إذا كانت هناك عمليات متعددة أو أجهزة متعددة تشارك نفس المفتاح، فإن منظم الذاكرة هذا غير كافٍ، حيث يتم حساب حدّ المعدل بشكل مجمع حسب المفتاح. في هذه الحالة، تحتاج إلى دلو رموز مشترك، والطريقة الشائعة هي استخدام Redis للعد، أو ببساطة توجيه جميع طلبات المعالجة الدفعية عبر خدمة استهلاك طابور منفصلة، والتي تتحكم في المخرج الموحد.

ماذا تفعل عند انقطاع البث المتدفق

اتصال البث المتدفق أكثر هشاشة من الطلبات العادية: فهو يستمر لفترة طويلة، ويمر عبر العديد من أجهزة الشبكة، وقد يقطعه أي مهلة زمنية للخمول في أي مرحلة. مبدأ المعالجة هو «المحتوى المستلم هو أصل»، لا تتجاهل كل شيء بسبب الانقطاع.

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

复制import asyncio
import os

from openai import AsyncOpenAI, APIConnectionError, APITimeoutError

client = AsyncOpenAI(
    base_url="https://api.llmzhongzhuan.com/v1",
    api_key=os.environ["API_KEY"],
    timeout=30.0,   # 流式场景下，这是相邻数据块之间允许的最长静默
    max_retries=0,
)


async def stream_once(messages):
    parts, usage = [], None
    stream = await client.chat.completions.create(
        model="uncensored", messages=messages, max_tokens=600, stream=True
    )
    try:
        async for chunk in stream:
            if chunk.usage:                      # 最后一个分片携带 usage
                usage = chunk.usage
            if chunk.choices and chunk.choices[0].delta.content:
                parts.append(chunk.choices[0].delta.content)
    except (APIConnectionError, APITimeoutError):
        return "".join(parts), None, False       # 中断：返回已收到的部分
    return "".join(parts), usage, True


async def stream_with_resume(question, max_resume=2):
    messages = [{"role": "user", "content": question}]
    text = ""
    for _ in range(max_resume + 1):
        part, usage, finished = await stream_once(messages)
        text += part
        if finished:
            return text, usage
        # 把已生成部分作为 assistant 消息，请模型接着写
        messages = [
            {"role": "user", "content": question},
            {"role": "assistant", "content": text},
            {"role": "user", "content": "请从上一条回复的结尾处继续，不要重复已写的内容。"},
        ]
    return text, None


if __name__ == "__main__":
    out, usage = asyncio.run(stream_with_resume("用三段话说明为什么日志要带 request id。"))
    print(out)
    print("usage:", usage)

عند الاستكمال، تذكر أن النص المرسل يُحسب أيضاً كرموز مدخلات، لذا فإن كل استكمال يكلف قليلاً إضافياً، ويحدّ من عدد مرات max_resume أمر ضروري.

التدرج، الدرع، وفحص ما قبل النشر

تعالج إعادة المحاولة الأعطال اللحظية، ولكن إذا استمر العطل لبضع دقائق، فإن الاستمرار في إعادة المحاولة سيؤدي إلى تراكم الطلبات. نوصي بإضافة طبقة درع فوق إعادة المحاولة: عندما يصل الفشل المتتالي إلى عتبة معينة، مثل عشرة فشل في الدقيقة، أوقف إرسال الطلبات للخارج لفترة قصيرة، وانتقل إلى منطق التدرج، مثل إرجاع نتيجة مخزنة مؤقتًا، أو إعلام المستخدم بالمحاولة لاحقًا، أو إعادة المهمة إلى الطابور للمعالجة لاحقًا. بعد انتهاء وقت التهدئة، اسمح فقط بطلبات استكشاف قليلة، واستعد للكم الكامل إذا نجحت.

يجب أيضًا التفكير في التدرج مسبقًا. يمكن لوظيفة الدردشة الموجهة للمستخدمين إرجاع رسالة مشغلة ودودة؛ أما مهام المعالجة الدفعية غير المتصلة بالشبكة فيجب عليها كتابة عناصر الفشل في جدول الفشل، وإكمالها دفعة واحدة بعد اكتمال التشغيل بأكمله، بدلاً من الانتظار بلا نهاية في العملية الرئيسية. في كلتا الحالتين، تأكد من أن الفشل لا يتم ابتلاعه بصمت، اترك على الأقل سطرًا من السجلات يحتوي على request id.

أخيرًا، نقدم قائمة مراجعة ما قبل النشر: هل تم إيقاف إعادة المحاولة الضمنية في SDK والاحتفاظ بإعادة المحاولة الخاصة بك فقط؟ هل تم التخلي عن أخطاء 400 و 401 و 402 و 403 مباشرة؟ هل شملت المهلة الزمنية الكلية جميع عمليات إعادة المحاولة؟ هل تحتوي المعالجة الدفعية على تحكم في المعدل بدلاً من الطلبات المتزامنة دفعة واحدة؟ هل تمت معالجة انقطاع البث؟ هل تم حفظ usage؟ هل يوجد تنبيه للرصيد؟ بعد اجتياز كل بند، قم بإجراء اختبار الضغط، وليس العكس.

المراقبة والمطابقة باستخدام usage

تحتوي كل استجابة على usage، والذي يظهر في الشريحة الأخيرة للبث. حفظه مع اسم الوظيفة هو أرخص وسيلة للمراقبة. سعر الإدخال في موقعنا هو 0.25 دولار لكل مليون رمز، والإخراج 1.00 دولار، يمكنك تقدير تكلفة كل طلب محليًا:

复制import csv
import time

LOG = "usage_log.csv"


def record_usage(feature, resp):
    u = resp.usage
    cost = u.prompt_tokens * 0.25 / 1_000_000 + u.completion_tokens * 1.00 / 1_000_000
    with open(LOG, "a", newline="", encoding="utf-8") as f:
        csv.writer(f).writerow(
            [int(time.time()), feature, u.prompt_tokens, u.completion_tokens, f"{cost:.6f}"]
        )

بعد الحفظ، يُنصح بمراقبة ثلاثة مؤشرات يومياً: الوسيط لرموز المدخلات لكل ميزة لاكتشاف تضخم الموجّهات؛ نسبة رموز المخرجات (حيث يكون سعر المخرجات أربعة أضعاف سعر المدخلات، لذا فهو عادةً الجزء الأكبر من الفاتورة)؛ ونسبة أخطاء 429 و503 (إذا ارتفعت النسبة، فهذا يعني أن معدل الإرسال أو استراتيجية إعادة المحاولة تحتاجان إلى تعديل). أضف أيضاً تنبيه الرصيد لإعلام المسؤول بالشحن قبل ظهور خطأ 402. لمزيد من التفاصيل حول التكامل، راجع إعداد إطار العمل، وتُجمع الأسئلة الشائعة في الأسئلة الشائعة.