使用服务端缓存

使用 stale-while-revalidate 和键值缓存 API 在你的智能体部署中服务端缓存值。

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

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

NOTE

swr 需要 Agent Server 运行时 v0.7.79 或更高版本，目前处于 beta 阶段。 cache_get 和 cache_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_for	timedelta(0)	将缓存值视为新鲜的时长。在此期间，swr 返回缓存值且不进行重新验证。
max_age	timedelta(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 完成。

检查缓存状态

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

如需不进行重新验证的简单 get/set 缓存，请直接使用 cache_get 和 cache_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

参数	类型	默认值	描述
key	str	必填	缓存键
value	Any	必填	要缓存的值。必须可 JSON 序列化
ttl	timedelta | None	None	存活时间（Time-to-live）。服务器将其上限设为 1 天。None 或零默认为 1 天

后续步骤

• 向你的部署添加自定义身份认证。
• 添加自定义生命周期事件，在服务器启动时初始化资源。
• 了解智能体服务器架构。

---