Webhook异步任务回调设计实战教程

AI API Webhook 与异步任务回调实战指南 2026：长任务、回调验证与失败重试设计

2026-09-03 · 约 11 分钟阅读

AI API 的调用时长差异极大——普通 Chat Completion 是秒级，但视频生成、批量推理、长链路 Agent 任务常常需要 30 秒到 30 分钟不等。这类长任务必须用异步 + Webhook 回调的方式，否则 HTTP 连接会超时、客户端会重试浪费配额、甚至整个任务被中断。本文讲清楚怎么设计一套可靠的"提交 - 回调 - 校验"闭环。

一、为什么 AI 长任务几乎必然用异步

同步调用的硬约束：

  • 浏览器 HTTP 超时：通常 30 秒 - 2 分钟
  • 网关超时：Nginx 默认 60 秒，云函数 5-15 分钟
  • 客户端耐心：用户等待超过 30 秒就开始刷新页面

而典型 AI 长任务耗时：

任务同步模式下耗时
单次 Chat Completion2-10 秒
长文档摘要 + 多步推理30-120 秒
Agent 多工具调用1-10 分钟
视频生成（10s 1080p）5-30 分钟
批量推理（10 万样本）几十分钟到数小时

结论：超过 30 秒的任务基本必须异步化。

二、两种异步化路径

2.1 厂商原生异步 API

主流厂商都提供"提交任务 + 轮询/回调"接口：

厂商异步入口通知机制
OpenAIclient.beta.messages.batches / Batch API轮询 status
AnthropicBatch API（消息批量）轮询 + Webhook（部分场景）
GoogleVertex AI Batch Prediction轮询
国内中转站普遍透传 OpenAI Batch API取决于具体平台

OpenAI 的 Batch API 是最成熟的异步范式，48 小时内完成、便宜 50%。

2.2 自建异步队列 + Webhook 回调

适用场景：模型本身只支持同步调用，但任务本身需要长链路。

客户端 → 你的 API（接收请求，立刻 202）
   └→ 推入任务队列（Redis/Kafka/SQS）
        └→ Worker 拉取任务，调 AI API（同步等结果）
             └→ 完成后向客户端 Webhook URL 推送结果
                  └→ 客户端收到回调，更新业务状态

2.3 怎么选

  • 任务量适中、任务本身是"一次性提交大批数据"：用厂商原生 Batch API
  • 任务规模中等、需要灵活触发、还要保留重试 / 监控：自建队列 + Worker
  • 视频生成、Agent 长链路：自建异步 + Webhook 回调

三、Webhook 回调设计核心要点

3.1 你的服务作为"Webhook 接收方"：签名校验

凡是接收外部 Webhook，必须做签名校验——否则任何人都能伪造请求给你刷数据。

import hmac
import hashlib
from fastapi import FastAPI, Header, HTTPException

app = FastAPI()
WEBHOOK_SECRET = "your-shared-secret-from-provider"

@app.post("/webhooks/ai-task")
async def handle_webhook(
    payload: bytes = Body(...),
    signature: str = Header(..., alias="X-AI-Signature"),
):
    # 1. 计算 HMAC-SHA256
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    # 2. 常量时间比较（防计时攻击）
    if not hmac.compare_digest(expected, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")
    
    # 3. 解析 payload
    event = json.loads(payload)
    process_event(event)
    return {"ok": True}

关键点：

  • 永远用常量时间比较（hmac.compare_digest），别用 ==
  • 签名计算时包含 raw body，不是解析后的 JSON（格式变化会让签名失败）
  • Secret 通过环境变量配置，不要写死在代码里

3.2 你的服务作为"Webhook 发送方"：让回调可信可重试

当你向客户端的回调地址发请求时，也要做对等的设计：

import time
import uuid
import httpx

async def deliver_webhook(url, payload, max_retries=5):
    headers = {
        "Content-Type": "application/json",
        "X-Webhook-Id": str(uuid.uuid4()),
        "X-Webhook-Timestamp": str(int(time.time())),
        "X-Webhook-Signature": sign_payload(payload),
    }
    
    for attempt in range(max_retries):
        try:
            async with httpx.AsyncClient(timeout=30) as client:
                resp = await client.post(url, json=payload, headers=headers)
                if 200 <= resp.status_code < 300:
                    return True
        except httpx.RequestError:
            pass
        
        # 指数退避
        await asyncio.sleep(2 ** attempt)
    
    # 重试耗尽 → 进入死信队列，告警人工处理
    dead_letter_queue.push(url=url, payload=payload)
    return False

3.3 客户端必须实现的：幂等去重

WebHook 协议本质是 "至少一次" 投递——同一个事件可能被投递多次（发送方重试、客户端网络抖动等）。接收端必须按 Webhook ID 去重。

seen_webhook_ids = set()  # 生产环境用 Redis + TTL

@app.post("/webhooks/ai-task")
async def handle_webhook(request: Request):
    webhook_id = request.headers["X-Webhook-Id"]
    if webhook_id in seen_webhook_ids:
        return {"ok": True, "duplicate": True}  # 静默丢弃
    
    # 业务处理
    ...
    
    seen_webhook_ids.add(webhook_id)
    return {"ok": True}

注意：内存 set 只适合 demo。生产必须用 Redis，TTL 至少 24 小时，覆盖最长的"客户端重试窗口"。

四、长任务的"超时续命"机制

异步任务最大的隐患是 "以为没跑完，其实早就跑完了" 和 "以为跑完了，其实没跑"。

4.1 三种超时控制策略

策略适用实现
硬超时客户端要立刻知道结果客户端 30 秒后丢弃等待，依赖 Webhook
软超时任务最长 1 小时Worker 端单次任务超时则进入失败队列
分阶段超时多步骤任务每步单独计时，超时则中断整链路

4.2 任务状态机

PENDING → RUNNING → SUCCEEDED / FAILED
              ↓
          CANCELLED (客户端主动取消)

每个状态都要可查询：

@app.get("/tasks/{task_id}")
async def get_task(task_id: str):
    return {
        "task_id": task_id,
        "status": "RUNNING",
        "submitted_at": "2026-09-03T10:00:00Z",
        "progress": 0.62,  # 0-1
        "estimated_remaining_sec": 180,
        "result_url": None,  # 只有 SUCCEEDED 才会有
    }

progress 字段对长任务用户体验极大——用户能看到"60% 完成"比"加载中"舒服得多。

4.3 心跳（Heartbeat）防止假死

对于 30 分钟以上的任务，建议加心跳上报：

async def long_running_task(task_id):
    last_heartbeat = time.time()
    
    while not done:
        result = do_step()
        
        # 每 30 秒上报一次心跳
        if time.time() - last_heartbeat > 30:
            await redis.set(f"task:{task_id}:heartbeat", time.time())
            last_heartbeat = time.time()
        
        if is_orphaned(task_id):
            break  # 任务被取消或客户端失联，提前退出

客户端侧可以监控"心跳超过 60 秒没更新"的任务，做超时处理。

五、OpenAI Batch API 实战示例

OpenAI 的 Batch API 是异步任务的代表。提交一批任务，48 小时内完成，结果通过文件返回。

from openai import OpenAI

client = OpenAI(api_key="***", base_url="https://你的中转站/v1")

# 1. 准备批量任务文件（jsonl）
import json
batch_requests = []
for i, prompt in enumerate(prompts):
    batch_requests.append({
        "custom_id": f"task-{i}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-5.6",
            "messages": [{"role": "user", "content": prompt}],
        },
    })

