《AI Agent 工程实战》7.2 重试、死信与幂等

from dataclasses import dataclass

class ModelError(Exception):
    def __init__(self, msg: str, status: int | None = None):
        super().__init__(msg)
        self.status = status


@dataclass(frozen=True)
class Retryable:
    reason: str


@dataclass(frozen=True)
class Fatal:
    reason: str


RETRY_STATUS = {408, 429, 500, 502, 503, 504}


def classify(exc: Exception) -> Retryable | Fatal:
    status = getattr(exc, "status", None)
    if status in RETRY_STATUS:
        return Retryable(f"HTTP {status}")
    if isinstance(exc, (ConnectionError, TimeoutError)):
        return Retryable(type(exc).__name__)
    return Fatal(type(exc).__name__)

import random
import time


def with_retry(fn, *, max_attempts=3, base=0.05, cap=0.5, jitter=0.2,
               sleep=time.sleep, rng=None, trace=None):
    rng = rng or random.Random(7)
    for attempt in range(1, max_attempts + 1):
        try:
            return fn(), attempt
        except Exception as exc:
            tag = classify(exc)
            if isinstance(tag, Fatal) or attempt == max_attempts:
                if trace is not None:
                    trace.append(f"  attempt {attempt} 失败（{tag.reason}）→ 放弃")
                raise
            delay = min(cap, base * 2 ** (attempt - 1))
            delay *= 1 + rng.uniform(-jitter, jitter)
            if trace is not None:
                trace.append(f"  attempt {attempt} 失败（{tag.reason}）→ 退避 {delay:.3f}s")
            sleep(delay)

def process_run(r, conn, run_id, prompt, model, *, max_attempts=3):
    # 幂等门：run 已完成就直接复用旧结果，绝不重复执行
    with conn.cursor() as cur:
        cur.execute("SELECT status FROM runs WHERE id=%s", (run_id,))
        status = cur.fetchone()[0]
    if status == "succeeded":
        print(f"[idempotent] run={run_id} 已 succeeded，跳过执行（幂等命中）")
        return
    # 原子认领 queued -> running
    with conn.cursor() as cur:
        cur.execute("UPDATE runs SET status='running', started_at=%s "
                    "WHERE id=%s AND status='queued' RETURNING id",
                    (now(), run_id))
        claimed = cur.fetchone()
    conn.commit()
    if claimed is None:
        print(f"[skip] run={run_id} 非 queued")
        return

def process_run(r, conn, run_id, prompt, model, *, max_attempts=3):
    # ...（幂等门与原子认领同上）
    trace: list[str] = []
    try:
        result, attempts = with_retry(
            lambda: model.complete(prompt), max_attempts=max_attempts, trace=trace)
    except Exception as exc:
        print("\n".join(trace))
        reason = f"{type(exc).__name__}: {exc}"
        dead = {"run_id": run_id, "reason": reason, "attempts": max_attempts}
        r.lpush(DEAD, json.dumps(dead, ensure_ascii=False))
        with conn.cursor() as cur:
            cur.execute("UPDATE runs SET status='failed', error=%s, attempts=%s, "
                        "finished_at=%s WHERE id=%s",
                        (reason, max_attempts, now(), run_id))
        conn.commit()
        print(f"[deadletter] 进死信：{json.dumps(dead, ensure_ascii=False)}")
        return
    # ...（成功分支：写 succeeded + assistant 消息）

== 场景一：前两次失败、第三次成功 ==
  attempt 1 失败（HTTP 503）→ 退避 0.046s
  attempt 2 失败（HTTP 503）→ 退避 0.086s
[ok] run=6ba6871b-82f7-4812-ba45-e6eb375494ae succeeded（第 3 次尝试）：围绕「对比 pgvector 与 Qdrant」，命中 3 条本地语料，结论见引用。

== 场景二：同一 run 被重复投递（worker 崩后重投）==
[idempotent] run=6ba6871b-82f7-4812-ba45-e6eb375494ae 已 succeeded，跳过执行（幂等命中）
  库中 run 状态: ('succeeded', 3)

== 场景三：永远失败 -> 死信 ==
  attempt 1 失败（HTTP 503）→ 退避 0.046s
  attempt 2 失败（HTTP 503）→ 退避 0.086s
  attempt 3 失败（HTTP 503）→ 放弃
[deadletter] 进死信：{"run_id": "f42a32a8-0853-45f3-9d64-ae17fd288f9c", "reason": "ModelError: upstream 503", "attempts": 3}
  库中 run 状态: ('failed', 3, 'ModelError: upstream 503')
  死信队列 queue:runs:dead: ['{"run_id": "f42a32a8-0853-45f3-9d64-ae17fd288f9c", "reason": "ModelError: upstream 503", "attempts": 3}']