以编程方式使用文档

代理服务器 包含一个内置缓存，您可以在部署的图中使用。调用 swr 并传入一个键和一个加载器函数，服务器会缓存结果、在后台重新验证过期条目，并在每次读取时返回新数据。

所有缓存 API 都是 **仅服务端** 且需要 LangGraph Agent Server 运行时。值必须可序列化为 JSON。

快速入门

传入一个键和一个异步加载器函数。 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  # deserialized automatically

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 completes.

检查缓存状态

swr 返回一个 SWRResult 包含值和缓存状态的对象：

result = await swr("my-key", my_loader)

result.value   # the cached or freshly loaded value
result.status  # "miss" | "fresh" | "stale" | "expired"

调用 .mutate() 来更新缓存值或强制重新验证：

await result.mutate(new_value)  # update the cache with a new value
await result.mutate()           # force revalidation by calling the loader

低级缓存 API

For simple get/set caching without revalidation, use cache_get 和 cache_set directly:

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 可序列化的
ttl`timedelta \None`None

后续步骤