LangGraph 코드 변경 시 호환성 유지하기
LangGraph 코드 변경 시 호환성 유지하기 (Backward compatibility)
프로덕션에서는 소프트웨어가 계속 변해야 해요. 새 요구사항, 버그 수정, 리팩토링은 언젠가 그래프 코드에 반영되거든요. LangGraph는 기존 스레드에 대해 영속화된 상태를 가진 최신 배포 그래프를 그대로 실행하므로, 내보내는 모든 변경이 사실상 기존 체크포인트에 대한 backward-compatible API 변경과 같아요. 그래프를 시작한 시점의 코드에 고정시키는 워크플로 엔진과 달리, LangGraph는 최신 그래프를 새 스레드는 물론 체크포인트에서 재개하는 스레드까지 모든 스레드에 즉시 적용해요. 그래서 버그 수정이 진행 중인 대화와 에이전트에 번거로움 없이 전파된다는 장점이 있지만, 그만큼 이전 버전 코드로 시작된 실행과 어떻게 상호작용하는지 신중히 따져야 해요. 주의할 호환성 이슈는 크게 세 종류예요.
출처: 공식문서
기술 호환성 (Technical compatibility)
기술 호환성은 마이크로서비스의 API breaking change에 해당하는 개념이에요. 여기서의 "API"는 그래프 코드와 기존 스레드를 위해 체크포인터가 이미 저장한 데이터 사이의 계약이에요. 스레드가 재개될 때 LangGraph는 저장된 상태를 역직렬화해서 이름으로 노드에 전달하고, 노드가 상태 스키마에 맞는 값을 반환하길 기대해요. 흔한 기술적 문제는:
- 노드 이름 변경·제거: 스레드가 그 노드에 진입하려는 순간 멈춰 있을 때(예:
interrupt지점이나 옛 이름으로 라우팅하는 체크포인트된 조건부 엣지) 문제가 돼요. 재개 시 저장된 이름으로 노드를 찾지 못하면 실행이 실패해요. - State 키 이름 변경·제거: 옛 체크포인트가 여전히 담고 있거나 하위 노드가 여전히 읽는 키를 없애면 안 돼요.
- State 필드 강화:
Optional필드를 필수로 바꾸거나, 타입을 좁히거나, 기본값 없는 필수 필드를 추가하면 기존 체크포인트가 새 스키마를 만족하지 못해요.
엣지 토폴로지 자체는 체크포인트에 저장되지 않아요. 존재하는 노드 사이의 엣지 추가·제거·라우팅 변경은 진행 중인 스레드에 안전해요. 요약하자면, 중단된 스레드를 깨뜨릴 수 있는 유일한 토폴로지 변경은 노드의 이름 변경·제거예요.
추천 패턴
- 새 상태 필드는
NotRequired(또는Optional[...] = None)로 추가해서 옛 체크포인트도 검증되게 해요:
from typing import NotRequired
from typing_extensions import TypedDict
class State(TypedDict):
messages: list
summary: NotRequired[str]
- 제거는 폐기(deprecation)처럼 다뤄요. 어떤 노드도 읽지 않아도 최소 한 드레인(drain) 주기 이상은 상태에 필드를 남겨둬서 기존 체크포인트가 계속 로드되게 해요.
- 이름 변경은 추가-후-제거(add-then-remove)로 해요. 새 필드·노드를 옛 것과 함께 추가하고, 폐기 기간 동안 둘 다에 쓰거나 라우팅한 뒤, 의존하는 진행 중 스레드가 없음을 확인하고 옛것을 제거해요.
- 노드 함수는 알 수 없는 키에 관대하게 해요.
TypedDict는 런타임에서 추가 키를 무시하므로, 노드가 명시적으로 없는 키를 읽지 않는 한 옛 코드 버전의 잔여 상태가 오류를 내지 않아요. - 배포 전에 타임트래블(Time travel)과
graph.get_state로 기존 스레드를 새 코드에 대해 스테이징에서 점검해요.
진행 중 스레드 감지
노드를 제거하거나 State 키를 바꾸기 전에, 곧 내려놓을 코드 버전에 어떤 스레드가 정차해 있는지 아는 게 좋아요. LangGraph 자체는 스레드 상태 검색 인덱스를 유지하지 않으므로 답은 그래프가 어디서 실행되는지에 달려 있어요. LangSmith에 배포했다면 Agent Server의 스레드 검색으로 status 필드를 필터링(idle, busy, interrupted, error)할 수 있어요. 어디서든 LangGraph를 실행한다면 LangSmith 트레이싱으로 어떤 노드가 접근 중인지 모니터링하면 돼요. thread_id가 이미 있다면 graph.get_state(config)로 최신 체크포인트(어느 노드에 멈춰 있는지·대기 중인 interrupt 포함)를, graph.get_state_history(config)로 전체 체크포인트 이력을 확인할 수 있어요.
비즈니스 호환성 (Business compatibility)
간혹 변경이 기술적으로 유효(모든 기존 체크포인트가 로드되고 모든 노드가 해석됨)하더라도, 새 그래프의 의미가 옛것과 다른 경우가 있어요. 새 동작은 새 스레드엔 맞지만, 옛 로직으로 시작한 스레드에는 소급 적용하고 싶지 않은 거예요. 예를 들어 그래프가 intake → triage → respond로 돌다가 triage와 respond 사이에 policy_check 단계를 끼워 넣는다면, 이미 triage를 지난 스레드는 옛 흐름대로 respond로 바로 가야 하고 새 스레드는 새 흐름을 따라야 해요. 권장 패턴은 스레드 시작 시 행동 버전(behavioral version) 을 상태에 기록하고, 조건부 엣지에서 분기하는 거예요:
from typing import NotRequired
from typing_extensions import TypedDict
from langgraph.graph import END, START, StateGraph
class State(TypedDict):
request: str
flow_version: NotRequired[int]
response: NotRequired[str]
def intake(state: State) -> dict:
# Stamp new threads with the current flow version. Existing threads
# that resume past `intake` keep whatever value was already saved.
return {"flow_version": state.get("flow_version", 2)}
def triage(state: State) -> dict: ...
def policy_check(state: State) -> dict: ...
def respond(state: State) -> dict: ...
def after_triage(state: State) -> str:
if state.get("flow_version", 1) >= 2:
return "policy_check"
return "respond"
builder = StateGraph(State)
builder.add_node("intake", intake)
builder.add_node("triage", triage)
builder.add_node("policy_check", policy_check)
builder.add_node("respond", respond)
builder.add_edge(START, "intake")
builder.add_edge("intake", "triage")
builder.add_conditional_edges("triage", after_triage, ["policy_check", "respond"])
builder.add_edge("policy_check", "respond")
builder.add_edge("respond", END)
graph = builder.compile()
옛 스레드는 triage 후 재개되면 저장된 상태에서 flow_version을 읽고(또는 v1 기본값으로 통과) policy_check를 건너뜁니다. 새 스레드는 intake에서 시작해 flow_version=2로 표시되고 새 경로를 탐니다. 모든 v1 스레드가 끝나면 버전 플래그와 조건부 엣지를 제거할 수 있어요. 이 패턴은 버전을 스레드 시작 시, 버전 지정이 필요한 분기 이전에 설정해야만 동작해요.
비결정성 (Non-determinism)
이 범주는 Functional API와, Graph API 노드 안의 tasks 또는 interrupt 호출에만 적용돼요. 일반 Graph API 노드는 재개 시 노드 함수 처음부터 다시 실행되므로, 부수효과는 멱등하게 설계하면 되고 task 호출 순서를 보존할 필요는 없어요. Functional API entrypoint는 단일 노드로 컴파일되는데, 실행이 재개되면 entrypoint 본문을 처음부터 재생하면서 캐시된 @task 결과를 사용해 이미 끝난 작업을 건너뜁니다. 이 모델을 깨뜨리는 두 종류의 변경:
- 재개 지점 이전에 오는
@task또는interrupt호출을 추가·제거·순서 변경. LangGraph는 캐시된 결과와 재개 값을 재생 중 호출의 위치로 매칭하므로, 위치가 바뀌면 잘못된 캐시 값이 다른 호출에 재생될 수 있어요. @task밖에서time.time(),random.random(), 네트워크 호출 같은 비결정적 연산 도입. 재생 시 첫 실행과 다른 값이 나와 제어 흐름이 바뀔 수 있어요.
진행 중인 실행이 있는 @entrypoint에 사소하지 않은 코드 변경이 필요하다면, 실행들이 drain될 때까지 기다리거나, 새 로직을 새 @task로 감싸 결과를 독립적으로 체크포인트하거나, langgraph.json에 새 그래프 이름으로 새 entrypoint를 등록해 새 스레드를 그쪽으로 라우팅하는 방법이 가장 안전해요.
더 알아보기 (Learn more)
- Graph migrations 요약 — 런타임이 기본 지원하는 그래프 토폴로지·상태 변경
- Determinism과 Common pitfalls — Functional API 가이드
- 타임트래블(Time travel) 사용법과
graph.get_state