纲要

• 速率限制的产生背景与影响
• 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

代码示例

import os, time
from langchain_openai import ChatOpenAI
from langchain_core.rate_limiters import InMemoryRateLimiter
from langchain_core.caches import InMemoryCache
from langchain_community.caches import SQLiteCache
import langchain

# ---------- 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 用量 ----------
def print_usage(response):
    usage = response.usage_metadata
    if usage:
        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 = "用一句话介绍深度学习。"
for i in range(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  # 清除全局缓存
for i in range(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()
for i in range(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")
for i in range(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 开发中，建议将速率限制与缓存结合使用，以保障服务的稳定性和经济性。