企业官网建设流程全解析

纲要

• 速率限制的产生背景与影响
• LangChain 内置速率限制器
  • InMemoryRateLimiter核心参数
• 缓存机制的必要性与原理
  • 短期缓存（内存）与长期缓存（持久化）
• 实战：速率限制与缓存结合
  • 项目结构
  • 速率限制器配置
  • 缓存方案对比（内存与 SQLite）
  • 代码示例
• 关键要点总结

速率限制：为什么需要关注

当前主流大模型提供商（OpenAI、DeepSeek、Anthropic 等）都会对 API 实施速率限制。例如 OpenAI 对免费用户限制每分钟请求数（RPM）和每分钟 Token 数（TPM），付费用户则有更高的配额。一旦请求超过限制，API 将返回错误（如 429），导致应用不可用，严重影响用户体验。

在 AI Agent 开发中，如果不做任何防护，高频调用很容易触发限流。因此，必须在客户端侧主动实施流量整形，确保请求速率始终在安全范围内。

LangChain 中的速率限制器

LangChain 提供了InMemoryRateLimiter作为内置速率控制组件。它位于langchain_core.rate_limiters模块中，通过令牌桶算法限制单位时间内的请求数。

核心参数如下：

使用时，只需在实例化模型组件时传入rate_limiter参数即可。

缓存机制：减少不必要的 API 调用

除了限制速率，缓存也是降低 API 调用频率的有效手段。其流程如下：

LangChain 的缓存分为两种：

• 内存缓存（InMemoryCache）：存储在进程内存中，重启即失，适合短期重用。
• 持久化缓存：如基于 SQLite 的实现，可长期保存，适合跨会话复用。

结合速率限制与缓存，可以大幅提升应用稳定性和成本效率。

项目结构

本示例将实现：

1. 使用InMemoryRateLimiter控制请求频率。
2. 对比内存缓存和 SQLite 缓存的实际效果。
3. 演示如何通过usage_metadata追踪 Token 消耗，验证缓存是否真正减少了 API 调用。

目录结构如下：

├── main.py └── requirements.txt

依赖清单（requirements.txt）：

langchain-openai>=0.3.0 langchain-core>=0.3.0 langchain-community>=0.3.0

代码示例

importos,timefromlangchain_openaiimportChatOpenAIfromlangchain_core.rate_limitersimportInMemoryRateLimiterfromlangchain_core.cachesimportInMemoryCachefromlangchain_community.cachesimportSQLiteCacheimportlangchain# ---------- 1. 配置速率限制器 ----------rate_limiter=InMemoryRateLimiter(requests_per_second=0.5,# 每 2 秒允许 1 次请求check_every_n_seconds=0.1,# 每 0.1 秒检查一次max_burst_size=1# 最大突发请求为 1)# ---------- 2. 初始化模型（使用 GPT-3.5 以降低成本） ----------llm=ChatOpenAI(model="gpt-3.5-turbo",temperature=0,max_tokens=100,api_key=os.getenv("OPENAI_API_KEY"),rate_limiter=rate_limiter# 注入速率限制器)# ---------- 3. 辅助函数：打印 Token 用量 ----------defprint_usage(response):usage=response.usage_metadataifusage:print(f" 输入:{usage.get('input_tokens')}, 输出:{usage.get('output_tokens')}, 总计:{usage.get('total_tokens')}")else:print(" 未获取到 Token 用量")# ---------- 4. 演示速率限制效果 ----------print("=== 速率限制演示：连续发送 5 个相同请求 ===")start_time=time.time()prompt="用一句话介绍深度学习。"foriinrange(5):t0=time.time()response=llm.invoke(prompt)t1=time.time()print(f"请求{i+1}: 耗时{t1-t0:.2f}s")print_usage(response)print(f"总耗时:{time.time()-start_time:.2f}s\n")# ---------- 5. 缓存机制对比 ----------# 先移除速率限制器以便专注于缓存测试llm_no_limit=ChatOpenAI(model="gpt-3.5-turbo",temperature=0,max_tokens=100,api_key=os.getenv("OPENAI_API_KEY"),)# 5.1 无缓存print("=== 无缓存：重复请求相同问题 ===")langchain.llm_cache=None# 清除全局缓存foriinrange(3):t0=time.time()res=llm_no_limit.invoke("什么是 Transformer？")t1=time.time()print(f"第{i+1}次: 耗时{t1-t0:.2f}s, 返回前 20 字:{res.content[:20]}...")# 5.2 内存缓存print("\n=== 启用内存缓存 ===")langchain.llm_cache=InMemoryCache()foriinrange(3):t0=time.time()res=llm_no_limit.invoke("什么是 Transformer？")t1=time.time()print(f"第{i+1}次: 耗时{t1-t0:.2f}s, 返回前 20 字:{res.content[:20]}...")# 第一次会调用 API，后续直接命中缓存# 5.3 SQLite 持久化缓存print("\n=== 启用 SQLite 缓存 ===")langchain.llm_cache=SQLiteCache(database_path=".langchain_cache.db")foriinrange(3):t0=time.time()res=llm_no_limit.invoke("什么是 Transformer？")t1=time.time()print(f"第{i+1}次: 耗时{t1-t0:.2f}s, 返回前 20 字:{res.content[:20]}...")# 如果之前已缓存，第一次也可能快速返回

运行说明：

• 代码依赖langchain-openai、langchain-core、langchain-community。
• 需要设置环境变量OPENAI_API_KEY为有效的 OpenAI API 密钥。
• 速率限制演示部分会因requests_per_second=0.5而明显降低请求速度。
• 缓存演示中，第一次调用会真正请求 API，后续调用几乎瞬间完成，表明缓存生效。
• SQLite 缓存文件.langchain_cache.db会在当前目录生成，可跨脚本持久化。

关键要点总结

• 几乎所有的商用大模型 API 都有速率限制，忽略它会导致应用频繁报错。
• LangChain 提供InMemoryRateLimiter用于客户端流量控制，只需传入rate_limiter参数即可。
• 缓存是减少 API 调用、降低成本的有效手段，LangChain 支持内存缓存和 SQLite 持久化缓存，切换非常方便。
• 在实际 AI Agent 开发中，建议将速率限制与缓存结合使用，以保障服务的稳定性和经济性。