使用服务端缓存

from langgraph_sdk.cache import swr

result = await swr("config:global", load_config)
config_data = result.value

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),
)

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  # 自动反序列化

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="缺少令牌")

    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

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

result.value   # 缓存值或刚加载的值
result.status  # "miss" | "fresh" | "stale" | "expired"

await result.mutate(new_value)  # 使用新值更新缓存
await result.mutate()           # 通过调用加载器强制重新验证

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))

async def cache_get(key: str) -> Any | None

async def cache_set(key: str, value: Any, *, ttl: timedelta | None = None) -> None