with open("batch_input.jsonl", "w") as f:
    for req in batch_requests:
        f.write(json.dumps(req) + "\n")

# 2. 上传文件
uploaded = client.files.create(file=open("batch_input.jsonl", "rb"), purpose="batch")

# 3. 提交 batch
batch_job = client.batches.create(
    input_file_id=uploaded.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)

# 4. 轮询状态（或注册 callback）
while True:
    status = client.batches.retrieve(batch_job.id)
    if status.status in ("completed", "failed", "cancelled"):
        break
    time.sleep(60)

# 5. 下载结果
if status.status == "completed":
    result = client.files.content(status.output_file_id)
    for line in result.text.splitlines():
        print(json.loads(line))

省钱：Batch API 比同步调用便宜 50%，代价是 24 小时内返回。

六、Agent 多步执行的异步化

Agent 场景特别复杂——一次"提交"可能涉及 5-10 次 LLM 调用 + 多次工具调用。直接同步跑会大量超时。

6.1 推荐架构

客户端 POST /agent/run → 立刻返回 task_id
Worker 异步处理：
  ├── Step 1: LLM Call (Plan)
  ├── Step 2: Tool Call (Search)
  ├── Step 3: LLM Call (Reflect)
  ├── Step 4: Tool Call (Code Exec)
  ├── Step 5: LLM Call (Final Answer)
  └── 完成后 → Webhook 推送结果

6.2 状态持久化

每一步的状态都要落库（PostgreSQL / Redis），不能只放在内存：

CREATE TABLE agent_tasks (
    task_id UUID PRIMARY KEY,
    user_id TEXT NOT NULL,
    status TEXT NOT NULL,           -- PENDING/RUNNING/SUCCEEDED/FAILED
    current_step INT DEFAULT 0,
    total_steps INT,
    steps JSONB,                   -- 每步的输入/输出/耗时
    final_result JSONB,
    error_message TEXT,
    submitted_at TIMESTAMP,
    updated_at TIMESTAMP
);

Worker 重启后能从数据库恢复任务继续执行，这是生产级异步系统的最低要求。

七、监控与告警

异步系统出问题比同步系统更隐蔽——客户端收不到回调但你不知道。建议监控：

指标阈值
Webhook 投递成功率< 99% 告警
Webhook P95 延迟> 5 秒告警
任务超时率> 1% 告警
死信队列堆积> 0 立刻告警
心跳超时任务数> 5 告警

八、自检清单

部署一个异步 + Webhook 系统前，对照检查：

  • [ ] 签名校验：接收方校验签名，发送方正确签名
  • [ ] 幂等去重：用 Webhook ID 去重，TTL 覆盖最长重试窗口
  • [ ] 超时控制：客户端、服务端、数据库三层都有超时
  • [ ] 状态持久化：所有中间状态落库
  • [ ] 重试策略：指数退避，最大次数，有死信队列
  • [ ] 监控告警：成功率、延迟、堆积、超时都有
  • [ ] 安全合规：Webhook 回调地址 HTTPS、IP 白名单（可选）

九、把异步任务跑起来

如果你正在搭建 AI 应用的异步任务系统，首先要选一个支持 OpenAI Batch API 兼容协议的中转平台——这样你的异步代码可以一份代码跑多家供应商。openairouter.net 实时汇总了主流 AI API 中转站的 Batch API 支持情况、价格折扣和稳定性，方便你按业务量级挑选最合适的合作伙伴。

找到最适合你的 AI API 中转站

收录 125+ 服务商，按价格、模型、标签一键筛选

查看所有中转站 →
← 返回所有教程