112 lines
5.6 KiB
Markdown
112 lines
5.6 KiB
Markdown
---
|
||
id: python.asyncio.scenario.concurrent-llm-calls
|
||
title: 并发调用多个模型和外部接口：限并发、超时、重试、部分失败
|
||
type: scenario
|
||
domain: python
|
||
stack: [asyncio, fastapi]
|
||
tags: [并发, 大模型调用, 超时, 重试, 限流, 部分失败]
|
||
level: intermediate
|
||
prerequisites: [python.asyncio.concurrency-control]
|
||
est_hours: 2
|
||
related: [python.asyncio.interview, backend.go.scenario.batch-third-party-calls]
|
||
week: [5, 12]
|
||
---
|
||
|
||
# 并发调用多个模型和外部接口：限并发、超时、重试、部分失败
|
||
|
||
## 业务背景与问题
|
||
|
||
客服工单系统里，每张新工单需要并行做四件事：摘要（大模型）、分类（大模型）、情绪识别（小模型服务）、查客户等级（CRM 接口）。第一版是顺序调用，单张工单处理要 8~15 秒；改成直接 `gather` 之后变快了，但遇到新问题：
|
||
|
||
- 早高峰一次进来几百张工单，同时发出上千个请求，模型服务商返回大量 429（限流）；
|
||
- 某一个接口卡住时，整张工单一直没有结果；
|
||
- 任何一个调用报错，整个 `gather` 抛异常，其余已经成功的结果也丢了。
|
||
|
||
## 约束
|
||
|
||
- 模型服务商有速率限制（每分钟请求数和 token 数）。
|
||
- 工单处理结果允许部分缺失（例如情绪识别失败，不影响分配客服），但必须标明缺失项。
|
||
- 每张工单的处理总时长不超过 20 秒，超时的部分以后台任务补齐。
|
||
|
||
## 方案设计
|
||
|
||
```mermaid
|
||
flowchart LR
|
||
T[新工单] --> F[fan_out]
|
||
F -->|Semaphore 限并发| S[摘要 LLM]
|
||
F --> C[分类 LLM]
|
||
F --> E[情绪服务]
|
||
F --> R[CRM 接口]
|
||
S & C & E & R --> M[合并结果：成功的字段 + 失败原因]
|
||
M --> D[分配客服]
|
||
M -->|有失败项| Q[补偿队列：稍后重试]
|
||
```
|
||
|
||
1. **全局并发上限**：所有模型调用共享一个 `Semaphore`（按服务商限额设置），而不是每张工单各自限制。多进程 / 多实例部署时，进一步用 Redis 做全局限流。
|
||
2. **单次超时 + 整体截止时间**：每次调用 5 秒超时；每张工单整体 20 秒，到时取消未完成的调用。
|
||
3. **有选择地重试**：只对超时、429、5xx 做指数退避重试（加随机抖动），最多 1~2 次；参数错误不重试。
|
||
4. **部分失败**：每个子任务的结果单独记录成功值或错误原因，不让一个失败影响全部；失败项进入补偿队列。
|
||
5. **可观测**：记录每个子任务的耗时、重试次数、失败原因，用于调整并发和超时参数。
|
||
|
||
## 关键代码
|
||
|
||
完整实现和测试见 [examples/fanout](../examples/fanout/)：
|
||
|
||
```python
|
||
async def fan_out(jobs, *, concurrency, timeout, retries=1, deadline=None):
|
||
sem = asyncio.Semaphore(concurrency)
|
||
results = {k: Result(key=k) for k in jobs}
|
||
|
||
async def run(key, fn):
|
||
async with sem:
|
||
try:
|
||
results[key].value, _ = await call_with_retry(fn, timeout=timeout, retries=retries)
|
||
except asyncio.CancelledError:
|
||
results[key].error = "整体超时，已取消"
|
||
raise # 取消要继续传播
|
||
except Exception as e: # 单个失败只记录
|
||
results[key].error = f"{type(e).__name__}: {e}"
|
||
|
||
tasks = [asyncio.create_task(run(k, fn)) for k, fn in jobs.items()]
|
||
done, pending = await asyncio.wait(tasks, timeout=deadline)
|
||
for t in pending:
|
||
t.cancel()
|
||
await asyncio.gather(*pending, return_exceptions=True)
|
||
return [results[k] for k in jobs]
|
||
```
|
||
|
||
演示输出（并发 2、单次超时 0.5 秒、重试 1 次）：
|
||
|
||
```text
|
||
摘要: 摘要 的回答（尝试 1 次）
|
||
分类: 分类 的回答（尝试 2 次） ← 第一次 429，重试成功
|
||
翻译: TimeoutError（尝试 2 次） ← 太慢，两次都超时
|
||
关键词: 关键词 的回答（尝试 1 次）
|
||
```
|
||
|
||
## 取舍
|
||
|
||
| 方案 | 优点 | 缺点 |
|
||
| --- | --- | --- |
|
||
| 顺序调用 | 简单 | 慢 |
|
||
| 裸 `gather` | 快、代码少 | 无并发上限、一个失败全部失败 |
|
||
| **Semaphore + 超时 + 重试 + 部分失败（本方案）** | 可控、稳定 | 代码多一些，参数要调 |
|
||
| 消息队列 + 独立 worker | 削峰、可水平扩展、失败可重放 | 架构更复杂，结果是异步的 |
|
||
|
||
请求量继续增大时，把“工单 → 子任务”拆成队列消息，由 worker 池消费，Semaphore 换成 Redis 全局限流。
|
||
|
||
## 易错点
|
||
|
||
- 每个请求各自创建一个 Semaphore：等于没有限制。
|
||
- 重试不加抖动：所有任务在同一时刻重试，再次触发限流。
|
||
- `except Exception` 之后不区分错误类型就重试：参数错误被重试多次，浪费额度。
|
||
- 捕获 `CancelledError` 后不重新抛出：整体截止时间失效。
|
||
- 截止时间到了只是“不等了”，没有取消任务：后台还在消耗模型额度。
|
||
- 只记录“失败”不记录原因：无法判断是限流、超时还是参数问题。
|
||
|
||
## 面试怎么讲
|
||
|
||
> “工单要同时调用几个模型和接口。我用 asyncio 并发，但加了全局 Semaphore 控制并发数以符合服务商的限流；每次调用有超时，只对超时和 429 做带抖动的指数退避重试；每个子任务单独记录结果和错误，单个失败不影响其他，失败项进补偿队列；整体有截止时间，到时取消剩余调用。处理时间从顺序调用的 8 到 15 秒降到了约等于最慢的那个调用。”
|
||
|
||
注意：面试中提到的耗时数据要来自你自己项目的实际测量，不要编造。和 Go 版本的同类场景对照：[批量调用第三方接口（Go）](../../../backend/go/scenarios/batch-third-party-calls.md)。
|