แนวปฏิบัติความเสถียรของ API กลาง: ไทม์เอาต์, การลองใหม่, คิวจำกัดอัตรา และการตัดการเชื่อมต่อสตรีมมิง

แบ่งข้อผิดพลาดออกเป็นสามประเภทก่อน

ขั้นตอนแรกสำหรับปัญหาความเสถียรไม่ใช่การเพิ่มการลองใหม่ แต่คือการรู้ว่าอะไรควรลองใหม่ ข้อผิดพลาดทั่วไปสามารถแบ่งออกเป็นสามประเภทตามวิธีการจัดการ:

ประเภท	ลักษณะทั่วไป	วิธีการจัดการ
กู้คืนได้ทันที	429 rate limit, 503 upstream_busy, การเชื่อมต่อถูก reset, หมดเวลา	ลองใหม่หลังจากถอยหลังแบบเอ็กซ์โพเนนเชียล จำกัดจำนวนครั้งทั้งหมด
คำขอมีปัญหา	400 (input รวม max_tokens เกิน 100,000, body ใหญ่เกินไป), 403 content_blocked	ไม่ต้องลองใหม่ แก้ไขคำขอหรือแจ้งผู้ใช้โดยตรง
ปัญหาสถานะบัญชี	401 คีย์ไม่ถูกต้อง, 402 no_credit	ไม่ต้องลองใหม่ แจ้งผู้รับผิดชอบ เติมเงินหรือเปลี่ยนคีย์

ตารางนี้ควรเขียนไว้ใน comment ของโค้ด ปัญหาทั่วไปคือ: เมื่อเครดิตหมด API จะตอบกลับ 402 อย่างเสถียร และถ้าลอจิกการ retry ของคุณไม่แยกแยะ status code การขอข้อมูลแต่ละรายการจะ retry ห้าครั้ง ทำให้ปริมาณข้อมูลเพิ่มขึ้นห้าเท่าโดยเปล่าประโยชน์ พร้อมกันนี้ยังทำให้ log เต็มไปด้วยสีแดง

ตั้งค่าไทม์เอาต์แบบชั้น

การตั้งค่าไทม์เอาต์รวมเพียงชั้นเดียวไม่เพียงพอ แนะนำให้แยกไทม์เอาต์ออกเป็นสามชั้นเพื่อพิจารณา:

• ไทม์เอาต์การเชื่อมต่อ: เวลาในการสร้างการเชื่อมต่อ TCP และ TLS โดยปกติ 5 ถึง 10 วินาทีก็เพียงพอ หากเชื่อมต่อไม่ได้ควรล้มเหลวให้เร็วที่สุด
• read timeout: เวลาในการรอรับข้อมูลตอบกลับ สำหรับการขอข้อมูลแบบไม่สตรีมมิง ต้องรอให้สร้างข้อความทั้งหมดเสร็จก่อนจึงจะตอบกลับ ดังนั้นค่านี้ต้องครอบคลุมความยาวการส่งออกสูงสุด; สำหรับการขอข้อมูลแบบสตรีมมิง จะกลายเป็น "ความเงียบที่ยาวนานที่สุดระหว่างข้อมูลสองบล็อก" ซึ่ง 10 ถึง 30 วินาทีจะเหมาะสมกว่า
• ไทม์เอาต์ธุรกิจรวม: เวลารอที่ยาวที่สุดที่ธุรกิจของคุณทนได้ ใช้ asyncio.wait_for หรือชั้นเกตเวย์เพื่อให้แน่ใจ รวมถึงการลองใหม่ทั้งหมด

ข้อความยาวขึ้นทำให้การตอบแบบไม่สตรีมมิงใช้เวลานานขึ้น หากคุณตั้งค่า max_tokens ไว้ที่ค่าสูงสุด 32,000 และตั้งค่าเวลาหมดอายุรวมไว้ที่ 30 วินาที โอกาสเกิดเวลาหมดอายุในสถานการณ์ที่สร้างข้อความยาวจะสูงขึ้นมาก วิธีที่ปลอดภัยกว่าคือใช้สตรีมมิงสำหรับข้อความยาวเสมอ ซึ่งผู้ใช้จะได้รับผลตอบกลับทันที และทำให้ความหมายของเวลาหมดอายุชัดเจนขึ้น พารามิเตอร์ timeout ใน SDK สามารถรับค่าตัวเลขหรือค่าการกำหนดค่าแบบแยกส่วนได้ โดยดูเอกสารตามเวอร์ชันที่คุณใช้งาน

การลองใหม่แบบถอยหลังแบบเอ็กซ์โพเนนเชียลสำหรับ 429 และ 503

การรีเทรต้องเพิ่มดีเลย์, ตั้งค่าสูงสุด, และเพิ่มจITTER เพื่อไม่ให้รีเทรพร้อมกันทับบริการ ฟังก์ชันนี้ปิดรีเทร 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) จำกัดจำนวน "คำขอที่กำลังดำเนินการ" เพื่อป้องกันไม่ให้การเชื่อมต่อและหน่วยความจำของเครื่องถูกใช้จนล้น ตัวควบคุมจังหวะ (rate limiter) จำกัด "อัตราการส่งคำขอ" เพื่อกระจายคำขออย่างสม่ำเสมอ ทั้งสองอย่างขาดกันไม่ได้: หากมีเพียงตัวจำกัดปริมาณ แต่คำขอแต่ละรายการเสร็จเร็ว จำนวนคำขออาจเกิน 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())

หากหลายโหนดใช้คีย์เดียวกัน ต้องใช้ token bucket ร่วมกัน เช่น ใช้ Redis หรือใช้ queue service แยกเพื่อจัดการคำขอทั้งหมด

จะทำอย่างไรหากการสตรีมมิงขาด

