概述

在构建基于大语言模型的 AI Agent 应用时，API 速率限制（Rate Limiting）与缓存（Caching）机制是保障系统稳定性和成本可控的核心技术环节。

本文基于 LangChain 框架，深入剖析速率限制的产生背景、LangChain 内置限流器的工作原理，以及不同缓存策略的适用场景与实现方案，并给出完整的可运行代码示例。

纲要

• 速率限制的产生背景与影响
• LangChain 速率限制器：InMemoryRateLimiter 核心参数与令牌桶算法
• 缓存机制的必要性与工作原理
  • 短期缓存：InMemoryCache
  • 长期持久化缓存：SQLiteCache
• 速率限制与缓存结合实践
  • 项目目录结构
  • 速率限制器配置与注入
  • 缓存方案对比（内存 vs SQLite）
  • 完整代码示例与运行说明
• 关键要点总结

速率限制：产生背景与影响

当前主流大模型提供商（包括 OpenAI、DeepSeek、Anthropic 等）对 API 请求均实施严格的速率限制策略。以 OpenAI 为例，免费用户受到每分钟请求数（RPM，Requests Per Minute）和每分钟 Token 数（TPM，Tokens Per Minute）的双重约束，付费用户则享有更高的配额。当请求超出限制时，API 会返回 HTTP 429 状态码（Too Many Requests），若客户端未做防护，将导致应用服务不可用，直接损害用户体验。

在 AI Agent 的开发实践中，Agent 的决策循环（Reasoning-Action-Observation）往往需要多次调用大模型，若不主动实施流量整形（Traffic Shaping），高频请求极易触发限流。因此，在客户端侧进行主动的速率控制是保证生产环境服务可用性的必要措施。

LangChain 中的速率限制器

LangChain 在 langchain_core.rate_limiters 模块中提供了 InMemoryRateLimiter 作为内置的速率控制组件。其底层基于令牌桶算法（Token Bucket Algorithm），通过维护一个固定容量的令牌桶，以恒定速率向桶中添加令牌，每个请求消耗一个令牌，当桶为空时请求将被阻塞直至新令牌生成。

核心参数如下：

参数	说明	示例值
requests_per_second	每秒允许的最大请求数	1.0（每秒 1 次）或 0.1（每 10 秒 1 次）
check_every_n_seconds	检查是否允许新请求的时间间隔	0.1（每 100 毫秒检查一次）
max_burst_size	允许的最大突发请求数（令牌桶容量）	10

使用方式非常简洁，只需在实例化 Chat Model 时通过 rate_limiter 参数注入即可。

缓存机制：原理与必要性

缓存是降低 API 调用频率、节省成本的另一关键手段。其核心流程如下：

LangChain 的缓存抽象支持多种后端实现，主要分为两类：

• 内存缓存（InMemoryCache）：将 LLM 响应存储在进程内存中，进程重启后缓存即丢失，适用于单次运行中的短期重用场景。
• 持久化缓存：例如基于 SQLite 的实现（SQLiteCache），将缓存持久化至磁盘文件，可跨会话、跨脚本复用，适合长期存储和高频重复请求的场景。

在实际 Agent 系统中，速率限制保证了请求的合规性，缓存则从源头上减少了请求数量，两者结合使用可以显著提升应用的稳定性和经济性。

实践：速率限制与缓存结合方案

项目结构

本示例项目结构如下：

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

依赖清单（requirements.txt）：

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

速率限制器配置

使用 InMemoryRateLimiter 控制请求频率，配置为每 2 秒允许 1 次请求：

from langchain_core.rate_limiters import InMemoryRateLimiter

rate_limiter = InMemoryRateLimiter(
    requests_per_second=0.5,
    check_every_n_seconds=0.1,
    max_burst_size=1
)

将限流器注入到 ChatOpenAI 实例中：

from langchain_openai import ChatOpenAI
import os

llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0,
    max_tokens=100,
    api_key=os.getenv("OPENAI_API_KEY"),
    rate_limiter=rate_limiter
)

缓存方案配置与对比

LangChain 的全局缓存通过设置 langchain.llm_cache 变量来控制。下面分别演示无缓存、内存缓存和 SQLite 持久化缓存的配置方式。

无缓存：

import langchain
langchain.llm_cache = None

内存缓存：

from langchain_core.caches import InMemoryCache
langchain.llm_cache = InMemoryCache()

SQLite 持久化缓存：

from langchain_community.caches import SQLiteCache
langchain.llm_cache = SQLiteCache(database_path=".langchain_cache.db")

完整代码示例

以下为完整的实践代码，包含速率限制演示、缓存机制对比以及 Token 用量追踪。

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

