Skip to main content
智能体服务器 内置了缓存功能,您可以在部署的图中使用。调用 swr 并传入一个键和一个加载函数,服务器将缓存结果、在后台重新验证过期的条目,并在每次读取时返回最新数据。 所有缓存 API 仅限服务端使用,需要 LangGraph 智能体服务器运行时环境。缓存值必须是 JSON 可序列化的。
swr 需要智能体服务器运行时 v0.7.79 或更高版本,目前处于 测试阶段cache_getcache_set 需要 v0.7.29 或更高版本。

快速开始

传入一个键和一个异步加载函数。swr 会返回缓存值(如果可用),否则调用您的加载函数来获取:
from langgraph_sdk.cache import swr

result = await swr("config:global", load_config)
config_data = result.value
首次调用时,swr 会等待 load_config() 并缓存结果。后续调用时,它会立即返回缓存值并在后台重新验证。

配置新鲜度

控制缓存值被视为新鲜的时间以及何时过期:
from datetime import timedelta
from langgraph_sdk.cache import swr

result = await swr(
    "config:global",
    load_config,
    fresh_for=timedelta(minutes=5),
    max_age=timedelta(hours=1),
)
参数默认值描述
fresh_fortimedelta(0)将缓存值视为新鲜的持续时间。在此窗口期内,swr 直接返回缓存值,无需重新验证。
max_agetimedelta(days=1)缓存条目的最长生存时间。超过此时间后,swr 会等待加载函数完成后再返回。上限为 1 天。

重新验证的工作原理

缓存状态条件行为
未命中键不在缓存中等待 loader() 完成,存储结果并返回。
新鲜age < fresh_for返回缓存值,不进行重新验证。
过期fresh_for <= age < max_age立即返回缓存值,触发后台刷新。
已失效age >= max_age等待 loader() 完成,存储结果并返回。

与 Pydantic 模型结合使用

传入 model 参数以自动序列化和反序列化 Pydantic 模型:
from pydantic import BaseModel
from langgraph_sdk.cache import swr

class UserProfile(BaseModel):
    name: str
    email: str
    role: str

result = await swr(
    f"profile:{user_id}",
    lambda: fetch_profile(user_id),
    model=UserProfile,
)
profile: UserProfile = result.value  # 自动反序列化
swr 在存储前调用 model_dump(mode="json"),在读取时调用 model.model_validate()

缓存认证凭据

您可以在自定义认证处理器中缓存凭据验证,以避免每次请求都访问身份提供商:
from datetime import timedelta
from langgraph_sdk import Auth
from langgraph_sdk.cache import swr

auth = Auth()

@auth.authenticate
async def authenticate(headers: dict) -> Auth.types.MinimalUserDict:
    token = (headers.get(b"authorization") or b"").decode()
    if not token:
        raise Auth.exceptions.HTTPException(status_code=401, detail="Missing token")

    result = await swr(
        f"auth:token:{token}",
        lambda: validate_and_fetch_user(token),
        fresh_for=timedelta(minutes=5),
        max_age=timedelta(hours=1),
    )
    return result.value
通过此设置,服务器将在 5 分钟内直接返回缓存的用户信息而无需重新验证,之后在长达 1 小时内在后台重新验证。1 小时后,下一次请求将等待 validate_and_fetch_user 完成。

检查缓存状态

swr 返回一个 SWRResult 对象,包含值和缓存状态:
result = await swr("my-key", my_loader)

result.value   # 缓存的或新加载的值
result.status  # "miss" | "fresh" | "stale" | "expired"
调用 .mutate() 来更新缓存值或强制重新验证:
await result.mutate(new_value)  # 使用新值更新缓存
await result.mutate()           # 通过调用加载函数强制重新验证

底层缓存 API

对于无需重新验证的简单获取/设置缓存,可以直接使用 cache_getcache_set
from datetime import timedelta
from langgraph_sdk.cache import cache_get, cache_set

value = await cache_get("my-key")

if value is None:
    value = await expensive_computation()
    await cache_set("my-key", value, ttl=timedelta(hours=1))

cache_get

async def cache_get(key: str) -> Any | None
返回反序列化的值,如果键不存在或已过期则返回 None

cache_set

async def cache_set(key: str, value: Any, *, ttl: timedelta | None = None) -> None
参数类型默认值描述
keystr必需缓存键
valueAny必需要缓存的值。必须是 JSON 可序列化的
ttltimedelta | NoneNone生存时间。服务器将其上限设为 1 天。None 或零值默认为 1 天

后续步骤