การเชื่อมต่อสตรีมมิงบอบบางกว่าคำขอปกติ: ใช้เวลานาน ผ่านอุปกรณ์เครือข่ายหลายตัว ความล่าช้าแบบว่างเปล่าในขั้นตอนใดขั้นตอนหนึ่งอาจตัดการเชื่อมต่อได้ หลักการจัดการคือ "เนื้อหาที่ได้รับแล้วคือสินทรัพย์" อย่าทิ้งทั้งหมดเนื่องจากมีการตัดขาด

การนำไปใช้ด้านล่างจะรวบรวมชิ้นข้อมูลที่รับมาสะสมกัน เมื่อจับข้อผิดพลาดการเชื่อมต่อและเวลาหมดอายุได้ จะนำข้อความที่มีอยู่ไปขอเป็นข้อความของ 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)

เมื่อเขียนต่อต้องจำไว้ว่า ข้อความที่ส่งไปแล้วจะนับเป็น input token ด้วย ดังนั้นการเขียนต่อแต่ละครั้งจะมีต้นทุนเพิ่มเติมเล็กน้อย การจำกัดจำนวนครั้งของ max_resume จึงเป็นสิ่งจำเป็น

การลดระดับ, การตัดวงจร และการตรวจสอบก่อนใช้งานจริง

การลองใหม่แก้ปัญหาชั่วคราว หากความล้มเหลวคงอยู่หลายนาที การลองใหม่ต่อไปเพียงแต่สะสมคำขอ แนะนำให้เพิ่มชั้นการตัดวงจรเหนือการลองใหม่: เมื่อความล้มเหลวต่อเนื่องถึงเกณฑ์ เช่น ล้มเหลวสิบครั้งในหนึ่งนาที ให้หยุดส่งคำขอออกไปชั่วคราว ในช่วงเวลานี้ใช้ตรรกะการลดระดับโดยตรง เช่น ส่งผลลัพธ์แคช, แจ้งผู้ใช้ให้ลองอีกครั้งในภายหลัง หรือส่งงานกลับเข้าคิวเพื่อจัดการภายหลัง เมื่อเวลาเย็นลงอนุญาตเฉพาะคำขอตรวจสอบจำนวนน้อย หากสำเร็จจึงกู้คืนปริมาณเต็ม

การลดระดับต้องคิดล่วงหน้าเช่นกัน ฟีเจอร์แชทที่面向ผู้ใช้สามารถส่งข้อความยุ่งๆ ที่เป็นมิตรได้; งานแบตช์ที่ออฟไลน์ควรเขียนรายการที่ล้มเหลวลงในตารางล้มเหลว รอให้รันเสร็จทั้งหมดแล้วค่อยรันใหม่รวมกัน แทนที่จะรออย่างไม่สิ้นสุดในกระบวนการหลัก ไม่ว่ากรณีใด ต้องรับประกันว่าความล้มเหลวจะไม่ถูกกลืนเงียบ อย่างน้อยต้อง留下一行带 request id 的日志。

สุดท้าย นี่คือรายการตรวจสอบก่อนการปล่อยใช้งาน: ได้ปิดการลองซ้ำอัตโนมัติของ SDK และเหลือจุดเดียวที่ควบคุมเองหรือไม่; ได้ละเว้นสถานะ 400, 401, 402, 403 ทันทีหรือไม่; เวลาหมดอายุรวมได้รวมเวลาของการลองซ้ำทั้งหมดแล้วหรือไม่; งานแบบ batch มีการควบคุมอัตราหรือไม่ แทนที่จะส่งคำขอพร้อมกันทั้งหมด; มีการจัดการการขัดจังหวะสตรีมมิงหรือไม่; มีการบันทึกข้อมูล usage แล้วหรือไม่; มีการแจ้งเตือนเมื่อยอดเงินเหลือต่ำหรือไม่ เมื่อผ่านรายการเหล่านี้ทั้งหมดแล้ว จึงค่อยทำการทดสอบแรงกด (load test) ไม่ใช่ทำในลำดับกลับกัน

ใช้ usage เพื่อการตรวจสอบและการตรวจสอบบัญชี

แต่ละการตอบกลับมี usage รวมถึงการสตรีมมิงซึ่งปรากฏในชิ้นส่วนสุดท้าย การบันทึกมันพร้อมกับชื่อฟังก์ชันเป็นวิธีการตรวจสอบที่ประหยัดที่สุด ราคา input ของเว็บไซต์นี้คือ $0.25 ต่อล้าน token, output คือ $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}"]
        )

หลังจากใช้งานแล้ว แนะนำให้ตรวจสอบสามตัวชี้วัดทุกวัน: ค่ามัธยฐานของ input token ในแต่ละฟีเจอร์ เพื่อตรวจจับว่าพรอมต์มีการขยายขนาดโดยไม่รู้ตัวหรือไม่ อัตราส่วนของ output token เนื่องจากราคาต่อ output token สูงกว่า input ถึง 4 เท่า จึงมักเป็นต้นทุนหลักในบิล และอัตราส่วนของ 429 ต่อ 503 หากอัตราส่วนนี้เพิ่มขึ้น แสดงว่าจำเป็นต้องปรับอัตราการส่งหรือกลยุทธ์การ retry เพิ่มการแจ้งเตือนยอดคงเหลือเพื่อแจ้งให้ผู้รับผิดชอบเติมเงินก่อนเกิดข้อผิดพลาด 402 รายละเอียดเพิ่มเติมเกี่ยวกับการเชื่อมต่อสามารถดูได้ที่ การกำหนดค่าเฟรมเวิร์ก และคำถามที่พบบ่อยรวบรวมไว้ใน คำถามที่พบบ่อย