ट्रांसिट API स्थिरता अभ्यास: टाइमआउट, रीट्राई, रेट लिमिट कतार और स्ट्रीमिंग डिस्कनेक्ट

पहले त्रुटियों को तीन श्रेणियों में बांटें

स्थिरता समस्याओं का पहला कदम रीट्राई बढ़ाना नहीं, बल्कि यह जानना है कि किसकी रीट्राई करनी है। संभालने के आधार पर सामान्य विफलताओं को तीन श्रेणियों में बांटा जा सकता है:

श्रेणी	विशिष्ट व्यवहार	संभालने की विधि
तत्काल पुनर्प्राप्य	429 रेट लिमिट, 503 upstream_busy, कनेक्शन रीसेट, टाइमआउट	एक्सपोनेंशियल बैकऑफ के बाद रीट्राई करें, कुल संख्या को सीमित करें
अनुरोध में ही समस्या है	400 (इनपुट + max_tokens 100,000 से अधिक, बड़ी रिक्वेस्ट बॉडी), 403 content_blocked	रीट्राई न करें, अनुरोध को ठीक करें या उपयोगकर्ता को सूचित करें
खाता स्थिति समस्या	401 अमान्य कुंजी, 402 no_credit	रीट्राई न करें, जिम्मेदार व्यक्ति को अलर्ट करें, टॉप-अप करें या कुंजी बदलें

इस टेबल को कोड कमेंट में लिखना सबसे अच्छा है। एक आम दुर्घटना यह है: बैलेंस खत्म होने के बाद API स्थिर रूप से 402 लौटाने लगता है, और आपका रीट्राई लॉजिक स्टेटस कोड में अंतर नहीं करता है, इसलिए हर अनुरोध पाँच बार रीट्राई करता है, जिससे पाँच गुना अनावश्यक ट्रैफ़िक बढ़ जाता है और लॉग लाल हो जाते हैं।

टाइमआउट को लेयर में सेट करें

केवल एक कुल टाइमआउट सेट करना पर्याप्त नहीं है। टाइमआउट को तीन परतों में विभाजित करने की सलाह दी जाती है:

• कनेक्शन टाइमआउट: TCP और TLS कनेक्शन स्थापित करने का समय, आमतौर पर 5 से 10 सेकंड पर्याप्त हैं। कनेक्शन न हो तो जल्दी फेल होना चाहिए।
• रीड टाइमआउट: प्रतिक्रिया डेटा की प्रतीक्षा का समय। नॉन-स्ट्रीमिंग अनुरोधों के लिए पूरा टेक्स्ट जनरेट होने तक इंतजार करना होता है, इसलिए इसे अधिकतम आउटपुट कवर करना चाहिए। स्ट्रीमिंग अनुरोधों में यह 'दो डेटा ब्लॉकों के बीच की अधिकतम सन्नाटा' बन जाता है, जो 10 से 30 सेकंड के बीच उचित है।
• व्यवसाय कुल टाइमआउट: आपकी व्यवसाय की सहनशीलता की अधिकतम प्रतीक्षा, जिसे asyncio.wait_for या गेटवे परत द्वारा सुनिश्चित किया जाता है, जिसमें सभी रीट्राई शामिल हैं।

आउटपुट जितना लंबा होगा, नॉन-स्ट्रीमिंग अनुरोध उतना ही अधिक समय लेंगे। यदि आप max_tokens को 32,000 की सीमा पर सेट करते हैं और केवल 30 सेकंड का कुल टाइमआउट रखते हैं, तो लंबे टेक्स्ट के मामले में बार-बार टाइमआउट आएगा। अधिक सुरक्षित तरीका यह है कि लंबे आउटपुट के लिए हमेशा स्ट्रीमिंग का उपयोग करें, जिससे उपयोगकर्ता को तुरंत प्रतिक्रिया मिलती है और टाइमआउट का अर्थ स्पष्ट होता है। SDK में timeout पैरामीटर एक संख्या या विभाजित कॉन्फ़िगरेशन ले सकता है, अपने संस्करण के दस्तावेज़ देखें।