# ---------- 1. 配置速率限制器 ----------
rate_limiter = InMemoryRateLimiter(
    requests_per_second=0.5,
    check_every_n_seconds=0.1,
    max_burst_size=1
)

# ---------- 2. 初始化带限流的模型 ----------
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]}...")

# 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，连续请求之间会有约 2 秒的间隔，整体耗时显著增加。
• 缓存验证：在缓存演示中，第一次调用会真正请求 API 并消耗 Token，后续调用几乎瞬时完成（耗时 < 0.1s），表明缓存生效。
• SQLite 缓存文件：会在当前目录生成 .langchain_cache.db 文件，该文件可跨脚本复用，实现持久化缓存。

API 速览

本节梳理代码中涉及的核心 API。

InMemoryRateLimiter

• 所属模块：langchain_core.rate_limiters
• 构造方法：InMemoryRateLimiter(requests_per_second: float, check_every_n_seconds: float, max_burst_size: int)
• 参数说明：
  • requests_per_second：每秒允许的请求数。
  • check_every_n_seconds：内部检查间隔。
  • max_burst_size：最大突发请求容量。
• 返回值：速率限制器实例。

InMemoryCache

• 所属模块：langchain_core.caches
• 构造方法：InMemoryCache()
• 说明：基于内存的缓存实现，进程重启后数据丢失。

SQLiteCache

• 所属模块：langchain_community.caches
• 构造方法：SQLiteCache(database_path: str = ".langchain_cache.db")
• 参数说明：
  • database_path：SQLite 数据库文件路径，默认为 .langchain_cache.db。
• 说明：基于 SQLite 的持久化缓存实现，数据可跨会话复用。

langchain.llm_cache

• 类型：全局变量，可赋值为任意 BaseCache 实例，用于设置 LangChain 的全局缓存。

Demo 示例

以下提供一个完整的可运行脚本，集成了上述所有功能，并包含详细的运行说明。

文件：main.py（完整内容见上方完整代码示例）

运行步骤：

1. 创建虚拟环境并安装依赖：

   python -m venv venv
   source venv/bin/activate  # Linux/macOS
   venv\Scripts\activate     # Windows
   pip install -r requirements.txt
   

2. 设置环境变量：

   export OPENAI_API_KEY="your-api-key"
   

3. 运行脚本：

   python main.py
   

技术点总结：

• 演示了 InMemoryRateLimiter 的配置与注入方法。
• 对比了无缓存、内存缓存、SQLite 缓存三种策略的实际耗时。
• 利用 usage_metadata 追踪 Token 消耗，验证缓存是否真正避免了 API 调用。
• 展示了 LangChain 全局缓存变量 langchain.llm_cache 的切换方式。

常见问题与陷阱

• 缓存粒度：LangChain 的默认缓存基于完整的请求参数（包括 prompt、model、temperature 等）计算缓存 key。若 temperature 设置为非 0 值，即使 prompt 相同，输出也可能不同，此时缓存仍会命中但返回的是历史结果，可能导致不符合预期。建议在确定性输出场景（temperature=0）下使用缓存。
• 速率限制器的作用范围：InMemoryRateLimiter 是进程内限流，如果部署多个实例，每个实例独立限流，无法实现全局速率控制。对于分布式场景，需要考虑使用 Redis 等集中式限流方案。
• 缓存与速率限制器的执行顺序：在实际请求流程中，LangChain 会先检查缓存，如果命中则直接返回，不会触发速率限制器。因此，缓存命中不会消耗速率限制器的令牌。
• SQLite 并发写入：SQLite 默认支持多线程读取，但并发写入需要适当的 PRAGMA 配置。在高并发场景下，建议使用更强大的数据库（如 PostgreSQL）作为缓存后端。

参考文档

官方文档：

• LangChain Core Rate Limiters
• LangChain Caching
• OpenAI Rate Limits

参考链接：

• OpenAI API 速率限制官方说明
• LangChain 0.3 迁移指南

总结

本文系统性地探讨了 AI Agent 开发中应对大模型 API 速率限制与实现缓存机制的技术方案。速率限制是保证服务稳定性的基础防护层，LangChain 提供的 InMemoryRateLimiter 基于令牌桶算法，能够有效平滑请求流量；缓存机制则从源头减少了重复请求的 API 调用次数，InMemoryCache 和 SQLiteCache 分别适用于短期与长期缓存场景。

两者结合使用，可以在保障可用性的同时显著降低 API 调用成本。在实际生产部署时，还需根据分布式部署、并发请求等场景进一步扩展限流与缓存策略。