用一把聚合 key 跑异步任务：只有同步接口时怎么模拟 webhook 回调

为什么需要模拟 webhook

许多 LLM API 聚合站只提供同步接口，没有服务端 webhook 回调机制。这意味着你无法在长时间生成任务完成时自动收到通知。但通过轮询加本地通知的薄壳方案，你依然可以构建接近异步的体验，而无需等待官方支持。

核心思路

1. 提交生成请求后，立即获得任务 ID 或请求 ID。
2. 使用一个轻量级轮询器定期检查任务状态。
3. 当检测到任务完成时，触发本地回调（如执行函数、发送消息到队列或调用本地 HTTP 端点）。
4. 回调中处理结果，模拟 webhook 的行为。

实现步骤

1. 提交任务并保存标识

调用聚合站的同步接口发起生成请求。由于接口是同步的，你需要在请求头或参数中指定异步模式（如果支持），或者直接使用返回的响应 ID。将 ID 与回调信息关联存储。

2. 构建轮询器

轮询器是一个简单的循环，按固定间隔查询任务状态。注意：

• 间隔不宜过短，避免不必要的负载。
• 设置超时和重试机制。
• 使用指数退避策略减少请求频率。

3. 触发本地回调

当轮询器发现任务状态为“完成”时，触发预先注册的回调函数。回调可以是：

• 一个本地函数调用（如 Node.js 中的 EventEmitter）。
• 一个 HTTP 请求到你的本地服务端点。
• 一条消息发送到消息队列（如 Redis、RabbitMQ）。

4. 错误处理与重试

如果任务失败或超时，回调应收到错误通知。实现重试逻辑，但注意不要无限重试。

示例代码结构

以下是一个简化的伪代码示例，展示轮询加回调的薄壳：

import time
import requests

def submit_task(prompt, callback_url):
    # 调用聚合站同步接口
    response = requests.post('https://api.aggregator.com/v1/generate', json={'prompt': prompt})
    task_id = response.json()['id']
    # 存储任务ID与回调URL的映射
    store[task_id] = callback_url
    return task_id

def poll_task(task_id):
    while True:
        status = requests.get(f'https://api.aggregator.com/v1/tasks/{task_id}').json()
        if status['state'] == 'completed':
            return status['result']
        elif status['state'] == 'failed':
            raise Exception('Task failed')
        time.sleep(5)  # 轮询间隔

def notify_callback(callback_url, result):
    requests.post(callback_url, json={'result': result})

# 使用示例
task_id = submit_task('写一篇关于AI的文章', 'http://localhost:8080/callback')
result = poll_task(task_id)
notify_callback(store[task_id], result)

注意事项

• 轮询会增加 API 调用次数，但聚合站按官方价 ×1.3 扣费，合理设计轮询频率可控制成本。
• 如果聚合站支持流式响应，可结合流式读取减少轮询。
• 确保回调端点安全，避免未授权访问。
• 对于长任务，考虑将轮询器部署为独立服务，提高可靠性。

总结

通过轮询加本地回调的薄壳方案，你可以在没有服务端 webhook 的聚合站上模拟异步任务处理。这种方法简单有效，适用于大多数场景。随着聚合站功能演进，未来可能直接支持 webhook，但当前方案已足够应对多数需求。