Python LangGraph 진행 상황 스트리밍: 오래 걸리는 노드의 중간 알림 보내기
오래 걸리는 LangGraph 노드의 진행 상황은 노드 안에서 get_stream_writer()로 보내고, 호출 측에서 custom 스트림으로 받으면 돼요. 최종 결과는 노드의 반환값으로 남기고, 작업 단계와 완료 건수는 실행 도중 별도로 전달하는 방식이에요. 외부 API 없이 실행하는 예제로 노드 종료 전에 알림을 받는지까지 확인해 볼게요. LangGraph 스트리밍 문서
실행 준비
# Python 3.11 이상에서 실행하세요.
# 가상환경을 만든 뒤 패키지를 설치합니다.
python -m venv .venv
# macOS / Linux
source .venv/bin/activate
# Windows PowerShell에서는 대신 실행:
# .venv\Scripts\Activate.ps1
python -m pip install "langgraph>=1.1,<2"
python -m pip show langgraph
# 다음 코드를 progress_demo.py로 저장한 뒤 실행합니다.
python progress_demo.py진행 알림과 수신 시점 검증 전체 코드
import asyncio
from time import monotonic
from typing import TypedDict
from langgraph.config import get_stream_writer
from langgraph.graph import END, START, StateGraph
class State(TypedDict):
items: list[str]
results: list[str]
node_finished_at: float
async def process_items(state: State) -> dict:
writer = get_stream_writer()
total = len(state["items"])
results: list[str] = []
def report(stage: str, completed: int) -> None:
writer({
"node": "process_items",
"stage": stage,
"completed": completed,
"total": total,
})
report("starting", 0)
for completed, item in enumerate(state["items"], start=1):
# 외부 API 대신 작업 대기를 흉내 냅니다.
await asyncio.sleep(0.6)
results.append(item.strip().upper())
report("processing", completed)
return {
"results": results,
"node_finished_at": monotonic(),
}
builder = StateGraph(State)
builder.add_node("process_items", process_items)
builder.add_edge(START, "process_items")
builder.add_edge("process_items", END)
graph = builder.compile()
async def main() -> None:
started_at = monotonic()
first_received_at: float | None = None
final_update: dict | None = None
completed_counts: list[int] = []
inputs: State = {
"items": [" alpha ", "beta", "gamma", "delta"],
"results": [],
"node_finished_at": 0.0,
}
async for part in graph.astream(
inputs,
stream_mode=["custom", "updates"],
version="v2",
):
received_at = monotonic()
elapsed = received_at - started_at
if part["type"] == "custom":
event = part["data"]
if first_received_at is None:
first_received_at = received_at
completed_counts.append(event["completed"])
print(
f"{elapsed:.2f}s 진행: {event['stage']} "
f"{event['completed']}/{event['total']}",
flush=True,
)
elif part["type"] == "updates":
update = part["data"].get("process_items")
if update is not None:
final_update = update
print(
f"{elapsed:.2f}s 결과: {update['results']}",
flush=True,
)
if final_update is None or first_received_at is None:
raise RuntimeError("진행 이벤트 또는 최종 결과가 없습니다.")
expected_counts = list(range(len(inputs["items"]) + 1))
if completed_counts != expected_counts:
raise RuntimeError(f"완료 건수 확인 필요: {completed_counts}")
expected_results = [item.strip().upper() for item in inputs["items"]]
if final_update["results"] != expected_results:
raise RuntimeError("최종 결과가 예상과 다릅니다.")
# 빈 입력은 즉시 종료할 수 있으므로 중간 수신 검사를 생략합니다.
if inputs["items"]:
if first_received_at >= final_update["node_finished_at"]:
raise RuntimeError("첫 이벤트를 노드 종료 전에 받지 못했습니다.")
print("검증 통과: 노드 종료 전에 진행 이벤트를 받았습니다.")
else:
print("검증 통과: 빈 입력을 처리했습니다.")
if __name__ == "__main__":
asyncio.run(main())
진행 알림과 작업 결과를 나누는 이유
파일 여러 개를 처리하는 노드라면 사용자는 마지막 결과뿐 아니라 지금 어느 파일까지 끝났는지도 알고 싶어요. 이때 모든 중간 메시지를 그래프 상태에 쌓기보다는, 화면에 필요한 알림과 다음 노드에 필요한 결과를 나눠 설계하는 편을 권해요. 예제에서도 알림은 writer로 보내고, 변환된 문자열 목록만 결과로 반환해요.
updates는 상태 변경을 전달하므로 긴 함수 내부의 반복문 진행을 자동으로 알려주지는 않아요. LLM 출력을 받는 messages 역시 일반 작업의 진행률을 대신 계산해 주는 기능은 아니에요. 어떤 시점을 작업 완료로 볼지는 작성자가 정해서 custom 이벤트를 보내야 해요. LangGraph 스트림 모드
이 예제는 Python 3.11 이상과 LangGraph 1.1 이상을 기준으로 해요. version="v2"를 지정하면 이벤트를 type과 data 필드로 구분할 수 있어요. 기존 코드가 (mode, data) 튜플을 풀어 받는 방식이라면 출력 형식 설정도 함께 확인하세요. LangGraph 출력 형식
출력 순서보다 실제 수신 시점을 확인해요
진행 이벤트가 최종 결과보다 먼저 출력됐다는 사실만으로는 충분하지 않아요. 모두 작업 종료 뒤에 수신했어도 출력 순서 자체는 같을 수 있기 때문이에요. ‘진행 알림과 수신 시점 검증 전체 코드’는 호출 측의 첫 수신 시각과 노드의 반환 직전 시각을 비교해요. 이 비교가 통과하면 해당 실행에서는 노드가 끝나기 전에 호출 측이 이벤트를 받았다는 뜻이에요.
시간 비교에는 같은 프로세스의 monotonic() 값을 사용해요. 이 시계는 뒤로 가지 않고 시스템 시각 변경의 영향을 받지 않으므로 경과 시간이나 앞뒤 시점을 비교하는 데 적합해요. 다만 이 값을 브라우저나 다른 서버의 시계와 직접 비교하는 용도로 확장하지는 마세요. Python time 문서
예제의 0.6초 대기는 처리 구간을 눈으로 구별하기 위한 설정값이에요. 실제 성능이나 스트림 지연을 측정한 숫자가 아니며, 표시되는 시간도 실행 환경에 따라 달라져요. 실행 로그를 제시하는 대신 검증 조건을 코드에 넣었으니, 자신의 환경에서 마지막 검사가 통과하는지 확인하면 돼요.
첫 이벤트의 조기 수신 검사는 모든 이벤트의 지연 시간을 보장하지는 않아요. 실제 작업으로 바꾼 뒤 중간 구간만 멈춘다면 이벤트마다 송신 시각을 추가하고 수신 시각과 비교해 보세요. 같은 프로세스에서 측정하는 동안에는 ‘노드가 늦게 보냈는지’와 ‘보낸 뒤 늦게 받았는지’를 구분하는 데 도움이 돼요.
퍼센트보다 먼저 완료의 기준을 정해요
예제 이벤트의 stage, completed, total은 이 글에서 정한 애플리케이션 규약이에요. 여기서는 문자열 하나의 변환을 마친 뒤 completed를 늘려요. 실제 문서 처리라면 ‘읽기 완료’인지 ‘분석과 저장까지 완료’인지부터 정해야 숫자가 일관돼요.
예를 들어 낱장 문서를 스캐너로 디지털화한 뒤 파일별 분석을 수행하는 작업을 생각해 볼게요. 스캔한 페이지 수와 분석을 마친 문서 수는 다른 단위예요. 화면에는 ‘스캔 20페이지’와 ‘분석 3문서’를 단계별로 표시하는 편을 권해요. 참고 사진의 낱장 급지 스캐너는 이런 입력 작업의 실물 사례이며, LangGraph 실행이나 장비 연동을 보여주는 사진은 아니에요. 문서 스캐너 참고 사진
전체 4건 중 2건을 마쳤다면 건수 기준으로는 50%예요. 하지만 첫 두 문서는 한 페이지, 나머지 두 문서는 수백 페이지라면 남은 시간도 절반이라고 말할 수는 없어요. 화면 문구를 ‘2/4건 처리’로 두면 측정한 내용만 전달할 수 있어요. 시간 예측이 필요한 서비스라면 별도의 추정 규칙을 마련하는 편이 좋아요.
페이지를 계속 조회하면서 처리 대상을 발견하는 작업이라면 total을 null로 보내는 규약을 권해요. 화면에는 ‘검색 중 · 27건 처리’처럼 단계와 건수만 표시하고요. total=0은 전체가 없는 경우에 남겨 두면 ‘아직 전체를 모름’과 ‘처리할 항목이 없음’을 구분할 수 있어요. 예제에서 입력 목록을 비워도 나눗셈 없이 빈 결과를 반환하도록 한 이유예요.
노드가 여러 개라면 완료 건수를 무작정 합산하지 마세요. 다운로드 10건과 색인 10건은 같은 문서 10개에 대한 서로 다른 단계일 수 있어요. 우선 node와 stage별로 표시하고, 전체 진행률이 꼭 필요할 때 단계별 가중치를 별도로 정하는 방식을 권해요.
비동기인데도 알림이 몰려 나온다면
async def만 붙였다고 내부 작업이 비동기가 되는 것은 아니에요. 코루틴 안에서 time.sleep()이나 동기 I/O를 직접 호출하면 이벤트 루프를 막을 수 있어요. 예제의 asyncio.sleep()은 기다리는 동안 다른 태스크가 실행될 수 있도록 제어를 넘겨요. 실제 작업으로 바꿀 때도 비동기 I/O를 사용하거나, 블로킹 I/O라면 asyncio.to_thread()로 분리하는 방법을 검토하세요. 일반적인 CPU 연산을 스레드로 옮기는 것까지 같은 해결책으로 취급하면 안 돼요. Python asyncio 문서
Python 3.11 미만의 비동기 실행에서는 get_stream_writer() 사용에 제약이 있어요. 해당 환경을 유지해야 한다면 공식 문서의 writer 인자 전달 방식을 확인해야 해요. 이 글의 실행 조건을 Python 3.11 이상으로 잡은 이유도 여기에 있어요. writer는 그래프가 실행하는 노드 안에서 얻도록 두세요. LangGraph 비동기 스트리밍 제약
또 하나는 표시 단계예요. 예제는 print(..., flush=True)로 Python 출력 버퍼를 바로 비우게 했어요. 수신 시각은 정상인데 터미널 표시만 늦는 경우와 이벤트 자체를 늦게 받는 경우를 나눠 살펴보면 원인 범위를 줄일 수 있어요. Python print 문서
실제 작업에 붙일 때는 성공 판정을 따로 둬요
실무 코드에 옮길 때는 ‘4/4건 처리’ 알림과 ‘전체 실행 성공’을 구분하는 설계를 권해요. 항목 처리를 마친 뒤 결과 저장이나 후속 단계가 남아 있을 수 있으니까요. 예제에서는 스트림 순회를 정상적으로 끝내고 최종 결과를 확인한 뒤에 성공 메시지를 출력해요. 오류가 난 경우에는 완료 화면으로 넘어가지 않도록 같은 기준을 적용하세요.
재시도를 도입한다면 이벤트에 작업 ID와 시도 번호를 붙이는 것도 좋아요. 화면에서 도착할 때마다 1을 더하기보다는 해당 시도의 completed 절대값을 표시하도록 설계하면 중복 알림을 다루기 쉬워요. 이벤트 빈도 역시 모든 내부 연산마다 보내기보다 항목 완료나 단계 전환처럼 사용자가 이해할 수 있는 지점부터 잡아 보세요.
이 예제의 수신자는 Python 호출 코드예요. 웹 화면까지 연결하려면 그 수신부에서 브라우저로 이벤트를 전달하는 서버 경로가 추가로 필요해요. 먼저 실행 예제의 검증을 통과시킨 다음, 문자열 변환 부분을 실제 작업 한 단위로 교체해 보세요. 그 상태에서도 노드 종료 전 수신이 확인되면 화면 연결을 진행하는 순서가 원인을 구분하기 편해요.