LangGraph RemainingSteps로 루프를 조기 종료하고 부분 결과 반환하기
Python LangGraph에서 반복 한도에 걸리기 전에 부분 결과를 반환하려면, 상태에 RemainingSteps를 선언하고 남은 단계가 적을 때 마무리 노드로 분기하면 돼요. 처리한 데이터는 상태에 쌓고, 반환값에는 완료 여부와 종료 이유를 함께 넣습니다. 공식 문서도 이 선제 종료 패턴을 안내해요. 외부 API 없이 숫자를 제곱하는 예제로 정상 완료, 예산 부족에 따른 종료, GraphRecursionError를 확인해 봅니다. LangGraph Graph API
설치와 실행 명령
# 새 가상환경에서 실행하세요.
python -m venv .venv
# macOS / Linux
source .venv/bin/activate
# Windows PowerShell에서는 대신 .venv\Scripts\Activate.ps1 실행
python -m pip install -U langgraph
# 전체 예제를 remaining_steps_demo.py로 저장한 뒤 실행
python remaining_steps_demo.py복사해서 실행하는 전체 예제
from importlib.metadata import version
from typing import Literal
from typing_extensions import TypedDict
from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph
from langgraph.managed import RemainingSteps
class State(TypedDict):
numbers: list[int]
next_index: int
squares: list[int]
reason: Literal['running', 'completed', 'step_budget']
response: dict
remaining_steps: RemainingSteps
def work(state: State) -> dict:
index = state['next_index']
numbers = state['numbers']
remaining = state['remaining_steps']
print(f'work: next_index={index}, remaining_steps={remaining}')
# 빈 입력 또는 이미 처리한 입력은 완료가 우선입니다.
if index >= len(numbers):
return {'reason': 'completed'}
# 다음 단계의 finish에 여유를 남깁니다.
if remaining <= 2:
return {'reason': 'step_budget'}
new_index = index + 1
return {
'next_index': new_index,
'squares': state['squares'] + [numbers[index] ** 2],
'reason': 'completed' if new_index == len(numbers) else 'running',
}
def route(state: State) -> Literal['work', 'finish']:
return 'work' if state['reason'] == 'running' else 'finish'
def finish(state: State) -> dict:
return {
'response': {
'values': state['squares'],
'processed': state['next_index'],
'total': len(state['numbers']),
'next_index': state['next_index'],
'completed': state['reason'] == 'completed',
'stop_reason': state['reason'],
}
}
builder = StateGraph(State)
builder.add_node('work', work)
builder.add_node('finish', finish)
builder.add_edge(START, 'work')
builder.add_conditional_edges('work', route)
builder.add_edge('finish', END)
graph = builder.compile()
def run(numbers: list[int], limit: int) -> dict:
if limit < 2:
raise ValueError('이 예제의 recursion_limit은 2 이상이어야 합니다.')
result = graph.invoke(
{
'numbers': numbers,
'next_index': 0,
'squares': [],
'reason': 'running',
'response': {},
},
config={'recursion_limit': limit},
)
return result['response']
# 비교용: 종료 조건 없이 반복하는 그래프
class EndlessState(TypedDict):
count: int
def increment(state: EndlessState) -> dict:
return {'count': state['count'] + 1}
bad_builder = StateGraph(EndlessState)
bad_builder.add_node('repeat', increment)
bad_builder.add_edge(START, 'repeat')
bad_builder.add_edge('repeat', 'repeat')
endless_graph = bad_builder.compile()
if __name__ == '__main__':
print('langgraph version:', version('langgraph'))
complete = run([1, 2, 3], limit=10)
assert complete['values'] == [1, 4, 9]
assert complete['completed'] is True
assert complete['stop_reason'] == 'completed'
print('정상 완료:', complete)
numbers = list(range(1, 101))
partial = run(numbers, limit=4)
count = partial['processed']
assert 0 < count < len(numbers)
assert partial['values'] == [n ** 2 for n in numbers[:count]]
assert partial['next_index'] == len(partial['values'])
assert partial['completed'] is False
assert partial['stop_reason'] == 'step_budget'
print('부분 결과:', partial)
empty = run([], limit=2)
assert empty['completed'] is True
assert empty['values'] == []
print('빈 입력:', empty)
tiny = run([7], limit=2)
assert tiny['completed'] is False
assert tiny['values'] == []
assert tiny['next_index'] == 0
print('작업 전 종료:', tiny)
try:
endless_graph.invoke({'count': 0}, config={'recursion_limit': 4})
except GraphRecursionError:
print('한도 초과: GraphRecursionError를 확인했습니다.')
else:
raise AssertionError('종료 조건 없는 그래프가 예상과 다르게 종료됐습니다.')
print('모든 검증문을 통과했습니다.')recursion_limit은 처리할 항목 수가 아니에요
GraphRecursionError는 그래프가 종료 조건에 도달하기 전에 실행 단계 한도에 걸렸다는 뜻이에요. 순환 연결이 잘못됐을 수도 있고, 정상적인 작업인데 한도가 작을 수도 있습니다. 따라서 한도를 올리기 전에 반복을 끝내는 조건이 실제로 참이 될 수 있는지 확인해야 해요. GRAPH_RECURSION_LIMIT 오류 설명
단계의 단위는 super-step이에요. 순차 실행하는 노드는 서로 다른 단계에 속하지만, 병렬로 실행하는 노드들은 같은 단계에 속할 수 있어요. 항목 하나를 검사하고 변환하는 두 노드가 순차 연결돼 있다면, 항목 수만 보고 실행 예산을 정할 수 없습니다. LangGraph Graph API
숫자를 세는 대상을 먼저 정한다는 점은 수동 계수기를 떠올리면 이해하기 쉬워요. 수동 계수기 실물 참고는 이 구분을 위한 사진이에요. 물건 하나를 처리할 때마다 누른다고 가정하면 표시값은 처리 건수입니다. 하지만 검사와 변환을 각각 한 번으로 세기로 했다면 같은 물건 수라도 숫자가 달라지겠죠. 예제에서도 next_index는 업무 진행률로, remaining_steps는 그래프 실행 예산으로 분리합니다.
관리 값은 읽고, 부분 결과는 직접 저장해요
RemainingSteps는 직접 1씩 줄이는 일반 상태 필드가 아니에요. LangGraph가 제공하는 관리 값이며, 현재 공개 구현은 실행 종료 경계에서 현재 단계를 뺀 값을 반환합니다. 입력에 remaining_steps를 채우거나 노드가 수정해서 반환하는 대신, 타입을 선언하고 필요한 시점에 읽으세요. RemainingSteps 구현
복사해서 실행하는 전체 예제에서는 종료 판단과 응답 조립을 나눴어요. work는 업무 상태를 갱신하고, route는 목적지만 결정하며, finish는 외부에 줄 응답을 만듭니다. 이렇게 두면 계산 로직을 바꿔도 응답 형식을 한곳에서 관리할 수 있어요.
squares에는 reducer를 붙이지 않았으므로 새 누적 리스트 전체를 반환합니다. 기본 상태 업데이트는 해당 키를 덮어써요. 나중에 operator.add 같은 누적 reducer를 붙인다면 새로 계산한 항목만 반환하도록 함께 수정해야 합니다. 전체 리스트를 다시 더하면 기존 값이 중복될 수 있어요. 상태 업데이트와 reducer
마무리에도 실행 예산을 남겨야 해요
remaining_steps <= 2는 모든 그래프에 적용하는 고정 공식이 아니에요. 공식 예제의 기준을, 여기서는 work에서 종료 결정을 내리고 다음 단계의 finish가 응답을 만드는 짧은 경로에 적용했어요. 선제 종료 예제
마무리 경로를 요약 → 검증 → 응답처럼 세 개의 순차 노드로 늘렸다면 더 일찍 종료 분기를 타도록 설계하세요. 종료 결정을 내리는 지점에서 END까지 가장 긴 경로를 확인하고, 작은 recursion_limit에서도 응답 조립까지 도달하는지 검사하면 됩니다. 조건부 분기 함수를 호출하는 것과 별도 검증 노드를 하나 더 실행하는 것을 같은 비용으로 취급하면 계산이 어긋나요.
run 함수가 2보다 작은 한도를 거부하는 것은 이 예제의 입력 계약이에요. LangGraph 전체의 최소 한도를 뜻하지 않습니다. 한 단계만 허용하는 구조가 필요하다면 현재 노드가 응답까지 조립하고 END로 보내도록 별도로 설계하는 편이 명확해요.
빈 결과와 완료 직전의 경계를 구분해요
부분 결과가 빈 리스트여도 오류라고 단정할 수 없어요. 처리 대상은 있지만 시작부터 예산이 부족하면 아무것도 처리하지 못한 채 정상 반환할 수 있습니다. 반대로 입력 자체가 비어 있다면 처리할 일이 없으므로 완료예요. 이 때문에 업무가 끝났는지를 예산 부족보다 먼저 판단합니다.
마지막 항목을 계산한 직후에 완료 여부를 갱신하는 것도 중요해요. 다음 반복에 들어가서야 완료를 알아차리게 만들면 불필요한 단계를 쓰게 됩니다. 예를 들어 결과가 이미 [1, 4, 9]이고 입력도 세 개뿐이라면, 다음 작업 시도 없이 마무리로 보내는 것이 이 예제의 의도예요.
부분 결과의 검증 기준은 특정 실행에서 몇 개가 나왔는지를 외우는 것이 아닙니다. 반환값이 원래 입력의 처리 완료 구간과 일치하는지 확인하세요. 숫자 대신 문서를 처리한다면 결과에 원본 식별자를 남기는 방식이 유용해요. 성공한 문서 세 개와 미처리 문서 두 개를 하나의 성공 목록으로 섞지 않아야 호출자가 후속 처리를 결정할 수 있습니다.
경계 검사를 더 늘리고 싶다면 같은 입력에 대해 limit을 2부터 조금씩 올려 보세요. 매번 processed와 결과 개수가 같고, completed가 참일 때는 processed와 total이 같은지 확인하면 됩니다. 코드를 실제 업무로 바꾼 뒤에도 유지할 수 있는 검증 조건이에요.
예외 처리와 부분 결과 반환은 역할이 달라요
한도 초과용 그래프에는 의도적으로 종료 분기가 없어요. GraphRecursionError를 잡아 메시지를 출력하지만, 이것이 실패한 invoke에서 중간 상태를 반환받았다는 뜻은 아닙니다. 부분 결과를 응답으로 제공하려면 정상 종료 경로에서 결과를 조립해야 해요. 루프 제어와 한도 처리
completed=False 응답의 next_index는 이 예제가 정한 다음 처리 위치이지, 실행 복원 토큰은 아니에요. 현재 run 함수는 호출할 때마다 처음부터 시작합니다. 이어 처리하는 기능이 필요하다면 동일한 입력과 누적 결과를 보존하는 규칙을 추가하세요. 입력이 바뀔 수 있다면 인덱스와 함께 작업 ID나 데이터 버전을 관리하는 방식을 권해요.
단계 예산은 이 예제의 work 호출 안에서 실행하는 계산량을 제한하지 않습니다. 숫자 제곱을 긴 반복문으로 교체했다면 그 반복문에도 중단 조건을 설계해야 해요. 실행 단계, 처리 건수, 작업 시간을 서로 다른 기준으로 다뤄야 합니다.
예제는 공식 문서와 공개 구현을 바탕으로 작성했으며 실행 로그를 실측 결과로 제시하지 않았어요. 실행하면 설치된 langgraph 버전과 각 검증 결과가 출력됩니다. 먼저 전체 예제의 검증문을 확인하고, squares 계산 부분을 실제 작업으로 교체한 뒤 같은 경계 조건을 다시 검사하세요.