批量工具调用与并发处理
一句话定义：当 Agent 需要处理一批同质任务（如批量导入多个 Excel）时，在工具层用 Python 并发机制替代串行逐个执行，提升吞吐并汇总结果。
1. 问题场景
典型场景：Agent 调用 Skill 创建航线模块并上传 Excel，每个 Excel 内含大量航线数据，需要调 Python 脚本执行导入。
用户："把这十几个 Excel 的航线都导进去"
→ Agent 调用 Skill（航线导入）
→ Skill 触发 Tool（import_routes）
→ Tool 调 Python 脚本
→ Python 读取 Excel → 解析航线 → 调 API 逐条创建
串行痛点：十几个 Excel 逐个跑，每个内部还要逐条调 API 创建航线，总耗时 = Σ(每个Excel处理时间)。如果单个 Excel 需要 30 秒，10 个就是 5 分钟，Agent 一直阻塞等待。
串行： [Excel1 30s] → [Excel2 30s] → [Excel3 30s] → ... → 总计 300s
并发： [Excel1 30s]
[Excel2 30s]
[Excel3 30s]
... → 总计 ~35s（取决于线程数）
2. 并发架构选择
关键决策点：在哪一层做并发？ 有三种方案，适用场景不同。
方案 A：LLM 层并行调用（Parallel Function Calling）
依赖模型的并行 Function Calling 能力，LLM 一次输出多个 tool call，宿主并行执行：
[
{"name": "import_routes", "arguments": {"file": "routes_01.xlsx"}},
{"name": "import_routes", "arguments": {"file": "routes_02.xlsx"}},
{"name": "import_routes", "arguments": {"file": "routes_03.xlsx"}}
]
宿主并行执行三个调用，汇总结果后一次性回传。
| 优点 | 缺点 |
|---|---|
实现简单，宿主用 Promise.all / asyncio.gather |
依赖模型支持并行调用 |
| 每个调用独立，天然隔离 | 文件多了 LLM 可能漏调或参数填错 |
| 并发度不可控（LLM 输出几个就几个） | |
| 每次调用独立拉起 Python 进程，进程启动开销大 |
适用：文件数量少（3-5 个）、已验证模型并行调用稳定、不想改工具内部逻辑。
方案 B：工具内部并发（推荐）
LLM 只调一次工具，把所有文件路径作为数组传入，工具内部用 Python 并发处理：
{
"name": "import_routes_batch",
"arguments": {
"files": ["routes_01.xlsx", "routes_02.xlsx", "routes_03.xlsx", "..."],
"concurrency": 5
}
}
工具内部 Python 用线程池/进程池并发处理所有文件，汇总后一次性返回。
| 优点 | 缺点 |
|---|---|
| 只一次 LLM 交互，减少往返 | 工具内部需实现并发逻辑 |
| 并发度可控（线程池大小） | 单个文件失败需设计容错 |
| 进程只拉起一次，无重复启动开销 | |
| 结果汇总在工具内完成，返回结构化汇总 | |
| 不依赖模型的并行调用能力 |
适用：文件数量多（10+）、需要精确控制并发度、需要统一汇总结果。本场景首选。
方案 C：混合模式
LLM 层并行调用 + 每个调用内部再并发。适用于超大规模（100+ 文件），LLM 分批输出 tool call，每个 tool call 处理一批文件。复杂度高，一般不需要。
决策流程
文件数 ≤ 5 且模型支持并行调用？
├─ 是 → 方案 A（LLM 层并行）
└─ 否 → 文件数 ≤ 50？
├─ 是 → 方案 B（工具内部并发）✓
└─ 否 → 方案 C（混合模式）
3. Python 并发机制选择
方案 B 确定后，工具内部用 Python 实现并发。Python 有三套并发机制：
| 机制 | 适用场景 | 航线导入场景 |
|---|---|---|
ThreadPoolExecutor |
I/O 密集（网络请求、文件读写） | ✅ 调 API 创建航线是 I/O 密集 |
ProcessPoolExecutor |
CPU 密集（大量计算、数据转换） | ⚠️ 如果 Excel 解析极重可考虑 |
asyncio |
高并发 I/O、已有异步代码库 | ✅ 适合，但与同步 API 混用较复杂 |
航线导入场景分析
单个 Excel 处理流程：
1. 读 Excel 文件 → I/O（磁盘读取）
2. 解析航线数据 → CPU（pandas/openpyxl 解析）
3. 数据校验 → CPU
4. 调 API 逐条创建航线 → I/O（网络请求）← 瓶颈在这
5. 记录结果 → I/O
瓶颈在第 4 步——调 API 是网络 I/O，等待响应时 CPU 空闲。用 ThreadPoolExecutor 在等待 API 响应期间切换到其他线程处理别的 Excel，是最佳选择。
Python 的 GIL（全局解释器锁）在 I/O 等待时会释放，所以多线程对 I/O 密集任务有效。纯 CPU 密集任务才需要多进程绕过 GIL。
线程数选择
# 经验公式
import os
# I/O 密集：线程数可以远大于 CPU 核数
# 常见经验值：min(文件数, CPU核数 × 5 ~ 10)
optimal_threads = min(len(files), (os.cpu_count() or 4) * 5)
| 文件数 | 建议线程数 | 说明 |
|---|---|---|
| 1-5 | 1-5 | 一一对应即可 |
| 6-20 | 5-10 | 控制在 API 限流范围内 |
| 20-50 | 10-15 | 注意 API 限流，必要时加限速 |
| 50+ | 15-20 | 配合分批 + 限速 |
线程数不是越大越好——后端 API 有并发限制（如 QPS 限制），线程太多会触发限流反而变慢。
4. 完整实现
4.1 工具定义
{
"name": "import_routes_batch",
"description": "批量导入多个 Excel 文件中的航线数据，并发处理并返回汇总结果",
"parameters": {
"type": "object",
"properties": {
"files": {
"type": "array",
"items": {"type": "string"},
"description": "Excel 文件路径列表"
},
"concurrency": {
"type": "integer",
"description": "并发线程数，默认 5",
"default": 5
},
"module_name": {
"type": "string",
"description": "航线模块名称"
}
},
"required": ["files", "module_name"]
}
}
4.2 Python 并发导入脚本
import concurrent.futures
import time
import traceback
from dataclasses import dataclass, field
from typing import Optional
# ============================================================
# 数据结构定义
# ============================================================
@dataclass
class FileResult:
"""单个文件的处理结果"""
file: str
success: bool
total: int = 0 # 该文件中航线总数
imported: int = 0 # 成功导入数
failed: int = 0 # 失败数
duration: float = 0.0 # 耗时（秒）
errors: list = field(default_factory=list) # 失败详情
@dataclass
class BatchResult:
"""批量处理汇总结果"""
total_files: int = 0
success_files: int = 0
failed_files: int = 0
total_routes: int = 0
imported_routes: int = 0
failed_routes: int = 0
total_duration: float = 0.0
details: list = field(default_factory=list) # 每个文件的 FileResult
# ============================================================
# 单文件处理函数（每个线程执行一次）
# ============================================================
def import_single_excel(file_path: str, module_name: str) -> FileResult:
"""
处理单个 Excel 文件：读取 → 解析 → 逐条调 API 创建航线
这个函数会被线程池并发调用
"""
result = FileResult(file=file_path, success=False)
start = time.time()
try:
# 1. 读取 Excel
import openpyxl
wb = openpyxl.load_workbook(file_path, read_only=True)
ws = wb.active
routes = []
for row in ws.iter_rows(min_row=2, values_only=True): # 跳过表头
if row[0] is not None:
routes.append({
"name": row[0],
"origin": row[1],
"destination": row[2],
"departure": row[3],
})
wb.close()
result.total = len(routes)
print(f"[{file_path}] 解析到 {result.total} 条航线")
# 2. 逐条调 API 创建航线
# 这里用 create_route_via_api 模拟 API 调用
for i, route in enumerate(routes):
try:
ok = create_route_via_api(module_name, route)
if ok:
result.imported += 1
else:
result.failed += 1
result.errors.append(f"第 {i+2} 行: API 返回失败")
except Exception as e:
result.failed += 1
result.errors.append(f"第 {i+2} 行: {str(e)}")
result.success = result.failed == 0
if result.failed > 0 and result.imported > 0:
result.success = True # 部分成功也算成功，详情看 errors
except Exception as e:
result.errors.append(f"文件处理异常: {traceback.format_exc()}")
finally:
result.duration = round(time.time() - start, 2)
return result
def create_route_via_api(module_name: str, route: dict) -> bool:
"""调用 API 创建单条航线（模拟）"""
import requests
resp = requests.post(
"https://api.example.com/routes",
json={"module": module_name, **route},
timeout=30,
)
return resp.status_code == 200
# ============================================================
# 并发调度 + 结果汇总（核心）
# ============================================================
def import_routes_batch(files: list, module_name: str, concurrency: int = 5) -> BatchResult:
"""
批量并发导入航线，汇总结果
"""
batch = BatchResult(total_files=len(files))
overall_start = time.time()
# 使用线程池并发处理
with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as executor:
# 提交所有任务
future_to_file = {
executor.submit(import_single_excel, f, module_name): f
for f in files
}
# as_completed：哪个任务先完成就先处理，不按提交顺序等待
for future in concurrent.futures.as_completed(future_to_file):
file_path = future_to_file[future]
try:
result: FileResult = future.result()
batch.details.append(result)
# 汇总计数
if result.success:
batch.success_files += 1
else:
batch.failed_files += 1
batch.total_routes += result.total
batch.imported_routes += result.imported
batch.failed_routes += result.failed
# 实时进度（输出到 stderr 供宿主捕获）
done = len(batch.details)
print(f"进度: {done}/{batch.total_files} | "
f"成功 {batch.imported_routes} / 失败 {batch.failed_routes}")
except Exception as e:
batch.failed_files += 1
batch.details.append(FileResult(
file=file_path, success=False,
errors=[f"线程异常: {traceback.format_exc()}"]
))
batch.total_duration = round(time.time() - overall_start, 2)
# 排序：按文件名排序，让结果可读
batch.details.sort(key=lambda x: x.file)
return batch
# ============================================================
# 结果格式化（返回给 LLM）
# ============================================================
def format_batch_result(batch: BatchResult) -> dict:
"""将汇总结果格式化为 LLM 可读的结构化返回"""
return {
"summary": {
"total_files": batch.total_files,
"success_files": batch.success_files,
"failed_files": batch.failed_files,
"total_routes": batch.total_routes,
"imported_routes": batch.imported_routes,
"failed_routes": batch.failed_routes,
"duration_seconds": batch.total_duration,
},
"details": [
{
"file": r.file,
"success": r.success,
"total": r.total,
"imported": r.imported,
"failed": r.failed,
"duration": r.duration,
"errors": r.errors[:5], # 每个文件最多保留 5 条错误，避免过长
}
for r in batch.details
],
}
# ============================================================
# 入口
# ============================================================
if __name__ == "__main__":
files = [
"routes_01.xlsx", "routes_02.xlsx", "routes_03.xlsx",
"routes_04.xlsx", "routes_05.xlsx", "routes_06.xlsx",
"routes_07.xlsx", "routes_08.xlsx", "routes_09.xlsx",
"routes_10.xlsx", "routes_11.xlsx", "routes_12.xlsx",
]
batch = import_routes_batch(files, module_name="华东航线", concurrency=5)
report = format_batch_result(batch)
# 打印汇总（实际场景中由工具框架返回给 LLM）
import json
print("\n========== 汇总结果 ==========")
print(json.dumps(report["summary"], ensure_ascii=False, indent=2))
print("\n========== 各文件详情 ==========")
for d in report["details"]:
status = "✓" if d["success"] else "✗"
print(f" {status} {d['file']}: {d['imported']}/{d['total']} 成功, "
f"{d['failed']} 失败, {d['duration']}s")
for err in d["errors"]:
print(f" → {err}")
4.3 执行流程图
LLM 调用 import_routes_batch(files=[12个文件], concurrency=5)
│
▼
Python: ThreadPoolExecutor(max_workers=5)
│
├── 线程1: import_single_excel("routes_01.xlsx") ──→ FileResult
├── 线程2: import_single_excel("routes_02.xlsx") ──→ FileResult
├── 线程3: import_single_excel("routes_03.xlsx") ──→ FileResult
├── 线程4: import_single_excel("routes_04.xlsx") ──→ FileResult
├── 线程5: import_single_excel("routes_05.xlsx") ──→ FileResult
│ ... (前5个完成后，线程空出继续处理剩余文件)
├── 线程N: import_single_excel("routes_12.xlsx") ──→ FileResult
│
▼
as_completed 逐个收集结果 → 汇总到 BatchResult
│
▼
format_batch_result → 返回结构化 JSON 给 LLM
│
▼
LLM 基于汇总结果继续推理（告知用户导入完成情况）
5. 结果汇总策略
5.1 as_completed vs map
# 方式一：as_completed（推荐）—— 谁先完成谁先处理
with ThreadPoolExecutor(max_workers=5) as executor:
futures = {executor.submit(fn, f): f for f in files}
for future in as_completed(futures):
result = future.result() # 拿到已完成任务的结果
# 可以实时输出进度
# 方式二：map —— 按提交顺序返回结果
with ThreadPoolExecutor(max_workers=5) as executor:
results = executor.map(fn, files) # 返回迭代器，按 files 顺序 yield
for r in results:
# 即使第3个先完成，也要等第1、2个完成才能拿到
| 对比 | as_completed |
map |
|---|---|---|
| 返回顺序 | 完成顺序（随机） | 提交顺序 |
| 实时进度 | ✅ 可实时输出 | ❌ 必须按序等 |
| 异常处理 | ✅ 逐个 try/catch | ❌ 一个异常中断迭代 |
| 推荐度 | ✅ 批量场景首选 | 简单场景可用 |
5.2 部分失败的容错策略
# 每个文件独立容错，一个失败不影响其他
for future in concurrent.futures.as_completed(future_to_file):
try:
result = future.result(timeout=300) # 单文件超时 5 分钟
except concurrent.futures.TimeoutError:
# 该文件处理超时，标记失败，继续处理其他
result = FileResult(file=file_path, success=False,
errors=["处理超时（300s）"])
except Exception as e:
# 该文件处理异常，标记失败，继续处理其他
result = FileResult(file=file_path, success=False,
errors=[f"异常: {str(e)}"])
batch.details.append(result) # 无论成败都记录
5.3 返回给 LLM 的结果裁剪
def format_batch_result(batch: BatchResult) -> dict:
return {
"summary": {
"total_files": batch.total_files,
"success_files": batch.success_files,
"failed_files": batch.failed_files,
"total_routes": batch.total_routes,
"imported_routes": batch.imported_routes,
"failed_routes": batch.failed_routes,
"duration_seconds": batch.total_duration,
},
# 只返回失败的详情，成功的只计数量
# 避免上下文膨胀
"failed_details": [
{"file": r.file, "errors": r.errors[:5]}
for r in batch.details if not r.success
],
}
返回裁剪原则：LLM 不需要看每个文件的每条成功记录。汇总数字 + 失败详情即可。成功详情让用户看工具日志。
6. 进阶：单文件内部也并发
上面的方案是「文件级并发」——每个 Excel 一个线程。如果单个 Excel 内航线很多（如 1000 条），可以在文件内部再做一层并发：
def import_single_excel(file_path: str, module_name: str) -> FileResult:
# ... 解析 Excel 得到 routes 列表 ...
# 文件内部：用线程池并发调 API
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as api_executor:
futures = {
api_executor.submit(create_route_via_api, module_name, route): i
for i, route in enumerate(routes)
}
for future in concurrent.futures.as_completed(futures):
idx = futures[future]
try:
if future.result():
result.imported += 1
else:
result.failed += 1
result.errors.append(f"第 {idx+2} 行: API 返回失败")
except Exception as e:
result.failed += 1
result.errors.append(f"第 {idx+2} 行: {str(e)}")
return result
两层并发的线程数协调
外层：5 个线程处理文件
内层：每个文件 10 个线程调 API
总并发：5 × 10 = 50 个 API 请求
→ 需确认后端 API 能承受 50 并发，否则触发限流
# 建议：总并发线程数控制在 API 限流范围内
# 假设 API 限制 30 QPS
file_concurrency = 3 # 外层 3 个文件并发
api_concurrency = 10 # 每个文件内 10 个 API 并发
# 总并发 = 3 × 10 = 30，刚好不超限
7. 进阶：API 限速控制
当并发量大时，需要加限速避免打崩后端：
import threading
import time
class RateLimiter:
"""简单的令牌桶限速器"""
def __init__(self, max_calls: int, period: float = 1.0):
self.max_calls = max_calls # 每周期最大调用数
self.period = period # 周期（秒）
self.calls = [] # 最近调用的时间戳
self.lock = threading.Lock()
def acquire(self):
with self.lock:
now = time.time()
# 清理过期的时间戳
self.calls = [t for t in self.calls if now - t < self.period]
if len(self.calls) >= self.max_calls:
# 需要等待
sleep_time = self.period - (now - self.calls[0])
if sleep_time > 0:
time.sleep(sleep_time)
self.calls.append(time.time())
# 使用
rate_limiter = RateLimiter(max_calls=30, period=1.0) # 30 QPS
def create_route_via_api(module_name: str, route: dict) -> bool:
rate_limiter.acquire() # 限速
import requests
resp = requests.post("https://api.example.com/routes",
json={"module": module_name, **route}, timeout=30)
return resp.status_code == 200
8. 进阶：用 asyncio 替代线程池
如果整个工具链是异步的，用 asyncio 更高效（协程比线程更轻量）：
import asyncio
import aiohttp
async def create_route_via_api_async(session, module_name, route):
async with session.post(
"https://api.example.com/routes",
json={"module": module_name, **route},
timeout=aiohttp.ClientTimeout(total=30),
) as resp:
return resp.status == 200
async def import_single_excel_async(file_path, module_name, session):
# ... 解析 Excel ...
result = FileResult(file=file_path, success=False)
# 用 semaphore 控制单文件内 API 并发度
sem = asyncio.Semaphore(10)
async def limited_create(route):
async with sem:
return await create_route_via_api_async(session, module_name, route)
tasks = [limited_create(r) for r in routes]
results = await asyncio.gather(*tasks, return_exceptions=True)
# ... 汇总 ...
return result
async def import_routes_batch_async(files, module_name, concurrency=5):
# 用 semaphore 控制文件级并发度
sem = asyncio.Semaphore(concurrency)
async with aiohttp.ClientSession() as session:
async def limited_import(file_path):
async with sem:
return await import_single_excel_async(file_path, module_name, session)
tasks = [limited_import(f) for f in files]
results = await asyncio.gather(*tasks, return_exceptions=True)
# ... 汇总 ...
return results
# 入口
if __name__ == "__main__":
results = asyncio.run(import_routes_batch_async(files, "华东航线", concurrency=5))
| 对比 | ThreadPoolExecutor | asyncio |
|---|---|---|
| 学习成本 | 低（同步代码） | 中（异步编程） |
| 性能 | 好（I/O 密集） | 更好（协程更轻量） |
| 库支持 | 所有同步库 | 需 async 版库（aiohttp 等） |
| 混用 | 可直接调用同步/异步 | 同步库需 run_in_executor 包装 |
| 推荐场景 | 已有同步代码、快速实现 | 全新建、高并发 |
如果已有同步的
requests代码，先用ThreadPoolExecutor最快落地。如果从头建且追求性能，用asyncio。
9. 与 Agent/Skill 的集成
完整调用链
Agent (LLM)
│
│ tool_call: import_routes_batch(files=[...], concurrency=5)
│
▼
Skill 执行器
│
│ 调用 Python 脚本：python import_routes.py --files ... --concurrency 5
│
▼
Python 脚本
│
│ ThreadPoolExecutor 并发处理
│ ├── 线程1: 读Excel → 解析 → 调API
│ ├── 线程2: 读Excel → 解析 → 调API
│ └── ...
│
│ as_completed 收集 → 汇总 → format_batch_result
│
▼
返回结构化 JSON → Skill 执行器 → 回传 LLM
│
▼
LLM 基于汇总结果回复用户：
"12 个文件全部处理完成，共导入 3,600 条航线，其中 15 条失败。
失败详情：routes_03.xlsx 第 5 行 API 超时，routes_07.xlsx 第 12 行..."
Skill 配置示例（SKILL.md 片段）
---
name: route-import
description: 批量导入航线数据，支持多 Excel 并发处理
---
## 工具
### import_routes_batch
批量导入多个 Excel 文件中的航线数据。
**参数：**
- `files`（array<string>, 必填）：Excel 文件路径列表
- `module_name`（string, 必填）：航线模块名称
- `concurrency`（int, 默认 5）：并发线程数
**执行：**
```bash
python scripts/import_routes.py \
--files "{{files}}" \
--module "{{module_name}}" \
--concurrency {{concurrency}}
返回：
{
"summary": {
"total_files": 12,
"success_files": 11,
"failed_files": 1,
"total_routes": 3600,
"imported_routes": 3585,
"failed_routes": 15
},
"failed_details": [...]
}
## 10. 注意事项
### 线程安全
- `requests` / `openpyxl` / `pandas` 读操作都是线程安全的。
- **写操作注意**：多线程写同一个文件需加锁；写不同文件无需加锁。
- 全局计数器需加锁或用 `threading.Lock`：
```python
counter_lock = threading.Lock()
with counter_lock:
batch.imported_routes += 1
实际上
as_completed在主线程汇总，子线程只返回自己的FileResult，不需要全局锁。
超时与重试
- 每个文件设独立超时，避免一个卡死全部等待：
result = future.result(timeout=300) # 单文件最多 5 分钟 - API 调用加重试（网络抖动常见）：
from tenacity import retry, stop_after_attempt, wait_exponential @retry(stop=stop_after_attempt(3), wait=wait_exponential(min=1, max=10)) def create_route_via_api(...): ...
资源控制
- 线程数不是越大越好：受 API 限流、数据库连接池、内存限制约束。
- 大文件注意内存：
read_only=True模式读 Excel 可减少内存占用。 - 进程数 vs 线程数：I/O 密集用线程（共享内存、轻量）；CPU 密集用进程（绕过 GIL）。
错误隔离
- 每个文件的处理用
try/except包裹，一个文件失败不影响其他。 as_completed+future.result()的异常不会传播到其他 future。- 汇总时区分「文件级失败」（Excel 损坏/格式错）和「航线级失败」（单条 API 调用失败）。
日志与可观测
- 每个线程输出带文件名的日志，便于排查：
import logging logger = logging.getLogger(f"import.{file_path}") logger.info(f"开始处理，共 {total} 条航线") - 实时进度输出到 stderr（宿主可捕获显示），最终结果输出到 stdout。
幂等性
- 如果可能重试（如 Agent 重新调用），确保导入操作幂等：
- 导入前检查航线是否已存在（按唯一标识查）。
- API 层面用
upsert（存在则更新，不存在则创建）。 - 或用批次 ID 标记，重试时先清理上次的部分导入。
11. 方案对比总结
| 维度 | 串行 | LLM 层并行 (方案A) | 工具内部并发 (方案B) | asyncio |
|---|---|---|---|---|
| 实现复杂度 | 最低 | 低 | 中 | 中高 |
| LLM 交互次数 | N 次 | 1 次 | 1 次 | 1 次 |
| 并发可控 | ✗ | ✗ | ✅ | ✅ |
| 结果汇总 | 手动 | 宿主处理 | 工具内自动 | 工具内自动 |
| 依赖模型能力 | ✗ | ✅ 需并行调用 | ✗ | ✗ |
| 适用文件数 | 1-3 | 3-5 | 5-50+ | 50+ |
| 推荐度 | ✗ | △ | ✅ 首选 | △ 高性能场景 |
12. 学习要点
- 并发在哪一层做是关键决策：LLM 层（并行 Function Calling）vs 工具层（Python 线程池）。文件多时工具层并发更可控。
- I/O 密集用线程池：航线导入的瓶颈在调 API（网络 I/O），
ThreadPoolExecutor是最简单有效的方案。 - as_completed 收集结果：谁先完成谁先处理，可实时输出进度，异常隔离好。
- 结果裁剪返回 LLM：汇总数字 + 失败详情，不要把所有成功记录塞回 LLM 上下文。
- 控制总并发量：外层文件并发 × 内层 API 并发 ≤ 后端 API 限制，必要时加限速器。
- 每个任务独立容错：一个文件/一条航线失败不影响其他，失败记录在汇总结果中。
13. 参考资料
- Python
concurrent.futures官方文档 - Python
asyncio官方文档 - GIL 与 Python 并发模型
tenacity重试库文档- 本系列《01-Function Calling》第 5 节「多函数调用」
评论
评论加载中…