429 और 503 के लिए एक्सपोनेंशियल बैकऑफ रीट्राई

बैकऑफ़ के तीन मुख्य बिंदु हैं: विलंबता विफलता की संख्या के साथ दोगुनी होनी चाहिए, एक ऊपरी सीमा निर्धारित करें, और यादृच्छिक ज़िटर (jitter) जोड़ें। बिना ज़िटर के, एक साथ विफल होने वाले सभी कार्य एक ही समय पर पुनः प्रयास करेंगे, जिससे हाल ही में बहाल सेवा फिर से दब जाएगी। निम्नलिखित फ़ंक्शन 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 मिलता है।

विश्वसनीय तरीका दो परतों का नियंत्रण है। सीमांकन (Semaphore) 'समानांतर में चल रहे' अनुरोधों की संख्या को सीमित करता है, जिससे स्थानीय कनेक्शन और मेमोरी भरने से बचा जाता है; टिकर (Ticker) 'निकासी दर' को सीमित करता है, अनुरोधों को समान रूप से फैलाता है। दोनों की आवश्यकता है: केवल सीमांकन होने पर, यदि एक अनुरोध तेज है, तो प्रति मिनट 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())

यदि कई प्रक्रियाएं या मशीनें एक ही कुंजी साझा करती हैं, तो मेमोरी-लेवल टिकर पर्याप्त नहीं होगा, क्योंकि रेट लिमिट कुंजी के आधार पर मिलाकर गिना जाता है। इस स्थिति में एक साझा टोकन बकेट की आवश्यकता होती है, जिसके लिए रेडिस का उपयोग करके गणना की जा सकती है, या फिर सभी बैच प्रोसेसिंग अनुरोधों को एक अलग कतार उपभोक्ता सेवा के माध्यम से भेजा जा सकता है, जो एकल निर्यात को नियंत्रित करे।

स्ट्रीमिंग इंटरप्ट के बाद क्या करें

स्ट्रीमिंग कनेक्शन साधारण अनुरोधों से अधिक नाजुक होता है: यह लंबे समय तक चलता है, कई नेटवर्क डिवाइसों से गुजरता है, और किसी भी घटक की आइडल टाइमआउट इसे काट सकती है। संभालने का सिद्धांत यह है कि 'प्राप्त सामग्री संपत्ति है', इंटरप्ट के कारण उसे पूरी तरह न छोड़ें।

निम्नलिखित कार्यान्वयन प्राप्त खंडों को जमा करता है, कनेक्शन और टाइमआउट अपवादों को पकड़ता है, और मौजूदा टेक्स्ट को assistant संदेश के रूप में भेजकर मॉडल को आगे लिखने देता है। यह एक सर्वोत्तम प्रयास का बचाव है, जिसमें जारी रखने का जुड़ाव पूर्ण नहीं हो सकता है, जो लंबे टेक्स्ट और चैट के लिए उपयुक्त है, लेकिन कठोर संरचना वाले आउटपुट जैसे JSON के लिए नहीं। संरचित आउटपुट में इंटरप्ट होने पर पूरी तरह से नया अनुरोध भेजना अधिक सुरक्षित है। ध्यान दें कि स्ट्रीमिंग के अंत में स्वचालित रूप से एक usage चिपक जुड़ जाता है, जिसे कोड में भी निकाला गया है:

复制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 त्रुटि आने से पहले जिम्मेदार व्यक्ति को टॉप-अप करने के लिए सूचित करें। अधिक एकीकरण विवरणों के लिए फ़्रेमवर्क कॉन्फ़िगरेशन देखें, और सामान्य प्रश्नों का सारांश सामान्य प्रश्न में दिया गया है।