AI Agent白手起家23: 大模型速率限制应对与缓存机制实践

纲要速率限制的产生背景与影响LangChain 内置速率限制器InMemoryRateLimiter核心参数缓存机制的必要性与原理短期缓存内存与长期缓存持久化实战速率限制与缓存结合项目结构速率限制器配置缓存方案对比内存与 SQLite代码示例关键要点总结速率限制为什么需要关注当前主流大模型提供商OpenAI、DeepSeek、Anthropic 等都会对 API 实施速率限制。例如 OpenAI 对免费用户限制每分钟请求数RPM和每分钟 Token 数TPM付费用户则有更高的配额。一旦请求超过限制API 将返回错误如 429导致应用不可用严重影响用户体验。在 AI Agent 开发中如果不做任何防护高频调用很容易触发限流。因此必须在客户端侧主动实施流量整形确保请求速率始终在安全范围内。LangChain 中的速率限制器LangChain 提供了InMemoryRateLimiter作为内置速率控制组件。它位于langchain_core.rate_limiters模块中通过令牌桶算法限制单位时间内的请求数。核心参数如下参数说明示例requests_per_second每秒允许的最大请求数1.0每秒 1 次或0.1每 10 秒 1 次check_every_n_seconds每隔多少秒检查一次是否允许新请求0.1每 100 毫秒检查max_burst_size允许的最大突发请求数10使用时只需在实例化模型组件时传入rate_limiter参数即可。缓存机制减少不必要的 API 调用除了限制速率缓存也是降低 API 调用频率的有效手段。其流程如下是否用户请求缓存命中?返回缓存结果调用大模型 API获取响应写入缓存LangChain 的缓存分为两种内存缓存InMemoryCache存储在进程内存中重启即失适合短期重用。持久化缓存如基于 SQLite 的实现可长期保存适合跨会话复用。结合速率限制与缓存可以大幅提升应用稳定性和成本效率。项目结构本示例将实现使用InMemoryRateLimiter控制请求频率。对比内存缓存和 SQLite 缓存的实际效果。演示如何通过usage_metadata追踪 Token 消耗验证缓存是否真正减少了 API 调用。目录结构如下├── main.py └── requirements.txt依赖清单requirements.txtlangchain-openai0.3.0 langchain-core0.3.0 langchain-community0.3.0代码示例importos,timefromlangchain_openaiimportChatOpenAIfromlangchain_core.rate_limitersimportInMemoryRateLimiterfromlangchain_core.cachesimportInMemoryCachefromlangchain_community.cachesimportSQLiteCacheimportlangchain# ---------- 1. 配置速率限制器 ----------rate_limiterInMemoryRateLimiter(requests_per_second0.5,# 每 2 秒允许 1 次请求check_every_n_seconds0.1,# 每 0.1 秒检查一次max_burst_size1# 最大突发请求为 1)# ---------- 2. 初始化模型使用 GPT-3.5 以降低成本 ----------llmChatOpenAI(modelgpt-3.5-turbo,temperature0,max_tokens100,api_keyos.getenv(OPENAI_API_KEY),rate_limiterrate_limiter# 注入速率限制器)# ---------- 3. 辅助函数打印 Token 用量 ----------defprint_usage(response):usageresponse.usage_metadataifusage:print(f 输入:{usage.get(input_tokens)}, 输出:{usage.get(output_tokens)}, 总计:{usage.get(total_tokens)})else:print( 未获取到 Token 用量)# ---------- 4. 演示速率限制效果 ----------print( 速率限制演示连续发送 5 个相同请求 )start_timetime.time()prompt用一句话介绍深度学习。foriinrange(5):t0time.time()responsellm.invoke(prompt)t1time.time()print(f请求{i1}: 耗时{t1-t0:.2f}s)print_usage(response)print(f总耗时:{time.time()-start_time:.2f}s\n)# ---------- 5. 缓存机制对比 ----------# 先移除速率限制器以便专注于缓存测试llm_no_limitChatOpenAI(modelgpt-3.5-turbo,temperature0,max_tokens100,api_keyos.getenv(OPENAI_API_KEY),)# 5.1 无缓存print( 无缓存重复请求相同问题 )langchain.llm_cacheNone# 清除全局缓存foriinrange(3):t0time.time()resllm_no_limit.invoke(什么是 Transformer)t1time.time()print(f第{i1}次: 耗时{t1-t0:.2f}s, 返回前 20 字:{res.content[:20]}...)# 5.2 内存缓存print(\n 启用内存缓存 )langchain.llm_cacheInMemoryCache()foriinrange(3):t0time.time()resllm_no_limit.invoke(什么是 Transformer)t1time.time()print(f第{i1}次: 耗时{t1-t0:.2f}s, 返回前 20 字:{res.content[:20]}...)# 第一次会调用 API后续直接命中缓存# 5.3 SQLite 持久化缓存print(\n 启用 SQLite 缓存 )langchain.llm_cacheSQLiteCache(database_path.langchain_cache.db)foriinrange(3):t0time.time()resllm_no_limit.invoke(什么是 Transformer)t1time.time()print(f第{i1}次: 耗时{t1-t0:.2f}s, 返回前 20 字:{res.content[:20]}...)# 如果之前已缓存第一次也可能快速返回运行说明代码依赖langchain-openai、langchain-core、langchain-community。需要设置环境变量OPENAI_API_KEY为有效的 OpenAI API 密钥。速率限制演示部分会因requests_per_second0.5而明显降低请求速度。缓存演示中第一次调用会真正请求 API后续调用几乎瞬间完成表明缓存生效。SQLite 缓存文件.langchain_cache.db会在当前目录生成可跨脚本持久化。关键要点总结几乎所有的商用大模型 API 都有速率限制忽略它会导致应用频繁报错。LangChain 提供InMemoryRateLimiter用于客户端流量控制只需传入rate_limiter参数即可。缓存是减少 API 调用、降低成本的有效手段LangChain 支持内存缓存和 SQLite 持久化缓存切换非常方便。在实际 AI Agent 开发中建议将速率限制与缓存结合使用以保障服务的稳定性和经济性。