API Tutorials

使用 CaptchaAI API 实现稳健的重试逻辑

自动化 Python 所有 CAPTCHA 类型
作者 CaptchaAI 中文团队
发布时间 Jun 15, 2026
使用 CaptchaAI API 实现稳健的重试逻辑
特色图片不可用

验证码识别请求偶尔失败很正常——队列打满、网络抖动、服务端过载都可能导致一次出错。关键是分清哪些错误值得重试、哪些重试也没用。本文贯穿一套统一思路：有限次数重试、带抖动的指数退避，以及 API 持续异常时自动熔断。

哪些错误该重试，哪些不该重试

先把错误分类，再动手写重试代码，能省下大量无意义的调试时间：

错误 是否重试 原因
ERROR_NO_SLOT_AVAILABLE 是 队列临时打满
HTTP 429 是 触发限流
HTTP 500/502/503 是 服务端临时故障
连接超时 是 网络抖动
CAPCHA_NOT_READY 继续轮询 任务仍在处理中
ERROR_WRONG_USER_KEY 否 配置错误，需修正 API Key
ERROR_KEY_DOES_NOT_EXIST 否 Key 无效
ERROR_ZERO_BALANCE 否 需先充值余额
ERROR_CAPTCHA_UNSOLVABLE 重新提交 用新参数发起全新任务

遇到双 11 这类流量高峰、国际出口带宽紧张的时段，连接超时和 HTTP 5xx 会更频繁——这正是重试逻辑体现价值的场景。

基础重试：指数退避加抖动

下面这段代码把"该不该重试"的判断和退避等待封装到一起，提交任务时直接调用即可：

import requests
import time
import random

API_KEY = "YOUR_API_KEY"
BASE_URL = "https://ocr.captchaai.com"

# Errors that should NOT be retried
PERMANENT_ERRORS = {
    "ERROR_WRONG_USER_KEY",
    "ERROR_KEY_DOES_NOT_EXIST",
    "ERROR_ZERO_BALANCE",
    "ERROR_BAD_PARAMETERS",
    "ERROR_WRONG_CAPTCHA_ID",
}

# Errors that should be retried
TRANSIENT_ERRORS = {
    "ERROR_NO_SLOT_AVAILABLE",
    "ERROR_TOO_MUCH_REQUESTS",
}


def submit_with_retry(method, max_retries=5, **params):
    """Submit task with retry on transient errors."""
    data = {"key": API_KEY, "method": method, "json": 1}
    data.update(params)

    for attempt in range(max_retries):
        try:
            resp = requests.post(
                f"{BASE_URL}/in.php", data=data, timeout=30,
            )

            # HTTP-level errors
            if resp.status_code in (429, 500, 502, 503):
                wait = _backoff(attempt)
                print(f"HTTP {resp.status_code}, retry in {wait:.1f}s")
                time.sleep(wait)
                continue

            result = resp.json()

            # Permanent errors — don't retry
            if result.get("request") in PERMANENT_ERRORS:
                raise RuntimeError(f"Permanent error: {result['request']}")

            # Transient errors — retry
            if result.get("request") in TRANSIENT_ERRORS:
                wait = _backoff(attempt)
                print(f"{result['request']}, retry in {wait:.1f}s")
                time.sleep(wait)
                continue

            # Success
            if result.get("status") == 1:
                return result["request"]

            # Unknown error
            raise RuntimeError(f"Unknown error: {result.get('request')}")

        except requests.ConnectionError:
            wait = _backoff(attempt)
            print(f"Connection error, retry in {wait:.1f}s")
            time.sleep(wait)

        except requests.Timeout:
            wait = _backoff(attempt)
            print(f"Timeout, retry in {wait:.1f}s")
            time.sleep(wait)

    raise RuntimeError(f"Failed after {max_retries} retries")


def _backoff(attempt, base=2, max_wait=60):
    """Exponential backoff with jitter."""
    wait = min(base ** attempt, max_wait)
    jitter = random.uniform(0, wait * 0.5)
    return wait + jitter

