Thực tiễn ổn định API trung gian: thời chờ, thử lại, hàng đợi giới hạn tốc độ và ngắt kết nối truyền phát

Bạn hãy chia lỗi thành ba loại trước

Bước đầu tiên để đảm bảo ổn định không phải là tăng số lần thử lại, mà là biết trường hợp nào nên thử lại. Bạn có thể chia các lỗi phổ biến thành ba loại theo cách xử lý:

Loại lỗi	Biểu hiện điển hình	Cách xử lý
Có thể khôi phục tức thời	Giới hạn tốc độ 429, upstream_busy 503, kết nối bị đặt lại, thời chờ	Bạn hãy thử lại sau khi lùi thời gian theo cấp số nhân, hạn chế tổng số lần thử lại
Yêu cầu có vấn đề	400 (tổng input cộng max_tokens vượt 100.000, thân yêu cầu quá lớn), 403 content_blocked	Bạn không nên thử lại, hãy sửa yêu cầu hoặc báo ngay cho người dùng
Vấn đề trạng thái tài khoản	401 khóa không hợp lệ, 402 no_credit	Bạn không nên thử lại, hãy gửi cảnh báo cho người phụ trách, nạp tiền hoặc thay khóa API

Bạn nên ghi bảng này vào chú thích mã. Một sự cố phổ biến là: khi hết dư lượng, API bắt đầu trả về 402 ổn định, nhưng logic thử lại của bạn không phân biệt mã trạng thái, nên mỗi yêu cầu đều thử lại năm lần, vô tình nhân đôi lưu lượng lên năm lần và làm đỏ trang nhật ký.

Bạn hãy đặt thời chờ theo từng lớp

Chỉ đặt một thời chờ tổng thể là chưa đủ. Bạn nên chia thời chờ thành ba lớp để cân nhắc:

• Thời chờ kết nối: thời gian thiết lập kết nối TCP và TLS, thường 5 đến 10 giây là đủ; nếu không kết nối được, bạn hãy cho lỗi nhanh.
• Thời chờ đọc: thời gian chờ dữ liệu phản hồi. Với yêu cầu không truyền phát, bạn phải chờ đến khi toàn bộ văn bản được tạo xong, nên thời gian này phải bao phủ độ dài đầu ra tối đa; với yêu cầu truyền phát, nó trở thành khoảng lặng dài nhất giữa hai khối dữ liệu, 10 đến 30 giây là hợp lý hơn.
• Thời chờ tổng nghiệp vụ: thời gian chờ dài nhất mà nghiệp vụ của bạn có thể chịu đựng, bạn hãy đảm bảo bằng asyncio.wait_for hoặc lớp gateway, bao gồm tất cả lần thử lại.

Đầu ra càng dài thì yêu cầu không truyền phát càng tốn thời gian. Nếu bạn đặt max_tokens lên giới hạn 32.000 mà chỉ đặt thời chờ tổng 30 giây, bạn sẽ thường xuyên gặp thời chờ trong các tình huống văn bản dài. Cách an toàn hơn là luôn dùng truyền phát cho đầu ra dài, vừa cung cấp phản hồi tức thì cho người dùng, vừa làm rõ ý nghĩa của thời chờ. Tham số timeout trong SDK có thể nhận một số hoặc cấu hình chi tiết, bạn hãy tra tài liệu theo phiên bản đang dùng.

Thử lại lùi thời gian theo cấp số nhân cho 429 và 503

Ba điểm chính của việc lùi thời gian là: độ trễ tăng gấp đôi theo số lần thất bại, đặt giới hạn trên và thêm nhiễu ngẫu nhiên. Khi không có nhiễu, các tác vụ cùng thất bại sẽ thử lại cùng lúc và đè lên dịch vụ vừa phục hồi. Hàm dưới đây tắt tính năng thử lại mặc định của SDK, để bạn tự kiểm soát thống nhất và chỉ thử lại với 429, 503 cùng các lỗi kết nối:

复制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 có nghĩa là máy chủ đang bận tạm thời, bạn nên thử lại sau vài giây; độ trễ cơ sở 1 giây, giới hạn trên 20 giây và tối đa 5 lần thử lại ở trên là đủ cho tình huống này. Với 429, nếu bạn thấy lỗi này xuất hiện thường xuyên, vấn đề không nằm ở việc thử lại mà là đầu gửi chưa kiểm soát tốc độ, bạn hãy xem phần tiếp theo.

Hàng đợi đồng thời dưới giới hạn 300 yêu cầu mỗi phút

Mỗi khóa API được phép 300 yêu cầu mỗi phút, tương đương trung bình 5 yêu cầu mỗi giây. Nhiệm vụ xử lý hàng loạt dễ mắc lỗi ở đây: bạn dùng asyncio.gather để gửi hàng nghìn yêu cầu cùng lúc, vài chục yêu cầu đầu sẽ dùng hết hạn ngạch ngay lập tức, số còn lại đều nhận 429.

Cách đáng tin cậy là kiểm soát hai lớp. Semaphore hạn chế số lượng yêu cầu đang xử lý đồng thời, ngăn máy cục bộ bị quá tải kết nối và bộ nhớ; bộ điều nhịp hạn chế tốc độ gửi, trải đều các yêu cầu. Cả hai đều cần thiết: chỉ có semaphore, nếu mỗi yêu cầu nhanh, bạn vẫn có thể vượt 300 yêu cầu mỗi phút; chỉ có bộ điều nhịp, nếu mỗi yêu cầu chậm, số yêu cầu đang xử lý sẽ tích tụ. Ví dụ đặt tốc độ ở 270 yêu cầu mỗi phút, dành 10% dự phòng cho nhiều phiên bản, thử lại và sai lệch đồng hồ:

