飞书速率限制技能

针对飞书/Lark API的智能速率限制处理。避免429错误，确保批量操作成功完成。

快速参考

触发条件

在以下情况下激活此技能：

• - 所有飞书API调用（feishudoc、feishubitable*、feishudrive、feishu_wiki等）
• 遇到429（速率限制）错误
• 对飞书数据进行批量操作
• 向飞书写入/查询大量数据
• 用户提及飞书API、Lark API、429、速率限制、飞书限流

飞书API限制

自定义应用限制

文档API特殊限制

官方文档：https://open.feishu.cn/document/platform-notices/platform-updates-/custom-app-api-call-limit

调用策略

基本间隔规则

初始间隔：1秒
最小间隔：1秒
最大间隔：10秒
警告阈值：50次调用/分钟

滑动窗口策略

python
class FeishuRateLimiter:
飞书API速率限制器

def init(self):
self.base_interval = 1.0 # 基础间隔1秒
self.current_interval = 1.0 # 当前间隔
self.max_interval = 10.0 # 最大间隔
self.recent_calls = [] # 近期调用记录
self.window_size = 60 # 统计窗口60秒
self.max_retries = 3 # 最大重试次数

def before_call(self):
调用前等待
time.sleep(self.current_interval)
self.recent_calls.append(time.time())
self.adjustinterval()

def adjustinterval(self):
根据近期调用频率动态调整间隔
now = time.time()
# 统计最近60秒内的调用次数
recent = [t for t in self.recentcalls if now - t < self.windowsize]
self.recent_calls = recent
call_count = len(recent)

# 根据频率调整间隔
if call_count > 50: # 接近限制（100次/分钟）
self.currentinterval = min(self.currentinterval * 1.5, self.max_interval)
elif call_count > 30:
self.currentinterval = min(self.currentinterval * 1.2, self.max_interval)
elif callcount < 10 and self.currentinterval > self.base_interval:
# 调用不频繁时减少间隔
self.currentinterval = max(self.currentinterval * 0.9, self.base_interval)

def onratelimit(self, retry_after=None):
遇到429错误时调用
if retry_after:
waittime = retryafter
else:
waittime = self.currentinterval * 2

self.currentinterval = min(waittime, self.max_interval)
time.sleep(self.current_interval)

def on_success(self):
成功后逐渐恢复正常间隔
if self.currentinterval > self.baseinterval:
self.currentinterval = max(self.currentinterval * 0.8, self.base_interval)

重试策略

遇到429错误时：

1. 1. 解析Retry-After头（如果存在）
2. 等待指定时间 或 current_interval × 2
3. 重试（最多3次）
4. 如果仍然失败，记录错误并通知用户

实现指南

1. 批量操作处理

markdown
❌ 错误做法：
一次性写入100条记录

✅ 正确做法：
for batch in chunks(records, 10):
for record in batch:
feishubitablecreate_record(record)
sleep(1) # 每条记录间隔1秒
sleep(5) # 批次间额外等待5秒

2. 优先使用批量API

markdown
优先级：

1. 1. 批量API（createrecords、batchupdate）
2. 单条API + 智能间隔
3. 并发请求（仅限读取，谨慎使用）

3. 智能缓存策略

markdown
对于频繁查询的数据：

• - 用户信息：缓存24小时
• 字段定义：缓存1小时
• 文档结构：缓存30分钟
• 部门列表：缓存2小时

错误处理流程

┌─────────────────┐
│ API调用 │
└────────┬────────┘
▼
┌─────────────────┐
│ 检查响应 │
└────────┬────────┘
▼
┌────┴────┐
│ 429? │
└────┬────┘
│
┌────┴────────────────────────────┐
│ 是 │ 否
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ 检查Retry-After │ │ 继续执行 │
└────────┬────────┘ └─────────────────┘
▼
┌─────────────────┐
│ 等待指定时间 │
│ 或间隔×2 │
└────────┬────────┘
▼
┌─────────────────┐
│ 重试次数 < 3? │
└────────┬────────┘
│
┌────┴────┐
│ 是 │ 否
▼ ▼
┌─────────┐ ┌─────────────────┐
│ 重试 │ │ 记录错误，通知 │
└─────────┘ └─────────────────┘

配置参数

yaml
feishuratelimit:
enabled: true
baseintervalms: 1000 # 基础间隔1秒
maxintervalms: 10000 # 最大间隔10秒
max_retries: 3 # 最大重试次数
windowsizesec: 60 # 统计窗口60秒
warning_threshold: 50 # 警告阈值（次/分钟）
batch_size: 10 # 批次大小
batchdelayms: 5000 # 批次间延迟

常见场景

场景1：批量创建多维表格记录

markdown
任务：创建200条记录

策略：

1. 1. 拆分：每批10条记录
2. 批次间隔：5秒
3. 记录间隔：1秒
4. 预计时间：约40秒

实现：
第1批：10条记录 → 等待5秒
第2批：10条记录 → 等待5秒
...
第20批：10条记录 → 完成

场景2：文档批量写入

markdown
任务：向文档写入50个段落

策略：

1. 1. 段落间隔：2-3秒（文档API限制更严格）
2. 每10个段落：额外等待10秒
3. 预计时间：约3-4分钟

注意：

• - 文档API限制更严格
• 遇到429时等待时间加倍

场景3：混合操作

markdown
任务：读取100条记录 + 更新50条记录

策略：

1. 1. 读取操作：可以并发
2. 更新操作：串行 + 1秒间隔
3. 混合操作：读取完成后等待再更新

建议：

• - 先批量读取（可缓存）
• 将更新拆分为多个批次

最佳实践