서버 측 캐싱 사용하기

배포된 그래프 안에서 stale-while-revalidate와 키-값 캐시 API를 사용해 값들을 서버 측에 캐시할 수 있어요. Agent Server는 배포된 그래프 안에서 사용할 수 있는 내장 캐시를 포함합니다. swr을 키와 로더 함수와 함께 호출하면 서버가 결과를 캐시하고, 오래된 항목을 백그라운드에서 재검증하며, 매 읽기마다 최신 데이터를 반환합니다.

출처: 문서

본문

Agent Server에는 배포된 그래프 안에서 사용할 수 있는 내장 캐시가 포함되어 있습니다. 키와 로더 함수와 함께 swr을 호출하면, 서버가 결과를 캐시하고, 오래된 항목을 백그라운드에서 재검증하며, 매 읽기마다 최신 데이터를 반환합니다.

모든 캐시 API는 서버 측 전용이며 LangGraph Agent Server 런타임이 필요합니다. 값은 JSON 직렬화 가능해야 합니다.

참고: swr은 Agent Server 런타임 v0.7.79 이상이 필요하며 현재 베타 상태입니다. 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일로 제한됩니다.

재검증 작동 방식

캐시 상태	조건	동작
Miss	캐시에 키 없음	`loader()`를 기다렸다 결과를 저장하고 반환합니다.
Fresh	`age < fresh_for`	캐시된 값을 반환, 재검증 없음.
Stale	`fresh_for <= age < max_age`	캐시된 값을 즉시 반환, 백그라운드 새로고침 트리거.
Expired	`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()를 호출합니다.

인증 자격 증명 캐시

커스텀 인증 핸들러에서 자격 증명 검증을 캐시해 매 요청마다 Identity Provider에 접근하는 것을 피할 수 있습니다:

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` 또는 0은 1일로 기본 설정됩니다.

다음 단계

배포에 커스텀 인증을 추가하세요.
서버 시작 시 리소스를 초기화하려면 커스텀 수명주기(lifespan) 이벤트를 추가하세요.
에이전트 서버 아키텍처에 대해 알아보세요.

더 알아보기

Agent Server 아키텍처는 Agent Server 문서를 참고하세요.
커스텀 인증 핸들러는 Custom auth 문서를 확인해 보세요.