PERMANENT_ERRORS 和 TRANSIENT_ERRORS 判断顺序不能反。_backoff 里的 jitter 不是装饰——没有随机量，大量客户端会同一秒集体重试，反而更容易触发限流。

轮询阶段怎么重试

提交任务只是第一步，拿到 token 要靠轮询 res.php。轮询逻辑和提交阶段类似，但要多维护一个"连续错误计数"，避免偶发的单次错误就直接判定失败：

def poll_with_retry(task_id, timeout=120, max_poll_errors=3):
    """Poll for result with error retry."""
    start = time.time()
    consecutive_errors = 0

    while time.time() - start < timeout:
        time.sleep(5)

        try:
            resp = requests.get(f"{BASE_URL}/res.php", params={
                "key": API_KEY, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15)

            if resp.status_code in (429, 500, 502, 503):
                consecutive_errors += 1
                if consecutive_errors >= max_poll_errors:
                    raise RuntimeError("Too many poll errors")
                time.sleep(_backoff(consecutive_errors))
                continue

            data = resp.json()
            consecutive_errors = 0  # Reset on success

            if data["request"] == "CAPCHA_NOT_READY":
                continue

            if data["request"] in PERMANENT_ERRORS:
                raise RuntimeError(f"Solve error: {data['request']}")

            return data["request"]

        except (requests.ConnectionError, requests.Timeout):
            consecutive_errors += 1
            if consecutive_errors >= max_poll_errors:
                raise RuntimeError("Too many poll connection errors")
            time.sleep(_backoff(consecutive_errors))

    raise TimeoutError(f"Task {task_id} timeout after {timeout}s")

consecutive_errors 在每次成功响应后会重置为 0——达标条件是"连续"失败，不是累计失败次数，改错了会让脚本在网络稍有波动时就提前判定超时。

打包成生产级 Solver 类

把提交和轮询合并进一个类，再加上统计信息，方便接入正式的自动化流程：

class RetrySolver:
    """Production-grade solver with comprehensive retry logic."""

    def __init__(self, api_key, max_submit_retries=5, max_poll_retries=3,
                 poll_timeout=120):
        self.api_key = api_key
        self.base = "https://ocr.captchaai.com"
        self.max_submit_retries = max_submit_retries
        self.max_poll_retries = max_poll_retries
        self.poll_timeout = poll_timeout
        self.stats = {
            "total": 0, "success": 0, "retry": 0,
            "permanent_error": 0, "timeout": 0,
        }

    def solve(self, method, **params):
        self.stats["total"] += 1

        # Submit with retry
        task_id = self._submit(method, **params)

        # Poll with retry
        try:
            token = self._poll(task_id)
            self.stats["success"] += 1
            return token
        except TimeoutError:
            self.stats["timeout"] += 1
            raise

    def _submit(self, method, **params):
        data = {"key": self.api_key, "method": method, "json": 1}
        data.update(params)

        for attempt in range(self.max_submit_retries):
            try:
                resp = requests.post(
                    f"{self.base}/in.php", data=data, timeout=30,
                )

                if resp.status_code in (429, 500, 502, 503):
                    self.stats["retry"] += 1
                    time.sleep(_backoff(attempt))
                    continue

                result = resp.json()

                if result.get("request") in PERMANENT_ERRORS:
                    self.stats["permanent_error"] += 1
                    raise RuntimeError(f"Permanent: {result['request']}")

                if result.get("request") in TRANSIENT_ERRORS:
                    self.stats["retry"] += 1
                    time.sleep(_backoff(attempt))
                    continue

                if result.get("status") == 1:
                    return result["request"]

            except (requests.ConnectionError, requests.Timeout):
                self.stats["retry"] += 1
                time.sleep(_backoff(attempt))

        raise RuntimeError("Submit failed after retries")

    def _poll(self, task_id):
        start = time.time()
        errors = 0

        while time.time() - start < self.poll_timeout:
            time.sleep(5)
            try:
                resp = requests.get(f"{self.base}/res.php", params={
                    "key": self.api_key, "action": "get",
                    "id": task_id, "json": 1,
                }, timeout=15)

                if resp.status_code in (429, 500, 502, 503):
                    errors += 1
                    if errors >= self.max_poll_retries:
                        raise RuntimeError("Poll errors exceeded limit")
                    time.sleep(_backoff(errors))
                    continue

                data = resp.json()
                errors = 0

                if data["request"] == "CAPCHA_NOT_READY":
                    continue
                if data.get("status") == 1:
                    return data["request"]
                raise RuntimeError(f"Solve error: {data['request']}")

            except (requests.ConnectionError, requests.Timeout):
                errors += 1
                if errors >= self.max_poll_retries:
                    raise

        raise TimeoutError("Poll timeout")

    def get_stats(self):
        return self.stats


# Usage
solver = RetrySolver("YOUR_API_KEY")

token = solver.solve(
    "userrecaptcha",
    googlekey="SITE_KEY",
    pageurl="https://example.com",
)

print(solver.get_stats())

get_stats() 排查问题很有用：retry 偏高但 success 正常说明只是网络不稳；permanent_error 持续增长多半是配置出了问题，再重试也没用。

用熔断器防止无意义的持续请求

API 连续多次失败时，继续按原频率重试只会浪费配额、拖慢流水线。熔断器的作用是：失败次数达到阈值就直接拒绝新请求，等冷却时间过后再尝试放行：

class CircuitBreaker:
    """Stop requests when the service appears down."""

    def __init__(self, failure_threshold=5, recovery_time=60):
        self.failure_threshold = failure_threshold
        self.recovery_time = recovery_time
        self.failures = 0
        self.last_failure = 0
        self.state = "closed"  # closed=normal, open=blocking

    def can_proceed(self):
        if self.state == "closed":
            return True
        # Check if recovery time has passed
        if time.time() - self.last_failure > self.recovery_time:
            self.state = "half-open"
            return True
        return False

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"
            print(f"Circuit OPEN — pausing for {self.recovery_time}s")


# Integrate with solver
breaker = CircuitBreaker(failure_threshold=5, recovery_time=60)


def solve_with_breaker(method, **params):
    if not breaker.can_proceed():
        raise RuntimeError("Circuit open — API appears unavailable")

    try:
        token = solver.solve(method, **params)
        breaker.record_success()
        return token
    except RuntimeError:
        breaker.record_failure()
        raise

state 有三种取值：closed（正常放行）、open（拒绝请求）、half-open（冷却后试探放行一次）。这能让整个集群一起降速，而不是几百个并发请求同时死磕一个不可用的接口。

常见故障排查

问题 原因 处理方式
把永久性错误也重试了 没有按错误类型过滤 核对请求结果是否在 PERMANENT_ERRORS 集合中
重试次数没有上限 缺少 max_retries 务必显式设置重试上限
退避间隔涨得太慢 用了固定延迟 改用带抖动的指数退避
重试多次结果都一样 问题本身不是暂时性的 检查 API Key、账户余额和请求参数是否正确

常见问题

提交和轮询分别应该重试几次？

提交阶段 3–5 次，轮询阶段连续错误 2–3 次。超过这个次数收益很有限——问题往往已经不是暂时性的了。

熔断器（Circuit Breaker）打开之后大概多久会恢复？

取决于 recovery_time。示例代码设为 60 秒：熔断器打开后等待 60 秒会自动切换到"半开"状态并放行下一次请求，成功就重新闭合，失败就再次打开。

轮询间隔为什么设置成 5 秒，可以更短吗？

5 秒是折中值——太短容易撞上限流，太长又拖慢整体识别速度。可按实际观察调整，但不建议低于 2–3 秒。

ERROR_CAPTCHA_UNSOLVABLE 要不要重试？

可以，但别重试同一个任务 ID——相同任务不会给出不同结果。正确做法是用新参数重新提交一个全新任务，对应上面表格里"重新提交"这一行。

相关指南

立即注册 CaptchaAI，把这套重试与熔断逻辑接入你的识别流水线。

该文章已禁用评论。