---
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)。