复制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())

Nếu nhiều tiến trình hoặc máy dùng chung một khóa API, bộ điều nhịp mức bộ nhớ ở trên là chưa đủ, vì giới hạn tốc độ được tính gộp theo khóa API. Lúc này bạn cần một thùng token chia sẻ, cách phổ biến là dùng Redis để đếm, hoặc đơn giản là cho tất cả yêu cầu xử lý hàng loạt đi qua một dịch vụ tiêu thụ hàng đợi riêng, dịch vụ này sẽ thống nhất xuất yêu cầu.

Phải làm gì khi truyền phát ngắt quãng

Kết nối truyền phát dễ vỡ hơn yêu cầu thông thường: thời gian kéo dài, đi qua nhiều thiết bị mạng, bất kỳ khoảng thời gian chờ nào cũng có thể ngắt kết nối. Nguyên tắc xử lý là “dữ liệu đã nhận là tài sản”, bạn đừng loại bỏ toàn bộ chỉ vì bị ngắt.

Cách triển khai dưới đây tích lũy các đoạn đã nhận, bắt ngoại lệ kết nối và thời chờ, sau đó gửi lại đoạn văn bản hiện có dưới dạng tin nhắn assistant để mô hình tiếp tục viết. Đây là cách xử lý cố gắng hết sức, sự chuyển tiếp khi tiếp tục viết có thể không hoàn hảo, phù hợp với văn bản dài và hội thoại, nhưng không phù hợp với đầu ra yêu cầu cấu trúc chặt chẽ như JSON. Khi đầu ra cấu trúc bị ngắt, cách an toàn hơn là yêu cầu lại từ đầu. Lưu ý rằng khi truyền phát kết thúc sẽ tự động kèm một phân đoạn usage, mã nguồn cũng trích xuất phân đoạn này:

复制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)

Khi tiếp tục viết, bạn hãy nhớ rằng văn bản đã gửi cũng tính vào token đầu vào, nên mỗi lần tiếp tục sẽ phát sinh chi phí phụ; việc hạn chế số lần max_resume là cần thiết.

Giảm tải, cắt mạch và tự kiểm tra trước khi lên bản chính

Thử lại giải quyết lỗi tức thời, nhưng nếu lỗi kéo dài vài phút, việc tiếp tục thử lại chỉ làm tích tụ yêu cầu. Bạn nên thêm lớp cắt mạch trên lớp thử lại: khi số lần thất bại liên tục đạt ngưỡng, ví dụ mười lần thất bại trong một phút, bạn hãy tạm ngừng gửi yêu cầu ra ngoài trong một khoảng thời gian ngắn, trong giai đoạn này hãy chuyển sang logic giảm tải, ví dụ trả về kết quả bộ nhớ đệm, thông báo cho người dùng thử lại sau hoặc đưa nhiệm vụ vào hàng đợi xử lý chậm. Sau thời gian làm mát, bạn chỉ cho phép một số ít yêu cầu thăm dò, nếu thành công thì khôi phục lưu lượng đầy đủ.

Bạn cũng cần lên kế hoạch giảm tải từ trước. Tính năng trò chuyện hướng người dùng có thể trả về một thông báo bận thân thiện; nhiệm vụ xử lý hàng loạt ngoại tuyến nên ghi các mục thất bại vào bảng lỗi và chạy bù tập trung sau khi toàn bộ nhiệm vụ hoàn tất, thay vì chờ đợi vô hạn trong luồng chính. Dù là cách nào, bạn cũng phải đảm bảo lỗi không bị nuốt lặng, ít nhất hãy ghi một dòng nhật ký kèm request id.

Cuối cùng, đây là danh sách tự kiểm tra trước khi lên bản chính: bạn đã tắt tính năng thử lại ngầm của SDK và chỉ giữ một điểm thử lại chưa? Bạn đã từ bỏ thử lại với 400, 401, 402, 403 chưa? Thời chờ tổng đã bao gồm tất cả lần thử lại chưa? Xử lý hàng loạt có kiểm soát tốc độ thay vì đồng loạt chưa? Truyền phát đã xử lý ngắt chưa? Usage đã được lưu đĩa chưa? Dư lượng có cảnh báo chưa? Bạn hãy vượt qua từng mục rồi mới ép tải, không làm ngược lại.

Bạn dùng usage để giám sát và đối chiếu

Mỗi phản hồi đều chứa usage, với truyền phát thì usage xuất hiện ở phân đoạn cuối. Lưu usage cùng tên chức năng vào đĩa là cách giám sát rẻ nhất. Giá đầu vào của trang web là 0,25 USD mỗi triệu token, giá đầu ra là 1,00 USD, bạn có thể ước tính chi phí mỗi yêu cầu tại chỗ:

复制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}"]
        )

Sau khi lưu, hãy kiểm tra 3 chỉ số mỗi ngày: trung vị token đầu vào để phát hiện prompt phình to; tỷ lệ token đầu ra (đắt gấp 4 lần); tỷ lệ 429/503 để điều chỉnh tốc độ. Thêm cảnh báo số dư để nạp tiền trước khi gặp 402. Xem cấu hình khung và FAQ.